Agent 知识库增量更新别每次改文档就重建索引一、文档改了一个标点为什么整个知识库都要重建索引RAG 类 Agent 的知识库通常用向量数据库存储文档分块chunks。问题在于大多数实现是全量重建每次有新文档或修改文档删掉旧的数据、重新切分、重新计算嵌入、重新写入。这导致资源浪费99% 的 chunks 没变化但被重新处理。1000 个文档的知识库修改其中 1 个文档全量重建需要重新计算所有 1000 个文档的嵌入——999 个文档的嵌入被白白重算延迟增高一个文档变更导致知识库不可用数分钟重建期间查询要么返回空结果要么返回新旧混合的不一致结果。对于实时性要求高的 Agent客服、金融分析几分钟的不可用是致命的成本膨胀重新计算所有 chunk 的 embeddingAPI 费用翻倍。1000 个文档 × 平均 10 chunks/文档 × $0.0001/chunk $1/次全量重建。每天更新 5 次 $5/天。增量更新只重算变化部分成本降至 $0.05/天——100 倍差距更具体的损失场景客服 Agent 的知识库有 2000 个 FAQ 文档。运营团队每天更新 5-10 个 FAQ修改答案、新增问题。全量重建每次约 5 分钟期间 Agent 检索功能完全不可用——用户问任何问题都得到抱歉知识库正在更新。一天 5 次更新 × 5 分钟不可用 25 分钟/天。增量更新每次约 3 秒不可用时间几乎为零。正确的方案是增量更新变更检测 → 差异计算 → 局部重建 → 原子替换。核心思想只处理变化的部分未变化的部分直接复用。flowchart TD A[文档变更事件] -- B{变更检测} B -- C[文件哈希对比] C -- D{有差异?} D --|否| E[跳过] D --|是| F[计算差异] F -- G[新增 chunks] F -- H[修改 chunks] F -- I[删除 chunks] G -- J[增量嵌入计算] H -- J I -- K[标记删除向量] J -- L[写入新 chunks] K -- M[标记旧 chunks tombstone] L M -- N[事务提交] N -- O{一致性检查} O --|通过| P[索引更新完成] O --|失败| Q[回滚到上一版本]二、增量更新的两个核心引擎文档变更检测不是比较文档内容太贵——1000 个文档逐字对比需要数秒是比较文件哈希。将每个源文档的文件名 内容 MD5 存下来变更时对比。只对哈希值变化的文件做重新处理。哈希对比的成本极低——单次 MD5 计算约 0.1ms1000 个文档约 100ms。但全量内容对比可能需要数秒甚至数分钟。这就是为什么先查变再看改什么是增量更新的第一步——它几乎零成本但消除了 90% 不必要的重计算。实际数据一个 5000 文档的知识库平均每次更新只涉及 2-5 个文档。哈希对比检查 5000 个文档耗时约 500ms但只触发 2-5 个文档的重新处理——从全量重建的数分钟降到增量更新的数秒。Chunk 差异计算文档内容变更后重新按固定策略切分为 chunks如按段落、按 token 数。和已有的 chunks 做 diff新增 chunk旧版本没有需要新计算 embedding。这是最耗时的操作——每次 embedding 计算需要调用一次 OpenAI API约 100ms API 成本删除 chunk新版本没有打 tombstone 标记。tombstone 是逻辑删除——不立即从存储中物理移除而是标记为已删除。查询时跳过 tombstone 标记的 chunk未改变 chunk复用已有 embedding——这是增量更新最大的节省点。如果一个文档有 20 个 chunks修改只影响 2 个那 18 个 chunks 的 embedding 直接复用节省 90% 的 API 调用Chunk 去重的关键细节chunks 的哈希值不仅包含文本内容还应包含位置信息。同一个句子出现在文档开头和结尾语义上下文不同不应被视为同一个 chunk做复用。位置信息的加入方式在哈希计算中拼接 chunk 的序号——hash(chunk_text : chunk_index)。三、增量索引同步器实现import asyncio import hashlib from dataclasses import dataclass, field from datetime import datetime from typing import Optional import json import redis.asyncio as redis from openai import AsyncOpenAI dataclass class DocumentMeta: 文档元数据——用于增量检测的最小单位。 doc_id: str file_path: str content_hash: str chunk_count: int updated_at: str def chunk_text(text: str, chunk_size: int 512, overlap: int 50) - list[str]: 文本分块——使用字符级分割而非 token 级简化实现。 为什么用字符级而非 token 级 1. 不需要额外的 tokenizer 依赖 2. 计算速度快纯字符串操作 vs tokenizer 的正则匹配 3. 对于增量更新场景字符级分割的结果更稳定—— 修改一个字只影响包含该字的 chunk而 token 级分割可能 因为 token 边界变化导致相邻 chunks 也被重新计算 overlap 的作用确保相邻 chunks 有一定重叠 避免关键信息恰好在 chunk 边界被切断。 chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) chunks.append(text[start:end]) start end - overlap return chunks class IncrementalIndexer: 增量知识库索引器。 设计原则 - 先查变再看改什么——哈希对比是最便宜的操作 - 只重建变化部分的 embedding——节省 API 成本 - 原子写入——新 chunk 全部就绪后才生效不产生中间状态 def __init__(self, redis_url: str): self._client AsyncOpenAI() self._redis redis.from_url(redis_url) self._chunk_size 512 async def sync_document(self, doc_id: str, content: str) - dict: 同步单个文档——增量更新的主入口。 # 第一步检查是否需要更新 new_hash hashlib.md5(content.encode()).hexdigest() old_meta await self._get_meta(doc_id) if old_meta and old_meta.content_hash new_hash: return {added: 0, removed: 0, unchanged: old_meta.chunk_count} # 第二步新分块 new_chunks chunk_text(content, self._chunk_size) new_chunk_hashes [ hashlib.md5(c.encode()).hexdigest() for c in new_chunks ] # 第三步获取旧 chunks old_chunk_hashes set() if old_meta: old_chunk_hashes set(await self._get_chunk_hashes(doc_id)) new_chunk_hashes_set set(new_chunk_hashes) # 第四步差异计算 to_add new_chunk_hashes_set - old_chunk_hashes to_remove old_chunk_hashes - new_chunk_hashes_set unchanged new_chunk_hashes_set old_chunk_hashes_set # 第五步处理新增 chunks added 0 for i, chunk_hash in enumerate(new_chunk_hashes): if chunk_hash in to_add: chunk_text_val new_chunks[i] embedding await self._compute_embedding(chunk_text_val) await self._write_chunk(doc_id, chunk_hash, chunk_text_val, embedding) added 1 # 第六步标记删除 for chunk_hash in to_remove: await self._mark_deleted(doc_id, chunk_hash) # 第七步更新文档元数据 await self._save_meta(DocumentMeta( doc_iddoc_id, file_path, content_hashnew_hash, chunk_countlen(new_chunks), updated_atdatetime.now().isoformat(), )) return { added: added, removed: len(to_remove), unchanged: len(unchanged), } async def search(self, query: str, top_k: int 5) - list[dict]: 知识检索——返回最相关的 chunks。 q_embedding await self._compute_embedding(query) results [] cursor 0 while True: cursor, keys await self._redis.scan( cursor, matchchunk:*:embedding, count100 ) for key in keys: chunk_key key.decode() content_key chunk_key.replace(:embedding, :content) # 跳过已标记删除的 deleted await self._redis.get(chunk_key.replace(:embedding, :deleted)) if deleted: continue emb_data await self._redis.get(chunk_key) if not emb_data: continue embedding json.loads(emb_data) similarity sum(a * b for a, b in zip(q_embedding, embedding)) content await self._redis.get(content_key) results.append({ chunk_key: chunk_key, content: content.decode() if content else , score: similarity, }) if cursor 0: break results.sort(keylambda x: x[score], reverseTrue) return results[:top_k] async def _compute_embedding(self, text: str) - list[float]: 计算文本嵌入——带重试和降级。 for attempt in range(3): try: resp await asyncio.wait_for( self._client.embeddings.create( modeltext-embedding-3-small, inputtext ), timeout10.0, ) return resp.data[0].embedding except asyncio.TimeoutError: if attempt 2: raise await asyncio.sleep(2**attempt) except Exception: if attempt 2: raise await asyncio.sleep(1) async def _write_chunk( self, doc_id: str, chunk_hash: str, content: str, embedding: list[float], ) - None: 原子写入 chunk 数据——使用 pipeline 保证 content 和 embedding 同时写入。 key_prefix fchunk:{doc_id}:{chunk_hash} pipeline self._redis.pipeline() pipeline.set(f{key_prefix}:content, content) pipeline.set(f{key_prefix}:embedding, json.dumps(embedding)) await pipeline.execute() async def _mark_deleted(self, doc_id: str, chunk_hash: str) - None: 标记 chunk 删除不立即物理删除方便回滚。 await self._redis.set(fchunk:{doc_id}:{chunk_hash}:deleted, 1) async def _get_meta(self, doc_id: str) - Optional[DocumentMeta]: data await self._redis.get(fdoc:{doc_id}:meta) if not data: return None d json.loads(data) return DocumentMeta(**d) async def _save_meta(self, meta: DocumentMeta) - None: await self._redis.set( fdoc:{meta.doc_id}:meta, json.dumps({ doc_id: meta.doc_id, file_path: meta.file_path, content_hash: meta.content_hash, chunk_count: meta.chunk_count, updated_at: meta.updated_at, }), ) async def _get_chunk_hashes(self, doc_id: str) - list[str]: 获取文档所有 chunk 哈希。 hashes [] cursor 0 while True: cursor, keys await self._redis.scan( cursor, matchfchunk:{doc_id}:*:content, count100 ) for key in keys: parts key.decode().split(:) if len(parts) 3: hashes.append(parts[2]) if cursor 0: break return hashes四、增量更新的代价哈希碰撞风险MD5 已不适合安全场景但对 chunk 去重来说128bit 哈希值碰撞概率约为 1/2⁶⁴ 次。100 万 chunks 中碰撞概率可以忽略。但如果真的发生碰撞两个不同内容的 chunk 产生相同的 MD5会导致新 chunk 被错误地复用旧 chunk 的 embedding——这是一个严重的质量问题。解决方案在碰撞风险不可接受的场景下如安全领域用 SHA-256 替代 MD5。SHA-256 的碰撞概率约 1/2¹²⁸代价是计算速度慢约 30%——对于 chunk 去重场景完全可接受。Tombstone 膨胀标记删除但不物理删除时间久了 Redis 里全是 tombstone。假设一个文档平均每 2 周更新一次每次产生 5 个 tombstone。1000 个文档一年产生约 260000 个 tombstone 条目——每个条目约 50 字节总计约 13MB。Redis 能 hold 住但查询时的过滤开销增加——每个查询都要检查 deleted 标记。解决方案定期清理如每天凌晨 2 点合并 tombstone。合并操作扫描所有 tombstone 标记的 key物理删除对应的数据content embedding tombstone 标记本身然后更新文档元数据中的 chunk_count。合并可以在低流量时段执行不影响正常查询。一致性保证如果新增 chunk 写入成功但元数据更新失败会怎么样答案是索引和元数据不一致——新 chunk 存在于 Redis 中但文档元数据不知道它们的存在。下次查询时这些 chunk 可能被遗漏。解决方案用 Redis 事务MULTI/EXEC或 Lua 脚本整体原子执行。将 chunk 写入和元数据更新放在同一个事务中——要么全部成功要么全部失败。Lua 脚本更适合这个场景因为它可以在 Redis 内部执行多步逻辑避免客户端和 Redis 之间的多次网络往返。Trade-off 讨论原子性 vs 性能MULTI/EXEC 保证原子性但阻塞 Redis事务执行期间其他命令排队。对于小规模更新几个 chunks没问题大规模更新可能影响查询性能。推荐日常增量更新不用事务因为 chunk 写入和元数据更新的时间差很小只在批量更新时使用事务。tombstone 清理频率 vs 存储开销每天清理一次最干净但有性能开销每周清理一次开销小但存储膨胀更多。推荐日活 1000 查询的场景每周清理日活 10000 的场景每天清理。哈希算法选择 vs 安全性MD5 快但碰撞风险高SHA-256 安全但慢 30%。知识库场景选 MD5碰撞概率可忽略安全场景选 SHA-256。五、总结增量更新的核心价值不在省了计算而在知识库的可用性。全量重建意味着更新期间 Agent 检索不到新文档或检索到一半新旧混合的结果。增量更新把重建时间从分钟级降到秒级让知识库真正做到了热更新。哈希对比是其中最被低估的一步——它几乎零成本但消除了 90% 不必要的重计算。工程实践中的建议先用 MD5 做快速对比碰撞风险不可接受时升级到 SHA-256tombstone 每天凌晨自动清理chunk 写入和元数据更新必须原子执行。
