"机器学习是人工智能的子领域"  → [0.23, -0.15, 0.87, ..., 0.42]  (1536维)
"深度学习属于AI研究领域"    → [0.21, -0.12, 0.85, ..., 0.40]  (1536维)
"今天的天气非常好"           → [-0.65, 0.33, -0.11, ..., 0.78]  (1536维)

cosine_similarity(A, B) = (A · B) / (|A| × |B|)

L2(A, B) = √(Σ(aᵢ - bᵢ)²)

IP(A, B) = Σ(aᵢ × bᵢ)

import chromadb
client = chromadb.Client()  # 纯内存，零配置

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

client = chromadb.HttpClient(host="localhost", port=8000)

pip install chromadb

pip install chromadb-client

import chromadb
print(chromadb.__version__)  # 应输出 1.5.9 或类似版本号

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

docker run -d \
  --name chroma-server \
  -v ./chroma-data:/data \
  -p 8000:8000 \
  chromadb/chroma:1.5.9

version: '3.8'
services:
  chroma:
    image: chromadb/chroma:1.5.9
    container_name: chroma-server
    ports:
      - "8000:8000"
    volumes:
      - chroma-data:/data
    environment:
      - IS_PERSISTENT=TRUE
      - ANONYMIZED_TELEMETRY=FALSE
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v2/heartbeat"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      resources:
        limits:
          memory: 4G
        reservations:
          memory: 2G

volumes:
  chroma-data:
    driver: local

client = chromadb.CloudClient(
    tenant="your-tenant",
    database="your-database",
    api_key="your-api-key"
)

import chromadb

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

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

client = chromadb.HttpClient(
    host="localhost",
    port=8000,
    # 可选：认证
    # settings=Settings(chroma_client_auth_credentials="your-token")
)

client = chromadb.AsyncHttpClient(host="localhost", port=8000)

# 所有操作都是 async/await 的
collection = await client.get_or_create_collection("docs")
results = await collection.query(query_texts=["hello"], n_results=5)

client = chromadb.CloudClient(
    tenant="my-tenant",
    database="default_database",
    api_key="ck-xxxxxxxxxxxxxxxx"
)

import chromadb

# 步骤 1：创建客户端（这里使用内存模式快速演示）
client = chromadb.Client()

# 步骤 2：创建集合
# Chroma 默认使用 all-MiniLM-L6-v2 嵌入函数
# 它会自动将文本转换为 384 维的嵌入向量
collection = client.create_collection(name="knowledge_base")

# 步骤 3：写入数据
collection.add(
    documents=[
        "Chroma 是一个开源的向量数据库，专为 AI 应用设计",
        "Python 是最流行的机器学习和数据科学编程语言",
        "向量数据库通过语义相似度检索非结构化数据",
        "HNSW 是目前最广泛使用的近似最近邻搜索算法",
        "RAG（检索增强生成）结合检索和大语言模型提升回答质量",
    ],
    ids=["doc1", "doc2", "doc3", "doc4", "doc5"]
)

# 步骤 4：语义查询
results = collection.query(
    query_texts=["什么是用于 AI 的数据库"],
    n_results=3
)

# 步骤 5：查看结果
print(results)

{
    'ids': [['doc1', 'doc3', 'doc5']],
    'documents': [[
        'Chroma 是一个开源的向量数据库，专为 AI 应用设计',
        '向量数据库通过语义相似度检索非结构化数据',
        'RAG（检索增强生成）结合检索和大语言模型提升回答质量'
    ]],
    'distances': [[0.3241, 0.4523, 0.6187]],
    'metadatas': [[None, None, None]],
    'embeddings': None
}

# 创建新集合 - 如果同名集合已存在则抛出异常
collection = client.create_collection(name="my_docs")

# 获取或创建 - 如果已存在则返回现有集合，否则创建新的
collection = client.get_or_create_collection(name="my_docs")

# 合法名称
"documents", "user-embeddings-v2", "my_app.docs"

# 非法名称
"ab"           # 太短（<3 字符）
"-docs"        # 以连字符开头
"docs."        # 以点结尾
"my collection" # 包含空格

collection = client.create_collection(
    name="text_docs",
    metadata={"hnsw:space": "cosine"}  # 可选: "cosine", "l2", "ip"
)

collection = client.create_collection(
    name="production_docs",
    metadata={
        "hnsw:space": "cosine",
        "description": "生产环境文档库",
        "version": "2.0",
        "owner": "ai-team"
    }
)

collection = client.create_collection(
    name="optimized_collection",
    metadata={
        # 距离度量空间
        "hnsw:space": "cosine",
        
        # 构建阶段参数
        "hnsw:construction_ef": 128,   # 构建时的搜索宽度
        "hnsw:M": 16,                  # 每个节点的连接数
        
        # 查询阶段参数
        "hnsw:search_ef": 64,          # 查询时的搜索宽度
        
        # 运行参数
        "hnsw:num_threads": 8,         # 并行线程数
        "hnsw:batch_size": 100,        # 批量操作大小
        "hnsw:sync_threshold": 1000,   # 同步到磁盘的阈值
        "hnsw:resize_factor": 1.2      # 索引扩容系数
    }
)

# 获取已有集合
collection = client.get_collection("my_docs")

# 如果指定了自定义嵌入函数，获取时需要传入
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction
ef = OpenAIEmbeddingFunction(api_key="sk-...", model_name="text-embedding-3-small")
collection = client.get_collection("my_docs", embedding_function=ef)

# 列出所有集合（支持分页）
collections = client.list_collections()       # 返回 Collection 对象列表
collections = client.list_collections(limit=10, offset=0)  # 分页

client.delete_collection("my_docs")

# 修改集合名称和元数据
collection = client.get_collection("old_name")
collection.modify(
    name="new_name",
    metadata={"hnsw:search_ef": 128, "description": "已重命名"}
)

