企业官网建设流程全解析

最近连续有几个人跑来问我同一个问题：LangGraph 和 LangChain 到底什么区别？问得多了，我发现大家不是不懂用法，而是卡在了同一个点上——链式调用写多了之后，根本想象不出"图"怎么组织 Agent。这篇就围绕标题里的三件事来写：StateGraph 状态图、条件路由、Agent 的工具调用循环。理解这三块，LangGraph 基本就算入门了，剩下的 API 都是查文档就能解决的问题。

这篇内容面向两类人：一类是用过 LangChain 但被 Agent 循环绕晕的老手，另一类是从没写过 LangGraph、想直接上手搭 Agent 的新人。我会把原理、代码、踩坑经验混在一起讲，不铺垫废话，直接拆。

1. 为什么是图：从顺序链到 LangGraph 的思维转变

1.1 链式调用的天花板

先说个实际场景。用 LangChain 写一个简单的客服 Agent，最初版通常长这样：

result = llm.invoke(user_input) if result.should_call_tool: tool_result = call_tool(result.tool_calls) final = llm.invoke([user_input] + [result] + [tool_result]) else: final = result

这段代码本身没问题，但你会发现一旦业务流程复杂起来，它就开始变形：要加记忆得在外部维护数组，要加重试机制得再包一层循环，要让不同的用户问题走不同的处理器得写一堆 if-else。我见过不少项目最后变成了一坨几百行的顺序代码，每个新需求都要往里面插一段逻辑，改一个地方就要顺带检查全流程。

这就是链式调用的天花板：它隐含假设执行路径是线性的。但真实 Agent 的决策流程是分叉的、会回环的。模型输出一个工具调用，你需要去执行工具，执行完再把结果喂回去，这本来就是循环；循环里还可能根据不同的工具结果走不同的分支，这又是条件跳转。用"链"去表达这些东西，代码结构和业务逻辑根本不匹配。

1.2 图模型的核心：节点、边、状态

LangGraph 换了一种表达方式：把整个流程画成一张图。图里有三个基本概念。

• 节点（Node）：一个执行单元，接收状态、处理逻辑、返回状态更新。在 Agent 里，"调用模型"是一个节点，"执行工具"是另一个节点。
• 边（Edge）：定义节点之间的跳转关系。普通边表示"跑完 A 直接去 B"，条件边表示"跑完 A 后根据某个结果决定去 B 还是去 C"。
• 状态（State）：贯穿全程的共享数据容器。每个节点都能读它、更新它，更新结果会叠加到状态上，传给下一个节点。

这个模型的妙处在于，你把"流程怎么走"和"每一步做什么"拆开了。节点只关心自己那部分逻辑：模型节点只负责生成回复，工具节点只负责执行工具。至于下一步去哪，由边来决定。这样改流程的时候，你不用动节点内部代码，只要重新连边就行。

1.3 LangGraph 和 LangChain 到底差在哪

很多人误以为 LangGraph 是 LangChain 的升级版，其实它俩定位不一样。LangChain 侧重的是"组件库"，它提供了模型封装、Prompt 模板、各种模型封装和工具接口，解决的是"怎么调用 LLM"的问题。LangGraph 侧重的是"编排层"，解决的是"多个组件之间的执行流程怎么组织"的问题。

一个类比是：LangChain 是零件仓库，LangGraph 是流水线图纸。你当然可以直接从仓库拿零件手搓一个流水线，但用图纸把工位、传送带、质检点画清楚，后续维护和扩展都会轻松很多。实际项目中，两者通常是配合使用的：用 LangChain 的工具接口和模型封装，用 LangGraph 把整个 Agent 流程串起来。

2. StateGraph 状态设计：Reducer、Schema 与共享内存

2.1 用 TypedDict 声明状态：第一步把"全局变量"定义清楚

StateGraph 里的状态就是一个 TypedDict，直接声明有哪些字段。入门时写复杂了容易懵，最简单的状态长这样：

