1. 什么是 ChromaDB？ #

ChromaDB 是一个轻量级的开源向量数据库，专门设计用于存储和检索向量数据。想象一下，你有一个包含数千篇文档的知识库，当用户提问"如何学习机器学习？"时，ChromaDB 能快速找到与这个问题最相关的文档，即使用户的提问和文档中的文字不完全一样。

向量数据库的作用：

• 语义搜索：根据意思搜索，而不仅仅是关键词匹配
• 相似度检索：找到与查询最相似的内容
• 高效存储：专门优化了向量数据的存储和检索

ChromaDB 的优势：

• 简单易用：API 简洁，上手快
• 轻量级：可以本地运行，无需复杂的服务器配置
• 自动向量化：内置向量化模型，无需手动转换
• 快速检索：使用 HNSW 算法实现高效的相似度搜索

前置知识补充：如果你不了解"向量"和"嵌入"，这里简单解释一下。向量是一串数字，比如 [0.1, 0.5, -0.3, ...]。文本可以通过"嵌入模型"转换成向量。相似的文本会有相似的向量，所以通过比较向量就能找到相似的文本。ChromaDB 就是专门用来存储和检索这些向量的数据库。

2. 环境准备 #

在学习 ChromaDB 之前，需要确保你的 Python 环境已经准备好。ChromaDB 支持 Python 3.8 及以上版本。

2.1 检查 Python 版本 #

在安装之前，先检查一下你的 Python 版本是否符合要求。

# Windows PowerShell：查看 Python 版本
python --version

# macOS 终端：查看 Python 版本
python3 --version

2.2 安装 ChromaDB #

ChromaDB 基于 Python，可以使用 pip 直接安装。安装过程会自动下载所需的依赖包。

# 说明：Windows PowerShell 安装 ChromaDB
# 先升级 pip 到最新版本，确保能正常安装
python -m pip install --upgrade pip
# 安装 ChromaDB 库
python -m pip install chromadb

# 说明：macOS / Linux 终端安装 ChromaDB
# 先升级 pip 到最新版本
python3 -m pip install --upgrade pip
# 安装 ChromaDB 库
python3 -m pip install chromadb

网络加速提示：如果从 PyPI 下载较慢，可以使用国内镜像加速：

• Windows: python -m pip install chromadb -i https://pypi.tuna.tsinghua.edu.cn/simple
• macOS: python3 -m pip install chromadb -i https://pypi.tuna.tsinghua.edu.cn/simple

2.3 验证安装 #

安装完成后，验证一下是否安装成功。

# -*- coding: utf-8 -*-
# 说明：验证 ChromaDB 是否安装成功

# 说明：导入 chromadb 库
import chromadb

# 说明：创建一个临时的内存客户端（不会保存到磁盘）
client = chromadb.EphemeralClient()

# 说明：创建一个测试集合
collection = client.create_collection(name="test")

# 说明：添加一条测试数据
collection.add(documents=["这是一个测试"], ids=["test_1"])

# 说明：查询数据，验证功能是否正常
results = collection.query(query_texts=["测试"], n_results=1)

# 说明：打印查询结果
print("安装成功！查询结果：", results["documents"])

# 说明：如果没有报错并输出了查询结果，说明安装成功

3. 核心概念 #

在使用 ChromaDB 之前，需要理解几个核心概念。这些概念类似于传统数据库中的概念，但专门针对向量数据设计。

• Client（客户端）：数据库的入口，决定数据存储在哪里

  • PersistentClient：持久化客户端，数据保存在磁盘上，程序重启后数据还在
  • EphemeralClient：临时客户端，数据只存在内存中，程序退出后数据就消失了

• Collection（集合）：类似于传统数据库中的"表"，用来存储同一类数据

  • 每个集合有自己的名称
  • 可以存储文档、向量和元数据

• Document（文档）：要存储的原始文本内容，比如一段话、一篇文章等

• Embedding（嵌入向量）：文档转换成的数字向量

  • ChromaDB 会自动将文档转换为向量
  • 也可以使用自定义的模型来生成向量

• Metadata（元数据）：文档的附加信息，比如分类、标签、来源等

  • 可以用来过滤查询结果
  • 格式是字典（键值对）

