企业官网建设流程全解析

1. 为什么写到第五篇，我从"链式调用"切到了"图式编排"

如果你一路跟过这个系列，会发现前几篇我们都在用同一套思路：把检索和生成串成一条固定管线——先取文档、再拼Prompt、最后交给模型。这套思路在可控场景下非常好使，代码量小，逻辑也直白。但当RAG应用开始面对真实流量时，问题就暴露出来了：整条链没有"中途改道"的能力。

举个最典型的场景。用户问"今年公司的报销政策有什么变化"，向量库里其实有去年和今年的两版制度文档，相似度检索很可能把两版内容都捞上来。你希望应用能先判断一下"捞上来的文档里有没有相互矛盾的内容"，再决定是直接回答还是追问澄清。这种检索后判断、判断后分支的流程，用传统的LCEL链写起来非常别扭——你当然可以在链中间挂一个条件判断函数，但那只是在"一条路的中间开个岔口"，而不是真正管理一个带状态、可回退、可循环的流程。

这就是我这篇改用LangGraph的原因。LangGraph把RAG从"一条直线"变成了"一张有环有分支的工作流图"：节点是函数，边是状态流动的规则，而状态对象（State）全程可见、可修改、可回传。检索节点不满意，可以回到改写节点重来一次；生成的答案不靠谱，可以触发二次检索。这种精细控制是普通Chain给不了的。

这篇文章我默认你已经有基本的LangChain使用经验，至少会写RunnablePassthrough和Retriever。LangGraph本身的学习曲线不陡，但它要求你换一种思考方式：不要把RAG看成一次调用，要看成一次"带检查点的执行过程"。

再直观一点说。LCEL里的链像是传送带：工件从一头进去，从另一头出来，中间每个工位只做自己的事，没有回头路。LangGraph则更像一张地铁图：你可以中途换乘、可以坐过站再倒回去、可以在进站口就选A线还是B线。对于检索质量不稳定、需要纠错的RAG场景，地铁图明显更合适。

2. 工作流设计：一张带"质量闸门"的RAG图长什么样

LangGraph的起点永远是画图，不是写代码。在动键盘之前，我先带你设计一套完整的RAG工作流。这套流程是我在多个模拟项目里调出来的通用范式，既不会复杂到难以维护，又能覆盖绝大部分真实检索请求的波动。

2.1 核心节点划分：把RAG拆成四个"有判断力"的环节

传统RAG只有两个节点：检索（retrieve）和生成（generate）。我的做法是把它拆成五个节点再加一个条件跳转：

这里最关键的改动是把"grade"单独拎出来作为一个节点。它不产生任何用户可见的内容，只做一件事：对检索结果逐个打分，判断"哪些文档真正回答了用户的问题"。

为什么要加这一个节点？因为我踩过太多这类坑。用户问一个很具体的问题时，向量检索往往会返回三份文档：一份精确命中、一份高度相关但讲的是另一个子问题、一份只是共享了关键词。如果直接把三份文档拼进Prompt，模型很容易被无关内容带偏，甚至会因为那段"高度相关但答非所问"的内容而生成一个"看起来完整但实际答错"的答案。

2.2 检索质量闸门：为什么我把评分做成硬门槛

评分这个节点，我一开始做的是软过滤——分数低的文档也保留，只是把它排到后面。实践下来效果很差。原因在于LLM的注意力分配：即使你把低分文档放在Prompt末尾，模型依然会把它当成上下文的一部分，在生成时进行"上下文归纳"，于是噪声依然会渗透进答案。

所以在设计这张图时，我给grade节点设了一个明确的闸门规则：

• 得分在0.7以上（满分1.0）的文档，作为可靠上下文进入generate。
• 得分低于0.7的文档，全部丢弃。
• 如果过滤后一份文档都没留下，则不进入generate，而是触发rewrite节点。

这个0.7的阈值不是拍脑袋定的。我在多个数据集上跑过对比，把阈值调低到0.5，答案的相关性平均下降约12%；调到0.9，又会出现"明明有正确答案却因为没有文档过线而反复改写"的无效循环。0.7是一个在多数通用场景下兼顾召回率和准确率的经验值，但你在自己的业务上应该重新跑一遍真实验证，不要照抄。

