企业官网建设流程全解析

把 RAG 应用从“一条链”改造成“一张图”，是我最近在 LangChain 生态里折腾 LangGraph 时最有体感的一件事。RAG 在 LangChain 中的实现，前面几篇用的基本都是 Chain 和 Runnable 的线性组合；到这篇，我直接把流程重构为 LangGraph 的有向图，让检索、生成、判断这些环节可以被显式编排，甚至支持循环和回退。如果你写过几条 RAG 链路，又觉得分支逻辑越来越难维护，那这篇文章正好适合你。我会把从零搭一个多轮 RAG 的完整过程、踩过的坑、以及生产环境里值得留意的设计取舍一次讲清楚。

1. 为什么 RAG 应用要选 LangGraph，而不是继续堆 Chain

1.1 线性 Chain 在复杂 RAG 场景下的窘境

早期的 RAG 实现通常是一条直线：用户问题进来，先做向量检索，把命中的文档片段拼进 Prompt，再交给大模型生成回答。用 LangChain 的 Chain 表达时，看起来非常简洁，代码只有几十行。可一旦进入真实业务，需求很快就会变得不“线性”：

• 可能需要先判断这轮问题需不需要检索，比如闲聊“你好”就直接答复，不查知识库；
• 检索结果可能为空，这时最好重写一遍问题再做一次召回；
• 多轮对话里，用户说的可能是“它性能怎么样”，这里的“它”要结合历史问题才能检索；
• 回答生成后可能还想校验一遍：上下文里是否真的有依据，没有依据就改走兜底话术。

这些逻辑塞进一条 Chain 里不是不行，但要不断用自定义函数拆步骤、塞 if、改 Runnable 分支。写出来的东西看起来还能跑，实际维护起来非常痛苦，排查问题时只能靠日志一步步猜，流程“死”在哪里全靠运气。我踩过几次这种坑之后，开始转向让控制流显式化的方案。

1.2 LangGraph 对 RAG 的真正价值：状态机加工作流

LangGraph 做的事情，本质上是把“业务逻辑节点”和“节点之间的连接关系”分开。节点就是普通函数，负责检索、生成、改写这类具体动作；边决定下一步走哪里，条件边则根据状态动态跳转。这样一来，RAG 从“一条预设好的直线”变成了“一张可以被运行和调试的图”。

对 RAG 来说，LangGraph 最有用的两个能力是循环和有状态。循环意味着检索质量不行时，可以沿着条件边重新回到检索节点再试一次，而不是必须一次性成功。有状态意味着整个图可以保存中间结果，多轮对话不再需要把所有历史一股脑塞进 Prompt，检查点机制会把每一步状态记下来，下次接着走。这种设计思路其实很像在工作流引擎里跑一个数据处理流程，只是节点变成了模型调用和检索调用。

1.3 不是所有 RAG 都需要上 LangGraph

这里得说句公道话：如果只是做一个“给某个文档做问答”的演示，用最朴素的 Chain 反而更快。LangGraph 的价值要在流程复杂之后才体现出来，它带来的额外成本是概念更多、调试链路更长、编译执行也多了一层。

我个人的判断标准很简单：只要流程里出现了一个以上的 if 分支，或者需要记录对话状态，我就会考虑用 LangGraph。如果只是固定顺序的检索加生成，老老实实写 Chain。工程上一个经验是，重构之前先在白板上把流程图画一遍，节点超过三四个、存在回退路径的时候，用 LangGraph 一定是更稳的选择。

2. 核心概念拆解：状态、节点、边与检查点

2.1 State：在节点之间传递和更新的“背包”

LangGraph 里最核心的概念是 State。它本质上是一个 TypedDict，定义这张图运行时所有节点共享的数据结构。我把常见 RAG 应用的 State 写成这样：

from typing import TypedDict, List class RAGState(TypedDict): question: str # 原始问题 history: List[dict] # 多轮对话历史 rewritten_question: str # 改写后的问题 documents: List[str] # 检索到的文档片段 answer: str # 最终答案 need_fallback: bool # 是否需要兜底回答

可以把它想成一个“背包”，每个节点从包里读自己需要的东西，再把想给后面节点用的东西塞回去。注意，LangGraph 的状态更新不是直接改字典，而是每个节点返回一个“要更新的部分”，图和已有状态做合并。

这里有三个容易踩的细节：第一，只返回你真正要修改的 key，不要试图把整个 State 重新构造一遍；第二，如果某个字段是列表，默认更新方式是覆盖，想要追加历史需要用到Annotated和operator.add；第三，State 中不要塞模型对象或不可序列化的东西，尤其当你要用检查点机制持久化时，这些对象会成为隐患。