• ID（标识符）：每条数据的唯一标识

  • 可以是字符串或整数
  • 如果不提供，ChromaDB 会自动生成

4. 创建客户端与集合 #

让我们通过一个简单的例子来快速了解 ChromaDB 的基本用法。

4.1 创建持久化客户端 #

持久化客户端会将数据保存到磁盘，即使程序关闭，数据也不会丢失。

# -*- coding: utf-8 -*-
# 说明：演示如何创建 ChromaDB 客户端和集合

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
# path 参数指定数据存储的目录路径
# 如果目录不存在，ChromaDB 会自动创建
persistent_client = chromadb.PersistentClient(path="./chroma_store")

# 说明：创建一个集合（类似于数据库中的表）
# name 参数是集合的名称，必须是唯一的
# metadata 参数可以添加集合的元数据，用于描述这个集合
collection = persistent_client.create_collection(
    name="notes",  # 集合名称
    metadata={"description": "示例笔记集合"}  # 集合的元数据
)

# 说明：列出所有集合，确认创建成功
# list_collections() 返回所有集合的列表
collections = persistent_client.list_collections()
print("当前所有集合：")
for col in collections:
    print(f"- {col.name}")

4.2 创建临时客户端 #

临时客户端只将数据存在内存中，适合测试或不需要持久化的场景。

# -*- coding: utf-8 -*-
# 说明：演示如何使用临时客户端（数据不保存到磁盘）

# 说明：导入 chromadb 库
import chromadb

# 说明：创建临时客户端（数据只存在内存中）
# EphemeralClient 不需要 path 参数
memory_client = chromadb.EphemeralClient()

# 说明：创建临时集合
temp_collection = memory_client.create_collection(name="temp_notes")

# 说明：添加测试数据
temp_collection.add(documents=["临时数据"], ids=["temp_1"])

# 说明：查询数据
results = temp_collection.query(query_texts=["数据"], n_results=1)
print("临时客户端查询结果：", results["documents"])

# 说明：注意：程序退出后，临时客户端中的数据会丢失

4.3 获取已存在的集合 #

如果集合已经存在，可以使用 get_collection() 或 get_or_create_collection() 方法。

# -*- coding: utf-8 -*-
# 说明：演示如何获取或创建集合

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：方法1：获取已存在的集合（如果不存在会报错）
try:
    existing_collection = client.get_collection(name="notes")
    print("集合已存在：", existing_collection.name)
except Exception as e:
    print("集合不存在：", e)

# 说明：方法2：获取或创建集合（推荐使用）
# 如果集合存在就获取，不存在就创建
collection = client.get_or_create_collection(
    name="notes",
    metadata={"description": "笔记集合"}
)
print("集合名称：", collection.name)

5. 数据写入与查询 #

数据写入和查询是 ChromaDB 的核心功能。ChromaDB 默认会自动将文档转换为向量，你只需要提供原始文本即可。

5.1 写入数据 #

# -*- coding: utf-8 -*-
# 说明：演示如何向 ChromaDB 写入数据

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取或创建集合
collection = client.get_or_create_collection(name="knowledge_base")

# 说明：准备要存储的文档（文本内容）
# documents 是一个列表，每个元素是一个文档（字符串）
documents = [
    "机器学习包含监督学习和无监督学习",
    "Python 拥有丰富的数据科学生态",
    "数据库可以持久化结构化或非结构化数据"
]

# 说明：准备元数据（可选）
# metadatas 是一个列表，每个元素是一个字典，对应一个文档的元数据
metadatas = [
    {"topic": "ml", "level": "intro"},  # 第一个文档的元数据
    {"topic": "python", "level": "beginner"},  # 第二个文档的元数据
    {"topic": "database", "level": "intro"}  # 第三个文档的元数据
]

# 说明：准备唯一标识符（可选）
# ids 是一个列表，每个元素对应一个文档的唯一 ID
# 如果不提供，ChromaDB 会自动生成
ids = ["doc_1", "doc_2", "doc_3"]

# 说明：将数据添加到集合
# add() 方法会将文档转换为向量并存储
collection.add(
    documents=documents,  # 文档列表
    metadatas=metadatas,  # 元数据列表
    ids=ids  # ID 列表
)

