Chroma DB 2026：面向开发者的 RAG 向量数据库，嵌入搜索速度提升 50 倍 — Python 实战指南

pip install chromadb

# 指定嵌入后端
pip install chromadb[sentence-transformers]

# 验证安装
python -c "import chromadb; print(chromadb.__version__)"
# Expected: 0.6.x or higher

import chromadb

# 纯内存模式 —— 进程退出后数据消失
client = chromadb.Client()

import chromadb

# 数据保存到 ./chroma_db 目录
client = chromadb.PersistentClient(path="./chroma_db")

# Docker 中运行 Chroma 服务器
docker run -d \
  --name chroma \
  -v ./chroma_data:/chroma/chroma \
  -p 8000:8000 \
  chromadb/chroma:latest

# 从 Python 连接
import chromadb
client = chromadb.HttpClient(host="localhost", port=8000)

import chromadb

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

# 创建或获取集合
# "documents" 是你的原始文本块
# "metadatas" 是用于过滤的键值对
# "ids" 是唯一标识符

collection = client.get_or_create_collection(name="knowledge_base")

documents = [
    "Chroma is a vector database designed for AI applications.",
    "RAG stands for Retrieval-Augmented Generation.",
    "HNSW indexing enables fast approximate nearest neighbor search.",
    "Embeddings convert text into high-dimensional vectors.",
]

collection.add(
    documents=documents,
    metadatas=[
        {"source": "docs", "topic": "database"},
        {"source": "docs", "topic": "ai"},
        {"source": "blog", "topic": "indexing"},
        {"source": "blog", "topic": "embeddings"},
    ],
    ids=["doc_1", "doc_2", "doc_3", "doc_4"]
)

print(f"Collection count: {collection.count()}")
# Output: Collection count: 4

# 简单相似度搜索
results = collection.query(
    query_texts=["What is a vector database?"],
    n_results=2
)

print(results["documents"])
# Output: [["Chroma is a vector database designed for AI applications."]]

# 带元数据过滤
results = collection.query(
    query_texts=["How does search work fast?"],
    where={"source": "blog"},  # 按元数据过滤
    n_results=2
)

print(results["documents"])
# Output: [["HNSW indexing enables fast approximate nearest neighbor search."]]

# 更新文档
collection.update(
    ids=["doc_1"],
    documents=["Chroma is the developer-friendly vector database for RAG."],
    metadatas=[{"source": "docs", "topic": "database", "updated": True}]
)

# 按 ID 删除
collection.delete(ids=["doc_4"])

print(f"Collection count after delete: {collection.count()}")
# Output: Collection count after delete: 3

pip install langchain-chroma langchain-openai

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# 初始化嵌入函数
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = Chroma(
    collection_name="langchain_docs",
    embedding_function=embeddings,
    persist_directory="./chroma_langchain"
)

# 添加文档
docs = [
    Document(page_content="Chroma integrates seamlessly with LangChain.", metadata={"source": "tutorial"}),
    Document(page_content="RAG pipelines combine retrieval with LLM generation.", metadata={"source": "guide"}),
]

vector_store.add_documents(docs)

# 搜索
results = vector_store.similarity_search("How do I use LangChain with Chroma?", k=2)
for doc in results:
    print(doc.page_content)

pip install llama-index-vector-stores-chroma

from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core import StorageContext, VectorStoreIndex
from llama_index.embeddings.openai import OpenAIEmbedding
import chromadb

# 设置
chroma_client = chromadb.PersistentClient(path="./chroma_llamaindex")
chroma_collection = chroma_client.get_or_create_collection("llamaindex")

vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 从文档构建索引
from llama_index.core import Document

documents = [Document(text="Chroma works great with LlamaIndex for RAG.")]
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
    embed_model=OpenAIEmbedding()
)

# 查询
query_engine = index.as_query_engine()
response = query_engine.query("What vector database should I use with LlamaIndex?")
print(response)

from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