2.2 Node：真正干活的普通函数

在图里，节点就是一个 Python 函数，输入是当前 State，输出是一个字典，就是这个节点想要更新的字段。LangGraph 对节点几乎没有魔法限制，你可以调用任意工具、查询数据库、调模型、做计算。

我之前刚用 LangGraph 时，容易把节点写得又大又杂，一个节点里既查库又拼 Prompt 还调模型。后来发现，节点拆得越细，调试和复用越方便。比如检索节点只负责返回 documents，生成节点只负责返回 answer，后面加一个校验节点也只返回 need_fallback。这样单个节点可以用传统函数测试，逻辑非常清楚。

def retrieve_node(state: RAGState) -> dict: docs = retriever.invoke(state["question"]) return {"documents": [d.page_content for d in docs]} def generate_node(state: RAGState) -> dict: answer = rag_chain.invoke({ "question": state["question"], "context": state["documents"] }) return {"answer": answer}

2.3 Edge 和 Conditional Edge：控制流的关键

有了节点还不够，必须规定节点之间的执行顺序。普通边用add_edge，表示无条件固定走下一步。条件边用add_conditional_edges，会根据函数返回值动态决定去向。

比如检索之后，如果 documents 为空，可以走 rewrite 节点；如果有内容，就直接生成。代码大致是：

def route_after_retrieve(state: RAGState) -> str: if not state["documents"]: return "rewrite" return "generate"

我最初犯过一个错误，以为条件边返回的是布尔值或者节点本身。实际上它返回的是字符串，这个字符串必须是图中已经注册过的节点名，或者使用END常量。如果写错，编译期不一定报错，运行时才会报“找不到目标节点”，这个问题很隐蔽。

2.4 Checkpointer：给图加上记忆能力

多轮对话是 RAG 应用逃不开的场景。LangGraph 的 Checkpointer 解决的就是“状态持久化”问题，它把每次图的执行结果保存下来，下次带上同一个会话标识继续执行。

最轻量的方案是MemorySaver，直接把状态放内存。生产环境建议用持久化存储，比如SqliteSaver或PostgresSaver。使用时，每次 invoke 都要传一个配置，里面包含thread_id，相当于会话 ID：

config = {"configurable": {"thread_id": "session-001"}} result = app.invoke({"question": "它性能怎么样"}, config=config)

注意，thread_id一定要按会话区分，否则不同用户会互相串上下文。这是我上线时踩过最狠的坑之一。

2.5 三种实现方式对比：Chain、Agent、LangGraph

我用一个表把这几种方式放在一起对比，方便选型：

对我来说，LangGraph 最舒服的一点是：模型只负责单个节点的推理，流程决策掌握在代码逻辑手里。对生产系统来说，这比完全交给模型自由发挥要安全得多。

3. 实操过程：从零编写一个多轮 RAG 应用

3.1 环境准备与依赖安装

先用 pip 装好依赖：

pip install langchain langgraph langchain-community langchain-chroma

这个组合用到的模块包括：文档加载、文本拆分、向量库、LangGraph 运行时。模型调用部分，我这里假设你已经在用一套兼容 OpenAI 接口的模型服务，环境变量按你自己的地址配置即可。嵌入模型同理，用你现有的向量模型就可以。

准备一个测试语料，我通常用一个简单的文本文件，内容可以是最新产品说明之类的内部资料，替换成你自己的文档即可。加载和切分：

from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter loader = TextLoader("kb_docs.txt") documents = loader.load() splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50 ) texts = splitter.split_documents(documents)

3.2 定义向量检索器

继续把这些切分好的文本写入向量库。这里我用的是 Chroma 这类本地向量库，方便演示；实际接入线上向量库时，只要对象对外提供invoke或similarity_search方法即可，LangGraph 完全不关心底层实现。

from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings() vectorstore = Chroma.from_documents( documents=texts, embedding=embeddings, persist_directory="./chroma_db" ) retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

向量检索器的k值我没有开很大。知识库召回不是越多越好，片段过多会让 Prompt 变长、响应变慢、噪声变大。4 到 6 是一个比较稳的起点，具体需要根据文档切分粒度调整。

3.3 定义图状态与节点函数

先定义状态结构，再写节点。节点数量这次控制在四个：改写节点、检索节点、生成节点、兜底节点。它们分别负责：

• 改写节点：结合历史问题，把用户当前提问补充为完整上下文；
• 检索节点：用改写后的问题查向量库；
• 生成节点：把检索片段和问题交给模型生成答案；
• 兜底节点：检索为空时返回一个固定话术。