# 说明：查看集合中的文档数量
# count() 方法返回集合中存储的文档总数
doc_count = collection.count()
print(f"写入完成，当前集合文档数：{doc_count}")

5.2 查询数据 #

查询是 ChromaDB 的核心功能，它可以根据语义找到最相似的文档。

# -*- coding: utf-8 -*-
# 说明：演示如何从 ChromaDB 查询数据

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取已存在的集合
collection = client.get_collection(name="knowledge_base")

# 说明：执行查询
# query_texts 是查询文本列表（可以是多个查询）
# n_results 是要返回的最相似结果数量
results = collection.query(
    query_texts=["如何入门机器学习？"],  # 查询文本
    n_results=2  # 返回最相似的 2 条结果
)

# 说明：查看查询结果的结构
# results 是一个字典，包含以下键：
# - "documents": 匹配的文档列表
# - "metadatas": 匹配文档的元数据列表
# - "distances": 相似度距离列表（越小越相似）
# - "ids": 匹配文档的 ID 列表

print("查询结果结构：")
print(f"找到 {len(results['documents'][0])} 条匹配结果\n")

# 说明：遍历并打印结果
# results["documents"][0] 表示第一个查询的结果列表
# zip() 用于同时遍历多个列表
for idx, (doc, metadata, distance, doc_id) in enumerate(zip(
    results["documents"][0],  # 文档内容
    results["metadatas"][0],  # 元数据
    results["distances"][0],  # 距离（越小越相似）
    results["ids"][0]  # 文档 ID
), 1):
    print(f"结果 {idx}:")
    print(f"  文档ID：{doc_id}")
    print(f"  匹配文档：{doc}")
    print(f"  附加信息：{metadata}")
    print(f"  相似度距离：{distance:.4f}（越小越相似）")
    print("-" * 50)

5.3 完整示例：写入并查询 #

下面是一个完整的示例，展示完整的写入和查询流程。

# -*- coding: utf-8 -*-
# 说明：完整的写入和查询示例

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取或创建集合
collection = client.get_or_create_collection(name="example")

# 说明：准备文档数据
documents = [
    "机器学习包含监督学习和无监督学习",
    "Python 拥有丰富的数据科学生态",
    "数据库可以持久化结构化或非结构化数据"
]

# 说明：准备元数据
metadatas = [
    {"topic": "ml", "level": "intro"},
    {"topic": "python", "level": "beginner"},
    {"topic": "database", "level": "intro"}
]

# 说明：准备 ID
ids = ["doc_1", "doc_2", "doc_3"]

# 说明：写入数据
print("正在写入数据...")
collection.add(documents=documents, metadatas=metadatas, ids=ids)
print(f"写入完成，共 {collection.count()} 条文档\n")

# 说明：执行查询
print("正在查询：'如何入门机器学习？'\n")
results = collection.query(
    query_texts=["如何入门机器学习？"],
    n_results=2
)

# 说明：打印查询结果
for idx, (doc, metadata, distance) in enumerate(zip(
    results["documents"][0],
    results["metadatas"][0],
    results["distances"][0]
), 1):
    print(f"匹配结果 {idx}:")
    print(f"  文档：{doc}")
    print(f"  元数据：{metadata}")
    print(f"  距离：{distance:.4f}")
    print()

6. 元数据过滤与数据管理 #

元数据可以帮助我们更精确地检索数据，比如只搜索某个分类的文档。同时，我们也可以更新和删除数据。

6.1 元数据过滤查询 #

使用元数据可以过滤查询结果，只返回符合特定条件的文档。

# -*- coding: utf-8 -*-
# 说明：演示如何使用元数据过滤查询结果

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取集合
collection = client.get_collection(name="knowledge_base")

# 说明：执行带过滤条件的查询
# where 参数用于指定元数据过滤条件
# {"topic": {"$eq": "python"}} 表示只返回 topic 等于 "python" 的文档
filtered_results = collection.query(
    query_texts=["数据分析"],  # 查询文本
    n_results=1,  # 返回最相似的 1 条结果
    where={"topic": {"$eq": "python"}}  # 过滤条件：topic 必须是 "python"
)

