RAG 系列 6：使用 Chroma 实现持久化向量检索

上一篇使用 FAISS 完成了本地向量检索。FAISS 主要管理向量和位置编号，原始文本、业务 ID 与 Metadata 通常需要由应用额外维护。

这一篇改用 Chroma 原生 API。Chroma 可以在一个 Collection 中同时保存 ID、Embedding、Document 和 Metadata，并原生支持持久化、过滤查询和增删改查。

为了让 FAISS、Chroma 和后面的 Milvus 结果可以直接比较，三篇文章统一使用：

• 相同的四条测试文档；
• 相同的本地 Qwen3 Embedding 模型；
• 相同的 HuggingFaceEmbeddings 参数；
• 文档统一调用 embed_documents()；
• 查询统一调用带 Qwen3 query prompt 的 embed_query()。

本文使用 PersistentClient 把数据保存在 data/p06_chroma_native/，不启动独立服务，也不使用 LangChain 的 Chroma VectorStore 包装。

1. Chroma 的整体读写流程

Chroma 把 FAISS 示例中分开管理的内容组织成 Collection 记录：

PersistentClient
└── Collection: life_notes
    ├── ID
    ├── Embedding
    ├── Document
    └── Metadata

写入时，应用把 ID、文档向量、原文和元数据一起交给 collection.add()。查询时，应用传入查询向量、Top-K 数量和可选过滤条件。

这里要始终区分两条 Embedding 路径：

Document → embed_documents() → collection.add/update()
Query    → embed_query()     → collection.query()

Qwen3 会为查询添加 query prompt。如果误用 embed_documents() 生成查询向量，虽然维度仍然是 1024，但向量语义空间的使用方式与 FAISS 示例不再一致。

2. 本地持久化结构

PersistentClient 接收目录路径。创建 Collection、添加或修改数据后，Chroma 自动把数据写入磁盘，不需要额外调用 persist()。

主要文件类似：

2fc4fb95-dc82-4008-8ef6-580c507cbc40 为 collection-id

SQLite 保存系统信息、Collection、Document 和 Metadata 等内容，UUID 目录保存 HNSW 索引文件。迁移或备份时应停止写入并复制完整目录，不能只复制 chroma.sqlite3。

3. 创建运行环境

Chroma 与后面的 PyMilvus 共用 .venv_vector_db：

cd /Users/bianhn/git/llm-learning

python3.12 -m venv .venv_vector_db
source .venv_vector_db/bin/activate
python -m pip install -U pip
python -m pip install -r rag/p06_chroma_native/requirements.txt
python -m pip check

核心依赖为：

chromadb==1.5.0
langchain-huggingface==1.2.2
pymilvus==2.6.9
sentence-transformers==5.1.2
torch==2.9.1
transformers==4.57.6

langchain-huggingface 只负责提供与 FAISS 相同的 Embedding 接口；Chroma 的数据库操作仍然全部使用原生 chromadb API。

4. 共用代码

为了让每种操作的输入、影响和输出更明确，新增、读取、过滤、修改、删除分别放在独立目录：

# Chroma 示例（p06）各脚本共用的配置：模型路径、测试数据、持久化目录。
#
# create_collection / read / filter / update / delete 都从这里 import，
# 保证所有脚本读写同一个 Collection、同一套 Embedding 配置。

from pathlib import Path

import torch
from langchain_huggingface import HuggingFaceEmbeddings

# parents[2]：p06_chroma_native → rag → llm-learning（项目根目录）
PROJECT_ROOT = Path(__file__).resolve().parents[2]
MODEL_PATH = Path("/Users/bianhn/git/llm/qwen3-embedding/model")
# Chroma 持久化目录；PersistentClient(path=DATA_DIR) 会把数据写到这里。
DATA_DIR = PROJECT_ROOT / "data" / "chroma"
# 所有脚本共用的 Collection 名称，get_collection 时必须一致。
COLLECTION_NAME = "life_notes"

# 四条测试文档，与 p05 FAISS 示例文本相同。
# id / text 写入 Chroma 的 ids、documents 字段；其余字段进 metadata（供 filter 演示）。
DOCUMENTS = [
    {
        "id": "doc-1",
        "text": "西湖边有步道和树荫，很适合周末散步。",
        "category": "travel",
        "rating": 4.8,
        "year": 2025,
        "is_public": True,
    },
    {
        "id": "doc-2",
        "text": "苹果富含膳食纤维，是常见的健康水果。",
        "category": "food",
        "rating": 4.6,
        "year": 2024,
        "is_public": True,
    },
    {
        "id": "doc-3",
        "text": "汽车需要定期更换机油并检查轮胎。",
        "category": "car",
        "rating": 4.2,
        "year": 2023,
        "is_public": False,
    },
    {
        "id": "doc-4",
        "text": "杭州植物园环境安静，适合慢慢游览。",
        "category": "travel",
        "rating": 4.7,
        "year": 2025,
        "is_public": True,
    },
]

