企业官网建设流程全解析

做 AI Agent 最怕什么？我自己的体验是：不是模型不够聪明，而是流程乱成一锅粥。传统做法把 prompt、工具调用、结果解析、重试逻辑全塞在一个 Python 函数里，看起来能跑，一旦改成多轮对话、工具循环、条件分支，代码就变成没人敢动的面条。所以这段时间我花了不少功夫折腾 LangGraph，核心目标只有一个：把 Agent 工作流从"函数递归"升级成"图执行 + 状态持久化"。这篇文章就围绕这个目标，把 LangGraph 从入门到实战的关键环节拆开讲清楚，包括 State、Node、Edge、Checkpointer 这些绕不开的概念，以及一个能直接改来用的带工具调用 Agent 完整示例。适合已经写过一点 Python、知道 LangChain 基础但还没碰过图工作流的开发者。

1. 为什么 Agent 工作流需要"有状态"：先看清传统链式调用的瓶颈

很多人会觉得，Agent 不就是"模型调用 + 工具调用"循环吗？我一开始也这么想，直到做了一个需要多轮交互的客服问答系统，才发现问题远没有那么简单。

1.1 多步调用早知道的状态管理到底难在哪

普通 LLM 调用是无状态的：你传一段 prompt，它返回一段文本，调用结束，一切归零。Agent 不一样，它需要把多轮对话、工具返回结果、当前执行到哪一步、已经重试过几次这些信息，在一轮完整工作流里持续传递下去。

用最朴素的方式写，就是层层传参：

def run_agent(question: str, history: list): messages = history + [user_message(question)] reply = llm.invoke(messages) if need_tool(reply): tool_result = call_tool(reply.tool_call) messages.append(tool_message(tool_result)) reply = llm.invoke(messages) # 再来一轮 return reply, messages

这段代码当然能跑，但一旦加一个"如果工具调用失败，换另一个工具再试"的分支，或者加一个"用户中途补充条件，需要回退到更早一步"的需求，你就开始疯狂 if/else。再往后，如果有多个 Agent 协作，每个人负责一段流程，还要共享上下文，代码复杂度会成倍上升。本质原因在于：状态不是被显式管理，而是被隐式塞进变量和调用栈里，流程一旦分支，状态就乱。

1.2 LangGraph 的核心思路：把流程画成图，把状态放在正中央

LangGraph 换了个思路，它把工作流建模成一张有向图。图中每个节点只做一件事，读取共享状态，处理完更新状态，然后沿边走到下一个节点。状态不再散落在各个函数参数里，而是作为一个全局对象在节点之间传递。

这个模式和 Excel 的计算逻辑很像。一个单元格负责一个计算，值变了，依赖它的单元格自动重新算；谁依赖谁，由表格结构决定，而不是靠代码里层层嵌套的调用顺序。LangGraph 的 State 就是这个"表格底部的数据区"，节点是"单元格"，边是"单元格之间的引用关系"。

运行时，LangGraph 会维护一个状态对象。节点函数接收当前状态的快照，返回你希望写入的增量更新，框架负责把这些更新合并回状态。这种设计带来三个直接好处：

• 分支和循环不再需要手写复杂逻辑，用边表达条件即可。
• 状态可以持久化，进程重启、服务崩溃后能恢复。
• 每个节点是纯函数式的，测试时把构造好的状态塞进去就行。

我自己用下来的体会是，LangGraph 最大的价值不是"多了一种写法"，而是逼你把 Agent 实现为可控制的状态机，而不是一段从入口走到出口的线性代码。

2. LangGraph 三大核心概念：State、Node、Edge 与 Checkpointer

没接触过图工作流的人，刚看 LangGraph 文档会有点懵，各种概念混在一起。拆开看其实就四样东西：State、Node、Edge、Checkpointer，其中前三样决定"流程怎么走"，最后一样决定"记忆怎么存"。

2.1 State：用 TypedDict 定义你的"记忆面板"

State 的定义方式很直接，就是一个 TypedDict，描述这份记忆里有哪些字段：

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_input: str remaining_retries: int

这里最关键的是Annotated这个类型标记。它表示：当有多个节点（或同一节点多次执行）都向同一个字段写入值时，这个字段该怎么合并。

比如messages字段用了add_messages，意思就是新消息追加到旧消息后面。user_input没有标记，那它就是普通的覆盖语义：新写入的值直接替换旧值。