from typing import Annotated, TypedDict import operator class State(TypedDict): messages: Annotated[list, operator.add] count: int

这里的messages是核心字段，我会在 Agent 循环里反复往里追加消息。count是临时加的计数器，用来演示普通字段的更新方式。

声明状态用 TypedDict 而不是 Pydantic 模型，是因为 TypedDict 足够轻量，LangGraph 内部可以灵活处理它的合并逻辑。如果你偏好数据校验，也可以换成 Pydantic 模型，但入门阶段没必要。

每次节点 return 的内容，LangGraph 会把它当作"对这个状态的一次更新"，然后根据字段对应的更新规则把新值合并进状态。如果字段没有声明合并规则，默认是"直接覆盖"；有了规则，就按规则来。

2.2 Reducer：告诉 LangGraph 同一字段被多次更新时怎么合并

Annotated[list, operator.add]是 LangGraph 里最容易卡住新人的写法。它的意思是：当多个节点或多次执行都要更新messages字段时，不做覆盖，而是用operator.add把新列表"追加"到旧列表后面。

为什么要这样做？因为 Agent 循环里的消息列表天然是只增不减的。模型节点追加一条 AI 消息，工具节点追加一条工具消息，每次更新都是追加操作。如果你把messages声明成普通字段，那么后一个节点更新的结果会把你上一轮的消息覆盖掉，整个对话历史就丢了。

自定义 Reducer 也不难，就是一个签名(current_value, new_value) -> final_value的函数：

def merge_lists(left: list, right: list) -> list: return left + right class State(TypedDict): messages: Annotated[list, merge_lists]

我个人的习惯是：所有"只追加、不覆盖"的字段都配上operator.add，普通标量字段才留默认覆盖逻辑。这样能在一开始就避开大量并发更新导致的冲突问题——后面第 5 节我会专门讲这个坑。

2.3 节点函数、add_edge 与编译运行的最小闭环

节点就是一个普通函数：输入 State，输出一个 dict，dict 的 key 必须是 State 声明的字段。

from langgraph.graph import StateGraph, START, END def node_a(state: State): return {"messages": ["A 已执行"], "count": state["count"] + 1} def node_b(state: State): return {"messages": ["B 已执行"], "count": state["count"] + 1} builder = StateGraph(State) builder.add_node("a", node_a) builder.add_node("b", node_b) builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("b", END) graph = builder.compile() result = graph.invoke({"messages": [], "count": 0}) print(result["messages"]) # ["A 已执行", "B 已执行"]

这里有两个细节值得注意。

START和END是新版的写法，早期文档里常见的是__start__和__end__。如果你在网上搜教程看到两种写法混着用，别慌，功能是一回事，只是不同版本的命名调整。建议新项目一律用新写法。

graph.compile()是必须的步骤。编译不只是做语法检查，它会把整张图的结构固化下来，生成内部执行计划。一些错误比如边指向不存在的节点，会在编译阶段就暴露出来，而不是等你运行到那一步才报错。

2.4 从"顺序执行"到"并行与分支"的推演

顺序图只是热身。StateGraph 真正厉害的地方在于，节点之间的边可以组成任意形状。

• 扇出：A 同时连接 B 和 C，A 执行完后 B 和 C 可以并行运行。
• 汇聚：B 和 C 都连接到 D，D 要等 B、C 全部完成后才启动。
• 自环：节点 A 连接到它自己，加上条件边，就构成了循环。

这里的关键词是"并行"。LangGraph 对并行分支的执行是异步调度、自动汇总的：多个分支各自更新同一份状态，各分支返回的字段更新会合并后再进入下一个节点。所以你在设计节点时，不用自己写并发代码，只要把图的结构搭对，并行能力是天然的。