collection.add(
    ids=["doc1", "doc2", "doc3"],
    documents=[
        "Chroma 支持多种嵌入模型",
        "向量检索比全文搜索更理解语义",
        "HNSW 算法构建分层导航图"
    ],
    metadatas=[
        {"source": "docs", "category": "database", "priority": 1},
        {"source": "blog", "category": "search", "priority": 2},
        {"source": "paper", "category": "algorithm", "priority": 3}
    ]
)

collection.upsert(
    ids=["doc1", "doc_new"],
    documents=[
        "Chroma 支持超过 20 种嵌入模型",  # 更新已有 doc1
        "这是一条全新的文档"                  # 插入新文档
    ]
)

# 只更新元数据，不改变文档内容和嵌入
collection.update(
    ids=["doc1"],
    metadatas=[{"source": "updated_docs", "category": "database", "priority": 5}]
)

# 更新文档内容 - 会自动重新生成嵌入向量
collection.update(
    ids=["doc1"],
    documents=["Chroma 支持超过 25 种嵌入模型，包括最新的多模态模型"]
)

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("large_dataset")

# 假设 documents 和 embeddings 是预先准备好的大列表
all_documents = [...]   # 10 万条文档
all_embeddings = [...]  # 对应的嵌入向量
all_ids = [f"doc_{i}" for i in range(len(all_documents))]

# 分批写入
batch_size = 200
for i in range(0, len(all_documents), batch_size):
    end = min(i + batch_size, len(all_documents))
    collection.add(
        ids=all_ids[i:end],
        documents=all_documents[i:end],
        embeddings=all_embeddings[i:end]
    )
    if (i // batch_size) % 50 == 0:
        print(f"已写入 {end}/{len(all_documents)} 条")

# 等价于不传 embedding_function
from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

ef = SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2",
    device="cpu"  # 可选 "cuda" 使用 GPU 加速
)

collection = client.create_collection(
    name="my_docs",
    embedding_function=ef
)

from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

ef = OpenAIEmbeddingFunction(
    api_key="sk-...",
    model_name="text-embedding-3-small"  # 1536 维，性价比高
    # model_name="text-embedding-3-large"  # 3072 维，精度最高
)

collection = client.create_collection(
    name="openai_docs",
    embedding_function=ef
)

from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

ef = OpenAIEmbeddingFunction(
    api_key="your-azure-key",
    api_base="https://your-resource.openai.azure.com/",
    api_type="azure",
    api_version="2024-02-01",
    model_name="text-embedding-3-small"
)

# Cohere
from chromadb.utils.embedding_functions import CohereEmbeddingFunction
ef = CohereEmbeddingFunction(api_key="...", model_name="embed-multilingual-v3.0")

# Google Gemini
from chromadb.utils.embedding_functions import GoogleGenerativeAiEmbeddingFunction
ef = GoogleGenerativeAiEmbeddingFunction(api_key="...", model_name="models/embedding-001")

# Ollama（本地运行）
from chromadb.utils.embedding_functions import OllamaEmbeddingFunction
ef = OllamaEmbeddingFunction(url="http://localhost:11434/api/embeddings", model_name="nomic-embed-text")

from chromadb import EmbeddingFunction
from chromadb.embeddings import Embeddings

class MyCustomEmbeddingFunction(EmbeddingFunction):
    def __init__(self, model_url: str, **kwargs):
        self.model_url = model_url
    
    def __call__(self, input: list[str]) -> Embeddings:
        """将文本列表转换为嵌入向量列表"""
        # 调用你的嵌入服务
        embeddings = call_my_embedding_service(self.model_url, input)
        return embeddings
    
    def name(self) -> str:
        return "my_custom_ef"
    
    def get_config(self) -> dict:
        return {"model_url": self.model_url}
    
    @staticmethod
    def build_from_config(config: dict) -> "MyCustomEmbeddingFunction":
        return MyCustomEmbeddingFunction(**config)

results = collection.query(
    query_texts=["什么是向量数据库"],  # 查询文本（可传多个）
    n_results=5,                        # 返回的最相似结果数量
    include=["documents", "metadatas", "distances"],  # 返回的字段
    # where={"category": "database"},   # 可选：元数据过滤
    # where_document={"$contains": "Chroma"}  # 可选：文档内容过滤
)

# 等于（默认运算符，可省略 $eq）
collection.query(
    query_texts=["..."],
    where={"category": "database"}           # 简写
    # where={"category": {"$eq": "database"}} # 完整写法
)

# 不等于
where={"status": {"$ne": "archived"}}

# 大于 / 大于等于
where={"priority": {"$gt": 5}}
where={"created_at": {"$gte": 1700000000}}

# 小于 / 小于等于
where={"price": {"$lt": 100.0}}
where={"score": {"$lte": 0.5}}

# 包含于列表
where={"category": {"$in": ["database", "search", "ml"]}}

# 不包含于列表
where={"status": {"$nin": ["deleted", "archived"]}}

# AND 组合
where={
    "$and": [
        {"category": "database"},
        {"priority": {"$gte": 3}},
        {"status": {"$ne": "archived"}}
    ]
}

# OR 组合
where={
    "$or": [
        {"source": "official_docs"},
        {"source": "blog", "priority": {"$gte": 5}}
    ]
}

# NOT
where={
    "$not": {"status": "draft"}
}

# 复杂嵌套
where={
    "$and": [
        {"$or": [
            {"category": "database"},
            {"category": "search"}
        ]},
        {"priority": {"$gte": 3}},
        {"$not": {"status": "deleted"}}
    ]
}

# 写入带数组元数据的数据
collection.add(
    ids=["doc1", "doc2"],
    documents=["Python 向量数据库教程", "Java Web 开发指南"],
    metadatas=[
        {"tags": ["python", "ai", "vector-db"], "level": "intermediate"},
        {"tags": ["java", "web", "spring"], "level": "beginner"}
    ]
)

# 查询包含特定标签的文档
results = collection.query(
    query_texts=["数据库教程"],
    where={"tags": {"$contains": "python"}},
    n_results=5
)

# 查询不包含某标签的文档
results = collection.get(
    where={"tags": {"$not_contains": "java"}}
)