def create_embeddings_model() -> tuple[HuggingFaceEmbeddings, str]:
    """创建 Embedding 模型，配置与 p05 FAISS 示例一致。

    返回 (embeddings_model, device)：
        embed_documents  → 文档向量（create / update 时用）
        embed_query      → 查询向量（filter 里向量检索时用，带 Qwen3 query prompt）
    """
    device = "mps" if torch.backends.mps.is_available() else "cpu"
    embeddings_model = HuggingFaceEmbeddings(
        model_name=str(MODEL_PATH),
        model_kwargs={"device": device},
        encode_kwargs={"normalize_embeddings": True},
        query_encode_kwargs={
            "prompt_name": "query",
            "normalize_embeddings": True,
        },
    )
    return embeddings_model, device


def metadata_of(item: dict) -> dict:
    """从 DOCUMENTS 的一条记录中提取写入 Chroma 的 metadata。

    text 和 id 由 collection.add 单独传入；metadata 只存标量字段，
    后续可用 where={"category": {"$eq": "travel"}} 等形式过滤。
    """
    return {
        "category": item["category"],
        "rating": item["rating"],
        "year": item["year"],
        "is_public": item["is_public"],
    }

共用代码位于 common.py。Embedding 初始化与 FAISS 完全一致：

def create_embeddings_model() -> tuple[HuggingFaceEmbeddings, str]:
    device = "mps" if torch.backends.mps.is_available() else "cpu"
    embeddings_model = HuggingFaceEmbeddings(
        model_name=str(MODEL_PATH),
        model_kwargs={"device": device},
        encode_kwargs={"normalize_embeddings": True},
        query_encode_kwargs={
            "prompt_name": "query",
            "normalize_embeddings": True,
        },
    )
    return embeddings_model, device

其中：

• encode_kwargs 控制文档向量，启用 L2 归一化；
• query_encode_kwargs 控制查询向量，除归一化外还指定 prompt_name="query"；
• Chroma 使用 cosine distance，归一化向量与 cosine 度量可以直接配合；
• 更新 Document 时必须重新调用 embed_documents()，不能继续保留旧向量。

四条测试文本和 category 与 FAISS 一致。为演示更多过滤表达式，额外增加了 rating、year、is_public：

ID	category	rating	year	is_public	Document
doc-1	travel	4.8	2025	true	西湖边有步道和树荫，很适合周末散步。
doc-2	food	4.6	2024	true	苹果富含膳食纤维，是常见的健康水果。
doc-3	car	4.2	2023	false	汽车需要定期更换机油并检查轮胎。
doc-4	travel	4.7	2025	true	杭州植物园环境安静，适合慢慢游览。

6. 新增：创建 Collection 并写入数据

入口为 create_collection.py。

"""重建持久化 Collection，并新增与 FAISS 示例一致的测试数据。"""

import shutil
import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

import chromadb

from rag.p06_chroma_native.common import (
    COLLECTION_NAME,
    DATA_DIR,
    DOCUMENTS,
    create_embeddings_model,
    metadata_of,
)

embeddings_model, device = create_embeddings_model()
document_vectors = embeddings_model.embed_documents(
    [item["text"] for item in DOCUMENTS]
)

# 本脚本是所有后续操作的起点；重新运行可恢复稳定的初始数据。
if DATA_DIR.exists():
    shutil.rmtree(DATA_DIR)
DATA_DIR.mkdir(parents=True, exist_ok=True)

client = chromadb.PersistentClient(path=str(DATA_DIR))
collection = client.create_collection(
    name=COLLECTION_NAME,
    configuration={"hnsw": {"space": "cosine"}},
)
collection.add(
    ids=[item["id"] for item in DOCUMENTS],
    embeddings=document_vectors,
    documents=[item["text"] for item in DOCUMENTS],
    metadatas=[metadata_of(item) for item in DOCUMENTS],
)

print(f"运行设备：{device}")
print(f"数据库目录：{DATA_DIR}")
print(f"Collection：{collection.name}")
print(f"新增记录数：{collection.count()}")

运行：

python -m rag.p06_chroma_native.create_collection

输出：

运行设备：mps
数据库目录：data/p06_chroma_native
Collection：life_notes
新增记录数：4

add() 要求 ID 不重复。如果业务上需要“存在就修改，不存在就新增”，应显式使用 upsert()，不能依赖 add() 覆盖旧记录。

7. 读取：按 ID 获取记录

入口为 read_records.py：