但并行也带来约束：如果有多个分支同时更新同一个字段，那个字段必须配上 Reducer，否则 LangGraph 不知道该听谁的，会抛异常。我见过太多人在这里翻车——稀里糊涂写了个普通字段，跑并行分支就报 Concurrent update 错误。解法就一句话：凡是多个分支会碰的字段，上 Reducer。

3. 条件路由是怎么"看图说话"的

3.1 路由函数的本质：一个返回"下一站"的普通函数

条件路由是 LangGraph 里最像"逻辑判断"的部分，但它的原理很简单：你写一个普通函数，函数根据当前状态决定"下一步去哪个节点"，然后返回值在图上被解释成一条边。

from typing import Literal def route_function(state: State) -> Literal["a", "b"]: if len(state["messages"]) < 5: return "a" return "b" builder.add_conditional_edges( "current_node", route_function, {"a": "node_a", "b": "node_b"}, )

add_conditional_edges接收三个参数：源节点名、路由函数、路径映射表。路由函数的返回值必须能在映射表里找到对应的目标节点。第三个参数的目的是解耦：路由函数里可以不写真实的节点名，而是返回一个抽象的"路线标记"，映射表再把标记翻译成具体节点。

这个设计很实用。假设你要替换节点实现，只要改映射表，不用动路由函数；或者你想给同一个路由逻辑配不同的执行策略，直接换张映射表就行。

3.2 三个典型路由场景：意图分流、校验重试、任务分类

第一个场景是意图分流。用户消息进来，先用分类节点判断走哪个处理流程：

def classify(state: State) -> str: text = state["query"].lower() if "天气" in text: return "weather" if "计算" in text: return "math" return "fallback" builder.add_conditional_edges( "classifier", classify, {"weather": "weather_node", "math": "math_node", "fallback": "fallback_node"}, )

第二个场景是校验重试。一个节点负责生成 SQL 语句，另一个节点负责执行和校验；如果校验失败，就路由回生成节点重来一遍。这种"当前节点的出边指向自己"的写法，就是重试循环的雏形。

第三个场景是任务分类。从一组待办里识别出哪些该走人工审核、哪些自动处理，然后分发给不同节点。核心都是用同一个路由函数，根据状态内容直接跳到对应分支。

3.3 条件路由踩坑清单

条件路由我踩过不少坑，列几条最典型的。

路由函数返回的值如果在映射表里不存在，运行时会抛异常，所以返回类型和映射表的 key 要保持严格一致。我建议路由函数的返回类型直接声明成Literal["分支名1", "分支名2"]，这样类型检查器能在写代码时就帮你兜底。

路由函数里读不到你想用的字段，优先检查 state 的 key 名。TypedDict 不会在运行时强制 key 存在，所以常常是"字段名拼错了但编译不报错，运行时报 KeyError"。如果真的遇到，先print(state.keys())看一遍真实字段。

第三个坑是映射表值和实际节点名不一致。编译阶段通常会校验这一点，但部分版本只在运行到条件边的时候才报错。遇到这种问题，先检查节点名有没有拼错，尤其注意大小写和下划线。

4. Agent 工具调用循环：让模型自己决定调用顺序

4.1 工具调用循环的最小闭环

Agent 的工具调用循环，本质就是一个环形图：模型节点生成回复，如果回复里带着工具调用请求，就进入工具节点去执行；工具执行完把结果写回状态，再回到模型节点，让模型看到结果后继续决策。

import operator from typing import Annotated, Literal, TypedDict, getattr from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI class State(TypedDict): messages: Annotated[list, operator.add] step_count: int @tool def add(a: float, b: float) -> float: """Add two numbers together.""" return a + b tools = [add] tool_node = ToolNode(tools) model = ChatOpenAI(model="gpt-4o-mini") model_with_tools = model.bind_tools(tools) def agent_node(state: State): response = model_with_tools.invoke(state["messages"]) return { "messages": [response], "step_count": state.get("step_count", 0) + 1, } def should_continue(state: State) -> Literal["continue", "end"]: last_message = state["messages"][-1] if getattr(last_message, "tool_calls", None): return "continue" return "end" builder = StateGraph(State) builder.add_node("agent", agent_node) builder.add_node("tools", tool_node) builder.add_edge(START, "agent") builder.add_conditional_edges( "agent", should_continue, {"continue": "tools", "end": END}, ) builder.add_edge("tools", "agent") graph = builder.compile() initial_state = {"messages": [{"role": "user", "content": "3 + 5 等于多少？"}], "step_count": 0} result = graph.invoke(initial_state) print(result["messages"][-1].content)