# 文档包含特定子串
collection.query(
    query_texts=["..."],
    where_document={"$contains": "Chroma"}
)

# 文档不包含特定子串
where_document={"$not_contains": "deprecated"}

# 正则表达式匹配
where_document={"$regex": "v\\d+\\.\\d+"}     # 匹配版本号
where_document={"$not_regex": "^\\[DRAFT\\]"}  # 排除草稿

# 组合文档过滤和元数据过滤
results = collection.query(
    query_texts=["向量检索"],
    where={"category": "tutorial"},
    where_document={"$contains": "HNSW"},
    n_results=5
)

# 按 ID 获取
results = collection.get(ids=["doc1", "doc2", "doc3"])

# 按元数据过滤获取（不做向量搜索）
results = collection.get(
    where={"source": "official_docs"},
    limit=100,
    offset=0,  # 分页
    include=["documents", "metadatas"]
)

# 预览集合中的前 N 条数据
results = collection.peek(limit=5)

total = collection.count()
print(f"集合中共有 {total} 条记录")

# 按 ID 删除
collection.delete(ids=["doc1", "doc2"])

# 按元数据条件删除
collection.delete(where={"status": "archived"})

# 按文档内容条件删除
collection.delete(where_document={"$contains": "DEPRECATED"})

TF_component = (f(q, D) × (k₁ + 1)) / (f(q, D) + k₁)

normalized_TF = (f(q, D) × (k₁ + 1)) / (f(q, D) + k₁ × (1 - b + b × |D|/avgdl))

w_j = max_i log(1 + ReLU(W_ij))

from chromadb.utils.embedding_functions import ChromaBM25

bm25_ef = ChromaBM25(
    k=1.2,              # 词频饱和参数
    b=0.75,             # 长度归一化参数
    avg_doc_length=256,  # 平均文档长度
    token_max_length=40  # 最大 token 长度
)

from chromadb import Search, K, Knn, Rrf

# 定义稠密向量搜索（语义匹配）
dense_rank = Knn(
    query="Chroma WAL3 写入日志架构",
    return_rank=True,  # RRF 要求必须设为 True
    limit=200          # 先取较多候选再融合
)

# 定义稀疏向量搜索（关键词匹配）
sparse_rank = Knn(
    query="Chroma WAL3 写入日志架构",
    key="sparse_embedding",  # 引用稀疏索引
    return_rank=True,
    limit=200
)

# RRF 融合
hybrid_rank = Rrf(
    ranks=[dense_rank, sparse_rank],
    weights=[0.7, 0.3],  # 70% 语义，30% 关键词
    k=60,                # 平滑参数
    normalize=True        # 权重归一化
)

# 执行搜索
results = collection.search(
    Search()
        .rank(hybrid_rank)
        .limit(10)
        .where(K("status") == "active")
)

RRF_score(d) = -Σ(wᵢ / (k + rᵢ(d)))

RRF(A) = -(0.7/(60+1) + 0.3/(60+5))
       = -(0.7/61 + 0.3/65)
       = -(0.01148 + 0.00462)
       = -0.01610

RRF(B) = -(0.7/(60+3) + 0.3/(60+1))
       = -(0.7/63 + 0.3/61)
       = -(0.01111 + 0.00492)
       = -0.01603

Loss = -log(exp(sim(image, text) / τ) / Σ exp(sim(image_i, text_j) / τ))

from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction

multimodal_ef = OpenCLIPEmbeddingFunction(
    model_name="ViT-B-32",          # 模型架构
    checkpoint="laion2b_s34b_b79k",  # 预训练权重
    device="cpu"                      # "cpu" 或 "cuda"
)

from chromadb.utils.data_loaders import ImageLoader

loader = ImageLoader()

# 创建多模态集合
collection = client.create_collection(
    name="image_gallery",
    embedding_function=multimodal_ef
)

# 通过 URI 添加图像
collection.add(
    ids=["sunset_001", "city_001", "forest_001"],
    uris=[
        "./images/sunset.jpg",
        "./images/city_skyline.jpg",
        "./images/forest_path.jpg"
    ],
    metadatas=[
        {"description": "海边日落", "location": "三亚", "type": "nature"},
        {"description": "城市夜景", "location": "上海", "type": "urban"},
        {"description": "森林小径", "location": "张家界", "type": "nature"}
    ],
    loader=loader
)

# 文本查图像 - "以文搜图"
results = collection.query(
    query_texts=["海边的夕阳美景"],
    n_results=3,
    loader=loader
)
# 返回最相似的图像（通过 URI 访问）

# 图像查图像 - "以图搜图"
results = collection.query(
    query_uris=["./images/query_sunset.jpg"],
    n_results=3,
    loader=loader
)

# 添加文本描述
collection.add(
    ids=["text_desc_001"],
    documents=["三亚的海滩是观赏日落的绝佳地点，金色的阳光洒在海面上"],
    metadatas=[{"type": "text_description", "location": "三亚"}]
)

# 现在可以用文本查询同时匹配文本和图像
results = collection.query(
    query_texts=["三亚日落"],
    n_results=5,
    loader=loader
)
# 结果可能同时包含图像记录和文本记录

from chromadb import Search, K, Knn, GroupBy, MinK

# 按 category 分组，每组保留分数最小的 3 个结果
search = (
    Search()
    .rank(Knn(query="机器学习算法教程"))
    .group_by(
        GroupBy(
            keys=K("category"),         # 按 category 字段分组
            aggregate=MinK(
                keys=K.SCORE,           # 按分数排序
                k=3                     # 每组保留 3 个
            )
        )
    )
    .limit(30)                           # 总共返回 30 个结果
)

results = collection.search(search)

# 在 Chroma Cloud 或分布式模式中使用
results = collection.search(
    Search()
    .rank(Knn(query="..."))
    .read_level("index_only")  # 低延迟模式
    .limit(10)
)

# 查询索引构建进度
progress = collection.index_progress()
print(f"已索引: {progress.indexed_count}")
print(f"待索引: {progress.pending_count}")
print(f"进度: {progress.percentage}%")

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,      # 每块最大字符数
    chunk_overlap=50,    # 相邻块重叠字符数
    separators=["\n\n", "\n", "。", "，", " ", ""]
)
chunks = splitter.split_text(full_text)