这个设计非常重要，因为它决定了状态更新的语义。如果你要维护一个历史列表，但不小心没加 reducer，节点第二次返回{"messages": [new_msg]}时，旧消息会被全部丢掉。反过来，如果你处理的是一个计数器，比如remaining_retries，你想每次都减一，那就只能靠节点自己读取旧值、计算新值再写回，因为默认覆盖语义不会帮你做"减一"。

我自己一般会把状态分成三类：会话历史（用add_messages合并）、流程参数（默认覆盖即可）、临时变量（最好别放进 State，后面说）。

2.2 Node：干活的最小单位

Node 就是一个普通 Python 函数，输入是当前 State，输出是一个字典，表示你想更新哪些字段。

def agent_node(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]}

注意几点：节点函数不直接修改传入的 state，而是返回增量更新，由框架决定怎么合并。这保证了图执行过程可控、可回放。节点可以做任何事：调模型、查数据库、调工具、写日志，甚至自己再调用另一个小模型。

2.3 Edge：决定下一步走向

Edge 就是节点之间的连线。add_edge表示无条件跳转，add_conditional_edges表示根据当前状态动态选择下一跳。

无条件边像流水线，上一道工序结束，下一道工序自动开始。条件边像一个岔路口，根据状态里某个字段的值去不同节点。条件边必须有明确的终点，否则图会不知道怎么走下去。

LangGraph 里有两个内置的特殊节点，一个是START，表示整个工作流的入口；另一个是END，表示终止。你所有的工作流，本质上就是从START出发，经过若干节点和边，最终到达END。

2.4 Checkpointer：让图拥有长期记忆

如果你编译图时不传 Checkpointer，每次invoke都是全新的状态，跟普通函数调用没有区别。要让 Agent 在多轮对话中记住上下文，必须有一个"存档机制"，这就是 Checkpointer。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer)

配合config里的thread_id，LangGraph 会把状态存档在某个会话下。同一个thread_id多次调用，状态是延续的；不同thread_id互不干扰。打个比方，thread_id就是游戏存档槽位，MemorySaver是内存存档，SqliteSaver是硬盘存档。

这一套组合拳下来，LangGraph 解决的核心问题就不是"把流程图画出来"，而是"流程执行过程中，状态如何被安全、可恢复地保存下来"。

3. 动手实现一个带工具调用的 Agent：核心步骤与可直接抄的代码

概念讲得再多，不如直接跑通一个例子。这里我用一个"天气查询 Agent"做演示：用户问天气，模型判断需要调用天气工具，工具返回结果，模型把结果整理成最终回答。

3.1 环境准备

先装基础依赖：

pip install langgraph langchain-core

模型部分，我不绑定具体服务商。你手头用的是什么模型 SDK，就初始化成什么，关键是它要能产出带tool_calls的消息对象。如果你用 LangChain 生态的模型类，直接传进来即可。

# 这里只是占位，换成你自己的模型初始化方式 llm = create_my_chat_model() # 返回一个 ChatModel 实例

要是你连模型调用还没配好，建议先跑通"模型能识别工具调用"这一步，再往下走，否则排错时你会分不清是图的问题还是模型的问题。

3.2 定义状态和工具

from typing import Annotated, Literal, TypedDict from langgraph.graph.message import add_messages from langchain_core.tools import tool class AgentState(TypedDict): messages: Annotated[list, add_messages] @tool def get_weather(city: str) -> str: """查询指定城市的当前天气。""" return f"{city}：晴，气温 23℃，东南风 2 级"

get_weather是一个工具节点，模型会在需要时生成调用参数，LangGraph 的ToolNode会帮我们把参数传进去。工具函数内部逻辑要写清楚返回格式，最好返回一个字符串或者 JSON 序列化后的文本，方便模型理解。

工具的定义里，docstring 一定要写清楚。模型根据函数名和 docstring 决定要不要调用、传什么参数，写得含糊，它就会频繁乱调或者不调。

3.3 构建图：节点、条件边、循环

这是整个示例的核心。我先把完整代码放出来，再逐步解释。