# 直接使用 OpenAI 嵌入模型
openai_ef = OpenAIEmbeddingFunction(
    api_key="your-openai-api-key",
    model_name="text-embedding-3-small"  # 1536 维，性价比优秀
)

collection = client.get_or_create_collection(
    name="openai_embedded",
    embedding_function=openai_ef
)

collection.add(
    documents=["OpenAI embeddings produce high-quality vectors for semantic search."],
    ids=["doc_openai_1"]
)

results = collection.query(
    query_texts=["Tell me about OpenAI vectors"],
    n_results=1
)

from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

# 完全本地运行 —— 无 API 调用，无限流
local_ef = SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2"  # 384 维，速度快，质量优秀
)

collection = client.get_or_create_collection(
    name="local_embeddings",
    embedding_function=local_ef
)

collection.add(
    documents=["Local embeddings are free and privacy-preserving."],
    ids=["local_1"]
)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import chromadb

app = FastAPI()
client = chromadb.PersistentClient(path="./chroma_api")
collection = client.get_or_create_collection("api_docs")

class QueryRequest(BaseModel):
    query: str
    n_results: int = 5

@app.post("/search")
def search_docs(request: QueryRequest):
    try:
        results = collection.query(
            query_texts=[request.query],
            n_results=request.n_results
        )
        return {
            "documents": results["documents"][0],
            "distances": results["distances"][0],
            "metadatas": results["metadatas"][0]
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
def health():
    return {"status": "ok", "count": collection.count()}

# Run: uvicorn main:app --reload

# 来自任意模型的预计算嵌入（如 OpenAI text-embedding-3-large）
import numpy as np

# 你的自定义嵌入 —— 3072 维
custom_embeddings = [
    np.random.randn(3072).tolist(),  # 替换为真实嵌入
    np.random.randn(3072).tolist(),
]

collection = client.get_or_create_collection("custom_dims")
collection.add(
    embeddings=custom_embeddings,
    documents=["Doc with custom embedding", "Another doc"],
    ids=["custom_1", "custom_2"]
)

# 复杂元数据查询
collection.add(
    documents=["Advanced filtering example"],
    metadatas=[{
        "category": "tutorial",
        "difficulty": "advanced",
        "year": 2026,
        "published": True,
        "tags": ["chroma", "filtering"]
    }],
    ids=["filter_demo"]
)

# 数值比较
results = collection.query(
    query_texts=["filtering"],
    where={"year": {"$gte": 2025}},
    n_results=5
)

# 逻辑运算符
results = collection.query(
    query_texts=["advanced tutorial"],
    where={
        "$and": [
            {"category": "tutorial"},
            {"difficulty": "advanced"}
        ]
    },
    n_results=5
)

# 每个用户/租户一个集合 —— 天然隔离
def get_user_collection(user_id: str):
    return client.get_or_create_collection(f"user_{user_id}_docs")

# 每个用户的数据完全隔离
user_a = get_user_collection("alice")
user_b = get_user_collection("bob")

user_a.add(documents=["Alice's private document"], ids=["alice_1"])
user_b.add(documents=["Bob's private document"], ids=["bob_1"])

# docker-compose.yml
version: "3.8"

services:
  chroma:
    image: chromadb/chroma:0.6.0
    ports:
      - "8000:8000"
    volumes:
      - chroma_data:/chroma/chroma
    environment:
      - IS_PERSISTENT=TRUE
      - PERSIST_DIRECTORY=/chroma/chroma
      - ANONYMIZED_TELEMETRY=FALSE
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 8G
        reservations:
          memory: 2G

volumes:
  chroma_data:

docker-compose up -d
# Chroma API 地址：http://localhost:8000

# Chroma 将所有数据存储在持久化目录中
# 使用标准工具即可备份

tar -czf chroma_backup_$(date +%Y%m%d).tar.gz ./chroma_data/

# 恢复只需解压到相同路径
tar -xzf chroma_backup_20260519.tar.gz