企业官网建设流程全解析

Agent 知识库增量更新：别每次改文档就重建索引

一、文档改了一个标点，为什么整个知识库都要重建索引？

RAG 类 Agent 的知识库通常用向量数据库存储文档分块（chunks）。问题在于大多数实现是全量重建：每次有新文档或修改文档，删掉旧的数据、重新切分、重新计算嵌入、重新写入。这导致：

1. 资源浪费：99% 的 chunks 没变化但被重新处理。1000 个文档的知识库，修改其中 1 个文档，全量重建需要重新计算所有 1000 个文档的嵌入——999 个文档的嵌入被白白重算
2. 延迟增高：一个文档变更导致知识库不可用数分钟（重建期间查询要么返回空结果，要么返回新旧混合的不一致结果）。对于实时性要求高的 Agent（客服、金融分析），几分钟的不可用是致命的
3. 成本膨胀：重新计算所有 chunk 的 embedding，API 费用翻倍。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.1ms，1000 个文档约 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_id=doc_id, file_path="", content_hash=new_hash, chunk_count=len(new_chunks), updated_at=datetime.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, match="chunk:*:embedding", count=100 ) 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(key=lambda x: x["score"], reverse=True) 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( model="text-embedding-3-small", input=text ), timeout=10.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 = f"chunk:{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(f"chunk:{doc_id}:{chunk_hash}:deleted", "1") async def _get_meta(self, doc_id: str) -> Optional[DocumentMeta]: data = await self._redis.get(f"doc:{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( f"doc:{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, match=f"chunk:{doc_id}:*:content", count=100 ) 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-256；tombstone 每天凌晨自动清理；chunk 写入和元数据更新必须原子执行。