import hashlib

def generate_chunk_id(doc_path: str, chunk_index: int) -> str:
    """生成确定性的分块 ID"""
    raw = f"{doc_path}::chunk_{chunk_index}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

# 增量同步
old_ids = set(collection.get()["ids"])
new_ids = set()

for doc_path, chunks in new_documents.items():
    for i, chunk in enumerate(chunks):
        chunk_id = generate_chunk_id(doc_path, i)
        new_ids.add(chunk_id)
        collection.upsert(
            ids=[chunk_id],
            documents=[chunk],
            metadatas=[{"source": doc_path, "chunk_index": i, "version": "2.0"}]
        )

# 删除不再存在的文档块
deleted_ids = list(old_ids - new_ids)
if deleted_ids:
    collection.delete(ids=deleted_ids)

version: '3.8'
services:
  chroma:
    image: chromadb/chroma:1.5.9
    container_name: chroma-prod
    ports:
      - "8000:8000"
    volumes:
      - chroma-data:/data
      - ./config.yaml:/config.yaml
    environment:
      - IS_PERSISTENT=TRUE
      - CHROMA_OPEN_TELEMETRY__ENDPOINT=http://otel-collector:4317
      - CHROMA_OPEN_TELEMETRY__SERVICE_NAME=chroma-prod
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v2/heartbeat"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 30s
    deploy:
      resources:
        limits:
          memory: 8G
          cpus: '4'
        reservations:
          memory: 4G
          cpus: '2'

volumes:
  chroma-data:
    driver: local
    driver_opts:
      type: none
      o: bind
      device: /mnt/ssd/chroma-data  # 使用 SSD 存储

chroma run --path /data/chroma_db

