智能体知识库怎么同步更新？用 Chroma 完成覆盖写入与旧片段删除

智能体知识库更新文档后，旧片段仍被检索到，常见原因是只把新片段追加进向量库，没有更新稳定 ID，也没有删除本次已消失的片段。用 Chroma 可以把同步拆成三步：读取同一来源的旧 ID、upsert 本次完整片段、删除旧 ID 与新 ID 的差集。

下面用一份虚构设备手册演示从“两段旧版”更新为“一段新版”。人工二维向量只用于验证存储变化，不代表语义检索效果。接口依据为 2026 年 10 月 1 日读取的官方文档；本文未对实际业务库执行更新。

先确定同步的是完整文档还是增量片段

差集删除只适用于“本次拿到了这个来源的完整现行片段”。如果抓取超时只拿到一半、上游只是发来一个修改通知，不能把其余片段误当作已经删除。

示例按顺序号构造 ID，方便观察两段变一段。生产环境如果能取得稳定段落标识，可用它减少无关重写；分段位置大幅改变时应按完整快照重新同步。无论怎么取 ID，都要保存来源和版本，不能只存一串向量。

创建本地演示库

在独立 Python 环境安装：python -m pip install -U chromadb。官方客户端说明提供 PersistentClient，它会将数据保存到指定本地目录并在下次启动时载入。下面使用新目录 ./chroma-sync-demo，避免误连真实业务数据。

import chromadb

client = chromadb.PersistentClient(path="./chroma-sync-demo")
collection = client.get_or_create_collection(
    name="manual_sync_demo", embedding_function=None
)

这里显式传入向量，不依赖默认嵌入函数。添加数据说明要求记录 ID 唯一，传入的嵌入维度应与集合一致。以后使用真实模型时，文档变化就重新生成对应向量，不能改原文却沿用原文不匹配的旧向量。

用完整快照执行覆盖写入和差集删除

将下列代码接在客户端初始化后。同步函数先校验所有条目，再读取旧 ID；它只处理传入 source_id 的资料，不删除其他来源。当前演示串行执行，不提供跨操作事务保证。

from math import isfinite

def sync_complete_snapshot(source_id, version, rows):
    # rows: [(chunk_id, text, embedding), ...]
    new_ids = [row[0] for row in rows]
    if len(new_ids) != len(set(new_ids)):
        raise ValueError("同一快照中出现重复片段 ID")
    for chunk_id, text, vector in rows:
        if not chunk_id.startswith(source_id + ":") or not text.strip():
            raise ValueError("片段来源或正文不合法")
        if len(vector) != 2 or not all(isfinite(x) for x in vector):
            raise ValueError("演示集合要求有限的二维向量")

    old = collection.get(where={"source_id": source_id}, include=["metadatas"])
    old_ids = set(old["ids"])
    if rows:
        collection.upsert(
            ids=new_ids,
            documents=[row[1] for row in rows],
            embeddings=[row[2] for row in rows],
            metadatas=[{"source_id": source_id, "version": version} for _ in rows],
        )
    removed = sorted(old_ids - set(new_ids))
    if removed:
        collection.delete(ids=removed)
    return {"written": len(rows), "deleted_ids": removed}

print(sync_complete_snapshot("manual-a", 1, [
    ("manual-a:0", "示例旧手册：设备 A 保修 6 个月。", [1.0, 0.0]),
    ("manual-a:1", "示例旧手册：保修凭纸质票据办理。", [0.8, 0.2]),
]))

print(sync_complete_snapshot("manual-a", 2, [
    ("manual-a:0", "示例新手册：设备 A 保修 12 个月，凭购买凭证办理。", [0.9, 0.1]),
]))

current = collection.get(
    where={"source_id": "manual-a"}, include=["documents", "metadatas"]
)
print(current)
assert current["ids"] == ["manual-a:0"]
assert current["metadatas"][0]["version"] == 2
assert "12 个月" in current["documents"][0]
assert not collection.get(ids=["manual-a:1"])["ids"]

官方更新说明区分 update 和 upsert：update 只更新已存在的 ID，缺失 ID 会被忽略；upsert 更新已有记录，也创建不存在的记录。完整快照既可能修改旧片段，也可能新增片段，所以这里使用 upsert，并同时提供 documents、embeddings 和 metadatas。

官方删除说明确认 delete 可按 ID 删除对应向量、原文和元数据。代码仅删除经过差集计算的 removed；先看返回的 deleted_ids 是否为 manual-a:1，再核对最终资料。

怎样验收同步成功

Query and Get 官方说明区分 get 的按 ID/条件读取与 query 的相似度检索。同步验收先用 get 确认精确存储状态，不先靠自然语言回答猜测更新有没有成功。

1. 本次来源只剩 manual-a:0，版本为 2，原文为“12 个月”。
2. manual-a:1 直接按 ID 读取应为空；“纸质票据”旧片段不再作为独立记录存在。
3. 再次执行同一新版完整快照，来源记录数仍为 1，不出现一份文档越同步越多的现象。
4. 接入真实嵌入模型后，再用“设备 A 保修多久”检索，检查返回 ID、版本和原文。数据库状态正确但问答仍用旧值时，继续检查应用缓存、查询所连集合和回答上下文。

如果任一断言失败，保存来源快照、写入响应、删除 ID 和 get 结果；不要把一条“同步完成”的日志当作验收。

生产同步需要补齐哪些边界

• 中途失败：upsert 和 delete 是两次操作，之间失败可能留下新旧片段并存。记录源版本及预期 ID 集合，重跑同一完整快照并回读；需要避免中间状态被查询时，采用新集合构建后切换，或设计经过验证的版本发布机制。
• 并发更新：同一来源必须串行或有明确的版本冲突处理，不能让晚到的旧版覆盖新版。
• 空快照：传入空 rows 会删除这个来源的全部片段。只有上游明确确认文档删除或确实为空时才允许，不把抓取失败包装成空列表。
• 模型变化：新模型即使维度一样，也不要混入旧嵌入空间；新建集合并重建索引。
• 权限边界：source_id、租户及访问条件由可信应用上下文决定，不能让模型构造任意删除范围。生产库执行前应先查看待删除清单。

先用演示库验证“两段变一段”和重复同步，再把自己的完整文档快照映射到同一接口。原文、向量、版本和删除清单能够互相对应，才算知识库更新真正闭合。