"""按 ID 读取 Chroma 中的 Document 和 metadata。"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

import chromadb

from rag.p06_chroma_native.common import COLLECTION_NAME, DATA_DIR

client = chromadb.PersistentClient(path=str(DATA_DIR))
collection = client.get_collection(COLLECTION_NAME)
rows = collection.get(
    ids=["doc-1", "doc-4"],
    include=["documents", "metadatas"],
)

print("按 ID 读取 doc-1 和 doc-4：")
for item_id, document, metadata in zip(
    rows["ids"],
    rows["documents"],
    rows["metadatas"],
):
    print(
        f"id={item_id} category={metadata['category']} "
        f"rating={metadata['rating']} text={document}"
    )

运行：

python -m rag.p06_chroma_native.read_records

输出：

按 ID 读取 doc-1 和 doc-4：
id=doc-1 category=travel rating=4.8 text=西湖边有步道和树荫，很适合周末散步。
id=doc-4 category=travel rating=4.7 text=杭州植物园环境安静，适合慢慢游览。

get() 是确定性读取，不计算向量距离。已知业务 ID 时应优先使用它，而不是执行语义搜索。

8. 查询: 多条件过滤数据

入口为 filter_queries.py。该目录集中演示 Metadata、Document 内容和“过滤条件 + 向量”的组合查询。

8.1 Metadata 过滤

Chroma 的 where 使用字典表达式：

类型	示例
等值	{"category": {"$eq": "travel"}}
比较	{"rating": {"$gte": 4.7}}
集合	{"category": {"$in": ["food", "car"]}}
AND	{"$and": [{"category": {"$eq": "travel"}}, {"rating": {"$gte": 4.8}}]}
OR	{"$or": [{"category": {"$eq": "food"}}, {"is_public": {"$eq": False}}]}

比较运算还包括 $gt、$lt、$lte、$ne，集合运算还包括 $nin。

8.2 Document 内容过滤

where_document 针对原始 Document，而不是 Metadata：

document_rows = collection.get(
    where_document={"$contains": "步道"},
    include=["documents"],
)

常用操作包括 $contains、$not_contains、$regex、$not_regex。Document 条件也可以使用 $and、$or 组合。

8.3 带过滤条件的向量查询

查询必须使用 embed_query()：

query = "周末想找一个适合散步的地方"
query_vector = embeddings_model.embed_query(query)

results = collection.query(
    query_embeddings=[query_vector],
    n_results=2,
    where={"category": {"$eq": "travel"}},
    include=["documents", "metadatas", "distances"],
)

这里先用 where 把候选限定为 travel，再在候选中按照 cosine distance 返回 Top-K。

8.4 完整代码

"""演示 metadata、Document 和带过滤条件的向量查询。"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

import chromadb

from rag.p06_chroma_native.common import (
    COLLECTION_NAME,
    DATA_DIR,
    create_embeddings_model,
)


def print_filtered_ids(collection, title: str, where: dict) -> None:
    """执行 metadata 过滤并稳定输出匹配 ID。"""
    rows = collection.get(where=where, include=["metadatas"])
    print(f"{title}：{sorted(rows['ids'])}")


embeddings_model, _ = create_embeddings_model()
query = "周末想找一个适合散步的地方"
query_vector = embeddings_model.embed_query(query)

client = chromadb.PersistentClient(path=str(DATA_DIR))
collection = client.get_collection(COLLECTION_NAME)

print_filtered_ids(
    collection,
    "category = travel",
    {"category": {"$eq": "travel"}},
)
print_filtered_ids(
    collection,
    "rating >= 4.7",
    {"rating": {"$gte": 4.7}},
)
print_filtered_ids(
    collection,
    "category in [food, car]",
    {"category": {"$in": ["food", "car"]}},
)
print_filtered_ids(
    collection,
    "travel 且 rating >= 4.8",
    {
        "$and": [
            {"category": {"$eq": "travel"}},
            {"rating": {"$gte": 4.8}},
        ]
    },
)
print_filtered_ids(
    collection,
    "food 或非公开记录",
    {
        "$or": [
            {"category": {"$eq": "food"}},
            {"is_public": {"$eq": False}},
        ]
    },
)

document_rows = collection.get(
    where_document={"$contains": "步道"},
    include=["documents"],
)
print(f"Document 包含“步道”：{sorted(document_rows['ids'])}")

results = collection.query(
    query_embeddings=[query_vector],
    n_results=2,
    where={"category": {"$eq": "travel"}},
    include=["documents", "metadatas", "distances"],
)
print(f"\n只在 category=travel 中执行向量查询：{query}")
for item_id, document, metadata, distance in zip(
    results["ids"][0],
    results["documents"][0],
    results["metadatas"][0],
    results["distances"][0],
):
    print(
        f"distance={distance:.4f} id={item_id} "
        f"rating={metadata['rating']} text={document}"
    )

运行：

python -m rag.p06_chroma_native.filter_queries

关键输出：