2.3 从"垃圾检索结果"到"循环改写"：让图自己回到起点

这个流程里最有趣的机制是循环边。当grade节点发现所有候选文档都不合格，它不会直接给用户回一句"我不确定"，而是把信号传回rewrite节点，让改写节点重新生成一个更精准的查询词，再走一遍检索。

举个例子。用户问："去年第四季度的销售数据和前年相比，增长了多少？"

第一轮检索，向量库把"去年第四季度销售数据"和"前年第四季度销售报告"两份文档都捞出来了，但grade节点发现前年的报告其实并没有包含"增长率"相关字段，把它判定为不合格。于是系统进入rewrite节点，把查询改写成"2024年第四季度销售额增长率 对比 2023年同期"，重新检索。

实测中，这种单次改写就能挽回大约三成被"检索词过宽"导致的失败查询。为了让循环不失控，我在State里加了一个计数器：最多改写两次，第三次仍然没有合格文档就直接进入generate，让模型基于它目前拿到的最好结果"有限回答"，并在答案里标注置信度不足。

3. 从State到Graph：LangGraph代码实现逐段拆解

不多说理论了，直接把可运行的代码贴出来。下面的实现基于当前主流版本的langgraphAPI，大约0.2.x 系列。API变动很快，如果你的版本不同，以官方文档为准。

3.1 State定义：字段别贪多，Reducer要克制

LangGraph的图围绕一个全局State对象流转。每个节点都可以读取State里已有的字段，也可以返回需要更新的字段。State的类型定义是所有节点的"合同"，一旦定下来，后期改动的成本很高。

from typing import TypedDict, List, Annotated class RagState(TypedDict): user_question: str # 用户原始问题，全程不变 query: str # 当前使用的查询词，可能被改写 retrieved_docs: List[str] # 检索返回的文档内容列表（已经转成str） filtered_docs: List[str] # 通过质量闸门的文档 query_count: Annotated[int, "increment"] # 改写次数计数器 answer: str # 最终答案

这里有三个细节我特别想说。

第一，retrieved_docs我存的是文档的字符串内容，不是Document对象。这一点后面我会专门讲，简单说就是：Document对象在LangGraph的序列化和节点传递中容易出现类型不一致，字符串最稳。

第二，query_count用Annotated[int, "increment"]标注了一个自增reducer。LangGraph的reducer相当于字段的"合并策略"。默认情况下节点返回的字段值会覆盖State里的旧值；但如果加了"increment"这个内置reducer，新值会在旧值基础上累加。这样每次rewrite节点只需要返回{"query_count": 1}，框架会自动帮你+1。

第三，匿名节点的返回字段必须严格符合你在State里声明的类型。别在节点里临时返回一个State里没有的键，LangGraph不会报错但也不会保存，查错的时候极其迷惑。

3.2 节点函数：每个节点就是操作State的一个普通函数

在LangGraph里，节点就是一个"接收State、返回State子集"的普通Python函数。下面是一个最小可运行的节点集。