from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, START, END class RAGState(TypedDict): question: str history: Annotated[List[dict], operator.add] rewritten_question: str documents: List[str] answer: str def rewrite_node(state: RAGState) -> dict: # 这里用简单规则：如果当前问题太短，就结合上轮问题拼接 # 更复杂可以用 LLM 做改写，这里先保持逻辑透明 if len(state["question"]) < 4 and state["history"]: rewritten = f'{state["history"][-1]["question"]}：{state["question"]}' else: rewritten = state["question"] return {"rewritten_question": rewritten} def retrieve_node(state: RAGState) -> dict: docs = retriever.invoke(state["rewritten_question"]) return {"documents": [d.page_content for d in docs]} def generate_node(state: RAGState) -> dict: prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的客服助理，只能根据给定资料回答，资料不足时直接说明。"), ("human", "资料：\n{context}\n\n问题：{question}") ]) chain = prompt | llm | StrOutputParser() answer = chain.invoke({ "context": "\n".join(state["documents"]), "question": state["rewritten_question"] }) return {"answer": answer} def fallback_node(state: RAGState) -> dict: return {"answer": "抱歉，资料库里没有找到相关信息，请换个说法再问一次。"}

3.4 组装图、配置条件边并编译

接下来是关键部分：把节点放进 StateGraph，连接边的走向，条件边决定检索后是去生成还是兜底。

def route_after_retrieve(state: RAGState) -> str: if not state["documents"]: return "fallback" return "generate" builder = StateGraph(RAGState) builder.add_node("rewrite", rewrite_node) builder.add_node("retrieve", retrieve_node) builder.add_node("generate", generate_node) builder.add_node("fallback", fallback_node) builder.add_edge(START, "rewrite") builder.add_edge("rewrite", "retrieve") builder.add_conditional_edges( "retrieve", route_after_retrieve, { "generate": "generate", "fallback": "fallback" } ) builder.add_edge("generate", END) builder.add_edge("fallback", END) app = builder.compile()

编译后，可以直接调用。第一次执行传入问题和空历史：

result = app.invoke({ "question": "数据同步失败怎么处理？", "history": [] }) print(result["answer"])

执行结果里可以看到rewritten_question、documents、answer这些字段都被填充，流程直接按我们设定的路径走完。

3.5 接入多轮对话记忆

上面的状态里已经定义了 history 字段，但真正要支持多轮，还需要用检查点把每一轮结果保存下来。我用最轻量的MemorySaver，并指定thread_id：

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-1001"}} # 第一轮 app.invoke({ "question": "你们支持哪些部署方式？", "history": [] }, config=config) # 第二轮问“有安全限制吗？” 这里的“它”需要联系上一轮 result = app.invoke({ "question": "有安全限制吗？", "history": [] }, config=config)

这里有一个容易踩的坑：每轮调用如果你都把 history 传空，检查点机制确实会自动保存状态，但你必须自己在节点里把历史取出来拼进上下文，否则模型看到的仍只是当前问题。正确做法是在状态里维护一个messages字段，或者像我之前一样，用Annotated把每轮问答追加进历史列表，再在生成节点里把所有历史信息拼进 Prompt。

4. 进阶玩法：让 RAG 具备路由、改写与可验证能力

4.1 增加问题类型路由节点

很多实际场景里，不是每个问题都需要走同样的流程。比如“你好”“感谢”这类寒暄，直接答一句就行；“这个功能怎么收费”可能根本不需要检索。在 RAG 图里加一个路由节点，先用轻量模型分类，再走不同分支：

def route_by_intent(state: RAGState) -> str: # 实际可以调用分类模型，这里用规则演示 if state["question"].startswith("你好"): return "chitchat" return "search"

这种路由的价值在于减少不必要的向量检索和模型调用，既省成本也避免噪声。生产里我用关键词和模型分类混合的方式，先用规则兜底，拿不准的再交给小模型，效果比较稳，也不会把流程拖慢太多。

4.2 检索为空时自动改写重试

我在第 3 节里做了一个简单 rewrite 节点，但它只是拼接历史，没有真正利用模型改写。更实用的做法是：第一次检索结果为空，进入改写节点，调用大模型把问题扩写为多个角度的查询词，再次检索，如果还是为空才走兜底。

这样的循环依赖 LangGraph 的条件边非常自然。需要注意改写节点要设置重试次数上限，我通常用状态里的retry_count字段控制，超过两次就不再循环，避免死循环导致超时。循环控制在两轮以内的原因是：第一次为空，多数是表述问题，改写后往往能命中；连续两轮为空，基本就是知识库里真没有，没必要继续烧 Token。