from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage class AgentState(TypedDict): messages: Annotated[list, add_messages] def make_agent_node(llm): def agent_node(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return {"messages": [response]} return agent_node def route_after_agent(state: AgentState) -> Literal["tools", "end"]: last_message = state["messages"][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return "tools" return "end" builder = StateGraph(AgentState) builder.add_node("agent", make_agent_node(llm)) builder.add_node("tools", ToolNode([get_weather])) builder.add_edge(START, "agent") builder.add_conditional_edges( "agent", route_after_agent, {"tools": "tools", "end": END} ) builder.add_edge("tools", "agent") checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer)

逐个说：

• make_agent_node(llm)是一个工厂函数，返回真正的节点函数。因为节点函数签名固定是(state) -> dict，想要传入 llm，用闭包是最简单的做法。
• route_after_agent是路由函数，它看messages里最后一个 AI 消息。如果模型决定调用工具，就跳转到tools节点；否则说明回答已经完整，进入END。
• ToolNode会读取最后一个AIMessage里的tool_calls，自动执行对应的工具，并把ToolMessage追加到状态里。
• tools节点执行完，通过add_edge("tools", "agent")回到agent节点。模型拿到工具结果后，生成最终回答。

这个循环结构就是最典型的 ReAct 模式：模型思考、决定调用工具、观察工具结果、再次思考，直到不需要工具为止。

这里有一个新手容易踩的坑：条件边的映射字典必须和路由函数返回值严格对应。路由函数返回"tools"，映射里要有"tools"这个键；返回"end"，映射里要有"end"这个键，值指向真实节点名或END。写错一个字符串，运行时会直接报错。

3.4 运行测试：连续对话如何保持上下文

图编译好之后，用invoke执行：

config = { "configurable": {"thread_id": "session-001"}, "recursion_limit": 15, } graph.invoke( {"messages": [HumanMessage(content="今天成都天气怎么样？")]}, config=config, ) graph.invoke( {"messages": [HumanMessage(content="那北京呢？")]}, config=config, )

这里第二次调用时，我没有重新传入第一轮的消息，但模型能通过thread_id对应状态里的完整messages看到之前的对话。这就是 Checkpointer 的作用。

如果去掉thread_id或者换一个thread_id，第二次调用就是全新的对话，模型不会记得成都那次查询。这个特性在做多用户、多会话场景时特别有用。

recursion_limit是兜底设置，防止 Agent 在"工具调用 → 模型 → 工具调用"之间无限循环。一般工具型 Agent 设 10-20 就够，太大容易让异常流程空转浪费 token，太小又会打断正常的多轮工具调用。

4. 生产环境实战：状态设计、持久化、并发与常见排查

Demo 跑通只是开始，真正难受的是把 Agent 放到生产环境。这一节我把自己踩过的一些坑整理出来，按场景分开说。

4.1 状态设计的三个避坑原则

第一个原则是：只放需要跨节点传递的数据。状态字段越多，合并语义越复杂，检查问题越困难。像计算过程中的中间结果，如果只在当前节点使用，就老老实实写成局部变量。我曾经把每一步的中间 JSON 都塞进状态，结果 Checkpointer 序列化时莫名其妙报错，排查半天发现是塞了一个不可序列化的对象进去。

第二个原则是：想清楚每个字段的合并语义。列表型数据默认不适合直接做状态字段，要么用add_messages这样现成的 reducer，要么自己写一个合并函数。计数器最好显式在节点内计算，不要指望默认覆盖能帮你做加减法。我自己写状态时一定会问：这个字段同时被多个节点写时，新的值应该覆盖还是叠加？一旦有多个节点并发写过同一个字段，覆盖语义很容易导致"后写者赢"但是丢掉关键数据。

第三个原则是：避免深层嵌套的状态结构。LangGraph 的 State 本身支持嵌套 TypedDict，但嵌套越深，调试打印越痛苦。宁可多拆几个一级字段，例如user_profile和search_result，也不要塞一个多层字典。

4.2 如何让循环安全退出

Agent 工作流里最常见的死循环是：模型反复要求调用同一个工具，或者工具报错后模型继续带着错误结果回到工具调用。我在项目里通常加两道保险。

第一道保险，在状态里放一个计数器，比如remaining_retries，每次工具调用后减一，路由函数在计数归零时强制走结束分支：

class AgentState(TypedDict): messages: Annotated[list, add_messages] remaining_retries: int def safe_route(state: AgentState): if state["remaining_retries"] <= 0: return "end" last_message = state["messages"][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return {"continue": True} return "end"

第二道保险，工具节点内部做异常捕获。ToolNode默认行为是把异常抛给上层，如果你希望工具失败后模型能换一种方式继续回答，最好自己做一个带容错的工具节点，把异常信息转成一句话写进状态，让模型看到"工具调用失败，原因是什么"，再决定下一步。

这个设计很符合 Agent 的交互逻辑：模型不是一遇到错误就崩，而是能根据错误反馈自我修正。

4.3 排查问题的几个实用手段

第一个手段是打印图结构。编译后调用graph.get_graph().print_ast()，能直观看到节点和边的连接关系，很多路由配置错误一眼就能发现。

第二个手段是单节点测试。图整体跑不通时，先把节点函数单独喂一个构造好的 State，看它返回的增量更新是否符合预期。因为是纯函数式设计，节点可以直接被测试，不用启动整个图。

第三个手段是关注 Checkpointer 的序列化。MemorySaver 使用的序列化要求状态字段里的对象能被正常序列化。出现Object of type X is not serializable之类错误时，优先检查状态里是否有自定义对象、函数、数据库连接等不可序列化的东西。我的习惯是状态里只放 JSON 友好的基础类型，复杂对象在外围处理结束后再写入。

下面这个速查表是我整理的高频问题：

4.4 生产化建议：从内存存档到真正的持久化

Demo 用MemorySaver没问题，但服务一重启，所有会话记忆就没了。生产环境我会换成 SQLite 或 PostgreSQL 系列的 Checkpointer，用法上和MemorySaver几乎没有差别，只是存档落在更可靠的地方。

如果你用 FastAPI 起服务，每个请求都可以带一个thread_id，这个值一般来自业务侧，比如用户 ID、会话 ID。同一个用户的后续请求使用同一个thread_id，Agent 就能无缝续接上下文。

并发时要格外小心：同一个thread_id的多次写操作需要串行化，否则可能写坏状态。不同thread_id之间可以放心并发。我一般会在应用层按thread_id做锁或者队列，避免同一会话出现写入竞争。

另外建议给图调用加监控。每个节点打一条日志，记录开始时间、结束时间、状态字段大小。Agent 出问题时，没有日志几乎没法排查，因为图执行路径是动态的。

5. 我踩过的坑与长期使用后的经验沉淀

最后说一些偏"手感"的经验。

5.1 状态就像数据库，不是全局变量

我后来总结的一句话是：默认覆盖语义相当于"赋值"，带 reducer 的字段相当于"ATOMIC UPSERT"。设计状态时，先把字段的写入模式写清楚。如果你发现自己总在节点里用很多 if 去处理历史数据，说明这些数据不适合放进 State，应该交给外部存储。

5.2 路由函数是 Agent 的大脑

节点主要负责"干活"，路由函数才真正决定"接下来干什么"。有条件的边写多了以后，路由函数会慢慢膨胀成一个大 if/else 块。这时候我会专门开一个模块放路由逻辑，并且给每个分支写注释，保证未来自己回来看得懂。

如果条件分支太多，可以考虑把路由结果也写进状态，例如next_action字段，方便审计"当前 Agent 为什么走到了这一步"。

5.3 先从 prebuilt 组件起步，再手写 StateGraph

如果你刚开始接触 LangGraph，又不确定自己的流程应该怎么拆，可以先用create_react_agent这类预置组件快速跑通核心功能，确认"模型 + 工具"这条路没问题，再换成自定义StateGraph做精细控制。

我自己就是这样：先用预置版验证 prompt 和工具是否好用，再拆节点、加状态字段、加条件边。一步一步来，比一上来就设计一个十几节点的复杂图稳妥得多。

5.4 一个小技巧：把人工审批也变成一个节点

Agent 执行到关键动作时，比如给用户发邮件、下单、删除数据，可以先走一个"待确认"节点，把请求挂起来，等外部系统确认后再继续。LangGraph 的状态持久化让这种挂起和恢复变得很自然，因为图执行到一半，状态是存着的。需要时用interrupt/断点机制，或者更简单地在节点内返回一个"模拟确认"的消息字段，由路由判断是否继续。

这个思路用顺手之后，你会发现自己对 Agent 的掌控力明显提升了：不再是一锤子买卖，而是可以在关键位置踩刹车。最后分享一个实际操作中的体会：不要把 LangGraph 当成一门新技术去"学"，而是把它当做一个更顺手的流程编排器。我见过太多人纠结 API 细节，却忽略了状态设计，结果写出来的图仍然像一锅粥。先把需求拆成"谁负责什么、状态需要记得什么、什么条件下流转到哪"，再落到代码上，整个开发体验会顺畅很多。