print("过滤结果（只返回 topic=python 的文档）：")
if filtered_results["documents"][0]:
    print(f"匹配文档：{filtered_results['documents'][0][0]}")
    print(f"元数据：{filtered_results['metadatas'][0][0]}")
else:
    print("未找到匹配的文档")

6.2 常用过滤操作符 #

ChromaDB 支持多种过滤操作符：

# -*- coding: utf-8 -*-
# 说明：演示 ChromaDB 支持的过滤操作符

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取集合
collection = client.get_collection(name="knowledge_base")

# 说明：示例1：等于（$eq）
# 只返回 topic 等于 "python" 的文档
results_eq = collection.query(
    query_texts=["Python"],
    n_results=10,
    where={"topic": {"$eq": "python"}}
)

# 说明：示例2：不等于（$ne）
# 返回 topic 不等于 "python" 的文档
results_ne = collection.query(
    query_texts=["编程"],
    n_results=10,
    where={"topic": {"$ne": "python"}}
)

# 说明：示例3：大于（$gt），小于（$lt）
# 注意：这些操作符主要用于数字类型的元数据
# 这里只是演示语法，实际使用时需要根据你的元数据结构调整
# results_gt = collection.query(
#     query_texts=["查询"],
#     n_results=10,
#     where={"score": {"$gt": 80}}  # score 大于 80
# )

print("过滤查询完成")

6.3 更新数据 #

ChromaDB 不支持直接更新文档，需要先删除再重新添加。

# -*- coding: utf-8 -*-
# 说明：演示如何更新 ChromaDB 中的数据

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取集合
collection = client.get_collection(name="knowledge_base")

# 说明：查看更新前的数据
print("更新前的文档数量：", collection.count())

# 说明：更新数据的方法：先删除再重新添加
# 步骤1：删除要更新的文档
# delete() 方法根据 IDs 删除文档
collection.delete(ids=["doc_2"])

# 说明：步骤2：重新添加更新后的文档
collection.add(
    documents=["Python 是构建 AI 原型的常见语言，广泛用于数据科学和机器学习"],
    metadatas=[{"topic": "python", "level": "intermediate"}],  # 更新元数据
    ids=["doc_2"]  # 使用相同的 ID
)

# 说明：查看更新后的数据
print("更新后的文档数量：", collection.count())

# 说明：验证更新是否成功
results = collection.get(ids=["doc_2"])
print("\n更新后的文档：", results["documents"][0])
print("更新后的元数据：", results["metadatas"][0])

6.4 删除数据 #

# -*- coding: utf-8 -*-
# 说明：演示如何删除 ChromaDB 中的数据

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取集合
collection = client.get_collection(name="knowledge_base")

# 说明：查看删除前的数据
print("删除前的文档数量：", collection.count())

# 说明：根据 ID 删除文档
# delete() 方法可以删除一个或多个文档
collection.delete(ids=["doc_1"])  # 删除单个文档
# collection.delete(ids=["doc_1", "doc_2"])  # 删除多个文档

# 说明：查看删除后的数据
print("删除后的文档数量：", collection.count())

# 说明：根据元数据条件删除文档
# 注意：ChromaDB 的某些版本可能不支持按元数据删除
# 可以先查询符合条件的 IDs，然后再删除

6.5 获取数据 #

除了查询，还可以直接根据 ID 获取数据。

# -*- coding: utf-8 -*-
# 说明：演示如何根据 ID 获取数据

# 说明：导入 chromadb 库
import chromadb

# 说明：创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_store")

# 说明：获取集合
collection = client.get_collection(name="knowledge_base")

# 说明：根据 ID 获取数据
# get() 方法可以根据 IDs 直接获取文档，不需要进行向量查询
results = collection.get(ids=["doc_2"])

# 说明：打印获取的数据
print("获取的文档：", results["documents"])
print("获取的元数据：", results["metadatas"])

# 说明：获取所有数据（不推荐，数据量大时会很慢）
all_data = collection.get()
print(f"\n集合中共有 {len(all_data['ids'])} 条数据")

7. 自定义嵌入函数 #

ChromaDB 默认使用内置的嵌入模型来将文档转换为向量。如果你想要更好的中文支持或使用特定的模型，可以自定义嵌入函数。

7.1 使用 Sentence Transformers #