这段代码跑起来之后，你观察到的执行顺序是：agent → tools → agent，然后模型判断没有更多工具调用，输出最终答案，走 END。工具调用的顺序完全由模型自己决定，你只负责搭好回环结构。

4.2 从 ReAct 模式理解 LangGraph 的循环结构

很多 Agent 教程都会提 ReAct 模式，它是一个经典的推理循环：Thought（思考该做什么）→ Action（调用工具）→ Observation（观察结果）→ 回到 Thought。LangGraph 的工具调用循环本质上就是 ReAct 的工程化实现。

区别在于，ReAct 在论文里描述的是一个形式化流程，落到代码里你需要自己去处理"模型是否请求工具""工具结果如何回传"这些细节。LangGraph 把最麻烦的骨架给你搭好了：ToolNode 负责执行工具和包装结果，条件边负责判断是否继续循环。你只需要提供模型和工具列表。

所以理解这个循环的关键是看消息列表的演进过程。每一轮，AI 消息可能携带 tool_calls，工具节点会把结果作为 ToolMessage 追加进去。模型每次调用时看到的不是单个结果，而是完整的历史上下文——这也是为什么messages必须用 Reducer 追加而不是覆盖。

4.3 循环终止的两个硬性保障

循环结构最大的风险是不终止。模型在极端情况下可能反复请求同一个工具，或者一直输出工具调用但永远不给出最终答案。这会导致运行时间无限拉长，甚至触发 LangGraph 默认的递归上限。

LangGraph 自带一个保护机制：recursion_limit，默认值是 25。超过这个步数，运行会抛 GraphRecursionError。很多初学者第一次遇到这个报错都以为是 bug，其实这是框架在替你踩刹车。

更好的做法是主动设置最大迭代步数。常见方案是在条件路由里加一个计数器判断：

MAX_STEPS = 5 def should_continue(state: State) -> Literal["continue", "end"]: if state["step_count"] >= MAX_STEPS: return "end" last_message = state["messages"][-1] if getattr(last_message, "tool_calls", None): return "continue" return "end"

注意step_count这个字段是普通 int，没有 Reducer。因为环形图中每个节点在单次执行里只会被访问一次（tools 走完才会到 agent，不存在两个分支同时更新它），默认覆盖逻辑完全够用。

4.4 带记忆的工具调用循环：从 stateless 到 stateful

上面的例子是无状态的：每次invoke都是全新上下文。如果想让 Agent 记住之前的对话，可以在编译时挂一个 Checkpointer。Checkpointer 的作用是持久化图的执行状态，并且支持按线程 ID 恢复。

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph_with_memory = builder.compile(checkpointer=memory) result_1 = graph_with_memory.invoke( {"messages": [{"role": "user", "content": "我的名字叫小明"}], "step_count": 0}, config={"configurable": {"thread_id": "user-001"}}, ) result_2 = graph_with_memory.invoke( {"messages": [{"role": "user", "content": "我叫什么名字？"}], "step_count": 0}, config={"configurable": {"thread_id": "user-001"}}, ) print(result_2["messages"][-1].content) # 模型能答出来

这里的核心变化是：状态被保存下来，下一次 invoke 会先加载历史状态再执行图。thread_id用来区分不同的会话，同一个线程 ID 共享一段记忆。

如果你用的是本地内存版本，重启进程记忆就丢了，生产环境建议换成持久化后端。这个方向涉及 Checkpointer 的选型，入门阶段先用 MemorySaver 理解机制就够了。