server {
    listen 443 ssl;
    server_name chroma.yourdomain.com;
    
    ssl_certificate /etc/ssl/certs/chroma.pem;
    ssl_certificate_key /etc/ssl/private/chroma.key;
    
    location / {
        # 验证 API Key
        if ($http_x_chroma_token != "your-secret-token") {
            return 401 '{"error": "Unauthorized"}';
        }
        
        proxy_pass http://chroma:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

# auth_config.yaml
roles:
  admin:
    actions:
      - "system:reset"
      - "collection:*"
  reader:
    actions:
      - "collection:query"
      - "collection:get"
  writer:
    actions:
      - "collection:add"
      - "collection:upsert"
      - "collection:query"
      - "collection:get"

users:
  - id: app-server
    role: writer
    tokens: ["app-writer-token-xyz"]
  - id: search-service
    role: reader
    tokens: ["search-reader-token-abc"]

networks:
  internal:
    driver: bridge
    internal: true  # 不可访问外部网络
  external:
    driver: bridge

services:
  chroma:
    networks: [internal]  # 只能被同网络的服务访问
  app:
    networks: [internal, external]  # 可以访问内外网络

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: chroma
  namespace: ai-platform
spec:
  serviceName: chroma
  replicas: 1  # Chroma 单节点模式
  selector:
    matchLabels:
      app: chroma
  template:
    metadata:
      labels:
        app: chroma
    spec:
      containers:
        - name: chroma
          image: chromadb/chroma:1.5.9
          ports:
            - containerPort: 8000
          volumeMounts:
            - name: data
              mountPath: /data
          resources:
            requests:
              memory: "4Gi"
              cpu: "2"
            limits:
              memory: "8Gi"
              cpu: "4"
          livenessProbe:
            httpGet:
              path: /api/v2/heartbeat
              port: 8000
            initialDelaySeconds: 30
            periodSeconds: 30
          readinessProbe:
            httpGet:
              path: /api/v2/heartbeat
              port: 8000
            initialDelaySeconds: 10
            periodSeconds: 10
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes: ["ReadWriteOnce"]
        storageClassName: ssd  # 使用 SSD 存储类
        resources:
          requests:
            storage: 100Gi
---
apiVersion: v1
kind: Service
metadata:
  name: chroma
  namespace: ai-platform
spec:
  selector:
    app: chroma
  ports:
    - port: 8000
      targetPort: 8000
  type: ClusterIP  # 集群内访问

                      ┌──────────┐
                      │ Gateway  │  ← API 入口、认证、路由
                      └────┬─────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
        ┌─────┴─────┐ ┌───┴───┐ ┌─────┴─────┐
        │  Log (WAL) │ │ SysDB │ │   Query   │
        │  写入日志   │ │ 目录库 │ │  查询节点  │
        └─────┬─────┘ └───────┘ └─────┬─────┘
              │                        │
        ┌─────┴─────┐                  │
        │ Compactor │                  │
        │  压缩节点  │                  │
        └─────┬─────┘                  │
              │                        │
        ┌─────┴────────────────────────┴─────┐
        │    Object Storage (S3 / GCS)       │
        │    对象存储（持久化）                 │
        └────────────────────────────────────┘

import hashlib

def get_shard_collection(client, doc_id: str, num_shards: int = 4):
    """根据文档 ID 的哈希值确定分片"""
    shard = int(hashlib.md5(doc_id.encode()).hexdigest(), 16) % num_shards
    return client.get_or_create_collection(f"docs_shard_{shard}")

# 写入性能对比测试
import time

# 不推荐：逐条写入
for doc_id, doc_text in zip(ids, documents):
    collection.add(ids=[doc_id], documents=[doc_text])
# 1 万条约需 300+ 秒

# 推荐：批量写入
batch_size = 200
for i in range(0, len(ids), batch_size):
    collection.add(
        ids=ids[i:i+batch_size],
        documents=documents[i:i+batch_size]
    )
# 1 万条约需 15-30 秒（10-20 倍提升）

# 大批量导入模板
collection = client.create_collection(
    name="million_docs",
    metadata={
        "hnsw:space": "cosine",
        "hnsw:construction_ef": 256,  # 高质量索引
        "hnsw:M": 24                  # 更多连接
    }
)

batch_size = 200
start = time.time()
for i in range(0, total_docs, batch_size):
    collection.add(
        ids=all_ids[i:i+batch_size],
        embeddings=all_embeddings[i:i+batch_size],
        documents=all_documents[i:i+batch_size],
        metadatas=all_metadatas[i:i+batch_size]
    )
    if (i // batch_size) % 100 == 0:
        elapsed = time.time() - start
        throughput = (i + batch_size) / elapsed
        print(f"进度: {i+batch_size}/{total_docs} "
              f"({(i+batch_size)/total_docs*100:.1f}%) "
              f"吞吐: {throughput:.0f} docs/s")

# 动态调整 search_ef
collection = client.get_collection("my_docs")
collection.modify(metadata={"hnsw:search_ef": 128})  # 增大搜索宽度

# 服务启动时预热索引
_ = collection.peek(limit=1)  # 触发索引加载
_ = collection.count()         # 确认集合可用

from chromadb import Search, Knn

searches = [
    Search().rank(Knn(query=q)).limit(10)
    for q in query_list  # 20 个查询
]
results = collection.search_batch(searches)
# results[i] 对应 searches[i] 的结果

每条向量的内存 ≈ M × 2 × sizeof(int) + dim × sizeof(float) + overhead
             = M × 8 + dim × 4 + 64  (字节)

每条向量 = 16 × 8 + 1536 × 4 + 64 = 128 + 6144 + 64 = 6336 字节 ≈ 6.2 KB
100 万条 = 6.2 GB
1000 万条 = 62 GB

import time
import numpy as np

# 构建测试数据
test_queries = ["query text 1", "query text 2", ...]  # 100+ 条
ground_truth = [...]  # 每条查询的真实最近邻 ID

# 测量延迟
latencies = []
for q in test_queries:
    start = time.perf_counter()
    results = collection.query(query_texts=[q], n_results=10)
    latency = time.perf_counter() - start
    latencies.append(latency)

print(f"平均延迟: {np.mean(latencies)*1000:.1f}ms")
print(f"P50 延迟: {np.percentile(latencies, 50)*1000:.1f}ms")
print(f"P95 延迟: {np.percentile(latencies, 95)*1000:.1f}ms")
print(f"P99 延迟: {np.percentile(latencies, 99)*1000:.1f}ms")

# 测量召回率
recalls = []
for q, truth in zip(test_queries, ground_truth):
    results = collection.query(query_texts=[q], n_results=10)
    retrieved_ids = set(results["ids"][0])
    recall = len(retrieved_ids & set(truth)) / len(truth)
    recalls.append(recall)

print(f"平均 Recall@10: {np.mean(recalls):.4f}")

# docker-compose.yml 环境变量
environment:
  - CHROMA_OPEN_TELEMETRY__ENDPOINT=http://otel-collector:4317
  - CHROMA_OPEN_TELEMETRY__SERVICE_NAME=chroma-production

# Prometheus alerting rules
groups:
  - name: chroma_alerts
    rules:
      - alert: HighQueryLatency
        expr: histogram_quantile(0.99, chroma_query_duration_seconds) > 0.5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Chroma P99 查询延迟超过 500ms"
      
      - alert: HighMemoryUsage
        expr: chroma_memory_usage_bytes / chroma_memory_limit_bytes > 0.85
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Chroma 内存使用率超过 85%"
      
      - alert: IndexBuildStalled
        expr: chroma_pending_index_count > 0 and rate(chroma_indexed_count[5m]) == 0
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "索引构建停滞，有待处理的数据但未进展"

# 比较集合记录数和预期数量
actual = collection.count()
expected = len(expected_ids)
if actual != expected:
    print(f"数据不一致: 预期 {expected}，实际 {actual}")

# 抽样验证
sample_ids = random.sample(expected_ids, min(100, len(expected_ids)))
results = collection.get(ids=sample_ids)
if len(results["ids"]) != len(sample_ids):
    missing = set(sample_ids) - set(results["ids"])
    print(f"丢失了 {len(missing)} 条记录: {list(missing)[:10]}")

# 备份（建议在低负载时段执行）
tar -czf chroma-backup-$(date +%Y%m%d).tar.gz /data/chroma_db/

# 恢复到新机器
tar -xzf chroma-backup-20260614.tar.gz -C /data/

docker exec chroma-server tar -czf /data/backup.tar.gz /data/
docker cp chroma-server:/data/backup.tar.gz ./backup.tar.gz

pip install langchain-chroma langchain-openai langchain-text-splitters

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader

# 1. 加载文档
loader = PyPDFLoader("technical_manual.pdf")
docs = loader.load()

# 2. 分块
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50
)
chunks = splitter.split_documents(docs)

# 3. 创建 Chroma 向量存储并导入数据
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    collection_name="tech_manual",
    persist_directory="./chroma_langchain_db"
)

# 4. 相似度搜索
results = vectorstore.similarity_search(
    query="如何配置 HNSW 索引参数",
    k=5
)

# 5. 带相关性分数的搜索
results_with_scores = vectorstore.similarity_search_with_relevance_scores(
    query="如何配置 HNSW 索引参数",
    k=5
)
for doc, score in results_with_scores:
    print(f"[{score:.4f}] {doc.page_content[:100]}")

# 将 Chroma 封装为 LangChain Retriever
retriever = vectorstore.as_retriever(
    search_type="similarity",       # 或 "mmr"（最大边际相关性）
    search_kwargs={
        "k": 5,
        "filter": {"source": "technical_manual.pdf"}  # 元数据过滤
    }
)

# 在 RAG 链中使用
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("""
基于以下参考资料回答问题。如果资料中没有相关信息，请说明。

参考资料：
{context}

问题：{question}

回答：
""")

llm = ChatOpenAI(model="gpt-4o", temperature=0)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("Chroma 的批量写入最优大小是多少？")

retriever_mmr = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,
        "fetch_k": 20,          # 先检索 20 个候选
        "lambda_mult": 0.5      # 0=最大多样性，1=最大相关性
    }
)

