chromadb
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 查询结果不相关
问题 ：查询返回的结果与预期不符。
可能原因和解决方案 ：
- 查询文本太短 ：尝试使用更完整的查询语句
- 文档质量问题 ：确保文档内容清晰、完整
- 相似度阈值 ：可以设置距离阈值过滤结果
# -*- 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 性能优化建议
- 批量添加数据 ：尽量一次性添加多条数据，而不是逐条添加
- 合理设置 n_results ：查询时不要设置过大的 n_results 值
- 使用索引 ：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