Sentence Transformers 是一个流行的文本嵌入库，支持多种语言。

# -*- coding: utf-8 -*-
# 说明：演示如何使用自定义嵌入函数（基于 Sentence Transformers）

# 说明：首先需要安装 sentence-transformers 库
# Windows: python -m pip install sentence-transformers
# macOS: python3 -m pip install sentence-transformers

# 说明：导入 SentenceTransformer
from sentence_transformers import SentenceTransformer

# 说明：导入 ChromaDB 的相关类
from chromadb import EmbeddingFunction, PersistentClient

# 说明：加载一个中文友好的嵌入模型
# 首次运行时会自动下载模型（可能需要一些时间）
# "shibing624/text2vec-base-chinese" 是一个专门针对中文优化的模型
print("正在加载中文嵌入模型...")
embedder = SentenceTransformer("shibing624/text2vec-base-chinese")
print("模型加载完成！")

# 说明：定义自定义嵌入函数类
# 必须继承 EmbeddingFunction 并实现 __call__ 方法
class STEmbedding(EmbeddingFunction):
    # 说明：实现 __call__ 方法
    # 当调用 collection.add() 时，会自动调用这个方法将文本转换为向量
    def __call__(self, texts):
        # 说明：使用 SentenceTransformer 将文本列表编码为向量
        # encode() 方法返回 numpy 数组
        embeddings = embedder.encode(texts)

        # 说明：将 numpy 数组转换为列表
        # ChromaDB 需要列表格式，不能直接使用 numpy 数组
        return embeddings.tolist()

# 说明：创建使用自定义嵌入的客户端
client = PersistentClient(path="./custom_store")

# 说明：创建集合时指定自定义嵌入函数
collection = client.create_collection(
    name="custom_embed",
    embedding_function=STEmbedding()  # 使用自定义嵌入函数
)

# 说明：添加文档时，会自动使用自定义的嵌入函数
collection.add(
    documents=["向量数据库提升语义检索效果"],
    metadatas=[{"tag": "vector"}],
    ids=["custom_1"]
)

print("使用自定义嵌入函数写入成功！")

# 说明：查询时也会使用相同的嵌入函数
results = collection.query(
    query_texts=["语义搜索"],
    n_results=1
)
print("\n查询结果：", results["documents"][0])

7.2 简化版自定义嵌入 #

如果你已经有现成的向量数据，可以直接使用。

# -*- coding: utf-8 -*-
# 说明：演示如何直接使用预计算的向量

# 说明：导入 chromadb 库
import chromadb

# 说明：创建客户端
client = chromadb.PersistentClient(path="./vector_store")

# 说明：创建集合时不指定 embedding_function（使用默认的）
# 但如果要直接提供向量，需要在添加数据时使用 embeddings 参数
collection = client.create_collection(
    name="precomputed_vectors",
    # 注意：如果直接提供向量，需要指定向量维度
    # 这里假设向量维度是 384（sentence-transformers 常用维度）
)

# 说明：准备文档
documents = ["这是第一个文档", "这是第二个文档"]

# 说明：准备预计算的向量（示例）
# 实际使用时，你需要使用你的嵌入模型生成这些向量
# 这里只是演示格式
example_vectors = [
    [0.1] * 384,  # 第一个文档的向量（384维）
    [0.2] * 384   # 第二个文档的向量（384维）
]

# 说明：直接添加向量（不使用默认的嵌入函数）
# 注意：如果使用 embeddings 参数，需要确保向量维度正确
# collection.add(
#     documents=documents,
#     embeddings=example_vectors,  # 直接提供向量
#     ids=["vec_1", "vec_2"]
# )

8. 实战案例：知识库检索系统 #

下面是一个完整的实战案例，封装成一个简单的知识库检索类，方便在项目中使用。

# -*- coding: utf-8 -*-
# 说明：实战案例 - 知识库检索系统
# 封装一个完整的知识库类，包含添加文档和查询功能

# 说明：导入 ChromaDB 的持久化客户端
from chromadb import PersistentClient

