Chroma向量数据库新手教程：从零构建高性能语义搜索系统

1次阅读

共计 2248 个字符，预计需要花费 6 分钟才能阅读完成。

在处理语义搜索、推荐系统等 AI 应用时，我们经常需要存储和查询高维向量（比如文本嵌入）。传统关系型数据库在这方面显得力不从心：

性能瓶颈：用 SQL 实现余弦相似度计算需要全表扫描，百万级数据查询可能需要数秒
开发复杂度 ：手动实现近似最近邻(ANN) 算法涉及复杂的数学和工程优化
扩展困难：无法有效利用现代硬件（如 GPU 加速）进行向量运算

FAISS：Facebook 开源的 ANN 库，适合内存计算但缺乏持久化能力
Milvus：功能全面的分布式向量数据库，但部署复杂度较高
Chroma的核心优势：
轻量级设计，API 简洁易用
内置持久化支持
完善的 Python 生态集成

# 安装最新版本
pip install chromadb

# 初始化客户端（默认使用本地持久化）import chromadb
client = chromadb.Client()

# 创建名为 "docs" 的集合，使用余弦相似度作为距离度量
collection = client.create_collection(
    name="docs",
    metadata={"hnsw:space": "cosine"}  # 指定 HNSW 索引的度量空间
)

# 准备测试数据
documents = ["狗是一种宠物", "猫讨厌洗澡", "汽车需要汽油"]
ids = ["doc1", "doc2", "doc3"]
embeddings = [...]  # 实际使用时应替换为真实嵌入向量

# 批量插入
collection.add(
    documents=documents,
    ids=ids,
    embeddings=embeddings
)

# 查询与 "宠物" 最相似的文档
results = collection.query(query_embeddings=[pet_embedding],  # 查询向量
    n_results=2                        # 返回 Top2 结果
)

print(results['documents'])  # 输出相似文档内容

# 创建高性能集合
performance_collection = client.create_collection(
    name="high_perf",
    metadata={
        "hnsw:space": "ip",       # 内积空间
        "hnsw:construction_ef": 128,  # 构建阶段搜索范围
        "hnsw:M": 16                  # 层间连接数
    }
)

# 分块处理大型数据集
batch_size = 1000
for i in range(0, len(docs), batch_size):
    batch = docs[i:i + batch_size]
    collection.add(
        documents=batch,
        ids=[f"doc_{j}" for j in range(i, i+len(batch))]
    )
    client.persist()  # 定期持久化释放内存

存储方案选择：
开发环境：使用默认的本地持久化（persist_directory参数）
生产环境：考虑云存储后端（如 S3）或专用分布式存储
多线程安全：
批量写入时启用 batch_mode=True 减少锁竞争
查询密集型场景可创建只读副本
监控指标：
定期检查 collection.count() 防止数据膨胀
监控查询延迟百分位（P99/P95）

结合 Sentence-BERT 构建完整流水线：

from sentence_transformers import SentenceTransformer

# 初始化模型
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

# 文本到向量
texts = ["苹果发布新手机", "特斯拉股价上涨"]
embeddings = model.encode(texts)

# 存储到 Chroma
collection.add(
    documents=texts,
    ids=["news1", "news2"],
    embeddings=embeddings.tolist())

# 实时查询
query = "科技公司动态"
query_embedding = model.encode([query])
results = collection.query(query_embeddings=query_embedding.tolist(),
    n_results=1
)

对于 Serverless 架构（如 AWS Lambda）：