企业官网建设流程全解析

做 RAG 知识库这两年，我最大的感受是：LangChain 上手很快，但真正想把检索流程做得复杂、可控、能应对生产环境，它那套链式写法会越来越拧巴。这个项目就是我在已有的 LangChain RAG 知识库基础上，整体迁移到 LangGraph 的一次完整改造。文章会直接讲清楚我为什么迁、怎么迁、迁完踩了哪些坑，以及实测的数据变化。

先交代下背景。我手头这个知识库是面向企业内部的售后技术问答，文档来源是产品手册、工单记录、历史运维案例，总量在几十万级 token。最早用 LangChain 搭的是经典三段式 RAG：查向量库、拼 Prompt、送给大模型回答。用了一段时间，业务方开始提各种要求——有些问题需要先判断类型再决定走哪个检索策略，有些问题首轮召回了但答案置信度不够要二次检索，还有一类工单问题必须走专门的结构化查询流程。这些需求叠加后，LangChain 里到处是 if-else、callback、自定义 retriever，代码难读到我自己都不想再打开那个文件。

于是我把目光转向了 LangGraph。这篇文章不是 LangGraph 的官方文档翻译，而是我基于实际项目做的迁移备忘，包含完整代码思路、节点设计、条件分支实现，以及改造前后延迟、答案质量的对比。适合已经用 LangChain 搭过 RAG、想升级到流程可控状态的新手，也适合正在纠结选型的团队做参考。

1. 为什么要把 RAG 迁到 LangGraph，而不是继续堆 LangChain

1.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_size=500，chunk_overlap=50，当时觉得很省事。但随着知识库文档类型增多，问题很快就来了：

一些产品手册里的表格、代码样例、列表结构被切断，造成语义碎片化。比如一个文档中某段代码是从第 499 个字符开始的，切块后上半截在 chunk A，下半截在 chunk B——检索召回时只拿到一半代码，答案自然没法看。

后来我换成了RecursiveCharacterTextSplitter，它最大的改进是会优先按“段落 -> 句子 -> 字符”的层级去切，尽量让每个块在语义上完整。

from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=600, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", "；", ";", "，", ",", " ", ""], )

注意这里我调整了分隔符优先级，把中文句号和标点提前了。默认的 separators 是为英文设计的，中文文档下不调整，它总是优先按英文句点切，效果不理想。

实测下来，同样的知识库，召回率提高了大约 7%，集中在包含表格和半结构化文本的文档上。这不是 LangGraph 的功劳，但却是后来图谱化流程能跑稳的基础——切块质量直接决定检索质量的瓶颈，后面所有流程优化都是在缓解这个瓶颈，而不是解决它。

不同业务场景的切块策略也有差异，这里列一下我实测过的选项：

我当时用父子切块做了个小范围实验：父块存整个小节，子块做检索，检索到子块后回取父块作为上下文。答案完整性确实变好了，但代价是向量库规模翻倍，对于我当时的场景性价比不高，所以最终只对产品手册类文档启用了父子切块，工单类文档仍然用递归字符切块。

2.2 向量化和检索策略：Embedding 选型与混合检索

知识库的底座是向量检索，Embedding 模型的选择影响非常大。我对比了几个主流方案后，综合效果、部署成本、中文支持度，最终用的是开源中文 Embedding 模型 + FAISS 向量库的组合。

主要对比数据如下：

我最终选了 M3E-base，主要是因为它能直接在 CPU 上跑，单台 16 核 32G 的服务器就能处理我这几十万级 token 的库，不需要额外申购 GPU。如果你的知识库规模大、并发高，我会建议加个 GPU 或者直接调 API，CP 的负载会明显影响检索延迟。

只靠向量检索还有一个痛点：关键词精准匹配能力弱。比如产品型号“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 拆成了五个关键节点：

1. question_analyzer：解析用户问题，判断问题类型、是否涉及产品型号、是否需要结构化查询等。
2. retrieve：根据分析结果，决定走哪个检索策略（纯向量、混合检索、结构化工单查询）。
3. rerank：统一对召回结果做重排序，截断噪音。
4. generate：拼装 Prompt，调用 LLM 生成答案。
5. 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, k=20) # BM25关键词检索 keyword_results = bm25_retriever.get_relevant_documents(question, k=20) # 合并去重 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 category

4.3 检索链路的维度问题：相似度阈值别乱设

有些 LangChain 实现里会给向量检索加 score_threshold，来过滤低分结果。这个阈值非常敏感，我一开始按线上的建议设成 0.6，结果召回率暴跌，很多有效文档被过滤掉了。调到 0.2 后，召回倒是全了，但噪音也进来了。

在 LangGraph 改造时我干脆把 score_threshold 逻辑挪到 rerank 之后。前面的粗召回阶段宁可多召回（k=20）也不设阈值，后面精排阶段再用 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 条测试问题做了评测，结果如下：

数据说明一个很关键的点：LangGraph 的改造提升主要在“能补救”的能力上。以前一次检索失败就直接给错误答案，现在通过质量评估和循环重查，把一部分失败场景救回来了。

5.2 可维护性的巨大提升

相比指标数字，我更看重的是日常迭代体验。改动前调一个检索策略，要顺着 chain 的 Lambda 一层一层找；现在改策略只需要替换一个节点函数，或者改一条条件边，其他部分完全不动。

也是这个项目的实际观察，LangGraph 的官方可视化工具能直接看到每一步的状态流转——这个节点输入了什么、输出了什么、为什么走了那条分支，全部一目了然。这在 LangChain 时代是不敢想的。做知识库问答的团队一定明白，错误的答案最难受的是不知道模型为何这么答，而图式流程至少给了你一条清晰的路径去回溯。

5.3 后续还可以继续扩展的方向

改造完成后，我又在做三件延伸的事：

一是把eval节点的判定从 LLM 硬判定改成规则 + LLM 混合模式，对高频问题走规则判定，减小延迟和成本。

二是把知识库更新流程也纳入图谱。现在文档的增删改是通过离线脚本做的，后续准备让 LangGraph 管理“文档更新触发 -> 增量切块 -> 向量库更新 -> 索引校验”这条链路。

三是尝试引入 ontology 概念，把知识库里文档之间的关系构建成语义图谱，让检索时不只是相似度匹配，还能沿关系链路做多跳检索。这类需求复杂度高，正好是 LangGraph 编排能力发挥价值的场景。

我个人的体会是：LangGraph 和 LangChain 不是对立关系，LangGraph 是 LangChain 编排能力的升维。这次迁移最值得的投入是，我重新认真梳理了整个 RAG 的流程，把原来隐藏在链式代码里的状态转换全部显性化了。如果你也在用 LangChain 做知识库，觉得流程越写越复杂、分支越来越多，我建议找个周末把流程图画出来，再对比下 LangGraph 的状态机思维，大概率你也会有和我一样的感受：早该迁了。