1. 从 LangChain 到 LangGraph为什么需要状态图1.1 一个真实的需求场景去年下半年我接了个内部工具的项目需求说起来不复杂用户输入一段自然语言系统自动判断意图然后决定是查数据库、调外部接口还是直接让模型回答。听起来就是个典型的 Agent 场景用 LangChain 的 AgentExecutor 搭一下就行。结果真上手才发现问题。LangChain 的 Agent 抽象把思考-行动-观察这个循环封装得很死我想在中间插一个人工确认的环节或者让某一步失败后走另一条分支就得跟框架的源码较劲。更麻烦的是调试——AgentExecutor 内部到底走了哪条路、传了什么参数日志里看不清楚出了问题只能靠猜。后来我把这套东西迁到了 LangGraph整个思路就顺了。LangGraph 的核心变化在于它不再把 Agent 当成一个黑盒循环而是让你显式地定义一张状态图——节点是处理步骤边是流转方向状态是在节点之间传递的数据。你想加分支就加分支想加人工介入就加个节点想循环就画条回边。控制权回到了开发者手里。这篇文章就围绕 LangGraph 最核心的三块内容展开StateGraph 怎么定义状态和节点、条件路由怎么实现分支决策、Agent 的工具调用循环怎么搭。适合已经用过 LangChain、想往更可控的 Agent 编排方向走的同学也适合完全没接触过 LangGraph 但想搞明白状态图这套思路的人。1.2 LangGraph 和 LangChain 到底是什么关系很多人第一次听到 LangGraph 会问这是不是要取代 LangChain其实不是。LangChain 提供的是组件层的东西——模型封装、提示词模板、工具定义、检索器、输出解析器这些在 LangGraph 里照样用。LangGraph 解决的是编排层的问题——多个步骤之间怎么串、怎么分支、怎么循环、状态怎么管。打个比方LangChain 像是给你一堆乐高积木LangGraph 是给你一张可以自己画的图纸。你可以用 LangChain 的积木按 LangGraph 的图纸搭出复杂的结构。所以实际项目里常见的组合是用 LangChain 定义工具和模型用 LangGraph 编排整个流程。那什么时候该用 LangGraph什么时候 LangChain 的 Chain 就够了我的判断标准很简单如果你的流程是线性的、不需要根据中间结果改变走向用 Chain 就行如果流程里有分支、有循环、有需要暂停等待外部输入的地方就上 LangGraph。Agent 的工具调用天然是个循环所以 LangGraph 在这个场景下优势特别明显。1.3 状态图这套思路的核心价值传统编程里我们写流程控制用的是 if-else、while、函数调用。LangGraph 把这套东西抽象成了图节点是函数边是跳转状态是共享的上下文。这个抽象的好处在于它把流程长什么样和每一步做什么解耦了。你可以先画图——有哪些节点、怎么连、在哪儿分叉——然后再逐个实现节点逻辑。改流程的时候只动图的结构不用去翻每个函数的内部实现。对于 Agent 这种流程本身就很动态的场景这种解耦带来的可维护性提升是实打实的。另一个价值是可观测性。因为流程是显式定义的每一步的输入输出状态都能被记录和检查。我在项目里加了个简单的日志节点每次状态更新都打一份快照调试的时候直接看状态怎么变的比在 AgentExecutor 里翻日志舒服太多。2. StateGraph 核心概念拆解2.1 State 的定义共享上下文怎么设计StateGraph 里的 State 是整个图的共享状态所有节点都能读到它、返回对它的更新。定义 State 有两种方式一种是用 TypedDict一种是用 Pydantic 模型。我一般用 TypedDict因为轻量而且 LangGraph 对它的支持最直接。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_intent: str tool_result: str retry_count: int这里有个关键点Annotated[list, add_messages]里的add_messages是个reducer。默认情况下节点返回的新值会覆盖旧值但加了 reducer 之后返回的消息会追加到原列表里而不是替换。这个设计对对话场景特别重要——你肯定不希望第二轮对话把第一轮的消息冲掉。我自己踩过的坑是一开始没加 reducer结果每轮对话模型只能看到最后一条消息上下文全丢了回答驴唇不对马嘴。后来才明白 reducer 的作用。除了add_messages你也可以自定义 reducer比如想让某个计数字段累加而不是覆盖就写个lambda x, y: x y。State 的字段设计也有讲究。我建议把对话消息、中间结果、控制变量分开存。对话消息用 messages 字段中间结果比如工具返回值单独放控制变量比如重试次数、当前步骤名也单独放。这样调试的时候一眼能看出流程走到哪了而不是把所有东西都塞进 messages 里。2.2 节点函数每个节点该做什么节点就是普通的 Python 函数接收 State返回一个字典表示对 State 的更新。注意是更新不是替换——你返回什么字段就更新什么字段没返回的字段保持不变。def classify_intent(state: AgentState): last_message state[messages][-1] # 调用模型判断意图 intent llm.invoke(f判断这句话的意图{last_message.content}) return {user_intent: intent.content}节点函数的设计原则我总结了几条。第一一个节点只做一件事。判断意图的节点就只判断意图不要顺手把工具也调了。这样图的结构才清晰出问题也好定位。第二节点要幂等。因为 LangGraph 支持重试和断点恢复同一个节点可能被执行多次如果它有副作用比如写数据库要保证重复执行不出问题。第三节点返回的字段要明确。别返回一堆用不上的字段增加状态复杂度。还有个细节节点函数可以是同步的也可以是异步的。如果你的节点里有 IO 操作调模型、查数据库建议用 async def这样多个节点可以并发执行整体吞吐会好很多。LangGraph 对 async 节点的支持很完善用ainvoke触发即可。2.3 边的类型普通边、条件边、入口出口边决定了节点之间怎么流转。LangGraph 里有几种边普通边用add_edge(source, target)定义表示 source 执行完一定走 target。这是最简单的线性流转。条件边用add_conditional_edges(source, router, mapping)定义表示 source 执行完后根据 router 函数的返回值决定走哪个 target。这是实现分支的核心下一节详细讲。入口和出口比较特殊。入口用set_entry_point(node)指定从哪个节点开始出口用add_edge(node, END)表示流程结束。END 是个特殊常量代表图的终点。from langgraph.graph import StateGraph, END graph StateGraph(AgentState) graph.add_node(classify, classify_intent) graph.add_node(call_tool, call_tool) graph.add_node(respond, generate_response) graph.set_entry_point(classify) graph.add_conditional_edges(classify, route_by_intent, { tool: call_tool, direct: respond }) graph.add_edge(call_tool, respond) graph.add_edge(respond, END) app graph.compile()这段代码画出来的图是从 classify 开始根据意图决定是走 call_tool 还是直接 respond如果走了 call_tool执行完再汇合到 respond最后结束。整个流程一目了然。2.4 编译与执行compile 之后发生了什么graph.compile()这一步会把图的结构校验一遍——有没有孤立节点、有没有指向不存在的节点、入口设了没——校验通过后返回一个可执行的对象。这个对象有invoke、ainvoke、stream、astream几个方法。invoke是同步执行传初始状态进去返回最终状态。stream是流式执行每走完一个节点就 yield 一次当前状态适合做实时进度展示。我在做前端联调的时候特别爱用 stream用户能看到正在判断意图正在调用工具这种进度提示体验好很多。result app.invoke({messages: [(user, 帮我查下明天的天气)]}) for chunk in app.stream({messages: [(user, 帮我查下明天的天气)]}): print(chunk)编译的时候还可以传 checkpointer用来做状态持久化。这个后面讲人工介入的时候会用到。另外 compile 支持interrupt_before和interrupt_after参数指定在某些节点前后暂停这是实现人工确认的关键机制。3. 条件路由让流程会拐弯3.1 路由函数的写法与返回值约定条件路由的核心是一个 router 函数它接收当前 State返回一个字符串这个字符串会被映射到具体的下一个节点。写法上没什么魔法就是个普通函数def route_by_intent(state: AgentState) - str: intent state[user_intent] if 查询 in intent or 搜索 in intent: return tool elif 计算 in intent: return tool else: return direct返回值必须是add_conditional_edges的 mapping 字典里出现过的 key否则运行时会报错。我建议把 mapping 的 key 定义成常量别用裸字符串不然改的时候容易漏。路由函数里可以做的事情很多读状态、调模型、查配置、做规则匹配。但有个原则——路由函数应该只做决策不做副作用操作。它不该去调工具、写数据库因为路由函数可能被多次调用比如在流式执行或重试时有副作用会出乱子。3.2 多分支路由的映射表设计当分支变多的时候mapping 表会变得很长。这时候我习惯把路由逻辑和映射分开管理ROUTE_MAP { search: search_node, calculate: calc_node, chat: chat_node, unknown: fallback_node, } def router(state): intent state.get(user_intent, unknown) return intent if intent in ROUTE_MAP else unknown graph.add_conditional_edges(classify, router, ROUTE_MAP)这样加分支的时候只改 ROUTE_MAP路由函数不用动。而且 ROUTE_MAP 可以做成配置从外部加载方便不同环境用不同的路由策略。还有个技巧给路由加个兜底分支。不管什么情况都要保证 router 有返回值且这个值在 mapping 里。我一般会加个 unknown 分支指向一个兜底节点避免因为模型输出异常导致整个图跑挂。3.3 循环路由Agent 循环的关键Agent 的工具调用本质是个循环模型决定调工具 → 执行工具 → 把结果给模型 → 模型再决定 → 直到模型不再调工具为止。这个循环在 LangGraph 里就是用条件边实现的。def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return continue return end graph.add_conditional_edges(agent, should_continue, { continue: tools, end: END }) graph.add_edge(tools, agent)这段代码画出来就是agent 节点判断要不要调工具要调就走 tools 节点tools 执行完回到 agent形成循环不要调就结束。这就是 ReAct 模式在 LangGraph 里的标准实现。循环路由最容易出的问题是死循环。模型可能一直觉得需要调工具或者工具一直返回错误让模型反复重试。我的做法是在 State 里加个retry_count每次循环加一超过阈值就强制走 end 分支。这个阈值设多少合适看场景一般 5 到 10 次够了超过这个数基本说明有问题该人工介入了。3.4 路由中的状态更新陷阱有个坑我踩过好几次在路由函数里改状态是无效的。路由函数只负责返回下一个节点的名字它返回的字符串不会被当成状态更新。如果你想在路由的时候顺便更新状态得在路由之前的节点里做或者专门加个节点。另一个坑是条件边的执行时机。条件边是在 source 节点执行完之后立即求值的此时 source 节点返回的状态更新已经合并进去了。所以路由函数能读到 source 节点刚写的字段。但如果你在 source 节点里做了异步操作还没 await 完路由函数可能读到旧值。这个在异步场景下要特别注意。还有个细节多个条件边指向同一个节点是允许的这叫多入边。LangGraph 会等所有指向该节点的边都满足条件才执行还是任一满足就执行默认是任一满足就执行但你可以通过配置改成等待所有。这个在并行分支汇合的场景下很有用。4. Agent 工具调用循环的完整实现4.1 工具定义与绑定工具定义用 LangChain 的tool装饰器这个和 LangGraph 是通用的from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。参数 city 是城市名。 # 实际调用天气 API return f{city}今天晴25度 tool def calculate(expression: str) - str: 计算数学表达式。参数 expression 是算式字符串。 return str(eval(expression)) tools [get_weather, calculate] llm_with_tools llm.bind_tools(tools)工具定义有几个要点。docstring 必须写清楚因为模型是根据 docstring 来判断什么时候调这个工具的。参数说明、功能描述都要写明白别偷懒。参数类型要标注LangChain 会根据类型生成 JSON Schema 给模型看。工具函数要能处理异常别让一个工具报错把整个图搞挂。bind_tools这一步是把工具的描述信息塞进模型的调用参数里让模型知道有哪些工具可用。注意不是所有模型都支持工具调用用之前确认下你用的模型有没有这个能力。4.2 Agent 节点的实现Agent 节点是循环的核心它负责调模型、拿决策def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]}就这么简单。模型返回的消息里如果带tool_calls说明它想调工具不带就说明它想直接回答。这个判断在路由函数里做。但实际项目里我会在这个节点里加点东西。比如加系统提示词告诉模型它的角色和可用工具比如做消息裁剪只把最近 N 轮消息给模型避免上下文超长比如记录调用日志方便排查问题。def agent_node(state: AgentState): messages [SystemMessage(contentSYSTEM_PROMPT)] state[messages][-10:] response llm_with_tools.invoke(messages) return {messages: [response]}消息裁剪这块要注意别把带 tool_calls 的消息裁掉了否则工具结果找不到对应的调用会报错。我一般是按轮裁保证一个完整的工具调用周期不被切断。4.3 工具执行节点的实现工具执行节点负责把模型请求的工具调用真正执行掉from langgraph.prebuilt import ToolNode tool_node ToolNode(tools)LangGraph 提供了ToolNode这个预置节点直接传工具列表进去就行它会自动解析模型消息里的 tool_calls挨个执行把结果包装成 ToolMessage 返回。省事。如果你想自己控制执行过程也可以手写def tool_node(state: AgentState): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: tool tool_map[tool_call[name]] try: result tool.invoke(tool_call[args]) except Exception as e: result f工具执行失败{e} results.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) return {messages: results}手写的好处是能加错误处理、能加日志、能对结果做后处理。我一般会在工具执行失败时返回一个描述错误的 ToolMessage而不是直接抛异常这样模型能看到错误信息有机会自己纠正。4.4 循环终止条件的设计循环什么时候停标准做法是看模型最后一条消息有没有 tool_calls。没有就停。但实际项目里我会加几个额外的终止条件终止条件触发场景处理方式模型不再调工具正常结束走 END重试次数超限模型反复调工具强制结束返回兜底回复工具连续失败工具本身有问题结束并提示用户达到最大步数防止无限循环结束并记录日志def should_continue(state: AgentState) - str: if state.get(retry_count, 0) MAX_RETRIES: return end last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return continue return end重试计数在 agent 节点里更新def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return { messages: [response], retry_count: state.get(retry_count, 0) 1 }这套终止条件组合下来基本能覆盖绝大多数异常情况。我实测下来正常对话很少超过 3 轮循环超过 5 轮基本就是出问题了。5. 实操中的常见问题与排查5.1 状态字段丢失或覆盖最常见的报错是KeyError: 某个字段不存在。原因通常是节点返回的字典里没包含这个字段而下游节点又去读它。解决办法有两个一是用state.get(field, default)读给个默认值二是在图的初始状态里把所有字段都初始化好。还有个隐蔽的坑reducer 用错导致数据被覆盖。比如 messages 字段忘了加add_messages结果每轮对话都只剩最后一条。这个问题的表现是模型失忆回答和上文对不上。排查方法是在每个节点后打印 state看 messages 的长度是不是在增长。5.2 条件路由返回了未映射的值报错信息一般是Invalid route: xxx。原因是 router 函数返回的字符串不在 mapping 字典的 key 里。排查步骤先打印 router 的返回值再对照 mapping 的 key 看差在哪。常见情况是模型输出的意图分类和预设的 key 对不上比如模型返回查询天气你 mapping 里写的是search。我的做法是在 router 里做一层归一化把模型输出映射到标准 keyINTENT_ALIAS { 查询: search, 搜索: search, 查找: search, 计算: calculate, } def router(state): raw state[user_intent] for keyword, intent in INTENT_ALIAS.items(): if keyword in raw: return intent return unknown5.3 工具调用参数解析失败模型生成的 tool_calls 里 args 是 JSON 格式如果模型生成的 JSON 不合法解析就会失败。这个在用小模型或者复杂参数结构时特别常见。缓解办法一是把工具参数设计得简单点别搞嵌套对象二是在工具定义里把参数说明写清楚给模型足够的提示三是在工具执行节点里加 try-except解析失败时返回错误信息让模型重试。5.4 异步节点的并发问题用 async 节点的时候如果多个节点共享了可变对象比如一个全局的 list并发执行会出问题。LangGraph 的异步执行是并发的同一个状态字段可能被多个节点同时更新。解决办法是避免在节点里改全局状态所有状态更新都通过返回值走 LangGraph 的合并机制。5.5 常见问题速查表问题现象可能原因排查方向KeyError 字段不存在节点没返回该字段检查各节点返回值模型失忆messages 没加 reducer检查 Annotated 配置Invalid routerouter 返回值未映射打印 router 返回值工具调用失败args 解析错误检查工具参数定义死循环终止条件没生效检查 retry_count 逻辑状态更新丢失异步并发冲突避免改全局状态6. 从能跑到好用几个进阶技巧6.1 用 checkpointer 做状态持久化LangGraph 支持把每一步的状态存到数据库这样图可以在任意节点暂停和恢复。这个能力对长流程和人工介入场景特别有用。配置方式是在 compile 的时候传 checkpointerfrom langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} app.invoke({messages: [...]}, config)thread_id 是会话标识同一个 thread_id 的多次 invoke 会共享状态。这意味着你可以先跑一半过一会儿再接着跑状态不会丢。做多轮对话的时候这个特别方便不用自己维护历史消息。6.2 人工介入节点的实现有些操作风险高需要人工确认后才能执行。LangGraph 用interrupt实现这个from langgraph.types import interrupt def human_approval_node(state: AgentState): decision interrupt(请确认是否执行该操作) return {approved: decision}图执行到这个节点会暂停把控制权交回调用方。调用方确认后用Command(resumeTrue)恢复执行。这个机制在做敏感操作比如转账、删除数据的 Agent 时是刚需。6.3 多 Agent 协作的图结构单个 Agent 搞不定复杂任务时可以拆成多个 Agent每个负责一块用图编排它们之间的协作。常见模式有主管模式一个 supervisor 节点负责分派任务多个 worker 节点各干各的干完汇报给 supervisor由 supervisor 决定下一步。这个模式适合任务可以明确拆分的场景。流水线模式多个 Agent 串成一条线前一个的输出是后一个的输入。适合有明确阶段划分的任务比如检索 → 分析 → 撰写 → 校对。辩论模式两个 Agent 对同一问题给出不同答案第三个 Agent 做裁判。适合需要多角度验证的场景。这几种模式在 LangGraph 里都是用节点和条件边组合出来的没有额外的抽象。理解了 StateGraph 的基本用法这些模式都能自己搭。6.4 性能优化的几个方向Agent 跑得慢是常见抱怨。优化方向有几个并行化——没有依赖关系的节点可以并发执行LangGraph 支持一个节点触发多个下游节点缓存——模型调用和工具调用都可以加缓存相同输入直接返回结果流式输出——用 stream 而不是 invoke用户能更早看到部分结果模型分级——简单判断用小模型复杂推理用大模型成本和质量兼顾。我自己项目里最有效的优化是减少不必要的模型调用。很多判断其实用规则就能做没必要每次都问模型。比如意图分类先用关键词匹配一遍匹配不上再调模型能省掉一大半调用。6.5 调试与可观测性LangGraph 的调试比 LangChain 的 Agent 友好很多因为流程是显式的。我常用的几个手段在每个节点入口打印 state 快照看数据怎么流动的用 stream 模式跑看每个节点的执行顺序给节点加计时找出耗时瓶颈记录模型输入输出排查模型行为异常。如果项目规模大了建议接入 LangSmith 这类可观测性平台能看到完整的调用链、每步的耗时和 token 消耗。不过小项目用 print 加日志文件就够了别过度工程。7. 我踩过的几个印象深刻的坑第一个坑是状态字段命名冲突。我在 State 里定义了个result字段结果某个工具返回的字典里也有result合并的时候把状态里的覆盖了。后来我养成了习惯状态字段加前缀比如agent_result、tool_result避免和工具返回值撞名。第二个坑是条件边和普通边混用导致流程不符合预期。我一开始给一个节点同时加了普通边和条件边以为会按顺序走结果 LangGraph 是并行触发的两个下游节点都执行了。后来才明白一个节点的出边要么全是普通边要么用条件边控制别混着来。第三个坑是异步节点里用了同步的模型调用。节点定义成 async def但里面调的是同步的llm.invoke结果整个图卡住不动。改成await llm.ainvoke才正常。这个坑很隐蔽因为代码不报错就是慢排查了半天才发现是同步阻塞了事件循环。第四个坑是checkpointer 的 thread_id 复用。我用同一个 thread_id 跑不同的会话结果状态串了上一个会话的消息出现在下一个会话里。后来改成每个会话生成唯一的 thread_id问题解决。这个在做多用户系统的时候要特别注意。8. 学习路径与资源建议如果你是刚接触 LangGraph我的建议是先别急着看官方文档的全部内容容易迷失在细节里。按这个顺序来先搞懂 StateGraph 的三个核心概念——State、Node、Edge用最简单的线性图跑通一个 hello world然后加条件边实现一个二分支的流程接着实现 Agent 循环把工具调用跑通最后再去看 checkpointer、interrupt 这些进阶特性。官方文档里langgraph的 tutorials 部分有几个很好的例子特别是 ReAct Agent 那个基本把核心用法都覆盖了。中文资料目前还比较少遇到问题建议直接看源码LangGraph 的代码量不大核心逻辑读一遍就清楚了。面试里如果被问到 LangGraph 和 LangChain 的区别我的回答框架是LangChain 是组件库提供模型、工具、检索这些积木LangGraph 是编排框架提供状态图这套机制来组织积木。两者是互补关系不是替代关系。LangGraph 解决的核心问题是复杂流程的可控性和可观测性特别是 Agent 这种有循环和分支的场景。最后分享一个我自己的体会LangGraph 最大的价值不是让你少写代码而是让你的 Agent 逻辑变得可见、可改、可调试。以前用 AgentExecutor 的时候流程藏在框架内部出了问题只能猜现在流程画在图上每一步都清清楚楚。这个心智负担的降低比省那几行代码重要得多。
LangGraph 状态图实战：StateGraph、条件路由与 Agent 工具调用循环
1. 从 LangChain 到 LangGraph为什么需要状态图1.1 一个真实的需求场景去年下半年我接了个内部工具的项目需求说起来不复杂用户输入一段自然语言系统自动判断意图然后决定是查数据库、调外部接口还是直接让模型回答。听起来就是个典型的 Agent 场景用 LangChain 的 AgentExecutor 搭一下就行。结果真上手才发现问题。LangChain 的 Agent 抽象把思考-行动-观察这个循环封装得很死我想在中间插一个人工确认的环节或者让某一步失败后走另一条分支就得跟框架的源码较劲。更麻烦的是调试——AgentExecutor 内部到底走了哪条路、传了什么参数日志里看不清楚出了问题只能靠猜。后来我把这套东西迁到了 LangGraph整个思路就顺了。LangGraph 的核心变化在于它不再把 Agent 当成一个黑盒循环而是让你显式地定义一张状态图——节点是处理步骤边是流转方向状态是在节点之间传递的数据。你想加分支就加分支想加人工介入就加个节点想循环就画条回边。控制权回到了开发者手里。这篇文章就围绕 LangGraph 最核心的三块内容展开StateGraph 怎么定义状态和节点、条件路由怎么实现分支决策、Agent 的工具调用循环怎么搭。适合已经用过 LangChain、想往更可控的 Agent 编排方向走的同学也适合完全没接触过 LangGraph 但想搞明白状态图这套思路的人。1.2 LangGraph 和 LangChain 到底是什么关系很多人第一次听到 LangGraph 会问这是不是要取代 LangChain其实不是。LangChain 提供的是组件层的东西——模型封装、提示词模板、工具定义、检索器、输出解析器这些在 LangGraph 里照样用。LangGraph 解决的是编排层的问题——多个步骤之间怎么串、怎么分支、怎么循环、状态怎么管。打个比方LangChain 像是给你一堆乐高积木LangGraph 是给你一张可以自己画的图纸。你可以用 LangChain 的积木按 LangGraph 的图纸搭出复杂的结构。所以实际项目里常见的组合是用 LangChain 定义工具和模型用 LangGraph 编排整个流程。那什么时候该用 LangGraph什么时候 LangChain 的 Chain 就够了我的判断标准很简单如果你的流程是线性的、不需要根据中间结果改变走向用 Chain 就行如果流程里有分支、有循环、有需要暂停等待外部输入的地方就上 LangGraph。Agent 的工具调用天然是个循环所以 LangGraph 在这个场景下优势特别明显。1.3 状态图这套思路的核心价值传统编程里我们写流程控制用的是 if-else、while、函数调用。LangGraph 把这套东西抽象成了图节点是函数边是跳转状态是共享的上下文。这个抽象的好处在于它把流程长什么样和每一步做什么解耦了。你可以先画图——有哪些节点、怎么连、在哪儿分叉——然后再逐个实现节点逻辑。改流程的时候只动图的结构不用去翻每个函数的内部实现。对于 Agent 这种流程本身就很动态的场景这种解耦带来的可维护性提升是实打实的。另一个价值是可观测性。因为流程是显式定义的每一步的输入输出状态都能被记录和检查。我在项目里加了个简单的日志节点每次状态更新都打一份快照调试的时候直接看状态怎么变的比在 AgentExecutor 里翻日志舒服太多。2. StateGraph 核心概念拆解2.1 State 的定义共享上下文怎么设计StateGraph 里的 State 是整个图的共享状态所有节点都能读到它、返回对它的更新。定义 State 有两种方式一种是用 TypedDict一种是用 Pydantic 模型。我一般用 TypedDict因为轻量而且 LangGraph 对它的支持最直接。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_intent: str tool_result: str retry_count: int这里有个关键点Annotated[list, add_messages]里的add_messages是个reducer。默认情况下节点返回的新值会覆盖旧值但加了 reducer 之后返回的消息会追加到原列表里而不是替换。这个设计对对话场景特别重要——你肯定不希望第二轮对话把第一轮的消息冲掉。我自己踩过的坑是一开始没加 reducer结果每轮对话模型只能看到最后一条消息上下文全丢了回答驴唇不对马嘴。后来才明白 reducer 的作用。除了add_messages你也可以自定义 reducer比如想让某个计数字段累加而不是覆盖就写个lambda x, y: x y。State 的字段设计也有讲究。我建议把对话消息、中间结果、控制变量分开存。对话消息用 messages 字段中间结果比如工具返回值单独放控制变量比如重试次数、当前步骤名也单独放。这样调试的时候一眼能看出流程走到哪了而不是把所有东西都塞进 messages 里。2.2 节点函数每个节点该做什么节点就是普通的 Python 函数接收 State返回一个字典表示对 State 的更新。注意是更新不是替换——你返回什么字段就更新什么字段没返回的字段保持不变。def classify_intent(state: AgentState): last_message state[messages][-1] # 调用模型判断意图 intent llm.invoke(f判断这句话的意图{last_message.content}) return {user_intent: intent.content}节点函数的设计原则我总结了几条。第一一个节点只做一件事。判断意图的节点就只判断意图不要顺手把工具也调了。这样图的结构才清晰出问题也好定位。第二节点要幂等。因为 LangGraph 支持重试和断点恢复同一个节点可能被执行多次如果它有副作用比如写数据库要保证重复执行不出问题。第三节点返回的字段要明确。别返回一堆用不上的字段增加状态复杂度。还有个细节节点函数可以是同步的也可以是异步的。如果你的节点里有 IO 操作调模型、查数据库建议用 async def这样多个节点可以并发执行整体吞吐会好很多。LangGraph 对 async 节点的支持很完善用ainvoke触发即可。2.3 边的类型普通边、条件边、入口出口边决定了节点之间怎么流转。LangGraph 里有几种边普通边用add_edge(source, target)定义表示 source 执行完一定走 target。这是最简单的线性流转。条件边用add_conditional_edges(source, router, mapping)定义表示 source 执行完后根据 router 函数的返回值决定走哪个 target。这是实现分支的核心下一节详细讲。入口和出口比较特殊。入口用set_entry_point(node)指定从哪个节点开始出口用add_edge(node, END)表示流程结束。END 是个特殊常量代表图的终点。from langgraph.graph import StateGraph, END graph StateGraph(AgentState) graph.add_node(classify, classify_intent) graph.add_node(call_tool, call_tool) graph.add_node(respond, generate_response) graph.set_entry_point(classify) graph.add_conditional_edges(classify, route_by_intent, { tool: call_tool, direct: respond }) graph.add_edge(call_tool, respond) graph.add_edge(respond, END) app graph.compile()这段代码画出来的图是从 classify 开始根据意图决定是走 call_tool 还是直接 respond如果走了 call_tool执行完再汇合到 respond最后结束。整个流程一目了然。2.4 编译与执行compile 之后发生了什么graph.compile()这一步会把图的结构校验一遍——有没有孤立节点、有没有指向不存在的节点、入口设了没——校验通过后返回一个可执行的对象。这个对象有invoke、ainvoke、stream、astream几个方法。invoke是同步执行传初始状态进去返回最终状态。stream是流式执行每走完一个节点就 yield 一次当前状态适合做实时进度展示。我在做前端联调的时候特别爱用 stream用户能看到正在判断意图正在调用工具这种进度提示体验好很多。result app.invoke({messages: [(user, 帮我查下明天的天气)]}) for chunk in app.stream({messages: [(user, 帮我查下明天的天气)]}): print(chunk)编译的时候还可以传 checkpointer用来做状态持久化。这个后面讲人工介入的时候会用到。另外 compile 支持interrupt_before和interrupt_after参数指定在某些节点前后暂停这是实现人工确认的关键机制。3. 条件路由让流程会拐弯3.1 路由函数的写法与返回值约定条件路由的核心是一个 router 函数它接收当前 State返回一个字符串这个字符串会被映射到具体的下一个节点。写法上没什么魔法就是个普通函数def route_by_intent(state: AgentState) - str: intent state[user_intent] if 查询 in intent or 搜索 in intent: return tool elif 计算 in intent: return tool else: return direct返回值必须是add_conditional_edges的 mapping 字典里出现过的 key否则运行时会报错。我建议把 mapping 的 key 定义成常量别用裸字符串不然改的时候容易漏。路由函数里可以做的事情很多读状态、调模型、查配置、做规则匹配。但有个原则——路由函数应该只做决策不做副作用操作。它不该去调工具、写数据库因为路由函数可能被多次调用比如在流式执行或重试时有副作用会出乱子。3.2 多分支路由的映射表设计当分支变多的时候mapping 表会变得很长。这时候我习惯把路由逻辑和映射分开管理ROUTE_MAP { search: search_node, calculate: calc_node, chat: chat_node, unknown: fallback_node, } def router(state): intent state.get(user_intent, unknown) return intent if intent in ROUTE_MAP else unknown graph.add_conditional_edges(classify, router, ROUTE_MAP)这样加分支的时候只改 ROUTE_MAP路由函数不用动。而且 ROUTE_MAP 可以做成配置从外部加载方便不同环境用不同的路由策略。还有个技巧给路由加个兜底分支。不管什么情况都要保证 router 有返回值且这个值在 mapping 里。我一般会加个 unknown 分支指向一个兜底节点避免因为模型输出异常导致整个图跑挂。3.3 循环路由Agent 循环的关键Agent 的工具调用本质是个循环模型决定调工具 → 执行工具 → 把结果给模型 → 模型再决定 → 直到模型不再调工具为止。这个循环在 LangGraph 里就是用条件边实现的。def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return continue return end graph.add_conditional_edges(agent, should_continue, { continue: tools, end: END }) graph.add_edge(tools, agent)这段代码画出来就是agent 节点判断要不要调工具要调就走 tools 节点tools 执行完回到 agent形成循环不要调就结束。这就是 ReAct 模式在 LangGraph 里的标准实现。循环路由最容易出的问题是死循环。模型可能一直觉得需要调工具或者工具一直返回错误让模型反复重试。我的做法是在 State 里加个retry_count每次循环加一超过阈值就强制走 end 分支。这个阈值设多少合适看场景一般 5 到 10 次够了超过这个数基本说明有问题该人工介入了。3.4 路由中的状态更新陷阱有个坑我踩过好几次在路由函数里改状态是无效的。路由函数只负责返回下一个节点的名字它返回的字符串不会被当成状态更新。如果你想在路由的时候顺便更新状态得在路由之前的节点里做或者专门加个节点。另一个坑是条件边的执行时机。条件边是在 source 节点执行完之后立即求值的此时 source 节点返回的状态更新已经合并进去了。所以路由函数能读到 source 节点刚写的字段。但如果你在 source 节点里做了异步操作还没 await 完路由函数可能读到旧值。这个在异步场景下要特别注意。还有个细节多个条件边指向同一个节点是允许的这叫多入边。LangGraph 会等所有指向该节点的边都满足条件才执行还是任一满足就执行默认是任一满足就执行但你可以通过配置改成等待所有。这个在并行分支汇合的场景下很有用。4. Agent 工具调用循环的完整实现4.1 工具定义与绑定工具定义用 LangChain 的tool装饰器这个和 LangGraph 是通用的from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。参数 city 是城市名。 # 实际调用天气 API return f{city}今天晴25度 tool def calculate(expression: str) - str: 计算数学表达式。参数 expression 是算式字符串。 return str(eval(expression)) tools [get_weather, calculate] llm_with_tools llm.bind_tools(tools)工具定义有几个要点。docstring 必须写清楚因为模型是根据 docstring 来判断什么时候调这个工具的。参数说明、功能描述都要写明白别偷懒。参数类型要标注LangChain 会根据类型生成 JSON Schema 给模型看。工具函数要能处理异常别让一个工具报错把整个图搞挂。bind_tools这一步是把工具的描述信息塞进模型的调用参数里让模型知道有哪些工具可用。注意不是所有模型都支持工具调用用之前确认下你用的模型有没有这个能力。4.2 Agent 节点的实现Agent 节点是循环的核心它负责调模型、拿决策def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]}就这么简单。模型返回的消息里如果带tool_calls说明它想调工具不带就说明它想直接回答。这个判断在路由函数里做。但实际项目里我会在这个节点里加点东西。比如加系统提示词告诉模型它的角色和可用工具比如做消息裁剪只把最近 N 轮消息给模型避免上下文超长比如记录调用日志方便排查问题。def agent_node(state: AgentState): messages [SystemMessage(contentSYSTEM_PROMPT)] state[messages][-10:] response llm_with_tools.invoke(messages) return {messages: [response]}消息裁剪这块要注意别把带 tool_calls 的消息裁掉了否则工具结果找不到对应的调用会报错。我一般是按轮裁保证一个完整的工具调用周期不被切断。4.3 工具执行节点的实现工具执行节点负责把模型请求的工具调用真正执行掉from langgraph.prebuilt import ToolNode tool_node ToolNode(tools)LangGraph 提供了ToolNode这个预置节点直接传工具列表进去就行它会自动解析模型消息里的 tool_calls挨个执行把结果包装成 ToolMessage 返回。省事。如果你想自己控制执行过程也可以手写def tool_node(state: AgentState): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: tool tool_map[tool_call[name]] try: result tool.invoke(tool_call[args]) except Exception as e: result f工具执行失败{e} results.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) return {messages: results}手写的好处是能加错误处理、能加日志、能对结果做后处理。我一般会在工具执行失败时返回一个描述错误的 ToolMessage而不是直接抛异常这样模型能看到错误信息有机会自己纠正。4.4 循环终止条件的设计循环什么时候停标准做法是看模型最后一条消息有没有 tool_calls。没有就停。但实际项目里我会加几个额外的终止条件终止条件触发场景处理方式模型不再调工具正常结束走 END重试次数超限模型反复调工具强制结束返回兜底回复工具连续失败工具本身有问题结束并提示用户达到最大步数防止无限循环结束并记录日志def should_continue(state: AgentState) - str: if state.get(retry_count, 0) MAX_RETRIES: return end last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return continue return end重试计数在 agent 节点里更新def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return { messages: [response], retry_count: state.get(retry_count, 0) 1 }这套终止条件组合下来基本能覆盖绝大多数异常情况。我实测下来正常对话很少超过 3 轮循环超过 5 轮基本就是出问题了。5. 实操中的常见问题与排查5.1 状态字段丢失或覆盖最常见的报错是KeyError: 某个字段不存在。原因通常是节点返回的字典里没包含这个字段而下游节点又去读它。解决办法有两个一是用state.get(field, default)读给个默认值二是在图的初始状态里把所有字段都初始化好。还有个隐蔽的坑reducer 用错导致数据被覆盖。比如 messages 字段忘了加add_messages结果每轮对话都只剩最后一条。这个问题的表现是模型失忆回答和上文对不上。排查方法是在每个节点后打印 state看 messages 的长度是不是在增长。5.2 条件路由返回了未映射的值报错信息一般是Invalid route: xxx。原因是 router 函数返回的字符串不在 mapping 字典的 key 里。排查步骤先打印 router 的返回值再对照 mapping 的 key 看差在哪。常见情况是模型输出的意图分类和预设的 key 对不上比如模型返回查询天气你 mapping 里写的是search。我的做法是在 router 里做一层归一化把模型输出映射到标准 keyINTENT_ALIAS { 查询: search, 搜索: search, 查找: search, 计算: calculate, } def router(state): raw state[user_intent] for keyword, intent in INTENT_ALIAS.items(): if keyword in raw: return intent return unknown5.3 工具调用参数解析失败模型生成的 tool_calls 里 args 是 JSON 格式如果模型生成的 JSON 不合法解析就会失败。这个在用小模型或者复杂参数结构时特别常见。缓解办法一是把工具参数设计得简单点别搞嵌套对象二是在工具定义里把参数说明写清楚给模型足够的提示三是在工具执行节点里加 try-except解析失败时返回错误信息让模型重试。5.4 异步节点的并发问题用 async 节点的时候如果多个节点共享了可变对象比如一个全局的 list并发执行会出问题。LangGraph 的异步执行是并发的同一个状态字段可能被多个节点同时更新。解决办法是避免在节点里改全局状态所有状态更新都通过返回值走 LangGraph 的合并机制。5.5 常见问题速查表问题现象可能原因排查方向KeyError 字段不存在节点没返回该字段检查各节点返回值模型失忆messages 没加 reducer检查 Annotated 配置Invalid routerouter 返回值未映射打印 router 返回值工具调用失败args 解析错误检查工具参数定义死循环终止条件没生效检查 retry_count 逻辑状态更新丢失异步并发冲突避免改全局状态6. 从能跑到好用几个进阶技巧6.1 用 checkpointer 做状态持久化LangGraph 支持把每一步的状态存到数据库这样图可以在任意节点暂停和恢复。这个能力对长流程和人工介入场景特别有用。配置方式是在 compile 的时候传 checkpointerfrom langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} app.invoke({messages: [...]}, config)thread_id 是会话标识同一个 thread_id 的多次 invoke 会共享状态。这意味着你可以先跑一半过一会儿再接着跑状态不会丢。做多轮对话的时候这个特别方便不用自己维护历史消息。6.2 人工介入节点的实现有些操作风险高需要人工确认后才能执行。LangGraph 用interrupt实现这个from langgraph.types import interrupt def human_approval_node(state: AgentState): decision interrupt(请确认是否执行该操作) return {approved: decision}图执行到这个节点会暂停把控制权交回调用方。调用方确认后用Command(resumeTrue)恢复执行。这个机制在做敏感操作比如转账、删除数据的 Agent 时是刚需。6.3 多 Agent 协作的图结构单个 Agent 搞不定复杂任务时可以拆成多个 Agent每个负责一块用图编排它们之间的协作。常见模式有主管模式一个 supervisor 节点负责分派任务多个 worker 节点各干各的干完汇报给 supervisor由 supervisor 决定下一步。这个模式适合任务可以明确拆分的场景。流水线模式多个 Agent 串成一条线前一个的输出是后一个的输入。适合有明确阶段划分的任务比如检索 → 分析 → 撰写 → 校对。辩论模式两个 Agent 对同一问题给出不同答案第三个 Agent 做裁判。适合需要多角度验证的场景。这几种模式在 LangGraph 里都是用节点和条件边组合出来的没有额外的抽象。理解了 StateGraph 的基本用法这些模式都能自己搭。6.4 性能优化的几个方向Agent 跑得慢是常见抱怨。优化方向有几个并行化——没有依赖关系的节点可以并发执行LangGraph 支持一个节点触发多个下游节点缓存——模型调用和工具调用都可以加缓存相同输入直接返回结果流式输出——用 stream 而不是 invoke用户能更早看到部分结果模型分级——简单判断用小模型复杂推理用大模型成本和质量兼顾。我自己项目里最有效的优化是减少不必要的模型调用。很多判断其实用规则就能做没必要每次都问模型。比如意图分类先用关键词匹配一遍匹配不上再调模型能省掉一大半调用。6.5 调试与可观测性LangGraph 的调试比 LangChain 的 Agent 友好很多因为流程是显式的。我常用的几个手段在每个节点入口打印 state 快照看数据怎么流动的用 stream 模式跑看每个节点的执行顺序给节点加计时找出耗时瓶颈记录模型输入输出排查模型行为异常。如果项目规模大了建议接入 LangSmith 这类可观测性平台能看到完整的调用链、每步的耗时和 token 消耗。不过小项目用 print 加日志文件就够了别过度工程。7. 我踩过的几个印象深刻的坑第一个坑是状态字段命名冲突。我在 State 里定义了个result字段结果某个工具返回的字典里也有result合并的时候把状态里的覆盖了。后来我养成了习惯状态字段加前缀比如agent_result、tool_result避免和工具返回值撞名。第二个坑是条件边和普通边混用导致流程不符合预期。我一开始给一个节点同时加了普通边和条件边以为会按顺序走结果 LangGraph 是并行触发的两个下游节点都执行了。后来才明白一个节点的出边要么全是普通边要么用条件边控制别混着来。第三个坑是异步节点里用了同步的模型调用。节点定义成 async def但里面调的是同步的llm.invoke结果整个图卡住不动。改成await llm.ainvoke才正常。这个坑很隐蔽因为代码不报错就是慢排查了半天才发现是同步阻塞了事件循环。第四个坑是checkpointer 的 thread_id 复用。我用同一个 thread_id 跑不同的会话结果状态串了上一个会话的消息出现在下一个会话里。后来改成每个会话生成唯一的 thread_id问题解决。这个在做多用户系统的时候要特别注意。8. 学习路径与资源建议如果你是刚接触 LangGraph我的建议是先别急着看官方文档的全部内容容易迷失在细节里。按这个顺序来先搞懂 StateGraph 的三个核心概念——State、Node、Edge用最简单的线性图跑通一个 hello world然后加条件边实现一个二分支的流程接着实现 Agent 循环把工具调用跑通最后再去看 checkpointer、interrupt 这些进阶特性。官方文档里langgraph的 tutorials 部分有几个很好的例子特别是 ReAct Agent 那个基本把核心用法都覆盖了。中文资料目前还比较少遇到问题建议直接看源码LangGraph 的代码量不大核心逻辑读一遍就清楚了。面试里如果被问到 LangGraph 和 LangChain 的区别我的回答框架是LangChain 是组件库提供模型、工具、检索这些积木LangGraph 是编排框架提供状态图这套机制来组织积木。两者是互补关系不是替代关系。LangGraph 解决的核心问题是复杂流程的可控性和可观测性特别是 Agent 这种有循环和分支的场景。最后分享一个我自己的体会LangGraph 最大的价值不是让你少写代码而是让你的 Agent 逻辑变得可见、可改、可调试。以前用 AgentExecutor 的时候流程藏在框架内部出了问题只能猜现在流程画在图上每一步都清清楚楚。这个心智负担的降低比省那几行代码重要得多。