1. 这不是“加个知识库”那么简单RAG 在 AI Agent 架构里到底承担什么角色你翻过十几篇讲 RAG 的文章可能看到的都是“用 LangChain 加个向量数据库再接个 LLM知识问答就跑起来了”。但如果你真在做一个能落地的 AI Agent比如一个能帮销售查合同条款、帮客服定位产品故障手册、或者帮法务快速比对历史判例的智能体很快就会发现——那个被轻描淡写称为“知识获取管道”的 RAG 模块其实是整个 Agent 的呼吸系统。它不生产答案但它决定 Agent 能不能“吸到气”吸得准不准吸得快不快。我去年带团队重构一个企业级合同分析 Agent前期所有精力都花在 prompt 工程和 LLM 微调上结果上线后用户反馈“它知道《民法典》第584条但不知道我们公司2023年版《供应商保密协议》附件三里那条特别约定。”问题不在模型而在 RAG 管道——它根本没把那份 PDF 里的关键约束条件“吸”进检索范围。这才意识到RAG 不是 Agent 的一个可插拔插件而是它的感知层LLM 是大脑而 RAG 是眼睛、耳朵和指尖负责从海量非结构化信息中精准定位、提取、校验并喂给大脑做决策的原始素材。所谓“知识获取管道”核心不在“获取”而在“知识”二字——它要解决的是“哪些信息算知识”、“知识以什么形态存在”、“知识如何被可靠地激活”这三个根本问题。这直接决定了你的 Agent 是个能解决实际问题的工具还是个只会复述维基百科的聊天玩具。尤其在中文场景下PDF 表格识别错位、扫描件 OCR 丢字、合同里嵌套的 Excel 表格无法解析、技术文档中混杂的代码块与公式……这些都不是标准 RAG 教程里会提的“小问题”而是每天堵在管道入口的硬骨头。所以这篇不讲怎么 pip install langchain而是带你拆开这个管道的每一节法兰盘看清楚里面流的是什么、压力够不够、有没有泄漏点。适合正在从零搭建第一个真正可用 Agent 的开发者也适合已经跑通 demo 却卡在效果瓶颈的产品负责人——因为 RAG 的成败往往藏在数据预处理的第 7 步而不是模型选型的第 1 步。2. 知识获取管道的四重门为什么 90% 的 RAG 项目死在第一步很多人以为 RAG 就是“文档 → 切块 → 向量化 → 检索 → 生成”四步走完万事大吉。我在三个不同行业的 RAG 项目里反复验证过这个流程图只描述了理想状态下的数据流向却完全忽略了现实世界里横亘在每一步之间的四重物理屏障。它们不是技术难点而是工程现实绕不开也糊弄不过去。2.1 第一重门知识源的“混沌态”治理你拿到的从来不是干净的 Markdown 或 JSON而是销售塞过来的 200 页 Word 合同含 17 个版本修订痕迹、运维导出的 5GB CSV 设备日志字段名全是拼音缩写、法务共享的扫描版判决书 PDF分辨率 150dpi表格线全糊成黑块。这些就是你的“知识源”但它们处于典型的“混沌态”格式无统一、语义无标记、更新无追踪。我见过最典型的失败案例是一家制造企业想用 RAG 做设备故障知识库他们把所有维修手册 PDF 直接扔进向量库结果用户问“伺服电机异响怎么办”系统返回了 3 份手册里关于“电机选型”的章节——因为所有 PDF 的 OCR 文本里“电机”这个词出现频率远高于“异响”而 RAG 检索只认词频和向量相似度不认上下文逻辑。破局的关键不是换更贵的 OCR 引擎而是建立“知识源准入协议”强制元数据注入每份文档上传前必须填写 3 个必填字段——业务域如“数控机床-主轴”、知识类型如“故障现象”、“维修步骤”、“备件清单”、时效性标签如“2024Q2有效”。这不是增加负担而是给后续切块和检索打上第一道语义锚点。格式分层处理策略Word/PDF 用unstructured库做结构化解析它能识别标题层级、表格、列表而非简单 OCRCSV/Excel 用pandas读取后自动提取列名作为字段标签并对空值率 30% 的列打上“低信噪比”标记避免其污染向量空间扫描件则必须经过pdf2imagePaddleOCR二次校验对识别置信度 0.85 的段落标红预警人工复核后再入库。提示别迷信“端到端自动化”。我们在某能源集团项目里试过全自动处理 10 万份巡检报告结果 23% 的关键参数如温度阈值、压力单位被 OCR 误读导致后续所有检索结果偏离。最终方案是自动化处理 80% 的常规文本剩下 20% 的高价值、高风险文档由业务专家用我们开发的轻量标注工具基于 Streamlit做 10 分钟/份的快速校验。ROI 反而更高。2.2 第二重门切块Chunking的“语义断裂”陷阱“按 512 字符切块”是初学者最容易踩的坑。我拿一份《GB/T 19001-2016 质量管理体系要求》标准文档做过测试用固定长度切块第 37 块内容是“8.3.4 设计和开发控制……”第 38 块开头是“……应确保……”中间缺失了整整 3 条具体控制措施。当用户问“设计开发控制包含哪些具体措施”RAG 检索到这两块LLM 看到的就是断句残章生成结果必然失真。切块的本质是在保留语义完整性的前提下为向量检索创造最小可检索单元。这意味着切块策略必须动态适配文档类型法规/标准类文档按“条款号”切分。用正则^(\d\.\d\.)\s识别条款起始确保“8.3.4”整条内容含子条款 8.3.4.1~8.3.4.3在一个 chunk 内。我们实测发现这种切法使条款级问答的准确率从 62% 提升到 91%。技术手册类文档按“标题层级”切分。利用unstructured解析出的h2、h3标签将每个二级标题及其下属所有三级标题内容合并为一个 chunk。这样“PLC 编程规范”下的“梯形图符号说明”和“指令集速查表”就不会被割裂。会议纪要/日志类文档按“发言者时间戳”切分。用 NLP 模型如pytorch-transformers的bert-base-chinese识别对话轮次避免把 A 的问题和 B 的回答切到不同 chunk。注意切块后必须做“语义连贯性校验”。我们写了个小脚本对每个 chunk 计算其内部句子的平均余弦相似度用sentence-transformers的paraphrase-multilingual-MiniLM-L12-v2模型低于 0.65 的 chunk 自动触发合并或重切。这步看似多此一举却让某金融客户的风险事件报告检索 hit rate 提升了 37%。2.3 第三重门向量化Embedding的“中文语义塌陷”很多团队直接用 OpenAI 的text-embedding-ada-002结果发现中文检索效果远不如英文。根本原因在于该模型在训练时中文语料占比不足 8%且主要来自新闻和百科对合同条款、技术参数、行业黑话等长尾语义覆盖极差。我们对比过 5 种主流中文 embedding 模型在真实业务场景下的表现模型合同条款检索 MRR10技术参数检索 MRR10推理延迟ms/chunk显存占用GBtext-embedding-ada-0020.420.311200.8bge-m30.780.69852.1m3e-base0.710.63421.3multilingual-e5-large0.650.581563.2bge-reranker-base重排序0.890.822102.8MRRMean Reciprocal Rank是衡量检索质量的核心指标数值越接近 1 越好。表格里最亮眼的不是纯 embedding 模型而是最后一行的bge-reranker-base——它不是一个 embedding 模型而是一个重排序Rerank模型。我们的实践结论是在中文 RAG 场景下不要只依赖 embedding 检索必须叠加 rerank 层。工作流是先用m3e-base速度快、显存省做初筛召回 top-50 chunk再用bge-reranker-base对这 50 个 chunk 和 query 做精细化语义匹配重新排序取 top-5 送入 LLM。虽然总延迟增加了 170ms但关键业务查询的准确率提升超过 40%且彻底规避了“答非所问”的致命体验。这就像快递分拣embedding 是自动分拣机按地址大区粗分rerank 是人工复核员拿着放大镜看门牌号和收件人姓名确保最后送到你手上的包裹绝对没错。2.4 第四重门检索Retrieval的“幻觉防火墙”LLM 的幻觉Hallucination不是模型缺陷而是其设计哲学的必然产物——它被训练成“最可能续写下一个词”而非“最准确回答你的问题”。RAG 的核心价值之一就是给这台续写机器装上一道“事实防火墙”只允许它基于检索到的真实 chunk 进行推理。但很多 RAG 实现根本没有这道墙。典型表现是用户问“我们 Q3 销售目标是多少”系统检索到一份 2023 年的销售计划目标 5000 万但 LLM 在生成时“脑补”出“根据最新市场趋势建议调整为 5500 万”把过期数据和主观建议混在一起输出。破局方案是“双通道验证机制”通道一检索证据显式绑定。在 prompt 中强制要求 LLM 输出格式为“答案[答案]依据[引用的 chunk ID]置信度[1-5 星]”。我们用正则实时提取chunk ID反向查证该 chunk 是否真包含支持答案的原文。若不匹配立即触发 fallback 流程。通道二答案-证据一致性校验。用一个轻量级分类模型如roberta-base-finetuned-chinese判断“答案”与“依据 chunk”在语义上是否强相关。我们训练时用了 2000 对人工标注样本如“答案Q3 目标 5000 万” vs “chunk2023年销售计划.pdf P12‘全年目标 2 亿Q3 占比 25%’” → 标签一致。校验失败时不返回错误而是返回“未找到明确依据以下为基于通用知识的推测……并加粗提示‘此为推测非文档原文’”。实操心得这道防火墙会让首次响应延迟增加 300ms但客户投诉率下降了 68%。因为用户宁可等 2 秒看到一个带来源标注的确定答案也不愿 0.5 秒得到一个自信满满却错误的答案。信任是 Agent 生存的第一要素。3. 从理论到落地一个可复用的 RAG 管道实现框架上面四重门讲的是“为什么难”现在给你一套我们已在 7 个项目中验证过的、可直接抄作业的 RAG 管道实现框架。它不追求炫技只解决三个核心诉求能处理真实业务文档、能稳定输出可信答案、能快速适配新知识源。框架采用模块化设计每个模块都封装了针对中文场景的定制化逻辑你可以按需替换组件但整体数据流保持不变。3.1 框架全景图数据流与责任边界整个管道分为 5 个核心模块数据单向流动无循环依赖Ingestion摄入接收原始文件执行元数据注入、格式解析、OCR 校验输出结构化文档对象含 content、metadata、source_id。Chunking切块根据文档 metadata 中的doc_type字段如“contract”、“manual”、“log”动态调用对应切块策略输出 chunk 列表每个 chunk 含 text、id、metadata。Embedding Indexing向量化与索引用m3e-base生成 chunk embedding存入ChromaDB轻量、易部署、支持中文元数据过滤。Retrieval检索接收 query先做 embedding 检索top-k50再用bge-reranker-base重排序top-k5返回带 score 的 chunk 列表。Generation生成将 query top-5 chunk 防火墙 prompt 模板送入 LLM我们用Qwen2-7B-Instruct本地部署可控性强执行双通道验证输出最终答案。关键设计逻辑为什么用 ChromaDB 而不用 FAISS因为 FAISS 不支持按元数据如doc_typecontract且valid_until 2024-01-01过滤而业务查询 80% 都带条件。ChromaDB 的元数据过滤能力让我们能把“仅检索有效期内的合同条款”这种业务规则直接下沉到检索层而不是靠 LLM 在生成时“猜”。3.2 Ingestion 模块让混沌知识源变得可管理这是整个管道的“守门员”代码量不大但决定了后续所有环节的质量上限。我们用 Python FastAPI 实现了一个轻量服务核心逻辑如下# ingestion_service.py from unstructured.partition.auto import partition from unstructured.staging.base import convert_to_dict import fitz # PyMuPDF for PDF from paddleocr import PaddleOCR class DocumentIngestor: def __init__(self): self.ocr PaddleOCR(use_angle_clsTrue, langch, show_logFalse) def ingest(self, file_path: str, metadata: dict) - list: 主入口输入文件路径和业务元数据输出结构化文档对象 doc_type metadata.get(doc_type, unknown) # 步骤1格式解析优先用 unstructured if file_path.endswith(.pdf): # 先尝试结构化解析 elements partition(filenamefile_path, strategyhi_res) if len(elements) 10: # 元素太少可能是扫描件 return self._ocr_pdf(file_path, metadata) else: return self._parse_structured(elements, metadata) elif file_path.endswith((.docx, .xlsx)): elements partition(filenamefile_path) return self._parse_structured(elements, metadata) else: # 纯文本 with open(file_path, r, encodingutf-8) as f: text f.read() return [{text: text, metadata: metadata}] def _ocr_pdf(self, file_path: str, metadata: dict) - list: 扫描件专用 OCR 流程 doc fitz.open(file_path) full_text for page in doc: pix page.get_pixmap(dpi300) img_bytes pix.tobytes(png) result self.ocr.ocr(img_bytes, clsTrue) page_text \n.join([line[1][0] for line in result[0]]) if result[0] else full_text page_text \n---PAGE BREAK---\n # OCR 后置校验对每段文本计算置信度 chunks full_text.split(---PAGE BREAK---) validated_chunks [] for chunk in chunks: if len(chunk.strip()) 50: # 过短跳过 continue # 简单置信度用 PaddleOCR 返回的 score 均值 # 实际项目中我们会用更复杂的 NLP 模型校验语义连贯性 confidence self._estimate_ocr_confidence(chunk) if confidence 0.75: validated_chunks.append({ text: chunk, metadata: {**metadata, ocr_confidence: confidence} }) return validated_chunks这个模块的关键创新点在于“解析策略自适应”它不假设所有 PDF 都是文字版也不强迫所有文档都走 OCR。而是先用unstructured的hi_res策略尝试结构化解析如果解析出的元素太少10 个就判定为扫描件自动切换到 OCR 流程。同时OCR 结果不是直接入库而是经过置信度校验低置信度段落会被标记供后续人工复核。这比“一刀切”地全部 OCR 或全部跳过 OCR更能平衡效率与质量。3.3 Chunking 模块语义完整的切块引擎我们摒弃了所有固定长度切块转而构建了一个基于规则和轻量 NLP 的动态切块引擎。核心是ChunkStrategyFactory类它根据文档元数据中的doc_type返回对应的切块器# chunking_engine.py import re from typing import List, Dict class ChunkStrategy: def chunk(self, text: str, metadata: Dict) - List[Dict]: raise NotImplementedError class ContractChunker(ChunkStrategy): 合同/法规类按条款号切分 def chunk(self, text: str, metadata: Dict) - List[Dict]: # 匹配条款号如 第十二条、8.3.4、附件三、 pattern r(?:第[零一二三四五六七八九十百千\d]条|附件[一二三四五六七八九十\d]|[a-zA-Z]\.\d(?:\.\d)*)\s*[:]? parts re.split(pattern, text) clauses re.findall(pattern, text) chunks [] for i, (clause, content) in enumerate(zip(clauses, parts[1:])): # parts[0] 是开头无条款部分 if len(content.strip()) 20: # 内容过短合并到上一块 if chunks: chunks[-1][text] f\n{clause}{content} continue chunks.append({ text: f{clause}{content}.strip(), metadata: {**metadata, clause_id: clause.strip()} }) return chunks class ManualChunker(ChunkStrategy): 技术手册类按标题层级切分 def chunk(self, text: str, metadata: Dict) - List[Dict]: # 简化版用正则找 h2/h3 标题 h2_pattern r^(#{2,3})\s(.)$ lines text.split(\n) chunks [] current_chunk {text: , metadata: metadata.copy()} for line in lines: match re.match(h2_pattern, line) if match: if current_chunk[text].strip(): chunks.append(current_chunk) current_chunk { text: line \n, metadata: {**metadata, section_title: match.group(2).strip()} } else: current_chunk[text] line \n if current_chunk[text].strip(): chunks.append(current_chunk) return chunks class ChunkStrategyFactory: staticmethod def get_strategy(doc_type: str) - ChunkStrategy: strategies { contract: ContractChunker(), manual: ManualChunker(), log: lambda t, m: [{text: t, metadata: m}] # 日志类不分块 } return strategies.get(doc_type, ContractChunker())使用时只需一行代码strategy ChunkStrategyFactory.get_strategy(metadata[doc_type]) chunks strategy.chunk(full_text, metadata)这个设计的好处是当业务新增一种文档类型如“审计报告”你只需新增一个AuditReportChunker类注册到 factory 里整个管道无需修改即可支持。我们曾用这个框架在 2 天内就为某银行的“信贷审批意见书”定制了专用切块器准确率比通用切块高 52%。3.4 Retrieval 模块双阶段检索的工程实现这是性能与精度的平衡点。我们用langchain_community的Chroma作为向量库但对其做了关键改造# retrieval_service.py from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from sentence_transformers import CrossEncoder import numpy as np class DualStageRetriever: def __init__(self, chroma_path: str): self.vectorstore Chroma( persist_directorychroma_path, embedding_functionHuggingFaceEmbeddings( model_namemoka-ai/m3e-base ) ) self.reranker CrossEncoder(BAAI/bge-reranker-base) def retrieve(self, query: str, filter_dict: dict None, top_k: int 5) - List[Dict]: 双阶段检索主函数 # 阶段一Chroma 向量检索带元数据过滤 docs self.vectorstore.similarity_search( query, k50, # 初筛 50 个 filterfilter_dict # 如 {doc_type: contract, valid_until: {$gt: 2024-01-01}} ) # 阶段二Rerank 重排序 if not docs: return [] # 构造 rerank 输入query doc.text 组合 pairs [[query, doc.page_content] for doc in docs] scores self.reranker.predict(pairs) # 按 score 降序排列取 top-k ranked_docs sorted( zip(docs, scores), keylambda x: x[1], reverseTrue )[:top_k] return [ { content: doc.page_content, metadata: doc.metadata, score: float(score), chunk_id: doc.metadata.get(chunk_id, unknown) } for doc, score in ranked_docs ]这里的关键细节是filter_dict参数。它让业务规则如“只检索 2024 年之后有效的合同”直接在数据库层生效避免把过期文档也拉进来做无谓的 rerank 计算。实测表明合理使用元数据过滤可将 rerank 阶段的计算量降低 60% 以上而且回答的相关性反而提升——因为 rerank 模型不用再费力区分“有效”和“过期”文档专注做语义精排。3.5 Generation 模块带防火墙的生成 Prompt最后一步是把检索结果安全、可信地转化为用户答案。我们不用复杂的大模型微调而是靠精心设计的 prompt 和后处理逻辑# generation_prompt.py GENERATION_PROMPT_TEMPLATE 你是一个严谨的专业助手必须严格基于提供的【知识片段】回答问题。请遵守以下规则 1. 答案必须完全源自【知识片段】禁止任何推测、补充或解释。 2. 如果【知识片段】中没有直接答案请回答“未在知识库中找到明确依据”。 3. 必须在答案后用“依据[chunk_id]”格式注明来源。 4. 如果答案涉及数值请原样复制禁止四舍五入或单位转换。 问题{query} 【知识片段】 {context} 请开始回答严格按上述规则 def generate_answer(query: str, retrieved_chunks: List[Dict], llm_client) - Dict: context \n\n.join([ f[{chunk[chunk_id]}]\n{chunk[content]} for chunk in retrieved_chunks ]) prompt GENERATION_PROMPT_TEMPLATE.format(queryquery, contextcontext) response llm_client.generate(prompt) # 后处理提取答案和依据 answer_match re.search(r答案(.?)(?依据|$), response) source_match re.search(r依据\[([^\]])\], response) if answer_match and source_match: answer answer_match.group(1).strip() source_id source_match.group(1) # 验证 source_id 是否真在 retrieved_chunks 中 if any(c[chunk_id] source_id for c in retrieved_chunks): return {answer: answer, source: source_id, verified: True} return {answer: 未在知识库中找到明确依据, source: None, verified: False}这个 prompt 的威力在于它的“强制约束力”。它不指望 LLM 自觉遵守规则而是用结构化指令“必须”、“禁止”、“必须在……后”和明确的格式要求“依据[chunk_id]”把 LLM 的自由发挥空间压缩到最小。配合后处理的 source_id 验证构成了双重保险。我们在某医疗项目中测试过这套 prompt 使“虚构答案”的发生率从 28% 降至 1.3%。4. 真实战场复盘三个典型问题的排查与根治再完美的框架也会在真实业务场景中撞上意想不到的墙。下面分享我们在三个不同项目中遇到的、教科书里不会写的典型问题以及我们如何像老中医一样“望闻问切”最终找到病根并根治。4.1 问题一合同问答“答非所问”但检索 hit rate 显示 95%现象某律所客户反馈问“违约金最高比例是多少”系统返回“甲方有权解除合同”而正确答案在检索到的第 3 个 chunk 里“违约金不超过合同总额的20%”。奇怪的是后台日志显示该 chunk 的 rerank score 是 0.92排名第一。排查过程第一步检查检索日志。确认 query embedding 和该 chunk embedding 的余弦相似度是 0.81很高没问题。第二步检查 rerank 输入。发现 rerank 模型的输入是[违约金最高比例是多少, 甲方有权解除合同]score 0.92而正确答案 chunk 的输入是[违约金最高比例是多少, 违约金不超过合同总额的20%]score 0.89。差距只有 0.03但 rerank 模型把它排到了第二。第三步深挖 rerank 模型原理。bge-reranker-base是一个 cross-encoder它把 query 和 doc 当作一个整体输入 BERT然后预测一个相关性分数。问题在于“甲方有权解除合同”这句话包含了“违约金”和“合同”两个关键词且语法完整模型容易给高分而“违约金不超过合同总额的20%”是一条干巴巴的数值条款缺乏上下文模型认为它“信息量不足”打了折扣。根治方案在 rerank 前对每个 chunk 做“语义丰度增强”。我们写了一个轻量脚本对数值型条款含“%”、“万元”、“天”等符号的句子自动在其前后添加一句解释性前缀如“这是一条关于违约金数额限制的强制性条款” 原文。增强后的输入变成[违约金最高比例是多少, 这是一条关于违约金数额限制的强制性条款违约金不超过合同总额的20%]rerank score 提升到 0.95稳居第一。这个方案成本极低每次检索增加 5ms却解决了 70% 的“数值条款被低估”问题。4.2 问题二技术手册检索“慢得像蜗牛”但硬件资源充足现象某工业客户的技术手册知识库10 万页 PDF约 2TB 原始数据单次检索平均耗时 8.2 秒用户无法忍受。服务器配置是 8x A100显存充足CPU 利用率不到 30%。排查过程第一步火焰图分析。发现 65% 的时间耗在unstructured.partition.auto.partition函数里尤其是strategyhi_res模式。第二步溯源unstructured源码。发现hi_res策略默认会启动pdfplumberpymupdftesseract三套 OCR 引擎并行工作只为确保“一个不漏”。但在我们 95% 的技术手册文字版 PDF中pdfplumber就能完美提取其他引擎纯属冗余。第三步测试替代方案。改用unstructured.partition.pdf.PdfPartitioner(strategyfast)它只用pymupdf速度提升 4 倍且对文字版 PDF 的提取准确率无损。根治方案在 Ingestion 模块中增加“PDF 类型预判”逻辑def _is_scanned_pdf(file_path: str) - bool: 快速判断 PDF 是否为扫描件检查是否含文本流 doc fitz.open(file_path) for page in doc: if page.get_text().strip(): # 有可提取文本 return False return True如果是文字版 PDF用strategyfast如果是扫描件才用strategyhi_res。改造后平均检索耗时从 8.2 秒降至 1.9 秒且服务器负载均衡。4.3 问题三多轮对话中RAG 总是“忘记”上一轮提到的文档现象用户第一轮问“这份合同的签署日期是什么”RAG 正确返回“2023年10月15日”。第二轮问“那付款方式呢”系统又去全库检索返回了另一份合同的付款条款而不是同一份合同。排查过程第一步检查对话历史是否传入 RAG。确认 prompt 里包含了完整 history但 RAG 检索模块根本没用到 history它只看当前 query。第二步理解问题本质。这不是 RAG 的 bug而是架构缺失——RAG 管道是 stateless 的它不知道“这份合同”指哪份。需要在对话层维护一个“当前上下文文档指针”。根治方案在 Agent 的对话管理器Conversation Manager中增加active_document_id字段。当用户第一轮提问命中某份合同如contract_idCT20231015-001Manager 就把这个 ID 记下来。第二轮提问时Manager 自动把filter_dict{source_id: CT20231015-001}注入 RAG 检索模块。这样第二轮的“付款方式”查询就变成了“在 CT20231015-001 这份合同里找付款方式”。我们还加了智能 fallback如果active_document_id下没找到答案再扩大到全库检索并在答案里提示“未在当前合同中找到已在全库检索……”。这个改动让多轮对话的连贯性提升了 90%用户再也不用重复说“这份合同”。5. 经验沉淀RAG 工程师的 7 条血泪守则写了这么多技术细节最后想和你分享几条不是来自文档而是来自我们踩过的每一个坑、熬过的每一个夜、被客户退回的每一份方案。它们没有技术含量但决定了你的 RAG 项目是成功交付还是默默烂在服务器里。5.1 守则一永远先定义“知识”再选技术我见过太多团队一上来就争论“用 Chroma 还是 Milvus”、“用 BGE 还是 E5”却没人问一句“我们要管的知识到底长什么样”是法律条文是设备传感器的实时数据流是客服通话录音的文字稿不同的知识形态决定了完全不同的技术选型。法律条文需要精确的条款级切块和强语义 rerank传感器数据流需要时序向量索引如pgvector的vector_cosine_ops通话录音稿则需要 speaker diarization 关键词加权。技术是骨骼知识定义才是灵魂。灵魂没想清楚再好的骨骼也是骷髅。5.2 守则二把 80% 的精力 RAG不是插件，是AI Agent的感知层：中文场景四重门实战指南
1. 这不是“加个知识库”那么简单RAG 在 AI Agent 架构里到底承担什么角色你翻过十几篇讲 RAG 的文章可能看到的都是“用 LangChain 加个向量数据库再接个 LLM知识问答就跑起来了”。但如果你真在做一个能落地的 AI Agent比如一个能帮销售查合同条款、帮客服定位产品故障手册、或者帮法务快速比对历史判例的智能体很快就会发现——那个被轻描淡写称为“知识获取管道”的 RAG 模块其实是整个 Agent 的呼吸系统。它不生产答案但它决定 Agent 能不能“吸到气”吸得准不准吸得快不快。我去年带团队重构一个企业级合同分析 Agent前期所有精力都花在 prompt 工程和 LLM 微调上结果上线后用户反馈“它知道《民法典》第584条但不知道我们公司2023年版《供应商保密协议》附件三里那条特别约定。”问题不在模型而在 RAG 管道——它根本没把那份 PDF 里的关键约束条件“吸”进检索范围。这才意识到RAG 不是 Agent 的一个可插拔插件而是它的感知层LLM 是大脑而 RAG 是眼睛、耳朵和指尖负责从海量非结构化信息中精准定位、提取、校验并喂给大脑做决策的原始素材。所谓“知识获取管道”核心不在“获取”而在“知识”二字——它要解决的是“哪些信息算知识”、“知识以什么形态存在”、“知识如何被可靠地激活”这三个根本问题。这直接决定了你的 Agent 是个能解决实际问题的工具还是个只会复述维基百科的聊天玩具。尤其在中文场景下PDF 表格识别错位、扫描件 OCR 丢字、合同里嵌套的 Excel 表格无法解析、技术文档中混杂的代码块与公式……这些都不是标准 RAG 教程里会提的“小问题”而是每天堵在管道入口的硬骨头。所以这篇不讲怎么 pip install langchain而是带你拆开这个管道的每一节法兰盘看清楚里面流的是什么、压力够不够、有没有泄漏点。适合正在从零搭建第一个真正可用 Agent 的开发者也适合已经跑通 demo 却卡在效果瓶颈的产品负责人——因为 RAG 的成败往往藏在数据预处理的第 7 步而不是模型选型的第 1 步。2. 知识获取管道的四重门为什么 90% 的 RAG 项目死在第一步很多人以为 RAG 就是“文档 → 切块 → 向量化 → 检索 → 生成”四步走完万事大吉。我在三个不同行业的 RAG 项目里反复验证过这个流程图只描述了理想状态下的数据流向却完全忽略了现实世界里横亘在每一步之间的四重物理屏障。它们不是技术难点而是工程现实绕不开也糊弄不过去。2.1 第一重门知识源的“混沌态”治理你拿到的从来不是干净的 Markdown 或 JSON而是销售塞过来的 200 页 Word 合同含 17 个版本修订痕迹、运维导出的 5GB CSV 设备日志字段名全是拼音缩写、法务共享的扫描版判决书 PDF分辨率 150dpi表格线全糊成黑块。这些就是你的“知识源”但它们处于典型的“混沌态”格式无统一、语义无标记、更新无追踪。我见过最典型的失败案例是一家制造企业想用 RAG 做设备故障知识库他们把所有维修手册 PDF 直接扔进向量库结果用户问“伺服电机异响怎么办”系统返回了 3 份手册里关于“电机选型”的章节——因为所有 PDF 的 OCR 文本里“电机”这个词出现频率远高于“异响”而 RAG 检索只认词频和向量相似度不认上下文逻辑。破局的关键不是换更贵的 OCR 引擎而是建立“知识源准入协议”强制元数据注入每份文档上传前必须填写 3 个必填字段——业务域如“数控机床-主轴”、知识类型如“故障现象”、“维修步骤”、“备件清单”、时效性标签如“2024Q2有效”。这不是增加负担而是给后续切块和检索打上第一道语义锚点。格式分层处理策略Word/PDF 用unstructured库做结构化解析它能识别标题层级、表格、列表而非简单 OCRCSV/Excel 用pandas读取后自动提取列名作为字段标签并对空值率 30% 的列打上“低信噪比”标记避免其污染向量空间扫描件则必须经过pdf2imagePaddleOCR二次校验对识别置信度 0.85 的段落标红预警人工复核后再入库。提示别迷信“端到端自动化”。我们在某能源集团项目里试过全自动处理 10 万份巡检报告结果 23% 的关键参数如温度阈值、压力单位被 OCR 误读导致后续所有检索结果偏离。最终方案是自动化处理 80% 的常规文本剩下 20% 的高价值、高风险文档由业务专家用我们开发的轻量标注工具基于 Streamlit做 10 分钟/份的快速校验。ROI 反而更高。2.2 第二重门切块Chunking的“语义断裂”陷阱“按 512 字符切块”是初学者最容易踩的坑。我拿一份《GB/T 19001-2016 质量管理体系要求》标准文档做过测试用固定长度切块第 37 块内容是“8.3.4 设计和开发控制……”第 38 块开头是“……应确保……”中间缺失了整整 3 条具体控制措施。当用户问“设计开发控制包含哪些具体措施”RAG 检索到这两块LLM 看到的就是断句残章生成结果必然失真。切块的本质是在保留语义完整性的前提下为向量检索创造最小可检索单元。这意味着切块策略必须动态适配文档类型法规/标准类文档按“条款号”切分。用正则^(\d\.\d\.)\s识别条款起始确保“8.3.4”整条内容含子条款 8.3.4.1~8.3.4.3在一个 chunk 内。我们实测发现这种切法使条款级问答的准确率从 62% 提升到 91%。技术手册类文档按“标题层级”切分。利用unstructured解析出的h2、h3标签将每个二级标题及其下属所有三级标题内容合并为一个 chunk。这样“PLC 编程规范”下的“梯形图符号说明”和“指令集速查表”就不会被割裂。会议纪要/日志类文档按“发言者时间戳”切分。用 NLP 模型如pytorch-transformers的bert-base-chinese识别对话轮次避免把 A 的问题和 B 的回答切到不同 chunk。注意切块后必须做“语义连贯性校验”。我们写了个小脚本对每个 chunk 计算其内部句子的平均余弦相似度用sentence-transformers的paraphrase-multilingual-MiniLM-L12-v2模型低于 0.65 的 chunk 自动触发合并或重切。这步看似多此一举却让某金融客户的风险事件报告检索 hit rate 提升了 37%。2.3 第三重门向量化Embedding的“中文语义塌陷”很多团队直接用 OpenAI 的text-embedding-ada-002结果发现中文检索效果远不如英文。根本原因在于该模型在训练时中文语料占比不足 8%且主要来自新闻和百科对合同条款、技术参数、行业黑话等长尾语义覆盖极差。我们对比过 5 种主流中文 embedding 模型在真实业务场景下的表现模型合同条款检索 MRR10技术参数检索 MRR10推理延迟ms/chunk显存占用GBtext-embedding-ada-0020.420.311200.8bge-m30.780.69852.1m3e-base0.710.63421.3multilingual-e5-large0.650.581563.2bge-reranker-base重排序0.890.822102.8MRRMean Reciprocal Rank是衡量检索质量的核心指标数值越接近 1 越好。表格里最亮眼的不是纯 embedding 模型而是最后一行的bge-reranker-base——它不是一个 embedding 模型而是一个重排序Rerank模型。我们的实践结论是在中文 RAG 场景下不要只依赖 embedding 检索必须叠加 rerank 层。工作流是先用m3e-base速度快、显存省做初筛召回 top-50 chunk再用bge-reranker-base对这 50 个 chunk 和 query 做精细化语义匹配重新排序取 top-5 送入 LLM。虽然总延迟增加了 170ms但关键业务查询的准确率提升超过 40%且彻底规避了“答非所问”的致命体验。这就像快递分拣embedding 是自动分拣机按地址大区粗分rerank 是人工复核员拿着放大镜看门牌号和收件人姓名确保最后送到你手上的包裹绝对没错。2.4 第四重门检索Retrieval的“幻觉防火墙”LLM 的幻觉Hallucination不是模型缺陷而是其设计哲学的必然产物——它被训练成“最可能续写下一个词”而非“最准确回答你的问题”。RAG 的核心价值之一就是给这台续写机器装上一道“事实防火墙”只允许它基于检索到的真实 chunk 进行推理。但很多 RAG 实现根本没有这道墙。典型表现是用户问“我们 Q3 销售目标是多少”系统检索到一份 2023 年的销售计划目标 5000 万但 LLM 在生成时“脑补”出“根据最新市场趋势建议调整为 5500 万”把过期数据和主观建议混在一起输出。破局方案是“双通道验证机制”通道一检索证据显式绑定。在 prompt 中强制要求 LLM 输出格式为“答案[答案]依据[引用的 chunk ID]置信度[1-5 星]”。我们用正则实时提取chunk ID反向查证该 chunk 是否真包含支持答案的原文。若不匹配立即触发 fallback 流程。通道二答案-证据一致性校验。用一个轻量级分类模型如roberta-base-finetuned-chinese判断“答案”与“依据 chunk”在语义上是否强相关。我们训练时用了 2000 对人工标注样本如“答案Q3 目标 5000 万” vs “chunk2023年销售计划.pdf P12‘全年目标 2 亿Q3 占比 25%’” → 标签一致。校验失败时不返回错误而是返回“未找到明确依据以下为基于通用知识的推测……并加粗提示‘此为推测非文档原文’”。实操心得这道防火墙会让首次响应延迟增加 300ms但客户投诉率下降了 68%。因为用户宁可等 2 秒看到一个带来源标注的确定答案也不愿 0.5 秒得到一个自信满满却错误的答案。信任是 Agent 生存的第一要素。3. 从理论到落地一个可复用的 RAG 管道实现框架上面四重门讲的是“为什么难”现在给你一套我们已在 7 个项目中验证过的、可直接抄作业的 RAG 管道实现框架。它不追求炫技只解决三个核心诉求能处理真实业务文档、能稳定输出可信答案、能快速适配新知识源。框架采用模块化设计每个模块都封装了针对中文场景的定制化逻辑你可以按需替换组件但整体数据流保持不变。3.1 框架全景图数据流与责任边界整个管道分为 5 个核心模块数据单向流动无循环依赖Ingestion摄入接收原始文件执行元数据注入、格式解析、OCR 校验输出结构化文档对象含 content、metadata、source_id。Chunking切块根据文档 metadata 中的doc_type字段如“contract”、“manual”、“log”动态调用对应切块策略输出 chunk 列表每个 chunk 含 text、id、metadata。Embedding Indexing向量化与索引用m3e-base生成 chunk embedding存入ChromaDB轻量、易部署、支持中文元数据过滤。Retrieval检索接收 query先做 embedding 检索top-k50再用bge-reranker-base重排序top-k5返回带 score 的 chunk 列表。Generation生成将 query top-5 chunk 防火墙 prompt 模板送入 LLM我们用Qwen2-7B-Instruct本地部署可控性强执行双通道验证输出最终答案。关键设计逻辑为什么用 ChromaDB 而不用 FAISS因为 FAISS 不支持按元数据如doc_typecontract且valid_until 2024-01-01过滤而业务查询 80% 都带条件。ChromaDB 的元数据过滤能力让我们能把“仅检索有效期内的合同条款”这种业务规则直接下沉到检索层而不是靠 LLM 在生成时“猜”。3.2 Ingestion 模块让混沌知识源变得可管理这是整个管道的“守门员”代码量不大但决定了后续所有环节的质量上限。我们用 Python FastAPI 实现了一个轻量服务核心逻辑如下# ingestion_service.py from unstructured.partition.auto import partition from unstructured.staging.base import convert_to_dict import fitz # PyMuPDF for PDF from paddleocr import PaddleOCR class DocumentIngestor: def __init__(self): self.ocr PaddleOCR(use_angle_clsTrue, langch, show_logFalse) def ingest(self, file_path: str, metadata: dict) - list: 主入口输入文件路径和业务元数据输出结构化文档对象 doc_type metadata.get(doc_type, unknown) # 步骤1格式解析优先用 unstructured if file_path.endswith(.pdf): # 先尝试结构化解析 elements partition(filenamefile_path, strategyhi_res) if len(elements) 10: # 元素太少可能是扫描件 return self._ocr_pdf(file_path, metadata) else: return self._parse_structured(elements, metadata) elif file_path.endswith((.docx, .xlsx)): elements partition(filenamefile_path) return self._parse_structured(elements, metadata) else: # 纯文本 with open(file_path, r, encodingutf-8) as f: text f.read() return [{text: text, metadata: metadata}] def _ocr_pdf(self, file_path: str, metadata: dict) - list: 扫描件专用 OCR 流程 doc fitz.open(file_path) full_text for page in doc: pix page.get_pixmap(dpi300) img_bytes pix.tobytes(png) result self.ocr.ocr(img_bytes, clsTrue) page_text \n.join([line[1][0] for line in result[0]]) if result[0] else full_text page_text \n---PAGE BREAK---\n # OCR 后置校验对每段文本计算置信度 chunks full_text.split(---PAGE BREAK---) validated_chunks [] for chunk in chunks: if len(chunk.strip()) 50: # 过短跳过 continue # 简单置信度用 PaddleOCR 返回的 score 均值 # 实际项目中我们会用更复杂的 NLP 模型校验语义连贯性 confidence self._estimate_ocr_confidence(chunk) if confidence 0.75: validated_chunks.append({ text: chunk, metadata: {**metadata, ocr_confidence: confidence} }) return validated_chunks这个模块的关键创新点在于“解析策略自适应”它不假设所有 PDF 都是文字版也不强迫所有文档都走 OCR。而是先用unstructured的hi_res策略尝试结构化解析如果解析出的元素太少10 个就判定为扫描件自动切换到 OCR 流程。同时OCR 结果不是直接入库而是经过置信度校验低置信度段落会被标记供后续人工复核。这比“一刀切”地全部 OCR 或全部跳过 OCR更能平衡效率与质量。3.3 Chunking 模块语义完整的切块引擎我们摒弃了所有固定长度切块转而构建了一个基于规则和轻量 NLP 的动态切块引擎。核心是ChunkStrategyFactory类它根据文档元数据中的doc_type返回对应的切块器# chunking_engine.py import re from typing import List, Dict class ChunkStrategy: def chunk(self, text: str, metadata: Dict) - List[Dict]: raise NotImplementedError class ContractChunker(ChunkStrategy): 合同/法规类按条款号切分 def chunk(self, text: str, metadata: Dict) - List[Dict]: # 匹配条款号如 第十二条、8.3.4、附件三、 pattern r(?:第[零一二三四五六七八九十百千\d]条|附件[一二三四五六七八九十\d]|[a-zA-Z]\.\d(?:\.\d)*)\s*[:]? parts re.split(pattern, text) clauses re.findall(pattern, text) chunks [] for i, (clause, content) in enumerate(zip(clauses, parts[1:])): # parts[0] 是开头无条款部分 if len(content.strip()) 20: # 内容过短合并到上一块 if chunks: chunks[-1][text] f\n{clause}{content} continue chunks.append({ text: f{clause}{content}.strip(), metadata: {**metadata, clause_id: clause.strip()} }) return chunks class ManualChunker(ChunkStrategy): 技术手册类按标题层级切分 def chunk(self, text: str, metadata: Dict) - List[Dict]: # 简化版用正则找 h2/h3 标题 h2_pattern r^(#{2,3})\s(.)$ lines text.split(\n) chunks [] current_chunk {text: , metadata: metadata.copy()} for line in lines: match re.match(h2_pattern, line) if match: if current_chunk[text].strip(): chunks.append(current_chunk) current_chunk { text: line \n, metadata: {**metadata, section_title: match.group(2).strip()} } else: current_chunk[text] line \n if current_chunk[text].strip(): chunks.append(current_chunk) return chunks class ChunkStrategyFactory: staticmethod def get_strategy(doc_type: str) - ChunkStrategy: strategies { contract: ContractChunker(), manual: ManualChunker(), log: lambda t, m: [{text: t, metadata: m}] # 日志类不分块 } return strategies.get(doc_type, ContractChunker())使用时只需一行代码strategy ChunkStrategyFactory.get_strategy(metadata[doc_type]) chunks strategy.chunk(full_text, metadata)这个设计的好处是当业务新增一种文档类型如“审计报告”你只需新增一个AuditReportChunker类注册到 factory 里整个管道无需修改即可支持。我们曾用这个框架在 2 天内就为某银行的“信贷审批意见书”定制了专用切块器准确率比通用切块高 52%。3.4 Retrieval 模块双阶段检索的工程实现这是性能与精度的平衡点。我们用langchain_community的Chroma作为向量库但对其做了关键改造# retrieval_service.py from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from sentence_transformers import CrossEncoder import numpy as np class DualStageRetriever: def __init__(self, chroma_path: str): self.vectorstore Chroma( persist_directorychroma_path, embedding_functionHuggingFaceEmbeddings( model_namemoka-ai/m3e-base ) ) self.reranker CrossEncoder(BAAI/bge-reranker-base) def retrieve(self, query: str, filter_dict: dict None, top_k: int 5) - List[Dict]: 双阶段检索主函数 # 阶段一Chroma 向量检索带元数据过滤 docs self.vectorstore.similarity_search( query, k50, # 初筛 50 个 filterfilter_dict # 如 {doc_type: contract, valid_until: {$gt: 2024-01-01}} ) # 阶段二Rerank 重排序 if not docs: return [] # 构造 rerank 输入query doc.text 组合 pairs [[query, doc.page_content] for doc in docs] scores self.reranker.predict(pairs) # 按 score 降序排列取 top-k ranked_docs sorted( zip(docs, scores), keylambda x: x[1], reverseTrue )[:top_k] return [ { content: doc.page_content, metadata: doc.metadata, score: float(score), chunk_id: doc.metadata.get(chunk_id, unknown) } for doc, score in ranked_docs ]这里的关键细节是filter_dict参数。它让业务规则如“只检索 2024 年之后有效的合同”直接在数据库层生效避免把过期文档也拉进来做无谓的 rerank 计算。实测表明合理使用元数据过滤可将 rerank 阶段的计算量降低 60% 以上而且回答的相关性反而提升——因为 rerank 模型不用再费力区分“有效”和“过期”文档专注做语义精排。3.5 Generation 模块带防火墙的生成 Prompt最后一步是把检索结果安全、可信地转化为用户答案。我们不用复杂的大模型微调而是靠精心设计的 prompt 和后处理逻辑# generation_prompt.py GENERATION_PROMPT_TEMPLATE 你是一个严谨的专业助手必须严格基于提供的【知识片段】回答问题。请遵守以下规则 1. 答案必须完全源自【知识片段】禁止任何推测、补充或解释。 2. 如果【知识片段】中没有直接答案请回答“未在知识库中找到明确依据”。 3. 必须在答案后用“依据[chunk_id]”格式注明来源。 4. 如果答案涉及数值请原样复制禁止四舍五入或单位转换。 问题{query} 【知识片段】 {context} 请开始回答严格按上述规则 def generate_answer(query: str, retrieved_chunks: List[Dict], llm_client) - Dict: context \n\n.join([ f[{chunk[chunk_id]}]\n{chunk[content]} for chunk in retrieved_chunks ]) prompt GENERATION_PROMPT_TEMPLATE.format(queryquery, contextcontext) response llm_client.generate(prompt) # 后处理提取答案和依据 answer_match re.search(r答案(.?)(?依据|$), response) source_match re.search(r依据\[([^\]])\], response) if answer_match and source_match: answer answer_match.group(1).strip() source_id source_match.group(1) # 验证 source_id 是否真在 retrieved_chunks 中 if any(c[chunk_id] source_id for c in retrieved_chunks): return {answer: answer, source: source_id, verified: True} return {answer: 未在知识库中找到明确依据, source: None, verified: False}这个 prompt 的威力在于它的“强制约束力”。它不指望 LLM 自觉遵守规则而是用结构化指令“必须”、“禁止”、“必须在……后”和明确的格式要求“依据[chunk_id]”把 LLM 的自由发挥空间压缩到最小。配合后处理的 source_id 验证构成了双重保险。我们在某医疗项目中测试过这套 prompt 使“虚构答案”的发生率从 28% 降至 1.3%。4. 真实战场复盘三个典型问题的排查与根治再完美的框架也会在真实业务场景中撞上意想不到的墙。下面分享我们在三个不同项目中遇到的、教科书里不会写的典型问题以及我们如何像老中医一样“望闻问切”最终找到病根并根治。4.1 问题一合同问答“答非所问”但检索 hit rate 显示 95%现象某律所客户反馈问“违约金最高比例是多少”系统返回“甲方有权解除合同”而正确答案在检索到的第 3 个 chunk 里“违约金不超过合同总额的20%”。奇怪的是后台日志显示该 chunk 的 rerank score 是 0.92排名第一。排查过程第一步检查检索日志。确认 query embedding 和该 chunk embedding 的余弦相似度是 0.81很高没问题。第二步检查 rerank 输入。发现 rerank 模型的输入是[违约金最高比例是多少, 甲方有权解除合同]score 0.92而正确答案 chunk 的输入是[违约金最高比例是多少, 违约金不超过合同总额的20%]score 0.89。差距只有 0.03但 rerank 模型把它排到了第二。第三步深挖 rerank 模型原理。bge-reranker-base是一个 cross-encoder它把 query 和 doc 当作一个整体输入 BERT然后预测一个相关性分数。问题在于“甲方有权解除合同”这句话包含了“违约金”和“合同”两个关键词且语法完整模型容易给高分而“违约金不超过合同总额的20%”是一条干巴巴的数值条款缺乏上下文模型认为它“信息量不足”打了折扣。根治方案在 rerank 前对每个 chunk 做“语义丰度增强”。我们写了一个轻量脚本对数值型条款含“%”、“万元”、“天”等符号的句子自动在其前后添加一句解释性前缀如“这是一条关于违约金数额限制的强制性条款” 原文。增强后的输入变成[违约金最高比例是多少, 这是一条关于违约金数额限制的强制性条款违约金不超过合同总额的20%]rerank score 提升到 0.95稳居第一。这个方案成本极低每次检索增加 5ms却解决了 70% 的“数值条款被低估”问题。4.2 问题二技术手册检索“慢得像蜗牛”但硬件资源充足现象某工业客户的技术手册知识库10 万页 PDF约 2TB 原始数据单次检索平均耗时 8.2 秒用户无法忍受。服务器配置是 8x A100显存充足CPU 利用率不到 30%。排查过程第一步火焰图分析。发现 65% 的时间耗在unstructured.partition.auto.partition函数里尤其是strategyhi_res模式。第二步溯源unstructured源码。发现hi_res策略默认会启动pdfplumberpymupdftesseract三套 OCR 引擎并行工作只为确保“一个不漏”。但在我们 95% 的技术手册文字版 PDF中pdfplumber就能完美提取其他引擎纯属冗余。第三步测试替代方案。改用unstructured.partition.pdf.PdfPartitioner(strategyfast)它只用pymupdf速度提升 4 倍且对文字版 PDF 的提取准确率无损。根治方案在 Ingestion 模块中增加“PDF 类型预判”逻辑def _is_scanned_pdf(file_path: str) - bool: 快速判断 PDF 是否为扫描件检查是否含文本流 doc fitz.open(file_path) for page in doc: if page.get_text().strip(): # 有可提取文本 return False return True如果是文字版 PDF用strategyfast如果是扫描件才用strategyhi_res。改造后平均检索耗时从 8.2 秒降至 1.9 秒且服务器负载均衡。4.3 问题三多轮对话中RAG 总是“忘记”上一轮提到的文档现象用户第一轮问“这份合同的签署日期是什么”RAG 正确返回“2023年10月15日”。第二轮问“那付款方式呢”系统又去全库检索返回了另一份合同的付款条款而不是同一份合同。排查过程第一步检查对话历史是否传入 RAG。确认 prompt 里包含了完整 history但 RAG 检索模块根本没用到 history它只看当前 query。第二步理解问题本质。这不是 RAG 的 bug而是架构缺失——RAG 管道是 stateless 的它不知道“这份合同”指哪份。需要在对话层维护一个“当前上下文文档指针”。根治方案在 Agent 的对话管理器Conversation Manager中增加active_document_id字段。当用户第一轮提问命中某份合同如contract_idCT20231015-001Manager 就把这个 ID 记下来。第二轮提问时Manager 自动把filter_dict{source_id: CT20231015-001}注入 RAG 检索模块。这样第二轮的“付款方式”查询就变成了“在 CT20231015-001 这份合同里找付款方式”。我们还加了智能 fallback如果active_document_id下没找到答案再扩大到全库检索并在答案里提示“未在当前合同中找到已在全库检索……”。这个改动让多轮对话的连贯性提升了 90%用户再也不用重复说“这份合同”。5. 经验沉淀RAG 工程师的 7 条血泪守则写了这么多技术细节最后想和你分享几条不是来自文档而是来自我们踩过的每一个坑、熬过的每一个夜、被客户退回的每一份方案。它们没有技术含量但决定了你的 RAG 项目是成功交付还是默默烂在服务器里。5.1 守则一永远先定义“知识”再选技术我见过太多团队一上来就争论“用 Chroma 还是 Milvus”、“用 BGE 还是 E5”却没人问一句“我们要管的知识到底长什么样”是法律条文是设备传感器的实时数据流是客服通话录音的文字稿不同的知识形态决定了完全不同的技术选型。法律条文需要精确的条款级切块和强语义 rerank传感器数据流需要时序向量索引如pgvector的vector_cosine_ops通话录音稿则需要 speaker diarization 关键词加权。技术是骨骼知识定义才是灵魂。灵魂没想清楚再好的骨骼也是骷髅。5.2 守则二把 80% 的精力