Chroma 向量数据库¶

为什么要学 Chroma¶

Chroma 是最易上手的开源向量数据库，专为 AI 应用设计。它既可以作为嵌入式数据库在 Python 中直接使用（无需额外服务），也可以部署为独立服务。对于 RAG 应用、语义搜索、推荐系统等需要向量相似度检索的场景，Chroma 提供了简洁的 API 和合理的默认配置，让开发者快速上手。

核心概念¶

概念	白话解释	用途
Collection	集合	类似数据库的"表"，存储一组向量
Embedding	嵌入向量	文本/图像的数字化表示
Document	原始文档	向量对应的原始文本
Metadata	元数据	附加的过滤信息（标签、分类等）
Query	查询	通过向量相似度搜索
Embedding Function	嵌入函数	将文本转为向量的模型

安装配置¶

安装¶

pip install chromadb

# 带客户端支持
pip install chromadb-client

使用模式¶

import chromadb

# 模式 1：内存模式（临时使用）
client = chromadb.Client()

# 模式 2：持久化模式（数据保存到磁盘）
client = chromadb.PersistentClient(path="./chroma_db")

# 模式 3：客户端-服务器模式
client = chromadb.HttpClient(host="localhost", port=8000)

启动服务器¶

# CLI 启动
chroma run --path ./chroma_data --port 8000

# Docker
docker run -p 8000:8000 \
  -v $PWD/chroma_data:/chroma/chroma \
  chromadb/chroma

# Docker Compose
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  chroma:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma-data:/chroma/chroma
    environment:
      - ANONYMIZED_TELEMETRY=False
      - CHROMA_SERVER_AUTH_PROVIDER=chromadb.auth.token_authn.TokenAuthenticationServerProvider
      - CHROMA_SERVER_AUTH_TOKEN_TRANSPORT_HEADER=Authorization
      - CHROMA_SERVER_AUTH_CREDENTIALS=your-secret-token
volumes:
  chroma-data:
EOF

快速上手¶

基本操作¶

import chromadb

client = chromadb.PersistentClient(path="./my_db")

# 创建集合
collection = client.create_collection(
    name="my_docs",
    metadata={"hnsw:space": "cosine"}  # 距离度量方式
)

# 添加文档（自动计算嵌入）
collection.add(
    documents=["Python 是一种解释型语言", 
               "JavaScript 运行在浏览器中",
               "Rust 注重内存安全"],
    metadatas=[{"lang": "python"}, {"lang": "js"}, {"lang": "rust"}],
    ids=["doc1", "doc2", "doc3"]
)

# 查询
results = collection.query(
    query_texts=["哪种语言最安全？"],
    n_results=2
)
print(results["documents"])     # 匹配的文档
print(results["distances"])     # 相似度距离
print(results["metadatas"])     # 元数据

使用自定义嵌入模型¶

from chromadb.utils import embedding_functions

# OpenAI
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="sk-your-key",
    model_name="text-embedding-3-small"
)

# Sentence Transformers（本地）
st_ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2"
)

# Ollama
ollama_ef = embedding_functions.OllamaEmbeddingFunction(
    url="http://localhost:11434/api/embeddings",
    model_name="nomic-embed-text"
)

# 在集合中使用
collection = client.create_collection(
    name="custom_embeddings",
    embedding_function=openai_ef
)

元数据过滤¶

# 添加带元数据的文档
collection.add(
    documents=["FastAPI 教程", "Django 入门", "Flask 基础"],
    metadatas=[
        {"framework": "fastapi", "level": "beginner", "year": 2024},
        {"framework": "django", "level": "beginner", "year": 2023},
        {"framework": "flask", "level": "beginner", "year": 2022}
    ],
    ids=["1", "2", "3"]
)

# 带过滤的查询
results = collection.query(
    query_texts=["Web 框架入门"],
    n_results=2,
    where={"year": {"$gte": 2023}},        # 元数据过滤
    where_document={"$contains": "入门"}    # 文档内容过滤
)

进阶用法¶

批量操作¶

# 批量添加（大数据量时使用）
batch_size = 100
for i in range(0, len(documents), batch_size):
    batch_docs = documents[i:i+batch_size]
    batch_ids = [f"doc_{j}" for j in range(i, i+len(batch_docs))]
    batch_meta = metadatas[i:i+batch_size]

    collection.add(
        documents=batch_docs,
        ids=batch_ids,
        metadatas=batch_meta
    )

更新和删除¶

# 更新文档
collection.update(
    ids=["doc1"],
    documents=["Python 3.12 是最新的稳定版本"],
    metadatas=[{"lang": "python", "version": "3.12"}]
)

# Upsert（存在则更新，不存在则添加）
collection.upsert(
    ids=["doc1", "doc4"],
    documents=["更新的文档1", "全新的文档4"],
    metadatas=[{"type": "updated"}, {"type": "new"}]
)

# 删除
collection.delete(ids=["doc1"])
collection.delete(where={"lang": "js"})  # 按条件删除

复杂过滤查询¶

# AND 条件
results = collection.query(
    query_texts=["问题"],
    where={
        "$and": [
            {"category": "tech"},
            {"year": {"$gte": 2023}},
            {"lang": {"$in": ["python", "rust"]}}
        ]
    }
)

# OR 条件
results = collection.query(
    query_texts=["问题"],
    where={
        "$or": [
            {"category": "tech"},
            {"priority": "high"}
        ]
    }
)

与 LangChain 集成¶

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    collection_name="langchain_docs",
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./chroma_langchain"
)

# 添加文档
vectorstore.add_texts(
    texts=["文档1", "文档2"],
    metadatas=[{"source": "web"}, {"source": "pdf"}]
)

# 相似度搜索
docs = vectorstore.similarity_search("查询", k=3)

# 作为 Retriever 使用
retriever = vectorstore.as_retriever(
    search_type="mmr",  # 最大边际相关性
    search_kwargs={"k": 5, "fetch_k": 20}
)

性能调优¶

# HNSW 索引参数调优
collection = client.create_collection(
    name="optimized",
    metadata={
        "hnsw:space": "cosine",        # 距离度量：cosine, l2, ip
        "hnsw:construction_ef": 200,   # 构建时精度（越高越准但越慢）
        "hnsw:search_ef": 100,         # 搜索时精度
        "hnsw:M": 16,                  # 连接数（影响内存和精度）
    }
)

常见问题¶

Q: Chroma 适合多大规模的数据？¶

单节点推荐百万级别以内的向量。超过百万建议使用 Qdrant、Weaviate 等专门的向量数据库。

Q: 数据安全如何保证？¶

# 启用认证
# 在服务端设置 auth token
# 客户端连接时传入
client = chromadb.HttpClient(
    host="localhost", port=8000,
    headers={"Authorization": "Bearer your-token"}
)

Q: 如何备份数据？¶

持久化模式下直接备份 persist_directory 目录即可：

cp -r ./chroma_db ./chroma_db_backup

Q: 支持多租户吗？¶

通过不同的 Collection 实现逻辑隔离，或部署多个 Chroma 实例实现物理隔离。

Q: 嵌入维度不匹配怎么办？¶

同一个 Collection 中所有向量维度必须一致。创建时确认 embedding function，不要中途更换模型。

参考资源¶

GitHub：https://github.com/chroma-core/chroma
文档：https://docs.trychroma.com/
API Reference：https://docs.trychroma.com/reference
Cookbook：https://cookbook.chromadb.dev/
Discord：https://discord.gg/MMeYNTmh3x