pip install llama-index-vector-stores-chroma llama-index

from llama_index.core import VectorStoreIndex, StorageContext, SimpleDirectoryReader
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# 初始化 Chroma
chroma_client = chromadb.PersistentClient(path="./chroma_llamaindex")
chroma_collection = chroma_client.get_or_create_collection("llama_docs")

# 创建向量存储
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 从文件加载并构建索引
documents = SimpleDirectoryReader("./my_docs").load_data()
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context
)

# 查询
query_engine = index.as_query_engine(similarity_top_k=5)
response = query_engine.query("Chroma 支持哪些嵌入模型？")
print(response)

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 父块（大）和子块（小）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=40)

parent_docs = parent_splitter.split_documents(raw_docs)
child_docs = []
parent_id_map = {}

for parent in parent_docs:
    parent_id = hashlib.md5(parent.page_content.encode()).hexdigest()[:12]
    children = child_splitter.split_documents([parent])
    for child in children:
        child.metadata["parent_id"] = parent_id
        child_docs.append(child)
    parent_id_map[parent_id] = parent

# 子块入 Chroma
vectorstore.add_documents(child_docs)

# 检索时：找到子块 → 回溯父块 → 用父块作为 LLM 上下文

npm install chromadb
# 或
yarn add chromadb

import { ChromaClient, OpenAIEmbeddingFunction } from "chromadb";

// 内存模式（仅开发测试用）
const client = new ChromaClient();

// HTTP 客户端（连接 Chroma 服务）
const httpClient = new ChromaClient({
    path: "http://localhost:8000",
});

// Chroma Cloud 客户端
const cloudClient = new ChromaClient({
    path: "https://api.trychroma.com",
    tenant: "your-tenant",
    database: "your-database",
    token: "your-api-key",
});

import { ChromaClient, OpenAIEmbeddingFunction } from "chromadb";

async function main() {
    const client = new ChromaClient({ path: "http://localhost:8000" });
    
    // 创建嵌入函数
    const embedder = new OpenAIEmbeddingFunction({
        openai_api_key: process.env.OPENAI_API_KEY!,
        openai_model: "text-embedding-3-small",
    });
    
    // 创建集合
    const collection = await client.getOrCreateCollection({
        name: "ts_documents",
        embeddingFunction: embedder,
        metadata: { "hnsw:space": "cosine" },
    });
    
    // 添加数据
    await collection.add({
        ids: ["doc1", "doc2", "doc3"],
        documents: [
            "TypeScript provides type safety for JavaScript",
            "Chroma supports multiple programming languages",
            "Vector databases enable semantic search",
        ],
        metadatas: [
            { category: "programming", lang: "typescript" },
            { category: "database", lang: "multi" },
            { category: "database", lang: "multi" },
        ],
    });
    
    // 语义查询
    const results = await collection.query({
        queryTexts: ["typesafe programming languages"],
        nResults: 3,
        where: { category: "programming" },
    });
    
    console.log("IDs:", results.ids);
    console.log("Documents:", results.documents);
    console.log("Distances:", results.distances);
    
    // 按 ID 获取
    const records = await collection.get({
        ids: ["doc1"],
        include: ["documents", "metadatas", "embeddings"],
    });
    
    // 统计
    const count = await collection.count();
    console.log(`Total records: ${count}`);
    
    // 更新
    await collection.update({
        ids: ["doc1"],
        documents: ["TypeScript is a typed superset of JavaScript"],
        metadatas: [{ category: "programming", lang: "typescript", updated: true }],
    });
    
    // 删除
    await collection.delete({ ids: ["doc3"] });
}

main().catch(console.error);

# config.yaml
cors:
  allow_origins: ["https://your-frontend.com"]
  allow_methods: ["*"]
  allow_headers: ["*"]

// Rust 客户端示例（概念性，API 可能随版本变化）
use chromadb::client::Client;

let client = Client::builder()
    .with_url("http://localhost:8000")
    .with_read_only_failover(true)  // v1.5.9: 自动故障转移到只读副本
    .build();

let collection = client.get_or_create_collection("docs").await?;

// 查询
let results = collection.query(
    vec!["What is vector search?"],
    5,
    None,  // where filter
    None,  // where_document filter
).await?;

# 创建 GitHub 同步源
curl -X POST https://sync.trychroma.com/api/v1/sources \
  -H "x-chroma-token: $CHROMA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "database_name": "my-db",
    "github": {
      "repo": "chroma-core/chroma",
      "branch": "main",
      "collection_name": "chroma-docs",
      "path_filter": "docs/**/*.md"
    }
  }'

curl -X POST https://sync.trychroma.com/api/v1/sources \
  -H "x-chroma-token: $CHROMA_API_KEY" \
  -d '{
    "database_name": "my-db",
    "web": {
      "url": "https://docs.trychroma.com",
      "collection_name": "chroma-docs-web",
      "max_depth": 3
    }
  }'

curl -X POST https://sync.trychroma.com/api/v1/sources \
  -H "x-chroma-token: $CHROMA_API_KEY" \
  -d '{
    "database_name": "my-db",
    "s3": {
      "bucket_name": "my-docs-bucket",
      "region": "us-east-1",
      "collection_name": "s3-docs",
      "aws_credential_id": 42,
      "path_prefix": "documents/"
    }
  }'

import chromadb
from pathlib import Path
from langchain_text_splitters import RecursiveCharacterTextSplitter