# 说明：定义知识库检索类
class MiniSearcher:
    """
    迷你知识库检索系统

    功能：
    - 添加文档到知识库
    - 根据查询文本检索最相关的文档
    """

    # 说明：初始化方法
    def __init__(self, db_path="./mini_db", collection_name="knowledge"):
        """
        初始化知识库检索系统

        参数:
            db_path (str): 数据库存储路径，默认为 "./mini_db"
            collection_name (str): 集合名称，默认为 "knowledge"
        """
        # 说明：创建持久化客户端
        self.client = PersistentClient(path=db_path)

        # 说明：获取或创建集合
        # 如果集合已存在就获取，不存在就创建
        self.collection = self.client.get_or_create_collection(collection_name)

        print(f"知识库已初始化，路径：{db_path}，集合：{collection_name}")

    # 说明：批量添加文档的方法
    def add(self, docs, metadatas=None):
        """
        批量添加文档到知识库

        参数:
            docs (list): 文档列表，每个元素是一个字符串
            metadatas (list, optional): 元数据列表，每个元素是一个字典
        """
        # 说明：生成自动 ID
        # 如果集合中已有数据，从现有数量开始编号
        existing_count = self.collection.count()
        ids = [f"doc_{existing_count + i}" for i in range(len(docs))]

        # 说明：如果没有提供元数据，为每个文档创建空的元数据字典
        if metadatas is None:
            metas = [{} for _ in docs]
        else:
            metas = metadatas

        # 说明：确保元数据列表长度与文档列表长度一致
        if len(metas) != len(docs):
            raise ValueError("元数据数量必须与文档数量一致")

        # 说明：添加文档到集合
        self.collection.add(documents=docs, metadatas=metas, ids=ids)

        print(f"成功添加 {len(docs)} 条文档到知识库")

    # 说明：搜索方法
    def search(self, query, top_k=3):
        """
        根据查询文本检索最相关的文档

        参数:
            query (str): 查询文本
            top_k (int): 返回最相关的文档数量，默认为 3

        返回:
            list: 包含文档、元数据和相似度分数的字典列表
        """
        # 说明：执行向量查询
        # query_texts 需要是列表格式
        resp = self.collection.query(query_texts=[query], n_results=top_k)

        # 说明：格式化查询结果
        hits = []
        # 说明：遍历查询结果，提取文档、元数据和距离
        for doc, meta, dist in zip(
            resp["documents"][0],  # 文档内容列表
            resp["metadatas"][0],  # 元数据列表
            resp["distances"][0]   # 距离列表（越小越相似）
        ):
            # 说明：将距离转换为相似度分数（距离越小，相似度越高）
            # 这里使用 1 - distance 作为相似度分数（假设距离范围在 0-1 之间）
            score = 1 - dist if dist <= 1 else 1 / (1 + dist)

            # 说明：构造结果字典
            hits.append({
                "document": doc,    # 文档内容
                "metadata": meta,   # 元数据
                "score": score      # 相似度分数（0-1之间，越大越相似）
            })

        return hits

    # 说明：获取知识库统计信息
    def stats(self):
        """
        获取知识库统计信息

        返回:
            dict: 包含文档数量的字典
        """
        return {
            "文档数量": self.collection.count()
        }


# 说明：主程序入口，演示如何使用
if __name__ == "__main__":
    # 说明：创建知识库实例
    mini = MiniSearcher()

    # 说明：添加文档到知识库
    mini.add(
        docs=[
            "Milvus 是分布式向量数据库，适合大规模数据场景",
            "Chroma 适合桌面快速开发，简单易用",
            "Faiss 是 Facebook 开发的 C++ 向量库，性能优异"
        ],
        metadatas=[
            {"type": "distributed", "language": "Python"},
            {"type": "local", "language": "Python"},
            {"type": "library", "language": "C++"}
        ]
    )

    # 说明：执行查询
    print("\n查询：'桌面原型向量数据库'\n")
    results = mini.search("桌面原型向量数据库", top_k=2)

    # 说明：打印查询结果
    for idx, item in enumerate(results, 1):
        print(f"结果 {idx}:")
        print(f"  文档：{item['document']}")
        print(f"  元数据：{item['metadata']}")
        print(f"  相似度：{item['score']:.4f}")
        print("-" * 50)

    # 说明：查看统计信息
    print("\n知识库统计：", mini.stats())

9. 常见问题与排查 #