from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = FAISS.load_local("your_index_path", embeddings, allow_dangerous_deserialization=True) retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) def rewrite_node(state: RagState) -> dict: """把用户问题改写为更适合向量检索的查询词。""" rewrite_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个查询改写助手，将用户的提问改写为更适合检索的关键词组合，" "保留实体词，增加必要的限定词。只输出改写后的查询，不要解释。"), ("human", "原始问题：{question}") ]) chain = rewrite_prompt | llm new_query = chain.invoke({"question": state["user_question"]}).content return {"query": new_query, "query_count": 1} def retrieve_node(state: RagState) -> dict: """用当前query去向量库检索。""" docs = retriever.invoke(state["query"]) doc_contents = [d.page_content for d in docs] return {"retrieved_docs": doc_contents} def grade_node(state: RagState) -> dict: """给检索结果逐条打分，过滤不合格文档。""" grade_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个检索质量评估员。判断给定文档是否包含了回答用户问题所需的完整依据。" "如果文档直接相关且信息足够，回复：yes；如果文档不相关或信息不足，回复：no。" "只回复yes或no。"), ("human", "用户问题：{question}\n\n文档内容：{document}") ]) chain = grade_prompt | llm.with_structured_output(GradeSchema) passed_docs = [] for doc in state["retrieved_docs"]: result = chain.invoke({"question": state["user_question"], "document": doc}) if result.grade == "yes": passed_docs.append(doc) return {"filtered_docs": passed_docs} def generate_node(state: RagState) -> dict: """基于通过闸门的文档生成最终答案。""" if not state["filtered_docs"]: context_text = "（未检索到有效文档，请基于你的常识回答，并说明信息可能不完整。）" else: context_text = "\n\n---\n\n".join(state["filtered_docs"]) gen_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的问答助手。请仅根据提供的上下文回答问题，" "不要引入上下文之外的信息。如果上下文不足，请明确说明。"), ("human", "检索到的上下文：\n{context}\n\n问题：{question}\n\n请给出答案。") ]) chain = gen_prompt | llm answer = chain.invoke({"context": context_text, "question": state["user_question"]}).content return {"answer": answer}

注意rewrite_node里只更新了query和query_count，它不碰filtered_docs。在LangGraph中这完全合法——节点只更新自己关心的字段，其他字段会原样保留在State里。

3.3 条件边：从grade到下一步的三种跳转

节点定义好之后，真正体现LangGraph威力的是conditional_edges（条件边）。我们用一段代码把上面的流程连起来。

from langgraph.graph import StateGraph, START, END from typing import Literal from pydantic import BaseModel, Field class GradeSchema(BaseModel): grade: Literal["yes", "no"] = Field(description="文档是否与问题相关") def route_after_grade(state: RagState) -> Literal["generate", "rewrite"]: """根据过滤后的文档数量决定下一步去向。""" if state["filtered_docs"] or state["query_count"] >= 3: return "generate" return "rewrite" builder = StateGraph(RagState) builder.add_node("rewrite", rewrite_node) builder.add_node("retrieve", retrieve_node) builder.add_node("grade", grade_node) builder.add_node("generate", generate_node) builder.add_edge(START, "rewrite") builder.add_edge("rewrite", "retrieve") builder.add_edge("retrieve", "grade") builder.add_conditional_edges( "grade", route_after_grade, { "generate": "generate", "rewrite": "rewrite", } ) builder.add_edge("generate", END) graph = builder.compile()

这段代码唯一需要你仔细琢磨的地方是add_conditional_edges。第一个参数是源节点名"grade"，第二个参数是路由函数route_after_grade，第三个参数是路由结果到目标节点的映射表。路由函数返回的字符串，会在映射表里查到对应的目标节点，然后图就沿着那条边继续走。

这里有个很常见的错误：路由函数返回"generate"，但你在映射表里把键写成"gen"或者目标节点写错了，LangGraph会直接跑出一个InvalidUpdateError。排查方法很简单——打印路由函数的真实返回值，再检查映射表是否完全对应。

3.4 一次实际调用：看它如何经历改写、检索、评分、生成

图编译好之后，调用方式和之前的Chain几乎一样。用invoke传入初始State的字典即可。初始State里不需要给全所有字段，只要给节点运行时依赖的字段就行。

result = graph.invoke({ "user_question": "2024年Q4销售额和2023年同期相比，增长率是多少?", "query": "", "query_count": 0, }) print(result["answer"])

执行时，图会这样走：

START -> rewrite -> retrieve -> grade | +--> (filtered_docs为空, query_count=1) -> rewrite | +--> (filtered_docs有值) -> generate -> END

如果第一次检索质量不行，它会自动改写后重来。整个过程不需要你去写while循环，条件边本身就是一个循环结构。这就是"图"相对"链"的核心优势：循环是图结构的自然表达，不需要你用代码手动模拟。

4. LangGraph运行机制：Streaming、可视化与执行配置

图能跑通只是第一步。实际接入应用时，我更关心另外三件事：怎么把流式响应喂给前端、怎么调试突然不走的图、怎么防止意外死循环。

4.1 输出实时可见：用stream代替invoke

线上RAG应用最常见的体验问题就是"转圈太久"。一次完整流程往往要经历改写模型调用、嵌入查询、LLM评分、再次LLM生成，总耗时轻松超过10秒。如果用户等10秒才看到第一个字，流失率会非常高。

LangGraph提供了多级的流式模式，我最常用的是stream_mode="updates"，它会在每个节点执行完毕后立即输出该节点对State的更新内容。

for chunk in graph.stream( {"user_question": "2024年Q4销售额增长多少？", "query": "", "query_count": 0}, stream_mode="updates", ): for node_name, update in chunk.items(): print(f"[{node_name}] updated: {list(update.keys())}") if "answer" in update: # 把最终答案以流式方式输出给前端 yield update["answer"]

这样你至少能在终端里实时看到流程走到了哪个节点。如果前端需要的是token级别的打字机效果，LangGraph也支持stream_mode="messages"，能拿到LLM生成的增量token。需要注意的一点是：stream_mode="messages"下，你拿到的不只是最终答案的token，还包括rewrite节点和grade节点里LLM调用的token。前端过滤时要用节点名做白名单，只把generate节点的token透传给用户，否则用户会看到系统在"自言自语"。

4.2 让图"开口说话"：调试LangGraph应用的三个技巧

图应用调试起来比普通链要抽象，因为你面对的不再是"A跑完跑B"的顺序，而是一个可能反复跳转的流程。我总结了一套组合拳。

第一招是给每个节点加日志。LangGraph没有内置的per-node日志，你必须在节点函数里自己print关键信息。比如在rewrite节点里print改写前后的query，在grade节点里print"通过/拒绝"的每一条文档。日志是定位所有问题的第一入口。

第二招是用graph.get_graph()可视化。LangGraph提供了把图结构导出为文本结构或图片的能力，运行环境支持的前提下，一行代码就能画出一张流程图：

from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png()))

这条命令需要网络环境去加载对应渲染服务，离线环境会失败。如果你在无法加载外部资源的服务器上开发，退而求其次用graph.get_graph().print_ascii()，它能直接打印出一张ASCII关系图，足够确认节点和边的连接关系。不要被"必须装额外可视化插件"的说法劝退，一个print_ascii就能覆盖80%的调试需求。

第三招是给invoke设置递归上限。LangGraph默认最大递归步数是25，一旦超过就会抛GraphRecursionError。如果你的图设计有环，这个上限就是保险丝。

result = graph.invoke( initial_state, config={"recursion_limit": 10}, )

我把10作为默认配置。正常的RAG流程最多经历5~6个节点步骤，如果你看到接近10的调用数，说明循环可能有问题。

4.3 执行配置与并行节点：后续扩展的伏笔

除了上面提到的，LangGraph还支持节点级并行。比如你有两个异构知识库（一个向量库、一个关系型数据库），可以设计两个并列的检索节点同时执行，最终汇总结果。这在StateGraph里只需要把两条边从同一个源节点分别指到两个检索节点，再让两个检索节点都指向grade节点即可。

我这次没有在示例里加入并行检索，是因为并行涉及额外的时间同步、去重合并成本，对一篇教程来说容易喧宾夺主。但如果你在实际项目里遇到了"多库检索慢"的问题，这就是LangGraph下一个值得深挖的方向。

5. 实战中踩过的坑：State设计、类型问题与版本兼容

写LangGraph这块，我大概花了两周时间才从"能跑"到"跑得稳"。中间踩了几个挺隐蔽的坑，写出来给你们省点时间。

5.1 不要在State里塞Document对象，存字符串

我的第一个版本在State里直接传List[Document]，节点之间流转的是LangChain的Document对象。这个设计在单节点测试时一切正常，但一旦跑整图就出问题——主要表现为：retrieve_node返回的文档里包含page_content和metadata，可到了grade_node里，metadata偶尔会丢失，甚至出现page_content内容被截断的诡异现象。

排查了两天，最终定位是LangGraph在节点间传递State时对复杂对象做了序列化处理。Document对象的序列化和反序列化在部分版本组合下不稳定。解决方案非常简单粗暴：节点之间只传字符串，所有需要保留的结构化信息在节点内部处理完再传给下一步。metadata如果需要使用，比如你要保留文档来源URL，用json.dumps把它变成字符串存进State，到了用的时候再解析。

5.2 条件边的返回值和映射表必须严格对应

这是一个特别容易踩的坑。add_conditional_edges的第二个参数（路由函数）返回"generate"或"rewrite"，第三个参数是一个字典，key是路由函数的返回值，value是目标节点名。如果路由函数返回了字典里没有的键，LangGraph会抛出一个看起来像是State更新的报错，实际原因是找不到目标节点。

还有一种隐蔽的崩溃方式：路由函数返回的值和目标节点名一样，比如route_after_grade直接返回"generate"，而你也确实定义了一个名为generate的节点。这种情况下代码能跑，但如果你哪天把节点改名成gen_final，却忘了同步修改路由函数的返回值，图就会在运行时崩溃。我的习惯是定义一组常量字符串，节点名、路由返回值、映射表key全部引用同一个常量，从源头上杜绝拼写不一致。

5.3 LangGraph API版本变化快，务必锁定版本

从早期版本到现在的正式版，LangGraph的API经历了好几轮调整。早期版本的StateGraph还在langchain包里，后来独立成langgraph；初始化方式和边添加方式也有多次变化。最典型的是START和END节点最初不是必需的，后续版本要求显式声明。

**我的建议是：项目一开始就用pip freeze把langgraph的版本固定写进requirements.txt，同时跟进官方更新日志。**另外，网上大量教程停留在旧版API，如果你照着旧代码在新的核心库上跑，会得到很多看起来莫名其妙的AttributeError。看到这种报错时，第一时间去查当前版本的官方文档，而不是盲目搜索旧的解答。

5.4 别让LLM频繁参与评分：成本与性能的平衡

我在grade节点用的是LLM评分，而不是传统的向量相似度阈值。LLM评分更准确，能理解语义相关性，但代价是每次检索后都要额外调用一次模型。假设一次检索返回5份文档，意味着grade节点要执行5次LLM调用——如果查询量大，这就是一笔不小的成本。

优化思路有两个方向。第一个是先跑便宜的过滤器：优先用向量相似度分数做第一轮粗筛，低于0.5的直接丢弃，再对剩余的文档做LLM精评。第二个是把多份文档的评分合并到一次LLM调用中，让模型一次判断多份文档的相关性，减少往返次数。我目前的方案是第一种，因为判断逻辑更清晰，也更方便调整阈值。

6. 下一步：把这张图扩展成更完整的RAG应用

写完上面这套基础图，你已经具备了用LangGraph编排RAG的核心能力。我觉得这个基础上还有三个方向，非常值得继续挖，我自己也在持续使用。

第一个是多路由检索。现在这张图只有一个retriever。你可以把retriever拆成多个并行节点，比如一个管技术文档、一个管客服工单、一个管产品手册，在grade之前加一个route_retrieval节点，根据用户问题的关键词判断该走哪个知识库。这种"按需路由"能把单库检索的噪声大幅降低。

第二个是引入对话记忆。目前的State里没有历史消息。你可以扩展RagState，增加一个messages字段，用add_messagesreducer自动追加消息。这样在多轮对话场景下，rewrite节点可以结合历史上下文来补全当前轮的问题指代——比如用户上一轮问了"报销票据要求"，这一轮问"有效期呢"，改写节点就能把"有效期"补全为"报销票据的有效期"。

第三个是生成结果的自校验。在generate之后再加一个verifier节点，用独立的LLM调用检查答案中是否有和检索文档冲突的内容。如果发现冲突，回到generate节点带提示"请修正以下冲突点"再生成一次。这会显著提升答案的可靠性，代价是增加一次模型的往返。

我自己的体会是，LangGraph写RAG真正改变的，不是代码量，而是设计时的思维方式。你不再问自己"这一步用什么Chain编排"，而是问"我的流程里，谁来决定下一步去哪，哪个节点有权喊停重来"。有了这种视角，RAG就不再是一条脆弱的直线，而是一个能自我纠错、能在运行时调整策略的流程系统。

如果你的项目恰好面临"检索结果波动大""多轮查询指代不清""需要多知识库路由"这几个问题中的任意一个，强烈建议把RAG迁移到LangGraph上。从一个带质量闸门的小图开始，跑通一条带循环的流程，你会发现之前那些"在链里绕不开"的设计难点，在图模型里基本都消失了。