去年年中我接手了一个企业知识库问答服务的改造原版本是用 LangChain 的 Chain 拼出来的。单轮问答跑得挺欢一上多轮对话和生产流量就各种露馅上下文追不上、检索结果不带引用、问题稍微绕一点就答非所问。后来我把整个 RAG 知识库流程迁移到了 LangGraph 上把原来一条直线改成了带分支、带状态、带人工审核的图。这篇文章就把改造过程里踩过的坑和最终落地方案完整记录下来围绕 LangChain、LangGraph、RAG 知识库这三个关键词展开适合正在做知识库问答、准备引入 Agent 编排或者想从 Chain 迁移到 Graph 的团队参考。1. 为什么从 LangChain 迁到 LangGraphRAG 改造的真实起点先说结论LangChain 没有过时但场景变了。你做一个 Demo、一个原型、一次性的问答脚本用 Chain 完全够。可一旦进入生产环境尤其是企业知识库这种对可控性、可追溯性、多轮交互都有要求的场景Chain 那种线性结构就开始拖后腿。1.1 老方案的问题链式编排的瓶颈LangChain 早期的核心抽象是 Chain一条链自上而下执行加载文档、切块、向量化、检索、拼 Prompt、调用模型、输出。逻辑简单代码也简单但问题恰恰出在“简单”上。第一个痛点是没法分叉。比如用户问“它支持哪些格式”你要先判断“它”指代什么再决定是直接检索还是先改写查询。在 Chain 里做这种条件路由得写 if-else 在外面包一层或者拼多个 Chain 手动串代码很快就乱成一团。第二个痛点是循环。第一次检索结果不理想想换个关键词重新检索Chain 不支持“回头再来一次”这种回路。第三个痛点是状态。多轮对话里对话历史、中间检索结果、重排序得分这些东西散落在各个变量里链跑完就丢了想持久化、想断点续跑基本靠自研。我用一个生活化的类比Chain 像一条传送带零件从一头进去沿固定路线走到另一头出来。你想在中间加一个质检环节质检不过就退回上一个工位重新加工传送带做不到你得把它拆开重新设计。LangGraph 就是帮你重新设计这套流程的框架。1.2 LangGraph 的核心理念从链到图LangGraph 的核心抽象就三样节点Node、边Edge、状态State。节点就是你的一段处理逻辑比如“查询改写”“检索”“生成答案”都是节点。边决定流程怎么走可以是普通边A 执行完一定走 B也可以是条件边A 执行完根据 State 里的某个字段决定走 B 还是走 C。状态是整个流程共享的数据对象所有节点都能读、能写节点之间通过 state 传递信息。这个设计直接解决了 Chain 的几个死穴条件路由用条件边表达重试用循环边表达多轮对话的历史放在 State 里随流程走Checkpointer 还能把每一步的状态落盘做到“跑到一半停下来人工审批完再继续”。而且 LangGraph 不是把 LangChain 推翻重来。LangChain 里的文档加载器、Text Splitter、VectorStore、Prompt Template 这些组件在 LangGraph 里照样能用。我改造的时候切块和向量化那部分代码几乎没动只把编排逻辑换掉了。1.3 什么时候该迁什么时候不用迁这可能是你最关心的问题。我的判断标准很简单RAG 流程里有没有“条件判断”“循环”“人工介入”这三个需求。只有条件判断用 LangGraph 的条件边值得迁。有循环重试必须用图Chain 做不了。需要人工审核生成结果LangGraph 的 interrupt 机制必须迁。以上都没有纯单轮问答检索一次就输出没必要迁Chain 写起来更简单维护成本更低。我当时迁移的触发点就是多轮对话需求上线加上业务方要求高危问题必须人工复核后才能回答这两个需求一叠加Chain 彻底扛不住了。2. RAG 知识库改造前的现状盘点与目标梳理动工之前我花了两天时间把旧链路的每个环节重新过了一遍把输入输出全部打点记录。这一步非常重要没有改造前的基线数据后面根本说不清楚改完到底有没有变好。2.1 旧版 RAG 链路回顾旧链路是标准的基础 RAG 流程文档加载 → 文本清洗 → 切块 → Embedding 向量化 → 存入向量库 → 检索 Top-K → 拼 Prompt → 大模型生成答案。文档来源主要是 Word、PDF、Markdown 格式的运维手册、产品说明、内部制度文件大概两百多份。切块用的RecursiveCharacterTextSplitter参数是chunk_size500、chunk_overlap50。Embedding 用的开源模型bge-m3向量库用的 Chroma。生成端走的是公司统一的 LLM 网关。这个配置跑单轮问答没问题一些小问题偶尔会出现但整体能用。真正崩掉的是两件事一是多轮对话用户说“那它支持分布式部署吗”系统根本不知道“它”是什么二是引用溯源业务方要求每个回答必须标出依据哪份文档旧链路只把检索到的文本拼进 Prompt模型输出根本不对应来源也没法校验。2.2 改造要解决的核心问题我把需求整理成了四类这四类也是绝大多数企业知识库 RAG 都会遇到的问题第一多轮对话与指代消解。用户经常说“这个”“它”“刚才提到的那个”。解决方案不是把所有历史都塞给模型而是加一个查询改写节点把对话历史 当前问题合并成一个独立的、自包含的问题再用改写后的问题做检索。第二引用完整性与可追溯。生成答案必须带来源编号且编号必须在检索结果里真实存在不能是模型编的。这里需要两个动作Prompt 里强制要求标注编号生成后做一个简单的规则校验发现引用了不存在的编号就触发重新生成。第三检索质量。旧链路只做向量检索召回结果对关键词不敏感专业术语稍微变个说法就召不回。改造后做成“向量检索 关键词检索”的混合检索再用 RRF 融合、接一个 rerank 模型把最终进入生成环节的文档卡到 3 到 5 篇。第四可控性与人工审核。对涉及权限、安全、操作类的回答配置人工审核节点。LangGraph 的interrupt_before机制可以完美支持“生成前停下来等审批”的流程。需求明确了改造的方向也就清晰了。接下来就是动手写代码。3. 用 LangGraph 重构 RAG核心实现细节这一章是全文的干货重点。我会按“状态定义 → 图结构 → 关键节点 → 持久化与人工介入”的顺序讲每一段代码都是改造后实际在跑的逻辑你照着搭就能出最小可用版本。3.1 图结构设计先画流程再写代码我建议你先别急着写代码用纸把流程图草稿画出来。我当时画的流程是这样的用户问题进来先做一个意图判断是闲聊、是普通知识问答、还是高危操作类问题普通问答走检索生成链路高危问题在生成之前停下来把检索结果发给审核人确认闲聊直接转给通用对话模型。进入检索链路后先判断当前问题有没有指代词有就去“查询改写”节点没有就直接检索。检索环节做混合检索结果做 RRF 融合和重排序最后进入生成节点生成后做引用校验校验不过重新生成最多重试两次。这个流程在 LangGraph 里对应的节点就是route_intent、rewrite_query、retrieve、rerank、generate、human_review。边上面有一些条件函数意图路由函数决定走哪个分支指代判断函数决定要不要改写查询引用校验函数决定是否重新生成。3.2 状态定义与图的构建LangGraph 里状态是一个 TypedDict我强烈建议所有字段都显式定义不要图省事只用一个大字典否则后面多人协作时会非常痛苦。from typing import TypedDict, Annotated, List from operator import add from langgraph.graph import StateGraph, START, END class RAGState(TypedDict): messages: List[dict] # 多轮对话历史 query: str # 原始用户问题 rewritten_query: str # 改写后的独立问题 retrieved_docs: Annotated[List[dict], add] # 检索结果允许多节点追加 reranked_docs: List[dict] # 重排序后的文档 answer: str # 最终答案 source_ok: bool # 引用是否校验通过 need_human_review: bool # 是否需要人工审核注意retrieved_docs我用了Annotated[List[dict], add]这意味着多个节点往这个字段写值时新值会追加到旧值后面而不是直接覆盖。这是在 LangGraph 开发里最容易踩的坑之一后面会详细说。 然后构建图builder StateGraph(RAGState) builder.add_node(route_intent, route_intent) builder.add_node(rewrite_query, rewrite_query) builder.add_node(retrieve, retrieve) builder.add_node(rerank, rerank) builder.add_node(generate, generate) builder.add_node(human_review, human_review) builder.add_edge(START, route_intent) builder.add_conditional_edges( route_intent, decide_next, # 返回字符串: chat / retrieve / human_review {chat: END, retrieve: rewrite_query, human_review: human_review} ) builder.add_conditional_edges( rewrite_query, should_rewrite, # 返回字符串: rewrite / retrieve {rewrite: rewrite_query, retrieve: retrieve} )decide_next和should_rewrite这两个函数的区别值得说一下。decide_next是主路由判断整轮对话的走向should_rewrite是指代消解开关判断当前问题需不需要改写。这两个不要混在一起否则后面加需求时逻辑会纠缠不清。3.3 查询改写与检索节点的实现查询改写的 Prompt 是改造效果提升的关键。我最开始用了一个很简单的 Prompt“请把用户问题改写成包含上下文的独立问题”效果很差模型经常把历史里的无关信息也吸收进来。后来改成了带结构化约束的写法def rewrite_query(state: RAGState) - dict: history state[messages][-4:] # 只看最近 4 轮防止历史过长引入噪声 current_question state[query] prompt f你负责把多轮对话中的当前问题改写为一个独立问题。 要求 1. 只补充必要的历史信息不要扩写不要加入知识库内容。 2. 如果当前问题本身信息完整直接原样输出。 3. 输出只有一句话不要解释。 对话历史 {format_history(history)} 当前问题{current_question} 改写结果 response llm.invoke(prompt) # 实际项目里我会在这里加一个判断改写结果与原始问题相同则置空 return {rewritten_query: response.content.strip()}这里有个我自己后来才想明白的细节不是每次都要改写。你可以在should_rewrite里先用简单规则判断一下比如检测“它”“这个”“上面提到的”这类指代词有再去调用大模型改写。这么做的好处是省一次 LLM 调用降低延迟和成本。说实话很多问题根本不含指代白白改写有时候还会把问题改歪。检索节点我做了混合检索向量召回和关键词召回并行最后用 RRF 融合。def retrieve(state: RAGState) - dict: query state.get(rewritten_query) or state[query] # 向量检索 vector_results vector_store.similarity_search_with_score(query, k10) # 关键词检索用 BM25 实现 keyword_results bm25_search(query, k10) # 统一 doc_id vector_ranked [{doc_id: d.id, content: d.page_content, score: s} for d, s in vector_results] keyword_ranked [{doc_id: d.id, content: d.page_content, score: s} for d, s in keyword_results] combined rrf_merge(vector_ranked, keyword_ranked, k60) return {retrieved_docs: combined[:10]}3.4 RRF 融合的隐藏坑默认实现去重逻辑有缺陷这里必须单独拿出来说因为我在这个坑里耗了整整一个下午。LangChain 和 LangChain4j 里都提供了 RRF 的默认实现但实测下来它们的去重逻辑存在缺陷。具体表现是同一份文档的不同 chunk 在融合时被当作独立结果参与排序而没有先按 doc_id 分组去重。结果就是一份出现过多次的文档会占据融合结果的前几名把其他真正相关的文档挤下去而且由于切块有 overlap相邻区块会生成大量重复内容导致 RRF 分数虚高。我后来直接改成了自定义实现def rrf_merge(result_lists, k60): result_lists: List[List[dict]]每个 dict 必须有 doc_id scores {} best_rank {} for docs in result_lists: for rank, doc in enumerate(docs): doc_id doc[doc_id] # 每个召回序列里同一个 doc 只取排名最高的那次 if doc_id not in best_rank or rank best_rank[doc_id]: best_rank[doc_id] rank for doc_id, rank in best_rank.items(): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)去重逻辑在融合之前不在融合之后。先保证每个召回序列里一个 doc 只参与一次计分再做 RRF。这个改动之后检索结果的多样性明显改善原来那种“一个文档的碎片把结果刷屏”的现象基本消失了。3.5 重排序节点最终决定生成质量的那道门槛RRF 融合出来的结果在真正交给大模型之前我会再做一次重排序。企业知识库里文档质量参差不齐向量召回和关键词召回都有各自的盲区用一个专门的交叉编码器 rerank 模型把相关性重新打一遍分取前 3 到 5 篇作为最终上下文效果提升非常明显。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-v2-m3) def rerank(state: RAGState) - dict: query state.get(rewritten_query) or state[query] docs state[retrieved_docs] if len(docs) 3: return {reranked_docs: docs} pairs [(query, doc[content][:512]) for doc in docs] scores reranker.predict(pairs) scored list(zip(docs, scores)) scored.sort(keylambda x: x[1], reverseTrue) return {reranked_docs: [d for d, s in scored[:5]]}重排序模型我只取content[:512]原因是交叉编码器的输入长度有限制而且长度越长推理越慢。你要是召回的文档太长可以先截断或者先用粗筛把候选集缩到 10 篇以内再做重排。3.6 生成节点与引用校验生成节点的 Prompt 我改过很多版最后稳定下来的核心要求是必须基于给出的资料回答每个关键结论后要标注来源编号编号格式必须与资料序号一致。生成之后我会跑一个简单的校验函数把模型输出的[编号]全部提取出来检查是否都在可用编号集合里如果出现非法编号强制重新生成。这里有个技巧重试次数一定要有限制我设的是最多重试两次。不设限制的话模型可能陷入死循环白白消耗 Token。而且重试时的 Prompt 要把错误原因说清楚让模型知道是编号格式问题不是回答内容问题。3.7 持久化与人工介入langgraph 比 Chain 值钱的地方最后是持久化和人工介入这是 LangGraph 相对 Chain 最值钱的两个能力。持久化用 Checkpointer 实现。我用的 SQLite 版本生产环境你可以换 PostgreSQL 版本但原理一样。from langgraph.checkpoint.sqlite import SqliteSaver graph builder.compile( checkpointerSqliteSaver.from_conn_string(rag.db), interrupt_before[generate] )interrupt_before[generate]的意思是流程跑到generate节点之前停下来等待外部确认。配合前面说的need_human_review标记就能做到正常问题自动生成答案高危、涉权限的问题检索完检索结果先给人看人点同意才调用生成节点人点拒绝就可以直接返回一个预设话术比如“该问题需要线下人工处理已转交相关同事”。在外部代码里你可以用graph.invoke(state, config)拿返回结果根据interrupt信息判断停在了哪里然后审核结束再带着审核结果graph.invoke(None, config)继续跑。这个模式在企业场景里非常实用。4. 实操过程中的踩坑记录与排查方法这段是我最想写的部分。LangGraph 的资料现在还比较杂中文文档也不够全很多坑是你看着官方教程根本发现不了的。我把这次改造里真实踩过的坑写下来你遇到了可以直接照方抓药。4.1 状态管理的坑字段被悄悄覆盖先说最经典的多个节点往同一个 State 字段写值后执行的节点会把先执行的节点写入的值整个覆盖掉。我第一次跑通全流程后发现retrieved_docs永远只包含最后一个节点写入的结果前面节点检索到的文档全都不见了。当时排查了很久最后翻开文档才想起来默认情况下节点返回值是直接赋给字段的不是追加。想追加必须用Annotated[List, add]或者自己写 reducer。我的建议是凡是“多个来源都要往里塞数据”的字段一律用Annotatedoperator.add标明是追加语义凡是“最终只有一份结果”的字段正常赋值就行。搞清楚每个字段的写入方式是 LangGraph 开发的第一课。4.2 checkpointer 的坑多轮对话的 config 要传入同一个 thread_id还有一个多轮对话特别容易踩的坑。LangGraph 按thread_id区分多轮对话历史你第一轮调用时传入config{configurable: {thread_id: uuid}}第二轮必须传同一个thread_id否则框架会视为新一轮对话之前的状态和历史全部丢失。这个坑相当隐蔽因为代码不会报错只是结果不对。我之前在本地测试时习惯随机生成一个 thread_id导致所有多轮测试全部失败白白浪费了大半天。config {configurable: {thread_id: customer-001}} result graph.invoke({messages: ...}, configconfig)4.3 切块参数需要按文档类型重新调旧链路沿用默认的chunk_size500, chunk_overlap50我以为改造时不用动结果重排序上线后才发现有些片段因为切得太大一句话里混了两个主题检索召回了但 rerank 得分很低反而拉低了整体准确率。后来我按文档类型做了差异化处理宽泛的制度文件chunk_size放到 600 也没问题操作手册这种步骤性强的文档chunk_size缩小到 300overlap提高到 80确保关键步骤不被切断。切块没有万能参数只能结合问答评测集去试先把测试集跑一遍看哪些问题因为切块召回不对再反向调。4.4 多轮问答的上下文污染查询改写节点上线后多轮准确率确实上来了但我很快发现一个新的质量问题用户问一句“和它有什么区别”模型改写时会把历史里所有话题都揉进新问题里导致改写结果又长又乱。解决方法是限制历史范围只取最近 4 轮并在 Prompt 里明确要求“只补充必要的历史信息”。另外加了一个防御逻辑如果改写后的问题与原始问题在字符相似度上超过 0.9 或完全一致就认为没有指代直接用原始问题。这样可以避免无意义的 LLM 改写调用。4.5 评估方式不要只看几个例子的感觉改造过程中最容易犯的错误是拿三五个例子试一下感觉变好了就上线。这次我特意建了一个 100 条问题的评测集覆盖单轮问答、多轮问答、指代消解、高危问题拦截、引用标注等场景每条问题都标注了预期答案和预期来源文档。改造前后用同一套评测集跑分结果才真正说明问题。这个评测集的构建方法比较简单从真实用户会话里抽问题后面再手动补充一些针对知识库边缘情况的题目比如专业术语变体、跨文档对比、否定式提问等。有条件的团队可以定期维护这个评测集后续每次改 Prompt、调检索参数、换模型都能用它做回归验证。5. 改造效果与对比数据说话改造后我跑了三个月左右的生产流量这里放一组基于评测集和线上抽样的对比数据你可以当参考。指标改造前Chain 版改造后LangGraph 版单轮问答准确率78%84%多轮问答准确率52%78%回答带引用比例20%92%高危问题人工拦截覆盖率0%100%平均响应耗时3.2s4.1s单轮问答提升了 6 个百分点主要贡献来自混合检索和重排序。多轮问答提升了 26 个百分点查询改写节点功不可没。引用比例从 20% 到 92%是 Prompt 约束 生成后校验的共同结果。剩余 8% 是因为部分知识条目本身没有明确来源文档属于合理情况。平均耗时多了 0.9 秒这是查询改写、重排序、人工审核机制付出的代价。对这个场景来说完全可以接受因为换来了更高的准确率。如果对延迟要求更高可以把重排序模型换成一个更小的版本或者只在召回结果超过一定数量时才触发重排平时直接走原顺序。改造完成后我还做了一次团队复盘得出一条很重要的结论LangGraph 带来的最大收益不是单点能力的提升而是把原来“不可控的黑盒流程”变成了“可编排、可干预、可插拔的显式流程”。业务方现在可以直观地看到每一步在干什么出了问题也能精确地指出是检索的问题还是生成的问题而不是只能对着一个 Chain 干瞪眼。6. 进阶方向从 RAG 到 Agentic RAG改造完成之后团队自然开始往更复杂的方向探索这里把思路也分享出来算是给后续留个引子。6.1 把 RAG 封装成 Agent 的工具知识库问答跑顺之后你会发现它只是整个业务系统里的一个能力。放着不管它就是一个被动等待的问答接口。往上走一步就是把 RAG 检索器封装成一个工具让 Agent 决定在合适的时机调用它。这就是热词里经常提到的“skill 怎么和 RAG 结合起来”的答案检索器就是 Agent 的一个 skill。用户说“查一下《XX 手册》里的部署要求”Agent 先查知识库再判断是否需要额外查在线文档。用户说“帮我写个告警处理总结”Agent 可能先查知识库里的告警指南再结合历史工单生成内容。本质上RAG 从一个主流程变成了 Agent 可选的工具。LangGraph 在这里的优势是工具调用、结果回填、再次判断这些逻辑都可以放进图里比 AgentExecutor 那种固定的 loop 模式更容易控制。6.2 Agentic RAG 的两种形态现在很多人张口就是“Agentic RAG”但落地时其实只有两种主流形态单 Agent 带工具、多 Agent 协作。单 Agent 带工具是最常见的起步形态一个主 Agent 负责理解用户意图决定要不要调用检索工具检索返回后再组织回答。复杂点是记忆管理长对话里 Agent 容易忘事需要频繁调用历史压缩节点。如果你的场景只是知识库问答单 Agent 就够了。多 Agent 协作适合任务复杂、需要分阶段处理的场景比如主管 Agent 拆解任务检索 Worker 去查知识库写作 Worker 基于检索结果写报告质检 Agent 负责审核。这种架构能力强调试难度也高。我的建议是先从单 Agent 起步等真的出现“一个 Agent 什么都干反而互相干扰”的情况再拆多 Agent。6.3 RAG 和 MCP 的区别与取舍不少人在问“RAG 和 MCP 有什么区别”。两者的定位完全不同RAG 是一种增强大模型生成的机制核心是把外部知识塞进上下文MCP 是一种模型与外部工具之间的连接协议解决的是“模型怎么调用工具、工具怎么返回结果”的问题。如果非要说关系它们更像是互补。你可以把知识库封装成一个 RAG 服务再通过 MCP server 暴露出去这样任何支持 MCP 的客户端都能以统一的方式调用你的知识库。内部高可控的问答场景直接用自己搭的 RAG 流程更稳对外想开放给多个 Agent 或平台走 MCP 是更好的选择。6.4 知识本体的价值最后提一下热词里的“ontology RAG”。我在改造后期尝试把企业术语表、产品线关系、组织架构这些信息构建成简单的知识本体在检索阶段注入约束。效果最明显的是专业术语消歧用户说“部署”知识库里有“软件部署”和“部署到生产环境”两个含义本体关系可以帮助检索器更精准地定位。这个方向目前还没有完全成熟但对知识密度高的行业来说是一个值得关注的增量。我在这次改造里体会最深的一点是迁移本身不难难的是搞清楚你为什么要迁。先把旧链路每个环节的输入输出盘点清楚把问题定义清楚再动手选型。不要为了用图而用图可以先从 3 到 5 个节点的小图开始跑跑通了再往上加复杂度。最后再分享一个小技巧改造期间每个节点的输入输出都打上结构化日志排错效率会翻倍这个习惯我一直保留到现在。 从LangChain迁移到LangGraph：企业RAG知识库改造实践
去年年中我接手了一个企业知识库问答服务的改造原版本是用 LangChain 的 Chain 拼出来的。单轮问答跑得挺欢一上多轮对话和生产流量就各种露馅上下文追不上、检索结果不带引用、问题稍微绕一点就答非所问。后来我把整个 RAG 知识库流程迁移到了 LangGraph 上把原来一条直线改成了带分支、带状态、带人工审核的图。这篇文章就把改造过程里踩过的坑和最终落地方案完整记录下来围绕 LangChain、LangGraph、RAG 知识库这三个关键词展开适合正在做知识库问答、准备引入 Agent 编排或者想从 Chain 迁移到 Graph 的团队参考。1. 为什么从 LangChain 迁到 LangGraphRAG 改造的真实起点先说结论LangChain 没有过时但场景变了。你做一个 Demo、一个原型、一次性的问答脚本用 Chain 完全够。可一旦进入生产环境尤其是企业知识库这种对可控性、可追溯性、多轮交互都有要求的场景Chain 那种线性结构就开始拖后腿。1.1 老方案的问题链式编排的瓶颈LangChain 早期的核心抽象是 Chain一条链自上而下执行加载文档、切块、向量化、检索、拼 Prompt、调用模型、输出。逻辑简单代码也简单但问题恰恰出在“简单”上。第一个痛点是没法分叉。比如用户问“它支持哪些格式”你要先判断“它”指代什么再决定是直接检索还是先改写查询。在 Chain 里做这种条件路由得写 if-else 在外面包一层或者拼多个 Chain 手动串代码很快就乱成一团。第二个痛点是循环。第一次检索结果不理想想换个关键词重新检索Chain 不支持“回头再来一次”这种回路。第三个痛点是状态。多轮对话里对话历史、中间检索结果、重排序得分这些东西散落在各个变量里链跑完就丢了想持久化、想断点续跑基本靠自研。我用一个生活化的类比Chain 像一条传送带零件从一头进去沿固定路线走到另一头出来。你想在中间加一个质检环节质检不过就退回上一个工位重新加工传送带做不到你得把它拆开重新设计。LangGraph 就是帮你重新设计这套流程的框架。1.2 LangGraph 的核心理念从链到图LangGraph 的核心抽象就三样节点Node、边Edge、状态State。节点就是你的一段处理逻辑比如“查询改写”“检索”“生成答案”都是节点。边决定流程怎么走可以是普通边A 执行完一定走 B也可以是条件边A 执行完根据 State 里的某个字段决定走 B 还是走 C。状态是整个流程共享的数据对象所有节点都能读、能写节点之间通过 state 传递信息。这个设计直接解决了 Chain 的几个死穴条件路由用条件边表达重试用循环边表达多轮对话的历史放在 State 里随流程走Checkpointer 还能把每一步的状态落盘做到“跑到一半停下来人工审批完再继续”。而且 LangGraph 不是把 LangChain 推翻重来。LangChain 里的文档加载器、Text Splitter、VectorStore、Prompt Template 这些组件在 LangGraph 里照样能用。我改造的时候切块和向量化那部分代码几乎没动只把编排逻辑换掉了。1.3 什么时候该迁什么时候不用迁这可能是你最关心的问题。我的判断标准很简单RAG 流程里有没有“条件判断”“循环”“人工介入”这三个需求。只有条件判断用 LangGraph 的条件边值得迁。有循环重试必须用图Chain 做不了。需要人工审核生成结果LangGraph 的 interrupt 机制必须迁。以上都没有纯单轮问答检索一次就输出没必要迁Chain 写起来更简单维护成本更低。我当时迁移的触发点就是多轮对话需求上线加上业务方要求高危问题必须人工复核后才能回答这两个需求一叠加Chain 彻底扛不住了。2. RAG 知识库改造前的现状盘点与目标梳理动工之前我花了两天时间把旧链路的每个环节重新过了一遍把输入输出全部打点记录。这一步非常重要没有改造前的基线数据后面根本说不清楚改完到底有没有变好。2.1 旧版 RAG 链路回顾旧链路是标准的基础 RAG 流程文档加载 → 文本清洗 → 切块 → Embedding 向量化 → 存入向量库 → 检索 Top-K → 拼 Prompt → 大模型生成答案。文档来源主要是 Word、PDF、Markdown 格式的运维手册、产品说明、内部制度文件大概两百多份。切块用的RecursiveCharacterTextSplitter参数是chunk_size500、chunk_overlap50。Embedding 用的开源模型bge-m3向量库用的 Chroma。生成端走的是公司统一的 LLM 网关。这个配置跑单轮问答没问题一些小问题偶尔会出现但整体能用。真正崩掉的是两件事一是多轮对话用户说“那它支持分布式部署吗”系统根本不知道“它”是什么二是引用溯源业务方要求每个回答必须标出依据哪份文档旧链路只把检索到的文本拼进 Prompt模型输出根本不对应来源也没法校验。2.2 改造要解决的核心问题我把需求整理成了四类这四类也是绝大多数企业知识库 RAG 都会遇到的问题第一多轮对话与指代消解。用户经常说“这个”“它”“刚才提到的那个”。解决方案不是把所有历史都塞给模型而是加一个查询改写节点把对话历史 当前问题合并成一个独立的、自包含的问题再用改写后的问题做检索。第二引用完整性与可追溯。生成答案必须带来源编号且编号必须在检索结果里真实存在不能是模型编的。这里需要两个动作Prompt 里强制要求标注编号生成后做一个简单的规则校验发现引用了不存在的编号就触发重新生成。第三检索质量。旧链路只做向量检索召回结果对关键词不敏感专业术语稍微变个说法就召不回。改造后做成“向量检索 关键词检索”的混合检索再用 RRF 融合、接一个 rerank 模型把最终进入生成环节的文档卡到 3 到 5 篇。第四可控性与人工审核。对涉及权限、安全、操作类的回答配置人工审核节点。LangGraph 的interrupt_before机制可以完美支持“生成前停下来等审批”的流程。需求明确了改造的方向也就清晰了。接下来就是动手写代码。3. 用 LangGraph 重构 RAG核心实现细节这一章是全文的干货重点。我会按“状态定义 → 图结构 → 关键节点 → 持久化与人工介入”的顺序讲每一段代码都是改造后实际在跑的逻辑你照着搭就能出最小可用版本。3.1 图结构设计先画流程再写代码我建议你先别急着写代码用纸把流程图草稿画出来。我当时画的流程是这样的用户问题进来先做一个意图判断是闲聊、是普通知识问答、还是高危操作类问题普通问答走检索生成链路高危问题在生成之前停下来把检索结果发给审核人确认闲聊直接转给通用对话模型。进入检索链路后先判断当前问题有没有指代词有就去“查询改写”节点没有就直接检索。检索环节做混合检索结果做 RRF 融合和重排序最后进入生成节点生成后做引用校验校验不过重新生成最多重试两次。这个流程在 LangGraph 里对应的节点就是route_intent、rewrite_query、retrieve、rerank、generate、human_review。边上面有一些条件函数意图路由函数决定走哪个分支指代判断函数决定要不要改写查询引用校验函数决定是否重新生成。3.2 状态定义与图的构建LangGraph 里状态是一个 TypedDict我强烈建议所有字段都显式定义不要图省事只用一个大字典否则后面多人协作时会非常痛苦。from typing import TypedDict, Annotated, List from operator import add from langgraph.graph import StateGraph, START, END class RAGState(TypedDict): messages: List[dict] # 多轮对话历史 query: str # 原始用户问题 rewritten_query: str # 改写后的独立问题 retrieved_docs: Annotated[List[dict], add] # 检索结果允许多节点追加 reranked_docs: List[dict] # 重排序后的文档 answer: str # 最终答案 source_ok: bool # 引用是否校验通过 need_human_review: bool # 是否需要人工审核注意retrieved_docs我用了Annotated[List[dict], add]这意味着多个节点往这个字段写值时新值会追加到旧值后面而不是直接覆盖。这是在 LangGraph 开发里最容易踩的坑之一后面会详细说。 然后构建图builder StateGraph(RAGState) builder.add_node(route_intent, route_intent) builder.add_node(rewrite_query, rewrite_query) builder.add_node(retrieve, retrieve) builder.add_node(rerank, rerank) builder.add_node(generate, generate) builder.add_node(human_review, human_review) builder.add_edge(START, route_intent) builder.add_conditional_edges( route_intent, decide_next, # 返回字符串: chat / retrieve / human_review {chat: END, retrieve: rewrite_query, human_review: human_review} ) builder.add_conditional_edges( rewrite_query, should_rewrite, # 返回字符串: rewrite / retrieve {rewrite: rewrite_query, retrieve: retrieve} )decide_next和should_rewrite这两个函数的区别值得说一下。decide_next是主路由判断整轮对话的走向should_rewrite是指代消解开关判断当前问题需不需要改写。这两个不要混在一起否则后面加需求时逻辑会纠缠不清。3.3 查询改写与检索节点的实现查询改写的 Prompt 是改造效果提升的关键。我最开始用了一个很简单的 Prompt“请把用户问题改写成包含上下文的独立问题”效果很差模型经常把历史里的无关信息也吸收进来。后来改成了带结构化约束的写法def rewrite_query(state: RAGState) - dict: history state[messages][-4:] # 只看最近 4 轮防止历史过长引入噪声 current_question state[query] prompt f你负责把多轮对话中的当前问题改写为一个独立问题。 要求 1. 只补充必要的历史信息不要扩写不要加入知识库内容。 2. 如果当前问题本身信息完整直接原样输出。 3. 输出只有一句话不要解释。 对话历史 {format_history(history)} 当前问题{current_question} 改写结果 response llm.invoke(prompt) # 实际项目里我会在这里加一个判断改写结果与原始问题相同则置空 return {rewritten_query: response.content.strip()}这里有个我自己后来才想明白的细节不是每次都要改写。你可以在should_rewrite里先用简单规则判断一下比如检测“它”“这个”“上面提到的”这类指代词有再去调用大模型改写。这么做的好处是省一次 LLM 调用降低延迟和成本。说实话很多问题根本不含指代白白改写有时候还会把问题改歪。检索节点我做了混合检索向量召回和关键词召回并行最后用 RRF 融合。def retrieve(state: RAGState) - dict: query state.get(rewritten_query) or state[query] # 向量检索 vector_results vector_store.similarity_search_with_score(query, k10) # 关键词检索用 BM25 实现 keyword_results bm25_search(query, k10) # 统一 doc_id vector_ranked [{doc_id: d.id, content: d.page_content, score: s} for d, s in vector_results] keyword_ranked [{doc_id: d.id, content: d.page_content, score: s} for d, s in keyword_results] combined rrf_merge(vector_ranked, keyword_ranked, k60) return {retrieved_docs: combined[:10]}3.4 RRF 融合的隐藏坑默认实现去重逻辑有缺陷这里必须单独拿出来说因为我在这个坑里耗了整整一个下午。LangChain 和 LangChain4j 里都提供了 RRF 的默认实现但实测下来它们的去重逻辑存在缺陷。具体表现是同一份文档的不同 chunk 在融合时被当作独立结果参与排序而没有先按 doc_id 分组去重。结果就是一份出现过多次的文档会占据融合结果的前几名把其他真正相关的文档挤下去而且由于切块有 overlap相邻区块会生成大量重复内容导致 RRF 分数虚高。我后来直接改成了自定义实现def rrf_merge(result_lists, k60): result_lists: List[List[dict]]每个 dict 必须有 doc_id scores {} best_rank {} for docs in result_lists: for rank, doc in enumerate(docs): doc_id doc[doc_id] # 每个召回序列里同一个 doc 只取排名最高的那次 if doc_id not in best_rank or rank best_rank[doc_id]: best_rank[doc_id] rank for doc_id, rank in best_rank.items(): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)去重逻辑在融合之前不在融合之后。先保证每个召回序列里一个 doc 只参与一次计分再做 RRF。这个改动之后检索结果的多样性明显改善原来那种“一个文档的碎片把结果刷屏”的现象基本消失了。3.5 重排序节点最终决定生成质量的那道门槛RRF 融合出来的结果在真正交给大模型之前我会再做一次重排序。企业知识库里文档质量参差不齐向量召回和关键词召回都有各自的盲区用一个专门的交叉编码器 rerank 模型把相关性重新打一遍分取前 3 到 5 篇作为最终上下文效果提升非常明显。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-v2-m3) def rerank(state: RAGState) - dict: query state.get(rewritten_query) or state[query] docs state[retrieved_docs] if len(docs) 3: return {reranked_docs: docs} pairs [(query, doc[content][:512]) for doc in docs] scores reranker.predict(pairs) scored list(zip(docs, scores)) scored.sort(keylambda x: x[1], reverseTrue) return {reranked_docs: [d for d, s in scored[:5]]}重排序模型我只取content[:512]原因是交叉编码器的输入长度有限制而且长度越长推理越慢。你要是召回的文档太长可以先截断或者先用粗筛把候选集缩到 10 篇以内再做重排。3.6 生成节点与引用校验生成节点的 Prompt 我改过很多版最后稳定下来的核心要求是必须基于给出的资料回答每个关键结论后要标注来源编号编号格式必须与资料序号一致。生成之后我会跑一个简单的校验函数把模型输出的[编号]全部提取出来检查是否都在可用编号集合里如果出现非法编号强制重新生成。这里有个技巧重试次数一定要有限制我设的是最多重试两次。不设限制的话模型可能陷入死循环白白消耗 Token。而且重试时的 Prompt 要把错误原因说清楚让模型知道是编号格式问题不是回答内容问题。3.7 持久化与人工介入langgraph 比 Chain 值钱的地方最后是持久化和人工介入这是 LangGraph 相对 Chain 最值钱的两个能力。持久化用 Checkpointer 实现。我用的 SQLite 版本生产环境你可以换 PostgreSQL 版本但原理一样。from langgraph.checkpoint.sqlite import SqliteSaver graph builder.compile( checkpointerSqliteSaver.from_conn_string(rag.db), interrupt_before[generate] )interrupt_before[generate]的意思是流程跑到generate节点之前停下来等待外部确认。配合前面说的need_human_review标记就能做到正常问题自动生成答案高危、涉权限的问题检索完检索结果先给人看人点同意才调用生成节点人点拒绝就可以直接返回一个预设话术比如“该问题需要线下人工处理已转交相关同事”。在外部代码里你可以用graph.invoke(state, config)拿返回结果根据interrupt信息判断停在了哪里然后审核结束再带着审核结果graph.invoke(None, config)继续跑。这个模式在企业场景里非常实用。4. 实操过程中的踩坑记录与排查方法这段是我最想写的部分。LangGraph 的资料现在还比较杂中文文档也不够全很多坑是你看着官方教程根本发现不了的。我把这次改造里真实踩过的坑写下来你遇到了可以直接照方抓药。4.1 状态管理的坑字段被悄悄覆盖先说最经典的多个节点往同一个 State 字段写值后执行的节点会把先执行的节点写入的值整个覆盖掉。我第一次跑通全流程后发现retrieved_docs永远只包含最后一个节点写入的结果前面节点检索到的文档全都不见了。当时排查了很久最后翻开文档才想起来默认情况下节点返回值是直接赋给字段的不是追加。想追加必须用Annotated[List, add]或者自己写 reducer。我的建议是凡是“多个来源都要往里塞数据”的字段一律用Annotatedoperator.add标明是追加语义凡是“最终只有一份结果”的字段正常赋值就行。搞清楚每个字段的写入方式是 LangGraph 开发的第一课。4.2 checkpointer 的坑多轮对话的 config 要传入同一个 thread_id还有一个多轮对话特别容易踩的坑。LangGraph 按thread_id区分多轮对话历史你第一轮调用时传入config{configurable: {thread_id: uuid}}第二轮必须传同一个thread_id否则框架会视为新一轮对话之前的状态和历史全部丢失。这个坑相当隐蔽因为代码不会报错只是结果不对。我之前在本地测试时习惯随机生成一个 thread_id导致所有多轮测试全部失败白白浪费了大半天。config {configurable: {thread_id: customer-001}} result graph.invoke({messages: ...}, configconfig)4.3 切块参数需要按文档类型重新调旧链路沿用默认的chunk_size500, chunk_overlap50我以为改造时不用动结果重排序上线后才发现有些片段因为切得太大一句话里混了两个主题检索召回了但 rerank 得分很低反而拉低了整体准确率。后来我按文档类型做了差异化处理宽泛的制度文件chunk_size放到 600 也没问题操作手册这种步骤性强的文档chunk_size缩小到 300overlap提高到 80确保关键步骤不被切断。切块没有万能参数只能结合问答评测集去试先把测试集跑一遍看哪些问题因为切块召回不对再反向调。4.4 多轮问答的上下文污染查询改写节点上线后多轮准确率确实上来了但我很快发现一个新的质量问题用户问一句“和它有什么区别”模型改写时会把历史里所有话题都揉进新问题里导致改写结果又长又乱。解决方法是限制历史范围只取最近 4 轮并在 Prompt 里明确要求“只补充必要的历史信息”。另外加了一个防御逻辑如果改写后的问题与原始问题在字符相似度上超过 0.9 或完全一致就认为没有指代直接用原始问题。这样可以避免无意义的 LLM 改写调用。4.5 评估方式不要只看几个例子的感觉改造过程中最容易犯的错误是拿三五个例子试一下感觉变好了就上线。这次我特意建了一个 100 条问题的评测集覆盖单轮问答、多轮问答、指代消解、高危问题拦截、引用标注等场景每条问题都标注了预期答案和预期来源文档。改造前后用同一套评测集跑分结果才真正说明问题。这个评测集的构建方法比较简单从真实用户会话里抽问题后面再手动补充一些针对知识库边缘情况的题目比如专业术语变体、跨文档对比、否定式提问等。有条件的团队可以定期维护这个评测集后续每次改 Prompt、调检索参数、换模型都能用它做回归验证。5. 改造效果与对比数据说话改造后我跑了三个月左右的生产流量这里放一组基于评测集和线上抽样的对比数据你可以当参考。指标改造前Chain 版改造后LangGraph 版单轮问答准确率78%84%多轮问答准确率52%78%回答带引用比例20%92%高危问题人工拦截覆盖率0%100%平均响应耗时3.2s4.1s单轮问答提升了 6 个百分点主要贡献来自混合检索和重排序。多轮问答提升了 26 个百分点查询改写节点功不可没。引用比例从 20% 到 92%是 Prompt 约束 生成后校验的共同结果。剩余 8% 是因为部分知识条目本身没有明确来源文档属于合理情况。平均耗时多了 0.9 秒这是查询改写、重排序、人工审核机制付出的代价。对这个场景来说完全可以接受因为换来了更高的准确率。如果对延迟要求更高可以把重排序模型换成一个更小的版本或者只在召回结果超过一定数量时才触发重排平时直接走原顺序。改造完成后我还做了一次团队复盘得出一条很重要的结论LangGraph 带来的最大收益不是单点能力的提升而是把原来“不可控的黑盒流程”变成了“可编排、可干预、可插拔的显式流程”。业务方现在可以直观地看到每一步在干什么出了问题也能精确地指出是检索的问题还是生成的问题而不是只能对着一个 Chain 干瞪眼。6. 进阶方向从 RAG 到 Agentic RAG改造完成之后团队自然开始往更复杂的方向探索这里把思路也分享出来算是给后续留个引子。6.1 把 RAG 封装成 Agent 的工具知识库问答跑顺之后你会发现它只是整个业务系统里的一个能力。放着不管它就是一个被动等待的问答接口。往上走一步就是把 RAG 检索器封装成一个工具让 Agent 决定在合适的时机调用它。这就是热词里经常提到的“skill 怎么和 RAG 结合起来”的答案检索器就是 Agent 的一个 skill。用户说“查一下《XX 手册》里的部署要求”Agent 先查知识库再判断是否需要额外查在线文档。用户说“帮我写个告警处理总结”Agent 可能先查知识库里的告警指南再结合历史工单生成内容。本质上RAG 从一个主流程变成了 Agent 可选的工具。LangGraph 在这里的优势是工具调用、结果回填、再次判断这些逻辑都可以放进图里比 AgentExecutor 那种固定的 loop 模式更容易控制。6.2 Agentic RAG 的两种形态现在很多人张口就是“Agentic RAG”但落地时其实只有两种主流形态单 Agent 带工具、多 Agent 协作。单 Agent 带工具是最常见的起步形态一个主 Agent 负责理解用户意图决定要不要调用检索工具检索返回后再组织回答。复杂点是记忆管理长对话里 Agent 容易忘事需要频繁调用历史压缩节点。如果你的场景只是知识库问答单 Agent 就够了。多 Agent 协作适合任务复杂、需要分阶段处理的场景比如主管 Agent 拆解任务检索 Worker 去查知识库写作 Worker 基于检索结果写报告质检 Agent 负责审核。这种架构能力强调试难度也高。我的建议是先从单 Agent 起步等真的出现“一个 Agent 什么都干反而互相干扰”的情况再拆多 Agent。6.3 RAG 和 MCP 的区别与取舍不少人在问“RAG 和 MCP 有什么区别”。两者的定位完全不同RAG 是一种增强大模型生成的机制核心是把外部知识塞进上下文MCP 是一种模型与外部工具之间的连接协议解决的是“模型怎么调用工具、工具怎么返回结果”的问题。如果非要说关系它们更像是互补。你可以把知识库封装成一个 RAG 服务再通过 MCP server 暴露出去这样任何支持 MCP 的客户端都能以统一的方式调用你的知识库。内部高可控的问答场景直接用自己搭的 RAG 流程更稳对外想开放给多个 Agent 或平台走 MCP 是更好的选择。6.4 知识本体的价值最后提一下热词里的“ontology RAG”。我在改造后期尝试把企业术语表、产品线关系、组织架构这些信息构建成简单的知识本体在检索阶段注入约束。效果最明显的是专业术语消歧用户说“部署”知识库里有“软件部署”和“部署到生产环境”两个含义本体关系可以帮助检索器更精准地定位。这个方向目前还没有完全成熟但对知识密度高的行业来说是一个值得关注的增量。我在这次改造里体会最深的一点是迁移本身不难难的是搞清楚你为什么要迁。先把旧链路每个环节的输入输出盘点清楚把问题定义清楚再动手选型。不要为了用图而用图可以先从 3 到 5 个节点的小图开始跑跑通了再往上加复杂度。最后再分享一个小技巧改造期间每个节点的输入输出都打上结构化日志排错效率会翻倍这个习惯我一直保留到现在。