在使用 ChromaDB 时，可能会遇到一些常见问题。本节提供解决方案。

9.1 中文查询效果不好 #

问题：使用默认嵌入模型查询中文时，效果不理想。

解决方案：使用自定义的中文嵌入模型（参考第 7 节）。

9.2 查询结果不相关 #

问题：查询返回的结果与预期不符。

可能原因和解决方案：

1. 查询文本太短：尝试使用更完整的查询语句
2. 文档质量问题：确保文档内容清晰、完整
3. 相似度阈值：可以设置距离阈值过滤结果

# -*- coding: utf-8 -*-
# 说明：演示如何设置距离阈值过滤查询结果

# 说明：导入 chromadb 库
import chromadb

# 说明：创建客户端和获取集合
client = chromadb.PersistentClient(path="./chroma_store")
collection = client.get_collection(name="knowledge_base")

# 说明：执行查询
results = collection.query(
    query_texts=["查询文本"],
    n_results=10  # 先获取较多结果
)

# 说明：过滤距离太大的结果（只保留相似度高的）
# 距离阈值可以根据实际情况调整
distance_threshold = 0.5  # 距离小于 0.5 才保留

filtered_docs = []
for doc, dist in zip(results["documents"][0], results["distances"][0]):
    if dist < distance_threshold:
        filtered_docs.append((doc, dist))

print(f"距离阈值 {distance_threshold} 下的结果数量：{len(filtered_docs)}")

9.3 数据持久化问题 #

问题：程序重启后数据丢失。

解决方案：

• 确保使用 PersistentClient 而不是 EphemeralClient
• 检查路径是否正确，有写入权限

# -*- coding: utf-8 -*-
# 说明：演示如何确保数据持久化

# 说明：导入 chromadb 库和 os 模块
import chromadb
import os

# 说明：确保使用持久化客户端
db_path = "./chroma_store"
client = chromadb.PersistentClient(path=db_path)

# 说明：检查路径是否存在（不存在会自动创建）
if not os.path.exists(db_path):
    print(f"创建数据库目录：{db_path}")
    os.makedirs(db_path, exist_ok=True)

# 说明：获取或创建集合
collection = client.get_or_create_collection(name="persistent_data")

# 说明：添加测试数据
collection.add(documents=["测试数据"], ids=["test_1"])

# 说明：验证数据已保存
print(f"数据已保存到：{os.path.abspath(db_path)}")
print(f"当前文档数：{collection.count()}")

# 说明：关闭客户端（可选，但建议在程序结束时关闭）
# client 会自动保存数据

9.4 性能优化建议 #

1. 批量添加数据：尽量一次性添加多条数据，而不是逐条添加
2. 合理设置 n_results：查询时不要设置过大的 n_results 值
3. 使用索引：ChromaDB 会自动创建索引，但首次查询可能较慢

# -*- coding: utf-8 -*-
# 说明：演示批量添加数据（更高效）

# 说明：导入 chromadb 库
import chromadb

# 说明：创建客户端和集合
client = chromadb.PersistentClient(path="./chroma_store")
collection = client.get_or_create_collection(name="batch_example")

# 说明：准备大量数据
documents = [f"这是第 {i} 个文档" for i in range(100)]
metadatas = [{"index": i} for i in range(100)]
ids = [f"doc_{i}" for i in range(100)]

# 说明：批量添加（一次性添加 100 条）
print("正在批量添加数据...")
collection.add(documents=documents, metadatas=metadatas, ids=ids)
print(f"批量添加完成，共 {collection.count()} 条文档")

# 说明：查询时合理设置结果数量
results = collection.query(
    query_texts=["文档"],
    n_results=5  # 只获取最相关的 5 条，而不是全部
)
print(f"查询到 {len(results['documents'][0])} 条结果")

10. 参考 #

• ChromaDB 官方文档：https://docs.trychroma.com/
• Sentence Transformers 文档：https://www.sbert.net/

11. 向量数据库对比 #