class DocumentPipeline:
    """文档处理流水线：解析 → 分块 → 向量化 → 入库"""
    
    def __init__(self, chroma_path: str, collection_name: str):
        self.client = chromadb.PersistentClient(path=chroma_path)
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50
        )
    
    def process_file(self, file_path: str):
        """处理单个文件"""
        text = Path(file_path).read_text(encoding="utf-8")
        chunks = self.splitter.split_text(text)
        
        ids = []
        documents = []
        metadatas = []
        
        for i, chunk in enumerate(chunks):
            chunk_id = f"{Path(file_path).stem}_chunk_{i}"
            ids.append(chunk_id)
            documents.append(chunk)
            metadatas.append({
                "source": str(file_path),
                "chunk_index": i,
                "total_chunks": len(chunks)
            })
        
        # 批量 upsert
        self.collection.upsert(
            ids=ids,
            documents=documents,
            metadatas=metadatas
        )
        return len(chunks)
    
    def process_directory(self, dir_path: str, pattern: str = "**/*.md"):
        """批量处理目录中的所有文件"""
        total = 0
        for path in Path(dir_path).glob(pattern):
            chunks = self.process_file(str(path))
            total += chunks
            print(f"已处理: {path} ({chunks} 个分块)")
        return total

# 使用
pipeline = DocumentPipeline("./chroma_db", "docs_collection")
pipeline.process_directory("./my_docs/")

import os
import json
from datetime import datetime

SYNC_STATE_FILE = "sync_state.json"

def load_sync_state():
    if os.path.exists(SYNC_STATE_FILE):
        return json.load(open(SYNC_STATE_FILE))
    return {}

def save_sync_state(state):
    json.dump(state, open(SYNC_STATE_FILE, "w"))

def incremental_sync(dir_path: str):
    state = load_sync_state()
    changed_files = []
    
    for path in Path(dir_path).glob("**/*.md"):
        mtime = os.path.getmtime(path)
        key = str(path)
        if key not in state or state[key] < mtime:
            changed_files.append(str(path))
            state[key] = mtime
    
    if changed_files:
        pipeline = DocumentPipeline("./chroma_db", "docs_collection")
        for f in changed_files:
            pipeline.process_file(f)
        save_sync_state(state)
        print(f"同步了 {len(changed_files)} 个变更文件")
    else:
        print("没有需要同步的变更")

def get_tenant_collection(client, tenant_id: str):
    return client.get_or_create_collection(
        name=f"tenant_{tenant_id}_docs",
        metadata={"tenant_id": tenant_id}
    )

shared = client.get_or_create_collection("all_tenants")

# 写入时标记 tenant_id
shared.add(
    ids=[f"t1_doc1", f"t2_doc1"],
    documents=["Tenant 1 doc", "Tenant 2 doc"],
    metadatas=[
        {"tenant_id": "t1"},
        {"tenant_id": "t2"}
    ]
)

# 查询时强制过滤 tenant_id
results = shared.query(
    query_texts=["search query"],
    where={"tenant_id": "t1"},  # 必须在中间件层强制注入
    n_results=5
)

class TenantMiddleware:
    def __init__(self, client, tenant_id):
        self.collection = client.get_or_create_collection("shared_data")
        self.tenant_id = tenant_id
    
    def _inject_filter(self, extra_where=None):
        tenant_filter = {"tenant_id": self.tenant_id}
        if extra_where:
            return {"$and": [tenant_filter, extra_where]}
        return tenant_filter
    
    def query(self, query_texts, where=None, **kwargs):
        return self.collection.query(
            query_texts=query_texts,
            where=self._inject_filter(where),
            **kwargs
        )
    
    def add(self, ids, documents, metadatas=None, **kwargs):
        metadatas = metadatas or [{} for _ in ids]
        for m in metadatas:
            m["tenant_id"] = self.tenant_id  # 强制注入
        return self.collection.add(ids=ids, documents=documents, metadatas=metadatas, **kwargs)

admin_client = chromadb.AdminClient(url="http://localhost:8000")

# 创建租户和数据库
admin_client.create_tenant("company_a")
admin_client.create_database("production", tenant="company_a")

# 客户端限定到特定租户
tenant_client = chromadb.HttpClient(
    host="localhost",
    port=8000,
    tenant="company_a",
    database="production"
)

import logging

audit_logger = logging.getLogger("chroma_audit")

class AuditedCollection:
    def __init__(self, collection, user_id):
        self.collection = collection
        self.user_id = user_id
    
    def query(self, *args, **kwargs):
        audit_logger.info(f"User {self.user_id} queried collection {self.collection.name}")
        return self.collection.query(*args, **kwargs)
    
    def add(self, *args, **kwargs):
        audit_logger.info(f"User {self.user_id} added data to {self.collection.name}")
        return self.collection.add(*args, **kwargs)
    
    def delete(self, *args, **kwargs):
        audit_logger.info(f"User {self.user_id} deleted from {self.collection.name}")
        return self.collection.delete(*args, **kwargs)

./chroma_data/
  ├── chroma.sqlite3              # 元数据、集合目录、倒排索引
  ├── {collection_uuid}/
  │   ├── *.bin                   # HNSW 索引文件（图 + 向量）
  │   ├── header.bin              # 索引元信息（维度、距离度量、参数）
  │   └── *.parquet               # 文档文本和嵌入的列式存储
  └── wal/                        # Write-Ahead Log 段

Layer 2（最稀疏，长程连接）：
    A ──────────> D ──────────> G

Layer 1（中等密度）：
    A ────> B ────> D ────> F ────> G

Layer 0（最稠密，包含所有节点）：
    A → B → C → D → E → F → G → H

l = floor(-ln(uniform(0,1)) × m_L)

                    ┌─────────────┐
                    │  Load       │
                    │  Balancer   │
                    └──────┬──────┘
                           │
            ┌──────────────┼──────────────┐
            │              │              │
     ┌──────┴──────┐ ┌────┴────┐  ┌──────┴──────┐
     │  Writer     │ │ Reader  │  │  Reader     │
     │  (主节点)    │ │ (副本1)  │  │  (副本2)    │
     └──────┬──────┘ └────┬────┘  └──────┬──────┘
            │              │              │
     ┌──────┴──────────────┴──────────────┴──────┐
     │            共享存储 (NFS / 对象存储)         │
     └───────────────────────────────────────────┘

import hashlib
import chromadb

