从手写循环到LangGraph:Agent状态图编排实战与避坑指南
发布时间:2026/9/29 18:41:15来源:尧图网络
1. 为什么我要把手写 Agent 循环换成 LangGraph1.1 手写循环的甜蜜期与阵痛期刚开始做 Agent 开发那会儿我和很多人一样觉得手写循环才是真男人的做法。一个while True里面塞上 LLM 调用、工具解析、结果回填几十行代码就能跑起来一个能查天气、能算数、能搜资料的智能体。那种一切尽在掌控的感觉确实很爽每一行逻辑都是自己写的出问题一眼就能定位。但甜蜜期没持续多久。第一个项目上线两周后需求开始膨胀要支持多轮工具调用、要在中间插入人工确认、要在失败时回退到上一步、要记录每一步的 token 消耗、要支持流式输出。我的那个while循环从 40 行涨到了 400 行里面嵌套了四层if-else状态变量散落在十几个地方改一个分支经常把另一个分支搞挂。最要命的是我想加一个并行调用两个工具再合并结果的功能发现整个循环结构根本撑不住——因为它是线性的没有显式的状态图概念。这就是手写循环的阵痛期当 Agent 的行为从线性对话演变成带分支、带循环、带并行、带中断恢复的状态机时裸写控制流会迅速失控。你写的其实不是业务逻辑而是一个越来越复杂的调度器而这个调度器本该由框架来管。1.2 LangGraph 到底解决了什么问题LangGraph 的核心价值用一句话概括它把 Agent 的执行过程显式建模成一张有向图节点是计算步骤边是控制流状态在图上流动。你不再写while循环而是声明有哪些节点节点之间怎么连状态怎么更新剩下的调度、循环、分支、中断、恢复、并行全交给框架。这里必须澄清一个高频困惑LangGraph 和 LangChain 到底啥区别。LangChain 更像一个组件库 链式编排它擅长把 prompt、模型、工具、检索器串成一条链Chain适合相对固定的流水线。而 LangGraph 是图式编排它擅长处理带循环和条件分支的复杂控制流天然适合 Agent 这种想一步、做一步、看结果、再想的迭代场景。两者不是替代关系LangGraph 里照样可以用 LangChain 的模型封装和工具定义只是把谁来驱动循环这件事从你手里交给了图。至于StateGraph和create_agent的关系也是新手最容易绕晕的点。StateGraph是底层原语你手动定义状态结构、手动加节点、手动连边控制粒度最细create_agent以及早期的create_react_agent是高层封装一行代码就能生成一个标准的 ReAct 风格 Agent 图适合快速起步。我的建议是先用create_agent跑通再逐步下沉到StateGraph定制这样既能快速见效又能在需要精细控制时不至于从零开始。1.3 这篇文章适合谁看如果你已经写过一个能跑的手写 Agent 循环现在被状态管理、分支控制、中断恢复这些问题折磨得够呛那这篇就是写给你的。如果你是完全的新手建议先理解 ReAct 的基本范式思考-行动-观察的循环再来看图式编排会顺畅很多。全文我会用从手写迁移到 LangGraph这条主线把状态设计、节点拆分、条件边、检查点、并行这些关键点一个个拆开讲每个点都配上我实际踩过的坑和可复制的代码。2. 迁移前的整体设计先想清楚图长什么样2.1 从循环思维切换到图思维手写循环时你的大脑里是一条时间线先调模型再判断有没有工具调用有就执行工具把结果塞回消息列表再调模型……如此往复。而图思维要求你把这条时间线拍扁成一张静态结构图有哪些状态、有哪些处理单元、单元之间在什么条件下跳转。我习惯用一个类比手写循环像流水账日记按时间顺序记LangGraph 像地铁线路图站是节点线是边换乘是条件跳转。你关心的不再是下一步执行哪行代码而是当前在哪个站满足什么条件去下一站。这个思维切换最大的收益是可观测性和可恢复性。因为图是显式的你可以把任意一次执行画出来、存下来、从中间某个节点恢复。手写循环里这些几乎都要自己造轮子。2.2 状态结构设计State 是整张图的血液在 LangGraph 里State是一个 TypedDict 或 Pydantic 模型它定义了在节点之间流动的数据。设计状态是整个迁移里最关键的一步因为节点之间只通过状态通信不通过函数参数。一个典型的 Agent 状态大概长这样from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] tool_calls_count: int user_intent: str final_answer: str这里有个必须讲透的点Annotated[list, add_messages]里的add_messages是reducer归约函数。默认情况下节点返回的新值会直接覆盖旧值但消息列表我们希望是追加而不是覆盖所以用add_messages声明合并策略。这是新手最容易忽略的细节——如果你不加 reducer模型每轮返回的消息会把历史消息冲掉Agent 立刻失忆。我踩过的坑一开始我把tool_calls_count也写成普通 int结果每个节点返回时都覆盖计数永远是 1。后来改成Annotated[int, operator.add]才能累加。凡是需要累积而非替换的字段都要显式指定 reducer这是状态设计的铁律。2.3 节点拆分粒度粗一点还是细一点节点拆得太细图会变成一团乱麻节点间跳转开销大、调试困难拆得太粗又退化成一个大节点里塞循环等于没迁移。我的经验法则是一个节点只做一件语义清晰的事且这件事的输入输出都能用状态描述。对标准 ReAct Agent我通常拆成三个核心节点agent节点调用 LLM决定是回答还是调用工具tools节点执行工具调用把结果写回消息should_continue条件函数判断下一步去tools还是结束这个三节点结构几乎能覆盖 80% 的场景。等遇到多智能体协作、人工审核、并行检索这些需求再往上加节点。2.4 条件边与循环Agent 的心跳在哪Agent 的本质是循环而 LangGraph 里循环是通过条件边指回上游节点实现的。should_continue这个条件函数返回一个字符串框架根据返回值决定走哪条边。如果返回tools就跳到工具节点工具节点执行完再连回agent节点形成闭环如果返回end就走到END终止。这个设计的美妙之处在于循环不再是代码里的while而是图结构里的一条回边。框架帮你管理迭代次数、状态传递、终止条件你只需要声明什么情况下继续、什么情况下停。想加最大迭代次数限制在条件函数里判断tool_calls_count就行不用改任何循环结构。3. 核心细节解析StateGraph 的每个零件怎么用3.1 定义状态与 reducer 的实战细节前面提了 reducer这里展开讲几个实战中高频用到的模式。除了add_messages还有几种常见需求字段类型需求reducer 写法消息列表追加而非覆盖Annotated[list, add_messages]计数器累加Annotated[int, operator.add]去重集合合并去重自定义函数最新值覆盖默认不加 Annotated自定义 reducer 的写法也很简单就是一个接收两个参数、返回合并结果的函数def merge_unique(left: list, right: list) - list: seen set(left) result list(left) for item in right: if item not in seen: result.append(item) seen.add(item) return result注意reducer 必须是纯函数不能有副作用否则在并行节点合并时会出诡异问题。我见过有人在 reducer 里写日志、发请求结果并行执行时日志乱序、请求重复排查了半天。3.2 节点函数的签名与返回值约定节点函数接收状态返回一个字典表示要更新的字段。这个字典会经过 reducer 合并回全局状态。签名固定为def agent_node(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]}这里有个容易犯的错返回的字典 key 必须是状态里已声明的字段否则框架会忽略或报错。我早期想临时存个中间变量直接返回了个没声明的 key结果数据凭空消失debug 了半小时才反应过来。另一个细节是返回值可以是部分更新。你不需要返回完整状态只返回变化的字段即可框架会自动合并。这让节点函数写起来很轻。3.3 条件边的路由函数怎么写才不出错路由函数conditional edge 的判断函数接收状态返回一个字符串这个字符串必须能映射到预先注册的边。写法def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return tools return end坑点在于路由函数的返回值必须和add_conditional_edges里映射表的 key 完全一致大小写、拼写错一个字符都会导致运行时找不到边。我建议把路由返回值定义成常量避免手写字符串。还有个隐蔽问题如果最后一条消息既没有tool_calls也没有内容比如模型返回空路由函数可能拿到None而崩溃。稳妥做法是加防御性判断把异常情况路由到end或专门的错误处理节点。3.4 检查点与中断让 Agent 能暂停续跑LangGraph 的检查点checkpointer机制是我最欣赏的功能之一。它把每一步的状态快照存下来配合interrupt就能实现执行到某节点暂停等人工确认后再继续。这在需要人工审核的高风险操作场景里是刚需。配置检查点很简单编译图时传入from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)调用时传入config{configurable: {thread_id: user-123}}同一个thread_id就能续接上次的状态。生产环境建议换成持久化后端如数据库MemorySaver只适合本地调试。提示中断恢复依赖状态可序列化。如果你在状态里塞了不可序列化的对象比如数据库连接、文件句柄检查点会失败。状态里只放数据不放资源。4. 完整实操从零搭一个带工具调用的 Agent 图4.1 环境准备与依赖安装先把环境搭起来。我习惯用 conda 建独立环境避免和系统 Python 打架conda create -n langgraph-demo python3.11 -y conda activate langgraph-demo pip install langgraph langchain langchain-openai选 Python 3.11 是因为它在异步和类型提示上比较成熟3.12 有些库还没完全跟上。langchain-openai只是模型接入层你也可以换成其他模型提供方的包LangGraph 本身不绑定具体模型。4.2 定义工具与模型先定义两个简单工具方便演示from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 fake_data {北京: 晴25度, 上海: 多云28度} return fake_data.get(city, 暂无数据) tool def calculate(expression: str) - str: 计算数学表达式例如 23*4。 try: return str(eval(expression)) except Exception as e: return f计算失败: {e} tools [get_weather, calculate]注意eval在生产环境有安全风险这里只为演示。真实项目请用安全的表达式解析库别直接 eval 用户输入。模型绑定工具from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools)bind_tools会把工具的描述转成模型能理解的格式模型据此决定是否调用、调用哪个。4.3 构建 StateGraph 的完整代码把前面的零件组装起来from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from typing import Annotated, TypedDict class AgentState(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: AgentState) - dict: response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState) - str: last state[messages][-1] if getattr(last, tool_calls, None): return tools return end builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue, { tools: tools, end: END, }) builder.add_edge(tools, agent) graph builder.compile()这段代码就是整个 Agent 的核心。注意ToolNode是 LangGraph 预置的工具执行节点它自动解析tool_calls、执行对应工具、把结果包装成ToolMessage写回状态。省去了手写工具分发的一大坨代码。4.4 运行与观察执行轨迹跑一个查询result graph.invoke({ messages: [(user, 北京天气怎么样顺便算一下 12*8)] }) for msg in result[messages]: print(type(msg).__name__, :, msg.content)你会看到消息序列大致是用户消息 → AI 消息带 tool_calls→ 工具消息 → AI 消息带 tool_calls→ 工具消息 → AI 最终回答。这个AI 决策-工具执行的往返就是图上的循环框架自动帮你转了两圈。想看得更清楚可以用graph.stream()逐步输出或者开启 LangSmith 追踪把每一步的输入输出、耗时、token 都可视化出来。调试复杂 Agent 时可视化追踪能省掉大量 print。4.5 用 create_agent 快速对比如果你只是想快速起一个标准 AgentLangGraph 提供了高层封装from langgraph.prebuilt import create_react_agent agent create_react_agent(llm, tools) result agent.invoke({messages: [(user, 上海天气如何)]})几行代码搞定内部帮你建好了和上面手写等价的图。我的建议是原型阶段用create_react_agent需要定制状态、加人工审核、改路由逻辑时再迁移到手写StateGraph。两者可以平滑过渡因为底层是同一套机制。5. 常见问题与排查技巧实录5.1 消息被覆盖导致 Agent 失忆这是最高频的问题。症状是 Agent 每轮只记得当前这一句历史全丢。根因几乎都是状态里的messages字段没加add_messagesreducer。检查方法打印状态里messages的长度如果每轮都是 1那就是被覆盖了。修复就是加上Annotated[list, add_messages]。5.2 无限循环停不下来Agent 反复调用同一个工具或者两个节点来回跳。原因通常是路由函数判断条件太宽松或者工具返回的结果让模型误以为没成功。排查思路先加最大迭代计数在路由函数里判断超过阈值就强制end再检查工具返回内容是否清晰比如失败时返回明确的错误信息而不是空字符串。def should_continue(state: AgentState) - str: if state.get(tool_calls_count, 0) 10: return end last state[messages][-1] return tools if getattr(last, tool_calls, None) else end5.3 工具调用参数解析失败模型生成的参数格式不对ToolNode执行时报错。常见于工具参数是复杂嵌套结构时。解决办法把工具的参数 schema 写清楚用 Pydantic 模型定义参数并在 docstring 里给出示例。模型对清晰的 schema 遵从度明显更高。5.4 检查点恢复后状态错乱用thread_id恢复时发现状态不对。多半是状态里有不可序列化对象或者多个thread_id混用了同一个 checkpointer 但没隔离。确保每个会话用独立thread_id状态里只放可序列化数据。问题现象可能原因排查方向Agent 失忆messages 无 reducer检查 Annotated 声明无限循环路由条件太松加迭代上限参数解析失败schema 不清晰用 Pydantic 定义参数恢复后错乱状态不可序列化清理状态字段节点找不到边路由返回值不匹配用常量替代字符串5.5 并行节点结果合并冲突当你用并行边同时跑多个节点它们返回的字段如果没配好 reducer会互相覆盖。原则是并行写入的字段必须用可交换、可结合的 reducer比如加法、集合并集否则结果依赖执行顺序不可预测。我一般避免让并行节点写同一个字段各写各的最后用一个汇总节点合并。6. 我踩过的坑和几条实在建议迁移到 LangGraph 这半年最大的体会是框架帮你管的是控制流但业务语义还得自己想清楚。图能保证循环正确、状态不丢、中断可恢复但它不知道你的 Agent 该在什么情况下停、工具失败了该怎么退。这些判断逻辑仍然要你写进路由函数和节点里。几条具体建议。第一状态字段宁少勿多每加一个字段都要问它真的需要在节点间共享吗能放局部变量的别放状态。第二路由函数保持纯粹只读状态做判断别在里面调模型或发请求否则并行和恢复时会出问题。第三尽早接入追踪工具Agent 的调试靠 print 是撑不住的可视化每一步的输入输出能让你少熬很多夜。第四别一上来就追求多智能体单 Agent 加几个工具能解决 90% 的问题多智能体的协调复杂度是指数级上升的。最后分享一个小技巧把常用的图结构ReAct、带审核、带并行检索封装成工厂函数项目里复用。我现在的做法是维护一个graph_patterns.py每个模式一个函数需要时改改参数就能用比每次从零搭图快得多。这个习惯让我在几个项目间切换时省了大量重复劳动。
网站建设高端定制企业官网