4.3 把节点设计成可替换接口

我发现不少做 RAG 的人一开始就把 LLM 调用直接写死在节点里，换个模型要改好几个函数。更好的做法是每个节点只依赖接口约定，Model 的提供方尽量从外部注入。比如检索节点面对的是一个“能根据 query 返回文本列表”的对象，生成节点面对的是一台“能接收消息列表并输出文本”的模型对象。

这样带来的好处很明显：换向量库、换模型服务、换提示词模板，都不需要改动图结构。节点逻辑和模型实现解耦之后，整个 RAG 应用很容易做 A/B 测试，同一套流程搭配不同模型直接对比效果。

4.4 让回答带上可验证的资料出处

企业场景里，RAG 回答没有出处，基本不会被信任。我会在状态里增加一个source_nodes字段，把每条检索到的片段对应的文件名或章节信息一并带回。生成节点里要求模型在答案末尾列出参考资料，最终展示结果时再把原始片段和答案放在一起参与校验。

这一步在 LangGraph 里只需要在 retrieve 节点多返回一个字段即可，不需要额外边，也不影响主流程。但确实会让生成节点更有说服力，后续做质量评估时也更容易定位是哪一条资料带偏了答案。

5. 常见问题与排查经验实录

5.1 节点更新不生效，状态“没变”

这类问题最常见的症状是：节点明明 return 了answer，下一步读取时却是空值。原因多半是 State 里没有定义answer字段，或者字段类型对不上。LangGraph 的 State 更新是浅合并，如果你返回的字典里有一个 key 没有被定义在 TypedDict 中，图可能不会报错，但等于没更新。

排查办法比较笨但有效：在节点入口和出口各打印一遍相关字段，看返回结果是否真的进了下一个节点。不要凭感觉猜，图流程一旦分支多起来，“哪个节点更新了哪个字段”必须画清楚。

5.2 条件边运行时提示找不到节点

出现InvalidUpdateError: no such node，基本都是add_conditional_edges第三个映射字典的 value 写错了。这个 value 必须是已经注册的节点名，或者END。要注意如果你把END写成了字符串"END"，运行时就真的会去找一个叫 “END” 的节点，然后报错。正确做法是在文件顶部from langgraph.graph import END，然后映射到END常量。

5.3 多轮对话中上下文越来越乱

如果发现第二轮开始模型回答“跑题”，大概率是历史拼接出了问题。一种情况是所有轮次都传了完整历史，导致 Prompt 越来越长，模型抓不住重点；另一种情况是历史记录只追加了用户问题，没有追加助手回答，模型看到的对话是断的。

正确的做法有两种。要么像我在第 3 节里那样，用Annotated[list, operator.add]积累消息，在生成节点里把最近几轮拼进 Prompt，超过阈值就裁剪；要么直接用 LangGraph 的消息列表状态配合模型绑定，让模型自己处理上下文。前者比较可控，后者省事。

5.4 并发会话互相串台

本地调试用 MemorySaver 时，两个会话如果使用同一个thread_id，数据就会混在一起。这其实是使用者的问题而不完全是框架的问题。处理方式是确保每个用户一个唯一的thread_id，如果是在分布式服务里，要改用共享存储的 CheckpointSaver，比如基于数据库的存储，否则进程重启后所有内存状态直接丢失。

5.5 常见问题排查速查表

这张表我每次搭建新的图都会对照检查一遍，省了不少调试时间。

6. 写在最后：把这个项目往前推进的几个方向

这一系列走到第五篇，我在实际项目里最大的体会是：RAG 的终点不是“能回答问题”，而是“能稳定地、可控地、带着证据地回答问题”。LangGraph 真正帮到我的地方，不是让代码看起来更酷，而是把流程里散落的判断逻辑显式地画了出来。图一旦画清楚，后续加路由、加校验、加记忆，都只是在原有结构上增加节点和边，不需要推翻重来。

如果你正在从传统 Chain 迁移到 LangGraph，我的建议是先拿一条最简单的 RAG 链路做试点，把它拆成检索、生成两个节点，加上一个条件边感受一下“图”的执行方式。等习惯了状态传递和条件路由，再逐步加入改写、校验、多轮记忆这些能力。

最后分享一个不起眼但很实用的技巧：在正式跑大模型之前，先把所有节点用“假数据”调试一遍，比如检索节点直接返回固定片段，生成节点直接返回一个测试字符串。这样能先跑通整个图的逻辑链路，再接入真实模型，排查问题会容易非常多。