category = travel：['doc-1', 'doc-4']
rating >= 4.7：['doc-1', 'doc-4']
category in [food, car]：['doc-2', 'doc-3']
travel 且 rating >= 4.8：['doc-1']
food 或非公开记录：['doc-2', 'doc-3']
Document 包含“步道”：['doc-1']

只在 category=travel 中执行向量查询：周末想找一个适合散步的地方
distance=0.3502 id=doc-1 rating=4.8 text=西湖边有步道和树荫，很适合周末散步。
distance=0.5340 id=doc-4 rating=4.7 text=杭州植物园环境安静，适合慢慢游览。

Chroma 的 cosine distance 等于 1 - cosine similarity，数值越小越接近。FAISS 示例使用归一化向量和 IndexFlatL2，返回平方 L2 距离；对单位向量有：

squared_l2_distance = 2 × cosine_distance

因此两篇示例的排序可以一致，但距离数值不会相同。

9. 修改：同步更新 Document、Metadata 和 Embedding

入口为 update_record.py。

本例把 doc-4 修改成更明确的“林间步道”描述。因为文本发生变化，必须先生成新的文档向量：

"""修改 doc-4，并同步更新文本对应的向量。"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

import chromadb

from rag.p06_chroma_native.common import (
    COLLECTION_NAME,
    DATA_DIR,
    create_embeddings_model,
)

updated_text = "杭州植物园有安静的林间步道，适合周末散步和慢慢游览。"
updated_metadata = {
    "category": "travel",
    "rating": 4.9,
    "year": 2026,
    "is_public": True,
}
embeddings_model, _ = create_embeddings_model()
updated_vector = embeddings_model.embed_documents([updated_text])[0]

client = chromadb.PersistentClient(path=str(DATA_DIR))
collection = client.get_collection(COLLECTION_NAME)
collection.update(
    ids=["doc-4"],
    embeddings=[updated_vector],
    documents=[updated_text],
    metadatas=[updated_metadata],
)

row = collection.get(
    ids=["doc-4"],
    include=["documents", "metadatas"],
)
print("修改后的 doc-4：")
print(
    f"id={row['ids'][0]} rating={row['metadatas'][0]['rating']} "
    f"text={row['documents'][0]}"
)

运行：

python -m rag.p06_chroma_native.update_record

输出：

修改后的 doc-4：
id=doc-4 rating=4.9 text=杭州植物园有安静的林间步道，适合周末散步和慢慢游览。

update() 只修改已经存在的 ID；upsert() 则会更新已有 ID，或在 ID 不存在时新增记录。

10. 删除：按 ID 和条件删除

代码目录为 delete/，入口为 delete_records.py。

按 ID 删除：

collection.delete(ids=["doc-3"])

按 Metadata 条件删除：

collection.delete(where={"category": {"$eq": "food"}})

完整代码:

"""分别演示按 ID 删除和按 metadata 条件删除。"""

import sys
from pathlib import Path

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

import chromadb

from rag.p06_chroma_native.common import COLLECTION_NAME, DATA_DIR

client = chromadb.PersistentClient(path=str(DATA_DIR))
collection = client.get_collection(COLLECTION_NAME)

collection.delete(ids=["doc-3"])
print("按 ID 删除 doc-3")

collection.delete(where={"category": {"$eq": "food"}})
print("按 metadata 删除 category=food")

remaining_rows = collection.get(include=["documents", "metadatas"])
print(f"剩余记录数：{collection.count()}")
print(f"剩余 ID：{sorted(remaining_rows['ids'])}")

运行：

python -m rag.p06_chroma_native.delete.delete_records

输出：

按 ID 删除 doc-3
按 metadata 删除 category=food
剩余记录数：2
剩余 ID：['doc-1', 'doc-4']

条件删除可能一次影响多条记录。生产代码应先用相同 where 调用 get()，确认匹配 ID 和数量后再删除。

11. 小结

检查项	结果
Chroma	1.5.0
Embedding	与 FAISS 相同的 HuggingFaceEmbeddings 配置
文档编码	embed_documents()
查询编码	embed_query()，启用 Qwen3 query prompt
测试数据	与 FAISS 相同的四条文本和 category
持久化	PersistentClient，重开客户端后仍可读取
新增	collection.add()
读取	collection.get(ids=...)
修改	collection.update()，同步更新向量
删除	按 ID、Metadata 条件删除
Metadata 过滤	等值、比较、集合、AND、OR
Document 过滤	contains、regex 等
向量查询	Metadata 过滤与 cosine Top-K 组合

到这里，Chroma 已经在一个本地持久化 Collection 中完成了新增、读取、过滤、修改和删除。每种操作都位于独立目录，并与 FAISS 使用相同的 Embedding 和测试数据，后续对比 Milvus 时不会再混入模型调用方式的差异。