class ShardedChroma:
    """应用层分片的 Chroma 管理器"""
    
    def __init__(self, hosts: list[str], collection_name: str):
        self.clients = [
            chromadb.HttpClient(host=host, port=8000)
            for host in hosts
        ]
        self.collection_name = collection_name
        self.num_shards = len(hosts)
    
    def _get_shard(self, doc_id: str) -> int:
        """一致性哈希确定分片"""
        h = int(hashlib.md5(doc_id.encode()).hexdigest(), 16)
        return h % self.num_shards
    
    def add(self, ids, documents, metadatas=None):
        """按分片分组写入"""
        shard_groups = {}
        for i, doc_id in enumerate(ids):
            shard = self._get_shard(doc_id)
            if shard not in shard_groups:
                shard_groups[shard] = {"ids": [], "documents": [], "metadatas": []}
            shard_groups[shard]["ids"].append(doc_id)
            shard_groups[shard]["documents"].append(documents[i])
            if metadatas:
                shard_groups[shard]["metadatas"].append(metadatas[i])
        
        for shard, data in shard_groups.items():
            collection = self.clients[shard].get_or_create_collection(self.collection_name)
            collection.add(**data)
    
    def query(self, query_texts, n_results=10):
        """并行查询所有分片并合并"""
        import concurrent.futures
        
        all_results = []
        
        def query_shard(client):
            collection = client.get_or_create_collection(self.collection_name)
            return collection.query(query_texts=query_texts, n_results=n_results)
        
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = [executor.submit(query_shard, c) for c in self.clients]
            for f in concurrent.futures.as_completed(futures):
                all_results.append(f.result())
        
        return self._merge_results(all_results, n_results)
    
    def _merge_results(self, results_list, n_results):
        """合并多个分片的查询结果"""
        # 收集所有结果并按距离排序
        merged_ids = []
        merged_docs = []
        merged_dists = []
        
        for results in results_list:
            for i in range(len(results["ids"][0])):
                merged_ids.append(results["ids"][0][i])
                merged_docs.append(results["documents"][0][i])
                merged_dists.append(results["distances"][0][i])
        
        # 按距离排序取 Top-N
        combined = list(zip(merged_dists, merged_ids, merged_docs))
        combined.sort(key=lambda x: x[0])
        top_n = combined[:n_results]
        
        return {
            "ids": [[item[1] for item in top_n]],
            "documents": [[item[2] for item in top_n]],
            "distances": [[item[0] for item in top_n]]
        }

# 数据导出脚本
def export_chroma_collection(collection, output_path: str):
    """导出 Chroma 集合的所有数据"""
    import json
    
    all_data = collection.get(include=["documents", "metadatas", "embeddings"])
    
    records = []
    for i in range(len(all_data["ids"])):
        records.append({
            "id": all_data["ids"][i],
            "document": all_data["documents"][i],
            "metadata": all_data["metadatas"][i],
            "embedding": all_data["embeddings"][i]
        })
    
    with open(output_path, "w") as f:
        json.dump(records, f)
    
    print(f"导出了 {len(records)} 条记录到 {output_path}")

quantized_value = round((value - min) / (max - min) × 255)

# 使用 Cohere Rerank
import cohere

co = cohere.Client("your-api-key")

# 第一阶段：粗排
initial_results = collection.query(
    query_texts=["Chroma 向量数据库的部署方案"],
    n_results=50
)

# 第二阶段：重排序
reranked = co.rerank(
    query="Chroma 向量数据库的部署方案",
    documents=[doc for doc in initial_results["documents"][0]],
    model="rerank-v3.5",
    top_n=10
)

# 使用重排序后的结果
final_docs = [initial_results["documents"][0][r.index] for r in reranked.results]

from openai import OpenAI

client = OpenAI()

def hyde_query(original_query: str) -> str:
    """生成假设性文档"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"请回答以下问题（即使不确定也请给出最佳猜测）：\n{original_query}"
        }],
        max_tokens=200
    )
    return response.choices[0].message.content

# 使用 HyDE
hypothetical_answer = hyde_query("Chroma 1.5.9 中的 MaxScore 索引有什么作用？")

# 用假设性答案的嵌入去检索（比原始问题的嵌入更接近真实文档）
results = collection.query(
    query_texts=[hypothetical_answer],
    n_results=10
)

def multi_query_expansion(query: str) -> list[str]:
    """多查询扩展：从不同角度重写查询"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"请从 3 个不同角度重写以下查询，以提升检索效果：\n{query}"
        }]
    )
    # 解析得到 3 个变体查询
    variants = parse_variants(response.choices[0].message.content)
    return variants

# 多路召回
variants = multi_query_expansion("如何优化 Chroma 的查询性能")
all_results = []
for v in variants:
    results = collection.query(query_texts=[v], n_results=10)
    all_results.extend(results["ids"][0])

# 去重 + RRF 融合
unique_ids = list(dict.fromkeys(all_results))  # 保持顺序去重

# 正确：使用 -v 挂载数据卷
docker run -v /host/data:/data -p 8000:8000 chromadb/chroma

# 错误：没有挂载数据卷
docker run -p 8000:8000 chromadb/chroma  # 容器重启后数据丢失！

# 正确访问方式
first_query_results = results["ids"][0]      # 第一个查询的结果列表
first_result_id = results["ids"][0][0]       # 第一个查询的第一个结果 ID

# 错误访问方式
first_result_id = results["ids"][0]          # 这拿到的是整个列表！

# 1. 备份现有数据
tar -czf chroma-backup-before-migration.tar.gz /data/chroma_db/

# 2. 升级 chromadb
pip install --upgrade chromadb>=1.5.9

# 3. 更新代码中的客户端创建方式
# 旧版：
# client = chromadb.Client(chromadb.Settings(persist_directory="./db"))
# 新版：
# client = chromadb.PersistentClient(path="./db")

# 4. 创建 YAML 配置文件替代环境变量
# config.yaml:
#   server:
#     host: "0.0.0.0"
#     port: 8000
#   storage:
#     persist_directory: /data/chroma_db

# 5. 验证数据完整性
python -c "
import chromadb
client = chromadb.PersistentClient(path='./db')
for c in client.list_collections():
    print(f'{c.name}: {c.count()} records')
"