项目名称	官网地址	介绍
FAISS	https://github.com/facebookresearch/faiss	Facebook 开源的高性能向量相似性搜索库，适合离线检索和实验研究
Chroma	https://www.trychroma.com/	开源、嵌入式或可持久化的向量数据库，适合快速开发和个人项目
Milvus	https://milvus.io/	企业级分布式向量数据库，支持大规模生产环境和多索引类型
Weaviate	https://weaviate.io/	支持多模态检索和图关系的分布式向量数据库，易于集成，支持GraphQL
Pinecone	https://www.pinecone.io/	云端全托管的向量检索服务，无需运维，适合高可用生产需求
PostgreSQL	https://www.postgresql.org/	经典关系型数据库，通过扩展（如pgvector）支持向量检索

11.1 对比 #

特性维度	FAISS	Chroma	Milvus	Weaviate	Pinecone	PostgreSQL
类型	库/Library	嵌入式/独立服务	分布式数据库	独立服务	云托管服务	关系型数据库
开源	✅ Facebook开源	✅ 开源	✅ LF AI & Data基金会	✅ 开源	❌ 商业云服务	✅ 开源
部署方式	Python库	嵌入式/独立服务	独立集群	独立服务	SaaS云服务	独立服务
向量索引算法	IVF, HNSW, PQ等	HNSW	IVF, HNSW, ANNOY等	HNSW	专有算法	ivfflat, hnsw (pgvector)
标量过滤	❌ 有限支持	✅ 支持	✅ 丰富支持	✅ 强大支持	✅ 支持	✅ 完整SQL支持
多模态支持	❌ 纯向量	✅ 文档+元数据	✅ 向量+结构化数据	✅ 向量+GraphQL	✅ 向量+元数据	✅ 通过扩展
分布式	❌ 单机	❌ 单机	✅ 原生分布式	✅ 集群模式	✅ 全托管分布式	✅ 需要配置
持久化	❌ 需手动保存	✅ 自动持久化	✅ 自动持久化	✅ 自动持久化	✅ 全托管	✅ 完整ACID
查询语言	Python API	Python/HTTP API	Python/REST/gRPC	GraphQL	Python/HTTP API	SQL
事务支持	❌	❌	✅ 有限	✅ 有限	❌	✅ 完整ACID
生产就绪	✅ 研究/中等规模	🟡 发展中期	✅ 企业级	✅ 生产就绪	✅ 企业级	✅ 工业级
社区生态	✅ 强大	🟡 成长中	✅ 活跃	✅ 活跃	❌ 商业支持	✅ 极其强大
学习曲线	中等	简单	较陡峭	中等	简单	陡峭（需SQL知识）
运维复杂度	低	低	高	中等	无	高
成本	免费	免费	免费（自托管）	免费（自托管）	$$$ 按使用量	免费（自托管）

11.2 存储能力 #

数据库	向量维度	最大数据量	元数据存储
FAISS	任意	内存限制	有限
Chroma	2,048	中等规模	✅ 完整
Milvus	32,768	PB级别	✅ 完整
Weaviate	未限制	TB级别	✅ 完整+图关系
Pinecone	20,000	自动扩展	✅ 完整
PostgreSQL	2,000 (pgvector)	TB级别	✅ 完整关系型

11.3 性能特点 #

数据库	搜索速度	写入速度	内存使用	扩展性
FAISS	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	高	有限
Chroma	⭐⭐⭐	⭐⭐⭐	中等	有限
Milvus	⭐⭐⭐⭐	⭐⭐⭐⭐	可配置	优秀
Weaviate	⭐⭐⭐⭐	⭐⭐⭐	中等	良好
Pinecone	⭐⭐⭐⭐	⭐⭐⭐⭐	托管	自动
PostgreSQL	⭐⭐	⭐⭐⭐	可配置	良好

11.4 选择建议 #

11.4.1 推荐场景 #

• FAISS: 研究原型、中等规模、需要最大搜索性能
• Chroma: 快速开始、简单应用、开发测试
• Milvus: 大规模生产环境、企业级需求
• Weaviate: 多模态搜索、图关系数据
• Pinecone: 无运维需求、快速上线、预算充足
• PostgreSQL: 已有PG环境、需要强事务一致性

11.4.2 技术栈考虑 #

• Python生态: FAISS, Chroma
• 云原生: Pinecone, Milvus
• 自托管: Milvus, Weaviate, PostgreSQL
• 全托管: Pinecone