5. 实测中的问题记录与个人经验

5.1 并行分支更新同一状态字段引发的冲突

我第一次写多分支图时遇到过这样的报错：Concurrent update to one or more channels detected, forcing a rewrite。我当时的状态里有两个分支同时往result这个普通字段里写内容，LangGraph 看到同一个字段收到了两份互相冲突的新值，无法决定覆盖顺序，直接中止执行。

这个报错教会我一件事：普通字段不是不能存在，而是要搞清楚哪些节点会写它。凡是可能被多个分支同时更新的字段，一律加 Reducer。如果你确实想做"多个并行结果汇总"，比如把多个工具响应收集到一个列表里再交给下一步，那就必须用Annotated[list, operator.add]或自定义合并函数。

顺带提醒一句：并行分支的更新顺序是不确定的，如果你的 Reducer 逻辑依赖顺序，比如"先到先得"，那结果可能不稳定。自定义 Reducer 尽量写成纯函数，不依赖执行顺序。

5.2 路由返回非法值：报错之后怎么快速定位

条件路由的报错通常长这样：ValueError: Node 'xxx' not found in graph，或者 "Invalid path 'xxx' in conditional edge"。第一反应别去翻框架源码，先检查三件事。

第一，路由函数的返回值是不是原样传给了映射表。如果路由函数里做了取模、拼接、大小写转换之类的操作，先打印一下最终返回值。第二，映射表的 key 是否与返回值完全一致，空格、大小写都算。第三，映射表的值是否都指向已存在的节点名，END 和 START 也要确认在当前版本里能直接用。

我的做法是在路由函数里临时加一句 debug：

def route_func(state): route = "a" if state["count"] > 0 else "b" print("route:", route) return route

看到打印值之后，问题通常一秒就定位了。

5.3 工具调用失败不等于循环结束

工具执行失败是很常见的，但不少初学者会在工具节点里让异常直接抛出，导致整个图崩溃。正确的思路是：把"工具执行失败"当作一个普通结果告诉模型，让模型决定接下来怎么办。ToolNode 默认会捕获部分异常并把错误信息封装成 ToolMessage 回传给模型，这是个很贴心的设计。

如果你用自定义工具节点，也建议自己包一层：

def safe_tool_node(state: State): try: return {"messages": [execute_tool(state["messages"][-1])]} except Exception as e: return {"messages": [ToolMessage(content=f"工具出错：{e}", tool_call_id="...")]}

把异常信息塞回消息列表，模型就有机会修正参数或换一个工具。这比直接崩溃要可靠得多，也是 Agent 类项目稳定性的关键一环。

5.4 调试和追踪：把每一步状态摊开来看

最后聊调试。LangGraph 的图结构容易让人产生错觉，以为它是透明的，其实节点多了之后，你得主动追踪每一步的输入输出。

最简单的调试方式是graph.invoke之后打印整个 state，但这只能看到最终结果，看不到中间过程。想逐步看，用graph.stream：

for chunk in graph.stream(initial_state): for node_name, update in chunk.items(): print(f"节点 {node_name} 的更新：", update)

这样每一步的返回值都会单独打出来，你在哪个节点出了什么问题一目了然。如果项目里开了 tracing 服务，还能看到更完整的调用链和耗时数据，但我个人经验是：Debug 阶段先用stream打印就够解决了。

最后分享一点个人体会。LangGraph 入门最大的障碍不是语法，而是思维方式。链式编程让人习惯"下一步是什么"；LangGraph 逼你思考"这一步之后可能有哪些分支，怎么判断去哪个分支，状态如何流动和合并"。等你把这套图式思维刻进脑子里，再去看 Agent 的复杂循环、多 Agent 协作、人工介入机制，都会清晰很多。我踩过的坑集中在状态合并和循环终止上，希望上面这些记录能让你少走几步弯路。