今年上半年我接手了部门内部的知识库问答项目代码库是早期同事用 LangChain 搭的。表面看一切正常文档灌进去问题丢进去答案出来。可真要往企业场景推的时候问题一个接一个多轮对话里用户说“那第二个方案呢”模型根本不知道“那”指的是什么检索召回的文档里混着不同业务线的资料但权限过滤逻辑散落在回调函数里根本没法统一管控更别提产品经理还要求“答不上的时候自动换个角度重新检索一次”。一条线性 Chain 根本塞不下这么多逻辑。后来我把底层迁到了 LangGraph同时对 RAG 知识库的检索链路做了一次系统重构。这篇文章不讲空话就记录这次改造里我踩过的坑、想通的逻辑和最终落地的方案。1. 从 LangChain 到 LangGraph不是追新是链式结构撑不住了先说结论如果你的 RAG 知识库只是“单文档问答”LangChain 足够别折腾。但只要有这些需求——多轮对话、路由分支、检索结果校验、多路召回后处理、人工审批节点——线性链的代码复杂度会指数上升。LangGraph 本质上解决的是“有状态、可控、可回退”的编排问题这不只是框架偏好而是架构选择。1.1 链式调用在长链路里的失控点最初的实现思路很典型加载文档 → 按固定大小切块 → embedding 入库 → 检索 → 组装 prompt → 调 LLM → 返回。每条链是一个LCEL表达式看起来干净利落rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() )单轮问答没问题。但一旦加入“问题改写”节点这条链就变成了chain ( {question: rewrite_chain, history: history_chain} | retriever_chain | prompt | llm )如果再加入“检索不到就回退”“来源权限过滤”“敏感词拦截”等分支LCEL 虽然支持RunnableBranch但本质上还是在一条线性管道里塞条件。真正让我崩溃的是两个场景多轮对话里我需要在改写、检索、生成之后把中间结果都记录到日志表里。链式调用里这些中间变量要么被包裹在子链内部要么靠全局缓存传递调试起来像在猜谜。产品要求“生成前先检查检索到的文档是否相关不相关就重新改写一次问题再检索”这种循环逻辑在 LCEL 里几乎没法自然表达。后来我意识到RAG 系统的复杂度不在于“一次性问答”而在于“状态管理”。LangGraph 把每一步变成图上的节点把上下文显式放进 State这条思路才真正匹配知识库场景。1.2 为什么不是直接换 Dify 或自研全部流程热词里很多人问 Dify 知识库流水线。确实Dify 这类平台能把切块、召回、生成配置化搭一个 Demo 无非半小时。但我这次改造的目标不是做个演示而是要在知识库问答里嵌入部门的权限模型还要对接内部单点登录、按项目维度隔离数据。Dify 的元数据过滤在界面上能做但细粒度的“每次检索前动态计算可见范围”这种需求在低代码平台里反而更绕。LangGraph 提供了自由度和可控性的平衡节点就是普通 Python 函数可以在里面写任何逻辑图结构又强制我显式表达流程。这也是我选择迁移而不是换平台的根本原因。2. 改造前架构复盘原先那套 RAG 知识库“能跑但不好养”动手之前先做了个老的架构体检。我把自己写过的 LangChain RAG 代码翻出来发现所有痛点其实都能追溯到三个设计问题。这里我把当初的实现和问题一起列出来如果你也是从类似状态出发应该会有共鸣。2.1 最早一版代码切块、向量化、检索、生成最早的实现技术栈是 LangChain OpenAI 兼容接口 Chroma 向量库。切块用的RecursiveCharacterTextSplitter每块 500 字符、重叠 50。向量化用的text-embedding-3-small。检索用similarity_searchTop-K 取 6。生成模型用gpt-4o-mini后来业务量上来换成了内部托管的推理服务。结构很简洁当时跑通 demo 的兴奋感还在。但这套流程有个隐藏的前提所有问题都是单轮的所有用户权限都一样所有检索结果都可信。这三个前提在企业级知识库里一个都不成立。2.2 三个“基础病”改写不准、状态丢失、排错困难多轮改写不准当时用query_rewriting_chain输入是chat_history latest_question。实际效果飘忽用户问“这个方案的成本呢”改写后经常变成“成本呢”完全没有指代消解。问题不在 LangChain而在 prompt 和状态拼接可当时的状态拼接就藏在 chain 里出了问题只能一层层拆。状态丢失检索出的文档、命中的 section、token 用量这些中间结果在链式调用里没有一个统一容器。做日志时得在子链里用RunnableConfig的 metadata 传递每次传参都要小心翼翼。排错困难最气人的是生产环境里偶发性答错。我想看具体召回了哪些文档发现日志里只有最终生成结果中间检索结果没落库。因为当初写的时候把中间变量都扔在函数作用域里了。2.3 结论需要的不是修修补补是把“过程”变成一等公民这次复盘让我下定决心与其继续在链式结构上堆 try-except 和分支判断不如把整个交互过程显式建模成一张图。节点、状态、边每一样东西都有名字日志天然就跟着 State 走。LangGraph 的核心抽象恰好就是我为这个项目需要的架构。3. LangGraph 的心智模型把 RAG 知识库当作一张流程图LangGraph 刚上手的时候容易被它的抽象绕晕。我当时看了半天官方文档脑子里一直在想“这不就是把 LangChain 的 chain 加上循环吗”后来我发现真正的关键是换一套思考方式把一次知识库问答看成一场有状态的工作流。3.1 节点Node就是一次决策或动作LangGraph 的节点就是一个普通函数输入是当前 State输出是 State 的部分更新。对应到 RAG 知识库我会把“问题改写”“文档检索”“生成回答”都定义成节点。每个节点只做一件事方便单独测试和复用。3.2 边Edge不只是串联更是条件路由在 LangChain 里边是隐式的一个 Runnable 的输出是另一个 Runnable 的输入。在 LangGraph 里边明显分成两种普通边上一个节点跑完直接进下一个节点。条件边根据当前 State 的内容决定下一跳是哪个节点。这一下就让“检索结果不合格就重新改写”这类需求变得非常自然不是靠 if-else 堆在函数里而是图里面的一个条件分支。3.3 状态State是贯穿全程的数据总线LangGraph 的 State 是一个 TypedDict。所有节点共享这同一个 State 对象。每个节点可以从 State 读数据也能往 State 写数据。最关键的机制是 reducer默认情况下节点返回的字段直接覆盖原字段但如果你想做“合并”比如把多路检索结果叠加到一个列表里可以用operator.add这类自定义 reducer。这个机制解决了我之前状态丢失的问题。所有中间结果都在 State 里调试的时候打印 State 等于拿到了整个流程的全貌。3.4 与 LangChain 组件怎么共存改造时我比较担心是不是要推翻所有 LangChain 的生态组件答案是不用。LangGraph 的节点里可以直接调用 LangChain 的 retriever、prompt template、LLM 封装。迁移成本其实主要在“流程编排”这层而不是“基础组件”这层。比如原来retriever | format_docs | prompt | llm是一条 LCEL 链现在变成在generate_node里按顺序写几行 Python 代码。这样反而更直观。4. 分步改造实操从 Chain 到 Graph 的具体落地下面是比较核心的部分我把改造后的代码骨架完整贴出来。这不是 demo是生产项目里能跑的那版只是隐去了业务字段。4.1 定义 State把所有过程数据显式装好from typing import Annotated, TypedDict import operator class RagState(TypedDict): # 会话历史 messages: list # 用户最新问题 query: str # 多轮改写后的问题 rewritten_query: str # 检索到的文档列表允许多个节点累加写入 retrieved_docs: Annotated[list, operator.add] # 是否需要走检索逻辑 need_retrieval: bool # 拼给最终生成模型的上下文 context: str # 最终回答 answer: str # 引用来源 sources: list这里我特别说明一个设计点retrieved_docs我用了operator.add作为 reducer。这样即使后续有多个检索节点并行执行返回的文档也会自动拼到同一个列表里而不是互相覆盖。如果有些字段你想保持“最后一次写入生效”则不声明 reducer默认行为就是覆盖。4.2 检索节点把 LangChain retriever 包进 Nodedef retrieve_node(state: RagState): query state[rewritten_query] or state[query] docs retriever.invoke(query) # 这里可以加日志、权限过滤、元数据辅助 return {retrieved_docs: docs}之所以要包一层节点而不是直接用原来的 LCEL是因为我想在检索前后做很多事情记录耗时、过滤用户无权访问的 doc metadata、统计命中来源。直接在节点里写原生 Python远比比在链式调用里插回调函数清爽。4.3 问题改写节点多轮对话的指代消解def rewrite_node(state: RagState): if not state[messages]: return {rewritten_query: state[query]} prompt ChatPromptTemplate.from_messages([ (system, 你是知识库问答的查询改写器只输出改写后的查询。), (human, 对话历史{history}\\n\\n最新问题{query}), ]) chain prompt | llm rewritten chain.invoke({ history: format_history(state[messages]), query: state[query], }) return {rewritten_query: rewritten.content}这里有个经验不要每次都用大模型改写。对于没有指代、没有省略的问题直接原样传递能省掉 200ms 到 500ms 延迟。我在改写的 prompt 里加了规则如果最新问题已经包含足够显式的主体和属性就原样输出。实测下来50% 以上的单轮问题都不会触发无意义改写。4.4 生成节点引用溯源靠 metadatadef generate_node(state: RagState): docs state[retrieved_docs] context \n\n.join([f[{i1}] {d.page_content} for i, d in enumerate(docs)]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的知识库问答助手请根据以下资料回答问题。 若资料无法回答请明确说明。回答末尾列出引用编号。), (human, 资料\\n{context}\\n\\n问题{query}), ]) chain prompt | llm response chain.invoke({context: context, query: state[rewritten_query]}) return { answer: response.content, sources: [d.metadata for d in docs], }引用溯源在知识库场景里必须做。我见过太多 RAG 项目只返回答案不返回来源结果用户发现答案是模型幻觉又无从验证。把d.metadata直接塞进 sources前端渲染时就能显示“参考了哪篇文档、哪个章节”。4.5 组装并编译图先跑通主流程from langgraph.graph import StateGraph, END graph StateGraph(RagState) graph.add_node(rewrite, rewrite_node) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, END) app graph.compile()第一次编译你会发现这个图和原来的 Chain 没本质区别甚至更啰嗦。别急它的优势在加分支的时候才体现出来。5. 检索与索引改造切块、元数据过滤、去重我一起做了光改架构不够RAG 知识库的“内容底座”也得跟着调。这次改造正好把之前积压的三个问题一起处理了切块策略、元数据权限过滤、多路召回去重。5.1 切块策略对比固定窗口、递归字符、父子块我做了三组实验数据来自内部几个大规模 Markdown 文档库大约 2000 篇技术文档。切块方式块大小/重叠Top-6 检索命中率上下文噪声实际感受固定长度切片512/6472%高经常切到表格中间递归字符切块字符分隔符递归81%中基本可用父子块切块父块 2000子块 40089%低检索子块、生成时喂父块最终我选了父子块方案用子块去匹配用户问题召回后合并所属父块内容作为上下文。这样既能精确定位到细节段落又不至于让 LLM 看到割裂的半句话。一个容易忽略的点切块必须保留文档结构的元数据比如标题层级、文件路径、项目编号。否则后面做权限过滤和来源展示时会发现不知道这段内容属于哪个项目。5.2 元数据过滤权限隔离不该在生成阶段才做之前项目的权限过滤是硬编码在 prompt 里的“忽略无权限的文档”。这非常不靠谱LLM 根本不知道哪些文档无权限。LangGraph 改造后我在retrieve_node节点里加载用户信息动态构造元数据过滤条件def retrieve_node(state: RagState): user state.get(user) allowed_projects get_user_projects(user) retriever vectorstore.as_retriever( search_kwargs{ k: 6, filter: {project_id: {$in: allowed_projects}} } ) docs retriever.invoke(state[rewritten_query]) return {retrieved_docs: docs}过滤越早做越好不能等文档都进上下文了再想办法。元数据过滤做在检索前既省钱又防止权限数据污染模型。5.3 去重逻辑LangChain 默认 RRF 实现有坑多路召回比如同时按向量和关键词召回时我最初用EnsembleRetriever自带的 RRF 融合。线上发现两个问题不同召回源的分数量纲不一致直接 RRF 有时会把同一篇文档的两个 chunk 都排到前面造成信息冗余。默认去重逻辑只看page_content完全一致很多文档只是格式差异比如行首空格、换行不同就会被当成两份内容。LangGraph 里我直接写了一个自定义节点做后处理def dedupe_and_rerank_node(state: RagState): docs state[retrieved_docs] seen set() unique_docs [] for d in docs: key normalize_text(d.page_content) if key not in seen: seen.add(key) unique_docs.append(d) return {retrieved_docs: unique_docs}另外我也把融合策略从 RRF 换成了量化分数加权每种检索排序转换成 0 到 1 的归一化分数再按权重相加。虽然 RRF 是经典方法但在我这个场景下显式的分数加权更容易调试可解释性更强。6. 从普通 RAG 升级为 Agentic RAG图结构带来的额外红利迁移图结构之后最明显的好处不是“能跑”而是“想加逻辑就加逻辑”。这一节分享我在主流程基础上扩展的几个 Agentic 分支也是搜索引擎里“agentic rag”这个概念落地的真实样子。6.1 条件分支这个 query 到底要不要检索并不是所有问题都该触发知识库检索。“你好”“谢谢”这种话直接回复就行不需要浪费 embedding 和检索的延迟。我在入口加了意图分类节点def route_by_intent(state: RagState): intent classify_intent(state[query]) if intent chitchat: return direct_reply return rewriteLangGraph 的条件边可以完全根据 State 字段决定走哪个节点。实测在日志里约 10% 的流量被分流到闲聊回复知识库检索压力小了不少。6.2 检索后校验分支不够就重写再检索这是最实用的一个改造。原链路里检索几次都返回空或者低相关文档模型还是会硬答。现在我在 “retrieve” 和 “generate” 之间加了一个grade_node对文档和 query 算一个粗粒度相关分数def grade_node(state: RagState): docs state[retrieved_docs] if not docs: return {need_retrieval: False} relevant is_relevant(state[rewritten_query], docs) if not relevant: if state.get(retry_count, 0) 1: return {need_retrieval: True, retry_count: state[retry_count] 1} return {need_retrieval: False}条件边再根据need_retrieval决定回到rewrite还是进入generate。最多重试一次避免死循环。这样系统不会在明显没答案的问题上硬编答案而是尝试换个角度再查一次仍然没有就诚实地回答“知识库中没有相关文档”用户体验比胡说八道强得多。6.3 并行检索一次查多个索引再合并企业内部的知识库往往散落在 wiki、工单系统、项目文档三个地方。我在 LangGraph 中用SendAPI 同时触发三个检索节点最终汇合到去重节点。并行把总耗时从原来的三次串行 2.4 秒压到了 0.9 秒左右。这一块是 LangGraph 的强项代码结构上只需要在动态边里构建Senddef spawn_retrievers(state: RagState): return [Send(retrieve_wiki, {source: wiki}), Send(retrieve_ticket, {source: ticket}), Send(retrieve_project, {source: project})]注意并行写入retrieved_docs时State 的 reducer 必须支持 append这就是我在第四章强调operator.add的原因。并发写同一个字段时如果 reducer 不是线程安全的可能丢文档。LangGraph 在此处会按入参聚合我踩过并发下文档数量不对的坑最终通过显式 reducer 和日志比对修复。7. 迁移后的实测结果与排错记录最后聊一下迁移后的真实数据以及几个 LangGraph 官方文档写得比较隐晦、但实践里一定会遇到的坑。7.1 我们项目里的效果对比以内部 2000 篇技术文档知识库为例测试集是 300 个典型问题包括多轮追问。对比的是改造前的 LangChain 链式版和改造后的 LangGraph 版指标改造前改造后多轮追问准确率61%83%检索无结果时的正确拒答率低常强行编答案高能明确说不知道端到端平均延迟2.8s1.6s并行检索意图分流引用来源完整率40%95%新增一个分支逻辑耗时半天到一天一小时内延迟下降主要是因为并行检索和意图分流而不是 LangGraph 本身更快。准确率提升则来自检索后校验和重写重试机制。7.2 坑一State reducer 的覆盖/追加语义容易搞混默认情况下节点返回什么State 对应字段就变成什么。很多新手在并行检索时忘了加 reducer后返回的节点直接把先返回的文档覆盖了。排查方式也很直接在每条边返回后打印 State 的retrieved_docs长度。def log_state(state: RagState): print( current state keys:, list(state.keys())) print( retrieved docs:, len(state[retrieved_docs])) return {}把这个调试节点插入到关键节点的上游比看抽象调用栈有用得多。7.3 坑二递归和重试上限必须显式控制LangGraph 允许循环但生产环境里死循环很危险。我建过一个“检查相关度”的子图忘记设置重试上限某次因为 Embedding 模型异常导致检索结果持续不相关系统陷入了无限重试把外部 API 的配额打爆了。后来我所有循环边都带一个retry_count字段并在条件边里判断上限。7.4 坑三不要把敏感配置写进 State这里提醒一句State 里不要放 API Key、数据库连接串之类的敏感配置。State 数据会随图执行被传递到多个节点日志系统也可能记录整个 State。我是用上下文对象单独传递密钥State 里只放业务字段。7.5 我的实操体会踩过这么多坑之后我对 RAG 知识库改造的判断是LangGraph 不一定会让你的回答质量凭空变好但它给了你一个可以持续迭代、随时插拔逻辑的骨架。对需要权限控制、多轮对话、路由分支、检索验证的企业知识库场景从 LangChain 迁过来是值得的如果只是做一个固定文档的问答 demoLangGraph 的复杂度反而是一种负担。最后分享一个小技巧改造完成后我把图结构画成了一张内部文档图每个节点对应一个负责人后续加需求时先改图再改代码。这套“流程即代码”的方式让整个知识库团队从“修链条”变成了“编排工作流”维护成本下降非常明显。 LangGraph重构RAG知识库：从Chain到Graph的企业级问答实践
今年上半年我接手了部门内部的知识库问答项目代码库是早期同事用 LangChain 搭的。表面看一切正常文档灌进去问题丢进去答案出来。可真要往企业场景推的时候问题一个接一个多轮对话里用户说“那第二个方案呢”模型根本不知道“那”指的是什么检索召回的文档里混着不同业务线的资料但权限过滤逻辑散落在回调函数里根本没法统一管控更别提产品经理还要求“答不上的时候自动换个角度重新检索一次”。一条线性 Chain 根本塞不下这么多逻辑。后来我把底层迁到了 LangGraph同时对 RAG 知识库的检索链路做了一次系统重构。这篇文章不讲空话就记录这次改造里我踩过的坑、想通的逻辑和最终落地的方案。1. 从 LangChain 到 LangGraph不是追新是链式结构撑不住了先说结论如果你的 RAG 知识库只是“单文档问答”LangChain 足够别折腾。但只要有这些需求——多轮对话、路由分支、检索结果校验、多路召回后处理、人工审批节点——线性链的代码复杂度会指数上升。LangGraph 本质上解决的是“有状态、可控、可回退”的编排问题这不只是框架偏好而是架构选择。1.1 链式调用在长链路里的失控点最初的实现思路很典型加载文档 → 按固定大小切块 → embedding 入库 → 检索 → 组装 prompt → 调 LLM → 返回。每条链是一个LCEL表达式看起来干净利落rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() )单轮问答没问题。但一旦加入“问题改写”节点这条链就变成了chain ( {question: rewrite_chain, history: history_chain} | retriever_chain | prompt | llm )如果再加入“检索不到就回退”“来源权限过滤”“敏感词拦截”等分支LCEL 虽然支持RunnableBranch但本质上还是在一条线性管道里塞条件。真正让我崩溃的是两个场景多轮对话里我需要在改写、检索、生成之后把中间结果都记录到日志表里。链式调用里这些中间变量要么被包裹在子链内部要么靠全局缓存传递调试起来像在猜谜。产品要求“生成前先检查检索到的文档是否相关不相关就重新改写一次问题再检索”这种循环逻辑在 LCEL 里几乎没法自然表达。后来我意识到RAG 系统的复杂度不在于“一次性问答”而在于“状态管理”。LangGraph 把每一步变成图上的节点把上下文显式放进 State这条思路才真正匹配知识库场景。1.2 为什么不是直接换 Dify 或自研全部流程热词里很多人问 Dify 知识库流水线。确实Dify 这类平台能把切块、召回、生成配置化搭一个 Demo 无非半小时。但我这次改造的目标不是做个演示而是要在知识库问答里嵌入部门的权限模型还要对接内部单点登录、按项目维度隔离数据。Dify 的元数据过滤在界面上能做但细粒度的“每次检索前动态计算可见范围”这种需求在低代码平台里反而更绕。LangGraph 提供了自由度和可控性的平衡节点就是普通 Python 函数可以在里面写任何逻辑图结构又强制我显式表达流程。这也是我选择迁移而不是换平台的根本原因。2. 改造前架构复盘原先那套 RAG 知识库“能跑但不好养”动手之前先做了个老的架构体检。我把自己写过的 LangChain RAG 代码翻出来发现所有痛点其实都能追溯到三个设计问题。这里我把当初的实现和问题一起列出来如果你也是从类似状态出发应该会有共鸣。2.1 最早一版代码切块、向量化、检索、生成最早的实现技术栈是 LangChain OpenAI 兼容接口 Chroma 向量库。切块用的RecursiveCharacterTextSplitter每块 500 字符、重叠 50。向量化用的text-embedding-3-small。检索用similarity_searchTop-K 取 6。生成模型用gpt-4o-mini后来业务量上来换成了内部托管的推理服务。结构很简洁当时跑通 demo 的兴奋感还在。但这套流程有个隐藏的前提所有问题都是单轮的所有用户权限都一样所有检索结果都可信。这三个前提在企业级知识库里一个都不成立。2.2 三个“基础病”改写不准、状态丢失、排错困难多轮改写不准当时用query_rewriting_chain输入是chat_history latest_question。实际效果飘忽用户问“这个方案的成本呢”改写后经常变成“成本呢”完全没有指代消解。问题不在 LangChain而在 prompt 和状态拼接可当时的状态拼接就藏在 chain 里出了问题只能一层层拆。状态丢失检索出的文档、命中的 section、token 用量这些中间结果在链式调用里没有一个统一容器。做日志时得在子链里用RunnableConfig的 metadata 传递每次传参都要小心翼翼。排错困难最气人的是生产环境里偶发性答错。我想看具体召回了哪些文档发现日志里只有最终生成结果中间检索结果没落库。因为当初写的时候把中间变量都扔在函数作用域里了。2.3 结论需要的不是修修补补是把“过程”变成一等公民这次复盘让我下定决心与其继续在链式结构上堆 try-except 和分支判断不如把整个交互过程显式建模成一张图。节点、状态、边每一样东西都有名字日志天然就跟着 State 走。LangGraph 的核心抽象恰好就是我为这个项目需要的架构。3. LangGraph 的心智模型把 RAG 知识库当作一张流程图LangGraph 刚上手的时候容易被它的抽象绕晕。我当时看了半天官方文档脑子里一直在想“这不就是把 LangChain 的 chain 加上循环吗”后来我发现真正的关键是换一套思考方式把一次知识库问答看成一场有状态的工作流。3.1 节点Node就是一次决策或动作LangGraph 的节点就是一个普通函数输入是当前 State输出是 State 的部分更新。对应到 RAG 知识库我会把“问题改写”“文档检索”“生成回答”都定义成节点。每个节点只做一件事方便单独测试和复用。3.2 边Edge不只是串联更是条件路由在 LangChain 里边是隐式的一个 Runnable 的输出是另一个 Runnable 的输入。在 LangGraph 里边明显分成两种普通边上一个节点跑完直接进下一个节点。条件边根据当前 State 的内容决定下一跳是哪个节点。这一下就让“检索结果不合格就重新改写”这类需求变得非常自然不是靠 if-else 堆在函数里而是图里面的一个条件分支。3.3 状态State是贯穿全程的数据总线LangGraph 的 State 是一个 TypedDict。所有节点共享这同一个 State 对象。每个节点可以从 State 读数据也能往 State 写数据。最关键的机制是 reducer默认情况下节点返回的字段直接覆盖原字段但如果你想做“合并”比如把多路检索结果叠加到一个列表里可以用operator.add这类自定义 reducer。这个机制解决了我之前状态丢失的问题。所有中间结果都在 State 里调试的时候打印 State 等于拿到了整个流程的全貌。3.4 与 LangChain 组件怎么共存改造时我比较担心是不是要推翻所有 LangChain 的生态组件答案是不用。LangGraph 的节点里可以直接调用 LangChain 的 retriever、prompt template、LLM 封装。迁移成本其实主要在“流程编排”这层而不是“基础组件”这层。比如原来retriever | format_docs | prompt | llm是一条 LCEL 链现在变成在generate_node里按顺序写几行 Python 代码。这样反而更直观。4. 分步改造实操从 Chain 到 Graph 的具体落地下面是比较核心的部分我把改造后的代码骨架完整贴出来。这不是 demo是生产项目里能跑的那版只是隐去了业务字段。4.1 定义 State把所有过程数据显式装好from typing import Annotated, TypedDict import operator class RagState(TypedDict): # 会话历史 messages: list # 用户最新问题 query: str # 多轮改写后的问题 rewritten_query: str # 检索到的文档列表允许多个节点累加写入 retrieved_docs: Annotated[list, operator.add] # 是否需要走检索逻辑 need_retrieval: bool # 拼给最终生成模型的上下文 context: str # 最终回答 answer: str # 引用来源 sources: list这里我特别说明一个设计点retrieved_docs我用了operator.add作为 reducer。这样即使后续有多个检索节点并行执行返回的文档也会自动拼到同一个列表里而不是互相覆盖。如果有些字段你想保持“最后一次写入生效”则不声明 reducer默认行为就是覆盖。4.2 检索节点把 LangChain retriever 包进 Nodedef retrieve_node(state: RagState): query state[rewritten_query] or state[query] docs retriever.invoke(query) # 这里可以加日志、权限过滤、元数据辅助 return {retrieved_docs: docs}之所以要包一层节点而不是直接用原来的 LCEL是因为我想在检索前后做很多事情记录耗时、过滤用户无权访问的 doc metadata、统计命中来源。直接在节点里写原生 Python远比比在链式调用里插回调函数清爽。4.3 问题改写节点多轮对话的指代消解def rewrite_node(state: RagState): if not state[messages]: return {rewritten_query: state[query]} prompt ChatPromptTemplate.from_messages([ (system, 你是知识库问答的查询改写器只输出改写后的查询。), (human, 对话历史{history}\\n\\n最新问题{query}), ]) chain prompt | llm rewritten chain.invoke({ history: format_history(state[messages]), query: state[query], }) return {rewritten_query: rewritten.content}这里有个经验不要每次都用大模型改写。对于没有指代、没有省略的问题直接原样传递能省掉 200ms 到 500ms 延迟。我在改写的 prompt 里加了规则如果最新问题已经包含足够显式的主体和属性就原样输出。实测下来50% 以上的单轮问题都不会触发无意义改写。4.4 生成节点引用溯源靠 metadatadef generate_node(state: RagState): docs state[retrieved_docs] context \n\n.join([f[{i1}] {d.page_content} for i, d in enumerate(docs)]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的知识库问答助手请根据以下资料回答问题。 若资料无法回答请明确说明。回答末尾列出引用编号。), (human, 资料\\n{context}\\n\\n问题{query}), ]) chain prompt | llm response chain.invoke({context: context, query: state[rewritten_query]}) return { answer: response.content, sources: [d.metadata for d in docs], }引用溯源在知识库场景里必须做。我见过太多 RAG 项目只返回答案不返回来源结果用户发现答案是模型幻觉又无从验证。把d.metadata直接塞进 sources前端渲染时就能显示“参考了哪篇文档、哪个章节”。4.5 组装并编译图先跑通主流程from langgraph.graph import StateGraph, END graph StateGraph(RagState) graph.add_node(rewrite, rewrite_node) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, END) app graph.compile()第一次编译你会发现这个图和原来的 Chain 没本质区别甚至更啰嗦。别急它的优势在加分支的时候才体现出来。5. 检索与索引改造切块、元数据过滤、去重我一起做了光改架构不够RAG 知识库的“内容底座”也得跟着调。这次改造正好把之前积压的三个问题一起处理了切块策略、元数据权限过滤、多路召回去重。5.1 切块策略对比固定窗口、递归字符、父子块我做了三组实验数据来自内部几个大规模 Markdown 文档库大约 2000 篇技术文档。切块方式块大小/重叠Top-6 检索命中率上下文噪声实际感受固定长度切片512/6472%高经常切到表格中间递归字符切块字符分隔符递归81%中基本可用父子块切块父块 2000子块 40089%低检索子块、生成时喂父块最终我选了父子块方案用子块去匹配用户问题召回后合并所属父块内容作为上下文。这样既能精确定位到细节段落又不至于让 LLM 看到割裂的半句话。一个容易忽略的点切块必须保留文档结构的元数据比如标题层级、文件路径、项目编号。否则后面做权限过滤和来源展示时会发现不知道这段内容属于哪个项目。5.2 元数据过滤权限隔离不该在生成阶段才做之前项目的权限过滤是硬编码在 prompt 里的“忽略无权限的文档”。这非常不靠谱LLM 根本不知道哪些文档无权限。LangGraph 改造后我在retrieve_node节点里加载用户信息动态构造元数据过滤条件def retrieve_node(state: RagState): user state.get(user) allowed_projects get_user_projects(user) retriever vectorstore.as_retriever( search_kwargs{ k: 6, filter: {project_id: {$in: allowed_projects}} } ) docs retriever.invoke(state[rewritten_query]) return {retrieved_docs: docs}过滤越早做越好不能等文档都进上下文了再想办法。元数据过滤做在检索前既省钱又防止权限数据污染模型。5.3 去重逻辑LangChain 默认 RRF 实现有坑多路召回比如同时按向量和关键词召回时我最初用EnsembleRetriever自带的 RRF 融合。线上发现两个问题不同召回源的分数量纲不一致直接 RRF 有时会把同一篇文档的两个 chunk 都排到前面造成信息冗余。默认去重逻辑只看page_content完全一致很多文档只是格式差异比如行首空格、换行不同就会被当成两份内容。LangGraph 里我直接写了一个自定义节点做后处理def dedupe_and_rerank_node(state: RagState): docs state[retrieved_docs] seen set() unique_docs [] for d in docs: key normalize_text(d.page_content) if key not in seen: seen.add(key) unique_docs.append(d) return {retrieved_docs: unique_docs}另外我也把融合策略从 RRF 换成了量化分数加权每种检索排序转换成 0 到 1 的归一化分数再按权重相加。虽然 RRF 是经典方法但在我这个场景下显式的分数加权更容易调试可解释性更强。6. 从普通 RAG 升级为 Agentic RAG图结构带来的额外红利迁移图结构之后最明显的好处不是“能跑”而是“想加逻辑就加逻辑”。这一节分享我在主流程基础上扩展的几个 Agentic 分支也是搜索引擎里“agentic rag”这个概念落地的真实样子。6.1 条件分支这个 query 到底要不要检索并不是所有问题都该触发知识库检索。“你好”“谢谢”这种话直接回复就行不需要浪费 embedding 和检索的延迟。我在入口加了意图分类节点def route_by_intent(state: RagState): intent classify_intent(state[query]) if intent chitchat: return direct_reply return rewriteLangGraph 的条件边可以完全根据 State 字段决定走哪个节点。实测在日志里约 10% 的流量被分流到闲聊回复知识库检索压力小了不少。6.2 检索后校验分支不够就重写再检索这是最实用的一个改造。原链路里检索几次都返回空或者低相关文档模型还是会硬答。现在我在 “retrieve” 和 “generate” 之间加了一个grade_node对文档和 query 算一个粗粒度相关分数def grade_node(state: RagState): docs state[retrieved_docs] if not docs: return {need_retrieval: False} relevant is_relevant(state[rewritten_query], docs) if not relevant: if state.get(retry_count, 0) 1: return {need_retrieval: True, retry_count: state[retry_count] 1} return {need_retrieval: False}条件边再根据need_retrieval决定回到rewrite还是进入generate。最多重试一次避免死循环。这样系统不会在明显没答案的问题上硬编答案而是尝试换个角度再查一次仍然没有就诚实地回答“知识库中没有相关文档”用户体验比胡说八道强得多。6.3 并行检索一次查多个索引再合并企业内部的知识库往往散落在 wiki、工单系统、项目文档三个地方。我在 LangGraph 中用SendAPI 同时触发三个检索节点最终汇合到去重节点。并行把总耗时从原来的三次串行 2.4 秒压到了 0.9 秒左右。这一块是 LangGraph 的强项代码结构上只需要在动态边里构建Senddef spawn_retrievers(state: RagState): return [Send(retrieve_wiki, {source: wiki}), Send(retrieve_ticket, {source: ticket}), Send(retrieve_project, {source: project})]注意并行写入retrieved_docs时State 的 reducer 必须支持 append这就是我在第四章强调operator.add的原因。并发写同一个字段时如果 reducer 不是线程安全的可能丢文档。LangGraph 在此处会按入参聚合我踩过并发下文档数量不对的坑最终通过显式 reducer 和日志比对修复。7. 迁移后的实测结果与排错记录最后聊一下迁移后的真实数据以及几个 LangGraph 官方文档写得比较隐晦、但实践里一定会遇到的坑。7.1 我们项目里的效果对比以内部 2000 篇技术文档知识库为例测试集是 300 个典型问题包括多轮追问。对比的是改造前的 LangChain 链式版和改造后的 LangGraph 版指标改造前改造后多轮追问准确率61%83%检索无结果时的正确拒答率低常强行编答案高能明确说不知道端到端平均延迟2.8s1.6s并行检索意图分流引用来源完整率40%95%新增一个分支逻辑耗时半天到一天一小时内延迟下降主要是因为并行检索和意图分流而不是 LangGraph 本身更快。准确率提升则来自检索后校验和重写重试机制。7.2 坑一State reducer 的覆盖/追加语义容易搞混默认情况下节点返回什么State 对应字段就变成什么。很多新手在并行检索时忘了加 reducer后返回的节点直接把先返回的文档覆盖了。排查方式也很直接在每条边返回后打印 State 的retrieved_docs长度。def log_state(state: RagState): print( current state keys:, list(state.keys())) print( retrieved docs:, len(state[retrieved_docs])) return {}把这个调试节点插入到关键节点的上游比看抽象调用栈有用得多。7.3 坑二递归和重试上限必须显式控制LangGraph 允许循环但生产环境里死循环很危险。我建过一个“检查相关度”的子图忘记设置重试上限某次因为 Embedding 模型异常导致检索结果持续不相关系统陷入了无限重试把外部 API 的配额打爆了。后来我所有循环边都带一个retry_count字段并在条件边里判断上限。7.4 坑三不要把敏感配置写进 State这里提醒一句State 里不要放 API Key、数据库连接串之类的敏感配置。State 数据会随图执行被传递到多个节点日志系统也可能记录整个 State。我是用上下文对象单独传递密钥State 里只放业务字段。7.5 我的实操体会踩过这么多坑之后我对 RAG 知识库改造的判断是LangGraph 不一定会让你的回答质量凭空变好但它给了你一个可以持续迭代、随时插拔逻辑的骨架。对需要权限控制、多轮对话、路由分支、检索验证的企业知识库场景从 LangChain 迁过来是值得的如果只是做一个固定文档的问答 demoLangGraph 的复杂度反而是一种负担。最后分享一个小技巧改造完成后我把图结构画成了一张内部文档图每个节点对应一个负责人后续加需求时先改图再改代码。这套“流程即代码”的方式让整个知识库团队从“修链条”变成了“编排工作流”维护成本下降非常明显。