做 RAG 知识库这两年我最大的感受是LangChain 上手很快但真正想把检索流程做得复杂、可控、能应对生产环境它那套链式写法会越来越拧巴。这个项目就是我在已有的 LangChain RAG 知识库基础上整体迁移到 LangGraph 的一次完整改造。文章会直接讲清楚我为什么迁、怎么迁、迁完踩了哪些坑以及实测的数据变化。先交代下背景。我手头这个知识库是面向企业内部的售后技术问答文档来源是产品手册、工单记录、历史运维案例总量在几十万级 token。最早用 LangChain 搭的是经典三段式 RAG查向量库、拼 Prompt、送给大模型回答。用了一段时间业务方开始提各种要求——有些问题需要先判断类型再决定走哪个检索策略有些问题首轮召回了但答案置信度不够要二次检索还有一类工单问题必须走专门的结构化查询流程。这些需求叠加后LangChain 里到处是 if-else、callback、自定义 retriever代码难读到我自己都不想再打开那个文件。于是我把目光转向了 LangGraph。这篇文章不是 LangGraph 的官方文档翻译而是我基于实际项目做的迁移备忘包含完整代码思路、节点设计、条件分支实现以及改造前后延迟、答案质量的对比。适合已经用 LangChain 搭过 RAG、想升级到流程可控状态的新手也适合正在纠结选型的团队做参考。1. 为什么要把 RAG 迁到 LangGraph而不是继续堆 LangChain1.1 LangChain 时代的 RAG 是什么样先说清楚原来用的 LangChain RAG 长什么样。如果你也搭过大概就是这一步from langchain_community.vectorstores import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough retriever vectorstore.as_retriever(search_kwargs{k: 4}) template 根据以下资料回答问题 {context} 问题{question} prompt ChatPromptTemplate.from_template(template) rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() )这套代码清爽、好懂做 demo 和内部小工具完全没问题。RunnablePassthrough()把用户问题传下去retriever异步去查向量库查完的结果和问题一起拼进模板让模型生成答案。但问题出在“这只是个顺序执行链”。它只能按固定顺序执行检索 - 拼装 - 生成。一旦流程里需要出现分支、循环、状态记忆代码就会迅速膨胀。比如我要做“问题分类后走不同检索策略”就需要在链外面先跑一个分类器再根据分类结果走不同的 chain效果类似这样def route_by_type(question): category classify_question(question) if category product: return product_rag_chain elif category maintenance: return maintenance_rag_chain else: return general_rag_chain这样把业务逻辑和链定义混在一起调用链之间互相嵌套调试时非常痛苦因为你看不到中间每一步到底发生了什么。LangChain 的日志输出是线性的想单独观察一次检索结果、重排序结果、Prompt 组装结果都得自己插桩打印。1.2 LangGraph 和 LangChain 的本质区别LangGraph 官方定位是 Low-level orchestration framework。它不打算替代 LangChain 的组件而是把 LangChain 里那些 Runnable、ChatModel、Retriever 当作积木用图结构来编排它们。我记得刚接触 LangGraph 时脑子里最大的转变是把 RAG 从“一根水管”想成“一个有状态的流程图”。LangGraph 的基本抽象只有几个State全局状态对象节点之间传递的数据载体。Node执行具体工作的函数接收上一步状态返回更新后的状态。Edge节点之间的连接决定下一步执行谁。Conditional Edge条件边根据当前状态中的某个值决定分支走向。这种设计天然适配 RAG 的多种复杂场景。举个最简单的例子我想实现“检索结果相关度不够时改写问题重新检索”。在 LangChain 中是很难优雅实现的因为你没法在一个线性链中间插一个循环。而在 LangGraph 里只需要加一个条件边判断状态里存的“结果质量评分”如果低于阈值就回到检索节点再来一轮。我迁移前实测的感受是LangGraph 并没有让单个检索跑得更快它解决的是流程层面的复杂度问题。检索还是那个检索、Embedding 还是那个 Embedding但你拥有了对流程的完全控制权。1.3 改造的第一个决定不要推翻重写这里有个经验想先分享。很多团队看到 LangGraph 的新概念会兴奋想直接把整个 RAG 推倒重写。我的建议是不要。LangGraph 和 LangChain 的组件是兼容的原始代码里最有价值的部分——文档加载、切块参数、向量库选择、Prompt 设计、LLM 调用——这些统统可以原样保留真正需要重写的是“编排层”。我的改造路径是先盘点现有 LangChain RAG 里的组件哪些可以留哪些要换成 LangGraph Node。这样风险最小也能在改造过程中逐步验证每一步的效果不会出现全量替换后系统不可用、连问题出在哪都找不到的窘境。提示LangGraph 不是 LangChain 的替代品而是它的编排层升级。组件复用是迁移的一个核心原则。2. 改造前的知识库架构盘点与优化在动手改之前我花了一天时间把现有 RAG 从“输入端”到“输出端”完整盘了一遍。这一步特别值得做因为很多隐患在没有复杂编排时暴露不出来一旦上了图流程状态传递会把问题放大。2.1 知识库切块我从固定长度换成了递归字符切块我最早用 LangChain 的TextSplitter是固定长度chunk_size500chunk_overlap50当时觉得很省事。但随着知识库文档类型增多问题很快就来了一些产品手册里的表格、代码样例、列表结构被切断造成语义碎片化。比如一个文档中某段代码是从第 499 个字符开始的切块后上半截在 chunk A下半截在 chunk B——检索召回时只拿到一半代码答案自然没法看。后来我换成了RecursiveCharacterTextSplitter它最大的改进是会优先按“段落 - 句子 - 字符”的层级去切尽量让每个块在语义上完整。from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size600, chunk_overlap80, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ], )注意这里我调整了分隔符优先级把中文句号和标点提前了。默认的 separators 是为英文设计的中文文档下不调整它总是优先按英文句点切效果不理想。实测下来同样的知识库召回率提高了大约 7%集中在包含表格和半结构化文本的文档上。这不是 LangGraph 的功劳但却是后来图谱化流程能跑稳的基础——切块质量直接决定检索质量的瓶颈后面所有流程优化都是在缓解这个瓶颈而不是解决它。不同业务场景的切块策略也有差异这里列一下我实测过的选项切块方式适用场景实测效果固定长度切块纯文本、无结构文档效率高但语义易断裂递归字符切块混合型文档手册、告警、工单平衡性最好推荐优先做父子切块Parent-Child大文档、需要保全局上下文时召回更准但存储成本高语义切块代码、日志、强逻辑文档效果最好但计算量较大我当时用父子切块做了个小范围实验父块存整个小节子块做检索检索到子块后回取父块作为上下文。答案完整性确实变好了但代价是向量库规模翻倍对于我当时的场景性价比不高所以最终只对产品手册类文档启用了父子切块工单类文档仍然用递归字符切块。2.2 向量化和检索策略Embedding 选型与混合检索知识库的底座是向量检索Embedding 模型的选择影响非常大。我对比了几个主流方案后综合效果、部署成本、中文支持度最终用的是开源中文 Embedding 模型 FAISS 向量库的组合。主要对比数据如下Embedding 模型中文理解检索准确率我的测试集部署成本OpenAI text-embedding-3-small中上0.81API 按量付费BGE-large-zh很好0.87需 GPU 或高配置 CPUM3E-base好0.83轻量CPU 可跑我最终选了 M3E-base主要是因为它能直接在 CPU 上跑单台 16 核 32G 的服务器就能处理我这几十万级 token 的库不需要额外申购 GPU。如果你的知识库规模大、并发高我会建议加个 GPU 或者直接调 APICP 的负载会明显影响检索延迟。只靠向量检索还有一个痛点关键词精准匹配能力弱。比如产品型号“X200-Pro”如果 Embedding 模型不够敏感检索出的结果可能是一堆带“X200”普通产品的内容反而丢了精确匹配。所以我在 LangChain 阶段就引入了混合检索向量检索 BM25 关键词检索用 RRF 把两路结果融合。这个策略保留到了 LangGraph 阶段实践证明它比单一检索方式稳定得多。2.3 检索后处理重排序是不可省的一环刚搭 RAG 时我有个误区向量库返回 top-k 就直接丢给 LLM觉得大模型自己能从上下文里挑答案。后来被现实教育了——向量返回的 top-k 里有一半是噪音尤其当切块质量不完美时。后面我接了一个 reranker。流程是先用向量和 BM25 各取 20 条候选合并去重后再用 reranker 根据语义相关性重新打分取 top 5 作为上下文。这一步让答案准确率提升非常明显。在 LangGraph 改造时我专门设计的retrieve节点和rerank节点将来中间插入评价逻辑、改写逻辑都是在这些节点之间做文章。3. 从 LangChain 到 LangGraph 的迁移改造实操3.1 改造总览核心流程节点识别迁移的第一步不是写代码而是画清楚流程。我把我现有的 LangChain RAG 拆成了五个关键节点question_analyzer解析用户问题判断问题类型、是否涉及产品型号、是否需要结构化查询等。retrieve根据分析结果决定走哪个检索策略纯向量、混合检索、结构化工单查询。rerank统一对召回结果做重排序截断噪音。generate拼装 Prompt调用 LLM 生成答案。answer_eval对生成的答案做质量判断不达标时回到 rewrite 节点改写问题后重新检索。最终流程是我想要的 agentic RAG 形态不是一次性检索生成而是带反馈闭环的多轮检索。LangGraph 里实现这个闭环并不需要很复杂的语法核心就是 State 节点函数 条件边。3.2 定义状态和节点LangGraph 里的 State 我一开始理解得比较浅以为就是 Python 字典随便装。实际上它需要定义每个字段的更新方式如果声明了 reducer同一个键可以多次被不同节点写入并合并否则新值会覆盖旧值。我的状态定义简略如下from typing import TypedDict, List, Annotated, Operator class RAGState(TypedDict): question: str # 原始问题 rewritten_question: str # 改写后的问题用于二次检索 category: str # 问题类型 retrieved_docs: List[Document] # 检索结果 ranked_docs: List[Document] # 重排序后的文档 answer: str # 最终生成答案 needs_rewrite: bool # 是否进入改写重查节点函数就是普通的 Python 函数接收 state 返回更新后的 dict。例如检索节点def retrieve_node(state: RAGState) - dict: question state.get(rewritten_question) or state.get(question) # 向量检索 vector_results vectorstore.similarity_search(question, k20) # BM25关键词检索 keyword_results bm25_retriever.get_relevant_documents(question, k20) # 合并去重 all_docs merge_and_deduplicate(vector_results, keyword_results) return {retrieved_docs: all_docs}这里有个细节rewritten_question为空时用原始question否则用改写后的问题。这样保证了第一次检索和循环重查走的是同一个节点代码不用复制两份。3.3 图的构建与条件边实现有了节点再把这些节点“粘”成图。LangGraph 的StateGraph定义节点、起点、边和条件边。核心代码如下from langgraph.graph import StateGraph, START, END graph StateGraph(RAGState) graph.add_node(analyzer, analyzer_node) graph.add_node(retrieve, retrieve_node) graph.add_node(rerank, rerank_node) graph.add_node(generate, generate_node) graph.add_node(rewrite, rewrite_node) graph.add_node(eval, eval_node) graph.add_edge(START, analyzer) graph.add_conditional_edges( analyzer, route_by_category, { general: retrieve, maintenance: retrieve, structured_query: structured_query_node, } ) graph.add_edge(retrieve, rerank) graph.add_edge(rerank, generate) graph.add_edge(generate, eval) graph.add_conditional_edges( eval, should_rewrite, { rewrite: rewrite, end: END } ) graph.add_edge(rewrite, retrieve) app graph.compile()注意条件边的函数route_by_category和should_rewrite它们都接收 state返回一个字符串这个字符串对应了边表中定义的走向def route_by_category(state: RAGState) - str: if state.get(category) structured_query: return structured_query return general def should_rewrite(state: RAGState) - str: if state.get(needs_rewrite) and state.get(rewrite_attempts, 0) 2: return rewrite return end这就是 LangGraph 控制力的来源。在 LangChain 里要写一堆 if-else 的地方现在全部变成了图上的条件边通过纯函数来表达独立可测。提示条件边函数里的返回值必须和映射表的键严格对应少一个都会在运行时报错我自己踩过这个坑后面还会细说。3.4 从 Simple RAG 到 Agentic RAG 的思维跃迁现在很多人提到 agentic RAG其实质就是让 RAG 流程具备“自我审视”和“动态路径规划”能力。理论上你可以用 LangChain 强行实现但返回值、回调、流式处理都很难受。在 LangGraph 里agentic RAG 只是一个小闭环。拿我做的eval节点举例def eval_node(state: RAGState) - dict: answer state.get(answer, ) question state.get(question, ) judge_prompt f 用以下标准评价答案质量 1. 是否直接回答了用户问题 2. 是否在引用文档中找到了依据不要臆造 如果答案质量不高返回 {needs_rewrite: true, reason: ...} 否则返回 {needs_rewrite: false} 问题{question} 答案{answer} verdict judge_llm.invoke(judge_prompt).content needs_rewrite true in verdict.lower() return {needs_rewrite: needs_rewrite}当答案质量不高时should_rewrite条件边让流程回到rewrite节点改写问题后重新走一遍retrieve - rerank - generate。这在传统 RAG 里是做不到的——传统流程每次回答完就结束了哪怕答案不完整也没有补救机会。3.5 一个多知识库路由的实例再举一个我实际做过的例子业务方有不同来源的知识库——产品文档库、运维工单库、历史案例库。如果所有内容混在一个向量库里检索时不同库之间的内容会互相干扰尤其是工单库和产品文档库的语言风格差异极大。我在分析了这个差异后用 LangGraph 的流程入口做了路由先用analyzer对问题做分类判断问题更可能属于哪类文档然后分别走不同的向量库必要时还可以并行检索后融合。节点设计为graph.add_conditional_edges( analyzer, route_to_database, { product: product_retrieve, ticket: ticket_retrieve, case: case_retrieve, mixed: multi_retrieve, } )每个检索节点内部逻辑相同但连接不同的 vectorstore 实例。这种“按需路由”的结构让我可以针对性调优不同库的检索参数比如工单库因为文本短切块策略就要小一点而产品文档库有长段落需要更大的 chunk_size 配合父子切块。4. 改造过程中踩过的坑与排查实录4.1 State 设计不合理导致的数据丢失第一次跑通 LangGraph 流程时我的 retrieve 节点返回的retrieved_docs总在下一个节点变成空列表。查了很久才发现原因我的 State 定义里retrieved_docs没有声明 reducer导致默认行为是“覆盖”,但我在 retrieve 节点内部合并 docs 时不小心把结果存到了局部变量而不是返回值里返回了空列表给下一个节点。这个问题理解之后很好修from typing import Annotated def merge_docs(left: List[Document], right: List[Document]) - List[Document]: if left is None: return right return list({doc.page_content: doc for doc in left right}.values()) class RAGState(TypedDict): retrieved_docs: Annotated[List[Document], merge_docs]现在每一次节点返回的retrieved_docs都会和新文档合并而非覆盖。经验是所有跨节点传递的字段都要想清楚语义是替换还是累加再用 reducer 明确表达这个意图。4.2 条件边返回值不匹配导致运行时异常LangGraph 在编译图时不会校验条件边映射表只有运行到那条边时才会抛错。我有一次在route_to_database里返回了product_docs但映射表里键是product结果直接抛了InvalidUpdateError类似的异常。排查方式倒是挺直观看异常信息里提示的边名、节点名然后对一下条件边函数返回值和映射表 key。这里建议加一层防御性校验def route_by_category(state: RAGState) - str: category state.get(category, general) allowed {general, product, ticket, case, mixed} if category not in allowed: return general return category4.3 检索链路的维度问题相似度阈值别乱设有些 LangChain 实现里会给向量检索加 score_threshold来过滤低分结果。这个阈值非常敏感我一开始按线上的建议设成 0.6结果召回率暴跌很多有效文档被过滤掉了。调到 0.2 后召回倒是全了但噪音也进来了。在 LangGraph 改造时我干脆把 score_threshold 逻辑挪到 rerank 之后。前面的粗召回阶段宁可多召回k20也不设阈值后面精排阶段再用 reranker 相关性分数做硬截断。这样既保证召回不丢又能控制最终上下文质量。经验值供参考FAISS 的余弦相似度在 0.3~0.5 区间普遍还是有价值的垂直领域文本可能整体偏低按自己测试集校准不要照搬开源项目的数字。4.4 节点的并行执行陷阱LangGraph 支持一个节点连接多个下游节点会并行执行。我试过把retrieve拆成vector_retrieve和keyword_retrieve两个并行节点再汇合这样确实缩短了单次检索时间。但并发带来的坑是如果两个并行节点都要写同一个 state 字段就必须用带 reducer 的 Annotated 类型否则后完成的节点会覆盖先完成的节点。我最初没有加 reducer导致retrieved_docs只有最后完成的那个节点 的数据。加上 reducer 后问题解决。4.5 和 Dify 这类可视化平台的关系改造过程中也有同事问现在有 Dify 这类可视化知识库流水线工具为什么不直接用我用 Dify 搭过一版快速原型说实话大部分常规 RAG 场景它真能搞定尤其是界面化配置知识库、多路检索、简单条件分支这些能力已经做得很成熟了。但我的项目里需要深度定制analyzer的策略、需要和老工单系统做结构化查询打通、还要在每次检索时动态调整检索参数——这些都是 Dify 的“黑盒流水线”难以覆盖的。LangGraph 的价值恰恰在灵活性上。如果你只是要做一个企业内部快速展示用的知识库问答Dify 会是更高性价比的选择如果你的核心需求是流程的可控性、可编程的节点逻辑、和现有系统深度集成那 LangGraph 这条路值得走。5. 改造效果与性能对比5.1 延迟变化不是变快了而是更稳了我迁移前最担心的是引入图和条件边会增加延迟。实际测试下来单轮问答的端到端延迟从平均 3.8 秒变成了 4.2 秒多了约 400ms主要开销来自多一轮 LLM 调用的分类节点。但这个增加换来的是显著的质量提升。对 300 条测试问题做了评测结果如下评测指标改造前 LangChain RAG改造后 LangGraph RAG首轮答案准确率68.3%74.7%重查后最终准确率68.3%81.3%关键信息缺失率23%12%端到端平均延迟3.8s4.2s多轮上下文保持能力弱中等偏强数据说明一个很关键的点LangGraph 的改造提升主要在“能补救”的能力上。以前一次检索失败就直接给错误答案现在通过质量评估和循环重查把一部分失败场景救回来了。5.2 可维护性的巨大提升相比指标数字我更看重的是日常迭代体验。改动前调一个检索策略要顺着 chain 的 Lambda 一层一层找现在改策略只需要替换一个节点函数或者改一条条件边其他部分完全不动。也是这个项目的实际观察LangGraph 的官方可视化工具能直接看到每一步的状态流转——这个节点输入了什么、输出了什么、为什么走了那条分支全部一目了然。这在 LangChain 时代是不敢想的。做知识库问答的团队一定明白错误的答案最难受的是不知道模型为何这么答而图式流程至少给了你一条清晰的路径去回溯。5.3 后续还可以继续扩展的方向改造完成后我又在做三件延伸的事一是把eval节点的判定从 LLM 硬判定改成规则 LLM 混合模式对高频问题走规则判定减小延迟和成本。二是把知识库更新流程也纳入图谱。现在文档的增删改是通过离线脚本做的后续准备让 LangGraph 管理“文档更新触发 - 增量切块 - 向量库更新 - 索引校验”这条链路。三是尝试引入 ontology 概念把知识库里文档之间的关系构建成语义图谱让检索时不只是相似度匹配还能沿关系链路做多跳检索。这类需求复杂度高正好是 LangGraph 编排能力发挥价值的场景。我个人的体会是LangGraph 和 LangChain 不是对立关系LangGraph 是 LangChain 编排能力的升维。这次迁移最值得的投入是我重新认真梳理了整个 RAG 的流程把原来隐藏在链式代码里的状态转换全部显性化了。如果你也在用 LangChain 做知识库觉得流程越写越复杂、分支越来越多我建议找个周末把流程图画出来再对比下 LangGraph 的状态机思维大概率你也会有和我一样的感受早该迁了。
