1. 从零理解 LangGraph它到底解决什么问题如果你已经用过 LangChain 的 Chain 和 Agent大概率会遇到一个共同的痛点流程一旦复杂起来代码就开始失控。一个带工具调用的 Agent背后其实是一个循环——模型思考、决定调不调工具、调哪个工具、拿到结果再思考直到给出最终答案。用传统 Chain 写你得手动拼 Prompt、解析输出、判断分支、维护循环状态写着写着就变成了一坨 if-else 面条。LangGraph 就是冲着这个问题来的。它把 Agent 的执行过程抽象成一张有向图节点是执行单元边是流转方向状态在节点之间传递和累积。你可以把它理解成给 LLM 应用装了一个流程引擎让多步骤、带分支、带循环的逻辑变得可视化、可调试、可持久化。我最初接触 LangGraph 的时候最直观的感受是它把控制流从 Prompt 里拿出来了。以前你靠 Prompt 里写如果用户问的是天气就调用天气工具现在你直接用代码定义条件路由模型只负责决策流程由图的拓扑结构保证。这个思路的转变是理解 LangGraph 的关键。这篇文章我会围绕三个核心概念展开StateGraph状态图、条件路由Conditional Edge、Agent 的工具调用循环。这三个东西串起来就是一个最小可用的 Agent 骨架。适合已经了解 LangChain 基础、想往 Agent 开发方向深入的人也适合完全没接触过 LangGraph 但想搞明白Agent 到底怎么跑起来的初学者。提示LangGraph 和 LangChain 不是替代关系。LangChain 提供模型封装、工具定义、Prompt 模板这些零件LangGraph 负责把这些零件按你想要的流程编排起来。两者配合使用不是二选一。2. StateGraph 核心机制拆解2.1 状态是什么一张在节点间流转的共享白板StateGraph 的核心是State。你可以把 State 想象成一块共享白板每个节点都能往上面写东西也能读到别人写的东西。在 LangGraph 里State 通常用一个TypedDict或者 Pydantic 模型来定义。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]这里有个关键细节Annotated[list, add_messages]里的add_messages是一个reducer归约函数。它的作用是定义当多个节点都想更新这个字段时怎么合并。add_messages的逻辑是追加而不是覆盖——新消息加到列表末尾而不是把整个列表替换掉。为什么这个设计很重要因为 Agent 的对话历史是累积的。如果每次节点更新都覆盖messages那模型就看不到之前的对话了。用add_messages做 reducer每个节点只需要返回我要新增的消息LangGraph 自动帮你合并到历史里。我踩过的一个坑一开始我自定义 State 的时候忘了加 reducer结果工具调用的结果把用户消息覆盖了模型完全不知道用户问了什么。排查了半天才意识到是 reducer 的问题。所以记住一句话需要累积的字段一定要加 reducer需要覆盖的字段用默认行为就行。2.2 节点与边图的骨架怎么搭节点Node就是一个函数接收当前 State返回要更新的字段。边Edge定义节点之间的流转关系。LangGraph 提供两种边普通边从 A 节点执行完直接到 B 节点无条件。条件边从 A 节点执行完后根据一个路由函数的返回值决定去 B 还是 C。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) 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 循环从 START 进入 agent 节点agent 决定是调工具还是结束如果调工具就去 tools 节点tools 执行完再回到 agent。should_continue就是条件路由函数。START和END是两个特殊节点分别代表图的入口和出口。compile()之后得到的graph是一个可执行对象调用graph.invoke({messages: [...]})就能跑起来。2.3 为什么用图而不是链三个实际好处第一循环是原生支持的。Chain 是线性的要做循环得自己套 while。图天然支持环Agent 的思考-行动-观察循环直接映射成agent - tools - agent的环。第二状态管理是自动的。每个节点的输入输出都经过 State 统一管理不用手动在函数之间传参。节点多了之后这个优势非常明显。第三可持久化和可中断。LangGraph 支持 Checkpointer可以把每一步的状态存下来实现断点续跑、人工审核中断human-in-the-loop。这个在 Chain 里做起来非常麻烦。注意图里的环一定要有退出条件否则会无限循环。条件路由函数必须能在某个时刻返回指向 END 的路由这是新手最容易犯的错误之一。3. 条件路由让 Agent 自己决定下一步3.1 路由函数的本质一个返回字符串的普通函数条件路由的核心是一个函数它读 State返回一个字符串LangGraph 根据这个字符串去映射表里找下一个节点。from langgraph.prebuilt import ToolNode def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return tools return end这个函数逻辑很直白看最后一条消息里有没有tool_calls。有就去执行工具没有就结束。tool_calls是模型返回的结构化字段当模型决定调用工具时它不会直接返回文本而是返回一个包含工具名和参数的tool_calls列表。映射表在add_conditional_edges的第三个参数里定义{tools: tools, end: END}。左边是路由函数的返回值右边是实际节点名。这个映射关系让路由逻辑和图的拓扑解耦——路由函数只管返回语义化的标签具体去哪个节点由映射表决定。3.2 多分支路由不止调工具和结束实际项目里路由往往不止两个分支。比如一个客服 Agent可能需要区分查订单、退款、转人工、直接回答。这时候路由函数可以返回更多标签。def route_by_intent(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: tool_name last_message.tool_calls[0][name] if tool_name in (query_order, refund): return order_tools return general_tools if 转人工 in last_message.content: return human_handoff return end映射表相应扩展{order_tools: order_tools, general_tools: general_tools, human_handoff: human_handoff, end: END}。这种设计的好处是路由逻辑集中在一个函数里改起来只改一处。而且路由函数是纯函数输入 State 输出字符串非常容易写单元测试。我一般会针对路由函数单独写测试用例构造不同的 State 看返回值对不对比跑整个图快得多。3.3 路由函数的常见陷阱第一个陷阱是读错消息。state[messages][-1]拿的是最后一条但有时候最后一条可能是 ToolMessage工具返回结果而不是 AIMessage。如果你判断tool_calls的时候拿的是 ToolMessage它没有这个字段就会报错。稳妥的写法是加个类型判断或者用getattr(last_message, tool_calls, None)。第二个陷阱是路由返回值没有对应的映射。比如路由函数返回了tools但映射表里写的是toolLangGraph 会直接抛错。这个错误信息还算清晰但如果你路由分支多很容易漏配。第三个陷阱是条件边和普通边混用导致死循环。比如 agent 节点既有条件边指向 tools又有一条普通边指向 tools那不管条件判断结果如何都会去 tools循环就出不来了。实操心得路由函数尽量保持简单只做读状态、返回标签这一件事。复杂的判断逻辑抽成独立的辅助函数路由函数里调用它们。这样路由函数本身可读性高辅助函数也好测试。4. Agent 工具调用循环的完整实现4.1 工具定义从函数到模型可调用的工具工具调用的第一步是定义工具。LangChain 提供了tool装饰器把一个普通 Python 函数变成模型能识别的工具。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 # 实际项目里这里调真实 API return f{city}今天晴气温 25 度。 tool def calculate(expression: str) - str: 计算数学表达式比如 2 3 * 4。 return str(eval(expression))tool装饰器会自动提取函数的名称、参数类型、docstring生成模型能理解的工具描述。docstring 特别重要——模型就是靠它来判断什么时候该调这个工具的。所以 docstring 要写清楚这个工具做什么、什么时候用而不是随便写一句。我见过很多人 docstring 写得很敷衍结果模型该调工具的时候不调不该调的时候乱调。工具描述的质量直接决定工具调用的准确率这一点值得花时间打磨。4.2 绑定工具到模型定义好工具后要把它们绑定到模型上from langchain_openai import ChatOpenAI tools [get_weather, calculate] llm ChatOpenAI(modelgpt-4o-mini) llm_with_tools llm.bind_tools(tools)bind_tools做的事情是把工具的 schema 转换成模型 API 能接受的格式附加到每次请求里。绑定之后模型在生成回复时就能看到这些工具并在需要时返回tool_calls。这里有个细节bind_tools返回的是一个新对象不是原地修改。所以你要用llm_with_tools而不是原来的llm。我第一次用的时候忘了这点调了半天发现模型根本不调工具后来才发现是绑定的对象没用上。4.3 工具执行节点ToolNode 的用法LangGraph 预置了ToolNode专门用来执行工具调用from langgraph.prebuilt import ToolNode tool_node ToolNode(tools)ToolNode会自动读取 State 里最后一条 AIMessage 的tool_calls逐个执行对应的工具把结果包装成 ToolMessage 追加到 messages 里。你不需要手动解析参数、匹配工具名、处理异常这些它都做了。如果你需要自定义工具执行逻辑比如加日志、加权限校验、加超时控制也可以自己写一个节点函数def custom_tool_node(state: AgentState): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: tool_fn tool_map[tool_call[name]] try: output tool_fn.invoke(tool_call[args]) except Exception as e: output f工具执行失败{e} results.append(ToolMessage(contentstr(output), tool_call_idtool_call[id])) return {messages: results}自定义的好处是可控性强坏处是要自己处理各种边界情况。新手建议先用 ToolNode跑通了再考虑自定义。4.4 组装完整循环把前面所有部分拼起来from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import MemorySaver def call_model(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end builder StateGraph(AgentState) builder.add_node(agent, call_model) 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) memory MemorySaver() graph builder.compile(checkpointermemory)跑起来config {configurable: {thread_id: user-1}} result graph.invoke( {messages: [(user, 北京天气怎么样顺便算一下 12 * 8)]}, configconfig ) print(result[messages][-1].content)执行流程是这样的用户消息进入 agent 节点模型看到有天气和计算两个工具可用返回带tool_calls的 AIMessage条件路由判断有工具调用转到 tools 节点ToolNode 执行两个工具把结果作为 ToolMessage 追加回到 agent 节点模型看到工具结果生成最终回复条件路由判断没有工具调用转到 END。MemorySaver是 Checkpointer它把每一步的状态按thread_id存下来。同一个thread_id的多次调用会共享对话历史实现多轮对话。这个机制在调试的时候特别有用——你可以随时查看某个 thread 的完整状态。5. 常见问题与排查技巧实录5.1 模型不调用工具怎么办这是最高频的问题。排查顺序如下第一检查工具是否真的绑定到了模型上。llm_with_tools llm.bind_tools(tools)之后用的是llm_with_tools吗第二检查工具的 docstring 是否清晰。模型靠 docstring 判断工具用途如果写得太模糊模型可能觉得不需要调工具也能回答。第三检查 Prompt。有时候系统提示词里写了你是一个聊天助手模型就倾向于直接聊天而不调工具。可以加一句需要实时信息或计算时优先使用工具。第四检查模型本身的能力。不是所有模型都支持工具调用小参数量的模型经常调不明白。换一个支持 function calling 的模型试试。5.2 工具调用参数解析失败模型返回的tool_calls里args是一个字典。如果工具的参数类型是int但模型传了字符串5Pydantic 校验会失败。解决办法有两个一是在工具函数里做类型转换二是用 Pydantic 模型定义参数并设置宽松模式。还有一种情况是模型编造了不存在的参数名。这个只能靠优化 docstring 和参数描述来减少没法完全避免。所以工具函数里最好加一层参数校验对未知参数做容错处理。5.3 循环停不下来如果 Agent 一直在 agent 和 tools 之间循环说明条件路由始终返回tools。可能的原因模型每次都返回tool_calls即使工具已经返回了结果。这通常是因为工具返回的内容让模型觉得还需要再调一次。解决办法是加一个最大循环次数限制。在 State 里加一个step_count字段每次 agent 节点执行时加一路由函数里判断超过阈值就强制返回end。class AgentState(TypedDict): messages: Annotated[list, add_messages] step_count: int def should_continue(state: AgentState) - str: if state.get(step_count, 0) 10: return end last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end5.4 常见问题速查表问题现象可能原因排查方向模型不调工具工具未绑定/描述不清/模型不支持检查 bind_tools、优化 docstring、换模型参数解析失败类型不匹配/参数名错误加类型转换、校验参数、优化描述无限循环路由始终返回 tools加 step_count 限制、检查工具返回内容状态被覆盖字段缺少 reducer给累积字段加 add_messages 等 reducer多轮对话失忆未配置 Checkpointer加 MemorySaver 并传 thread_id路由报错返回值无对应映射检查 add_conditional_edges 映射表避坑技巧调试 LangGraph 的时候把graph.stream()用起来。它会逐步输出每个节点的执行结果比invoke一次性返回全部结果直观得多。你能清楚看到每一步 State 变成了什么样哪个节点做了什么决策。6. 从最小可用到生产可用的扩展方向跑通最小循环之后实际项目里还有几个方向值得深入。持久化存储。MemorySaver只存在内存里进程重启就没了。生产环境要换成SqliteSaver或PostgresSaver把状态存到数据库。这样服务重启后对话历史还在也能支持多实例部署。人工审核中断。LangGraph 支持在某个节点前中断等人工确认后再继续。比如工具调用涉及资金操作时先中断让运营审核。这个通过interrupt_before参数配置配合 Checkpointer 使用。多 Agent 协作。把多个 Agent 作为节点放进一张图里每个 Agent 有自己的工具集和职责通过条件路由决定任务分给谁。这是 LangGraph 相比单 Agent 框架最大的优势——它天然支持把复杂任务拆成多个专职 Agent 协作完成。流式输出。graph.stream()支持流式返回前端可以实时显示 Agent 的思考过程。对于需要展示正在调用工具这类状态的场景流式输出体验好很多。可观测性。LangGraph 可以接入 LangSmith 做链路追踪每一步的输入输出、耗时、token 消耗都能看到。调试复杂图的时候这个比打日志高效得多。我在实际项目里的体会是先用最小循环跑通业务逻辑再逐步加持久化、中断、多 Agent 这些能力。一上来就搭大框架很容易在细节里迷失。LangGraph 的学习曲线主要在前面的概念理解上一旦理解了 State、Node、Edge、Conditional Edge 这四个东西后面就是组合的问题了。最后分享一个小技巧把图的结构用graph.get_graph().draw_mermaid()打印出来虽然这里不画图但你可以把它贴到支持 Mermaid 的编辑器里看。可视化之后流程哪里有问题一目了然比盯着代码强。
LangGraph 从零构建 Agent：StateGraph 与条件路由实战
1. 从零理解 LangGraph它到底解决什么问题如果你已经用过 LangChain 的 Chain 和 Agent大概率会遇到一个共同的痛点流程一旦复杂起来代码就开始失控。一个带工具调用的 Agent背后其实是一个循环——模型思考、决定调不调工具、调哪个工具、拿到结果再思考直到给出最终答案。用传统 Chain 写你得手动拼 Prompt、解析输出、判断分支、维护循环状态写着写着就变成了一坨 if-else 面条。LangGraph 就是冲着这个问题来的。它把 Agent 的执行过程抽象成一张有向图节点是执行单元边是流转方向状态在节点之间传递和累积。你可以把它理解成给 LLM 应用装了一个流程引擎让多步骤、带分支、带循环的逻辑变得可视化、可调试、可持久化。我最初接触 LangGraph 的时候最直观的感受是它把控制流从 Prompt 里拿出来了。以前你靠 Prompt 里写如果用户问的是天气就调用天气工具现在你直接用代码定义条件路由模型只负责决策流程由图的拓扑结构保证。这个思路的转变是理解 LangGraph 的关键。这篇文章我会围绕三个核心概念展开StateGraph状态图、条件路由Conditional Edge、Agent 的工具调用循环。这三个东西串起来就是一个最小可用的 Agent 骨架。适合已经了解 LangChain 基础、想往 Agent 开发方向深入的人也适合完全没接触过 LangGraph 但想搞明白Agent 到底怎么跑起来的初学者。提示LangGraph 和 LangChain 不是替代关系。LangChain 提供模型封装、工具定义、Prompt 模板这些零件LangGraph 负责把这些零件按你想要的流程编排起来。两者配合使用不是二选一。2. StateGraph 核心机制拆解2.1 状态是什么一张在节点间流转的共享白板StateGraph 的核心是State。你可以把 State 想象成一块共享白板每个节点都能往上面写东西也能读到别人写的东西。在 LangGraph 里State 通常用一个TypedDict或者 Pydantic 模型来定义。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]这里有个关键细节Annotated[list, add_messages]里的add_messages是一个reducer归约函数。它的作用是定义当多个节点都想更新这个字段时怎么合并。add_messages的逻辑是追加而不是覆盖——新消息加到列表末尾而不是把整个列表替换掉。为什么这个设计很重要因为 Agent 的对话历史是累积的。如果每次节点更新都覆盖messages那模型就看不到之前的对话了。用add_messages做 reducer每个节点只需要返回我要新增的消息LangGraph 自动帮你合并到历史里。我踩过的一个坑一开始我自定义 State 的时候忘了加 reducer结果工具调用的结果把用户消息覆盖了模型完全不知道用户问了什么。排查了半天才意识到是 reducer 的问题。所以记住一句话需要累积的字段一定要加 reducer需要覆盖的字段用默认行为就行。2.2 节点与边图的骨架怎么搭节点Node就是一个函数接收当前 State返回要更新的字段。边Edge定义节点之间的流转关系。LangGraph 提供两种边普通边从 A 节点执行完直接到 B 节点无条件。条件边从 A 节点执行完后根据一个路由函数的返回值决定去 B 还是 C。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) 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 循环从 START 进入 agent 节点agent 决定是调工具还是结束如果调工具就去 tools 节点tools 执行完再回到 agent。should_continue就是条件路由函数。START和END是两个特殊节点分别代表图的入口和出口。compile()之后得到的graph是一个可执行对象调用graph.invoke({messages: [...]})就能跑起来。2.3 为什么用图而不是链三个实际好处第一循环是原生支持的。Chain 是线性的要做循环得自己套 while。图天然支持环Agent 的思考-行动-观察循环直接映射成agent - tools - agent的环。第二状态管理是自动的。每个节点的输入输出都经过 State 统一管理不用手动在函数之间传参。节点多了之后这个优势非常明显。第三可持久化和可中断。LangGraph 支持 Checkpointer可以把每一步的状态存下来实现断点续跑、人工审核中断human-in-the-loop。这个在 Chain 里做起来非常麻烦。注意图里的环一定要有退出条件否则会无限循环。条件路由函数必须能在某个时刻返回指向 END 的路由这是新手最容易犯的错误之一。3. 条件路由让 Agent 自己决定下一步3.1 路由函数的本质一个返回字符串的普通函数条件路由的核心是一个函数它读 State返回一个字符串LangGraph 根据这个字符串去映射表里找下一个节点。from langgraph.prebuilt import ToolNode def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return tools return end这个函数逻辑很直白看最后一条消息里有没有tool_calls。有就去执行工具没有就结束。tool_calls是模型返回的结构化字段当模型决定调用工具时它不会直接返回文本而是返回一个包含工具名和参数的tool_calls列表。映射表在add_conditional_edges的第三个参数里定义{tools: tools, end: END}。左边是路由函数的返回值右边是实际节点名。这个映射关系让路由逻辑和图的拓扑解耦——路由函数只管返回语义化的标签具体去哪个节点由映射表决定。3.2 多分支路由不止调工具和结束实际项目里路由往往不止两个分支。比如一个客服 Agent可能需要区分查订单、退款、转人工、直接回答。这时候路由函数可以返回更多标签。def route_by_intent(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: tool_name last_message.tool_calls[0][name] if tool_name in (query_order, refund): return order_tools return general_tools if 转人工 in last_message.content: return human_handoff return end映射表相应扩展{order_tools: order_tools, general_tools: general_tools, human_handoff: human_handoff, end: END}。这种设计的好处是路由逻辑集中在一个函数里改起来只改一处。而且路由函数是纯函数输入 State 输出字符串非常容易写单元测试。我一般会针对路由函数单独写测试用例构造不同的 State 看返回值对不对比跑整个图快得多。3.3 路由函数的常见陷阱第一个陷阱是读错消息。state[messages][-1]拿的是最后一条但有时候最后一条可能是 ToolMessage工具返回结果而不是 AIMessage。如果你判断tool_calls的时候拿的是 ToolMessage它没有这个字段就会报错。稳妥的写法是加个类型判断或者用getattr(last_message, tool_calls, None)。第二个陷阱是路由返回值没有对应的映射。比如路由函数返回了tools但映射表里写的是toolLangGraph 会直接抛错。这个错误信息还算清晰但如果你路由分支多很容易漏配。第三个陷阱是条件边和普通边混用导致死循环。比如 agent 节点既有条件边指向 tools又有一条普通边指向 tools那不管条件判断结果如何都会去 tools循环就出不来了。实操心得路由函数尽量保持简单只做读状态、返回标签这一件事。复杂的判断逻辑抽成独立的辅助函数路由函数里调用它们。这样路由函数本身可读性高辅助函数也好测试。4. Agent 工具调用循环的完整实现4.1 工具定义从函数到模型可调用的工具工具调用的第一步是定义工具。LangChain 提供了tool装饰器把一个普通 Python 函数变成模型能识别的工具。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 # 实际项目里这里调真实 API return f{city}今天晴气温 25 度。 tool def calculate(expression: str) - str: 计算数学表达式比如 2 3 * 4。 return str(eval(expression))tool装饰器会自动提取函数的名称、参数类型、docstring生成模型能理解的工具描述。docstring 特别重要——模型就是靠它来判断什么时候该调这个工具的。所以 docstring 要写清楚这个工具做什么、什么时候用而不是随便写一句。我见过很多人 docstring 写得很敷衍结果模型该调工具的时候不调不该调的时候乱调。工具描述的质量直接决定工具调用的准确率这一点值得花时间打磨。4.2 绑定工具到模型定义好工具后要把它们绑定到模型上from langchain_openai import ChatOpenAI tools [get_weather, calculate] llm ChatOpenAI(modelgpt-4o-mini) llm_with_tools llm.bind_tools(tools)bind_tools做的事情是把工具的 schema 转换成模型 API 能接受的格式附加到每次请求里。绑定之后模型在生成回复时就能看到这些工具并在需要时返回tool_calls。这里有个细节bind_tools返回的是一个新对象不是原地修改。所以你要用llm_with_tools而不是原来的llm。我第一次用的时候忘了这点调了半天发现模型根本不调工具后来才发现是绑定的对象没用上。4.3 工具执行节点ToolNode 的用法LangGraph 预置了ToolNode专门用来执行工具调用from langgraph.prebuilt import ToolNode tool_node ToolNode(tools)ToolNode会自动读取 State 里最后一条 AIMessage 的tool_calls逐个执行对应的工具把结果包装成 ToolMessage 追加到 messages 里。你不需要手动解析参数、匹配工具名、处理异常这些它都做了。如果你需要自定义工具执行逻辑比如加日志、加权限校验、加超时控制也可以自己写一个节点函数def custom_tool_node(state: AgentState): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: tool_fn tool_map[tool_call[name]] try: output tool_fn.invoke(tool_call[args]) except Exception as e: output f工具执行失败{e} results.append(ToolMessage(contentstr(output), tool_call_idtool_call[id])) return {messages: results}自定义的好处是可控性强坏处是要自己处理各种边界情况。新手建议先用 ToolNode跑通了再考虑自定义。4.4 组装完整循环把前面所有部分拼起来from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import MemorySaver def call_model(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end builder StateGraph(AgentState) builder.add_node(agent, call_model) 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) memory MemorySaver() graph builder.compile(checkpointermemory)跑起来config {configurable: {thread_id: user-1}} result graph.invoke( {messages: [(user, 北京天气怎么样顺便算一下 12 * 8)]}, configconfig ) print(result[messages][-1].content)执行流程是这样的用户消息进入 agent 节点模型看到有天气和计算两个工具可用返回带tool_calls的 AIMessage条件路由判断有工具调用转到 tools 节点ToolNode 执行两个工具把结果作为 ToolMessage 追加回到 agent 节点模型看到工具结果生成最终回复条件路由判断没有工具调用转到 END。MemorySaver是 Checkpointer它把每一步的状态按thread_id存下来。同一个thread_id的多次调用会共享对话历史实现多轮对话。这个机制在调试的时候特别有用——你可以随时查看某个 thread 的完整状态。5. 常见问题与排查技巧实录5.1 模型不调用工具怎么办这是最高频的问题。排查顺序如下第一检查工具是否真的绑定到了模型上。llm_with_tools llm.bind_tools(tools)之后用的是llm_with_tools吗第二检查工具的 docstring 是否清晰。模型靠 docstring 判断工具用途如果写得太模糊模型可能觉得不需要调工具也能回答。第三检查 Prompt。有时候系统提示词里写了你是一个聊天助手模型就倾向于直接聊天而不调工具。可以加一句需要实时信息或计算时优先使用工具。第四检查模型本身的能力。不是所有模型都支持工具调用小参数量的模型经常调不明白。换一个支持 function calling 的模型试试。5.2 工具调用参数解析失败模型返回的tool_calls里args是一个字典。如果工具的参数类型是int但模型传了字符串5Pydantic 校验会失败。解决办法有两个一是在工具函数里做类型转换二是用 Pydantic 模型定义参数并设置宽松模式。还有一种情况是模型编造了不存在的参数名。这个只能靠优化 docstring 和参数描述来减少没法完全避免。所以工具函数里最好加一层参数校验对未知参数做容错处理。5.3 循环停不下来如果 Agent 一直在 agent 和 tools 之间循环说明条件路由始终返回tools。可能的原因模型每次都返回tool_calls即使工具已经返回了结果。这通常是因为工具返回的内容让模型觉得还需要再调一次。解决办法是加一个最大循环次数限制。在 State 里加一个step_count字段每次 agent 节点执行时加一路由函数里判断超过阈值就强制返回end。class AgentState(TypedDict): messages: Annotated[list, add_messages] step_count: int def should_continue(state: AgentState) - str: if state.get(step_count, 0) 10: return end last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end5.4 常见问题速查表问题现象可能原因排查方向模型不调工具工具未绑定/描述不清/模型不支持检查 bind_tools、优化 docstring、换模型参数解析失败类型不匹配/参数名错误加类型转换、校验参数、优化描述无限循环路由始终返回 tools加 step_count 限制、检查工具返回内容状态被覆盖字段缺少 reducer给累积字段加 add_messages 等 reducer多轮对话失忆未配置 Checkpointer加 MemorySaver 并传 thread_id路由报错返回值无对应映射检查 add_conditional_edges 映射表避坑技巧调试 LangGraph 的时候把graph.stream()用起来。它会逐步输出每个节点的执行结果比invoke一次性返回全部结果直观得多。你能清楚看到每一步 State 变成了什么样哪个节点做了什么决策。6. 从最小可用到生产可用的扩展方向跑通最小循环之后实际项目里还有几个方向值得深入。持久化存储。MemorySaver只存在内存里进程重启就没了。生产环境要换成SqliteSaver或PostgresSaver把状态存到数据库。这样服务重启后对话历史还在也能支持多实例部署。人工审核中断。LangGraph 支持在某个节点前中断等人工确认后再继续。比如工具调用涉及资金操作时先中断让运营审核。这个通过interrupt_before参数配置配合 Checkpointer 使用。多 Agent 协作。把多个 Agent 作为节点放进一张图里每个 Agent 有自己的工具集和职责通过条件路由决定任务分给谁。这是 LangGraph 相比单 Agent 框架最大的优势——它天然支持把复杂任务拆成多个专职 Agent 协作完成。流式输出。graph.stream()支持流式返回前端可以实时显示 Agent 的思考过程。对于需要展示正在调用工具这类状态的场景流式输出体验好很多。可观测性。LangGraph 可以接入 LangSmith 做链路追踪每一步的输入输出、耗时、token 消耗都能看到。调试复杂图的时候这个比打日志高效得多。我在实际项目里的体会是先用最小循环跑通业务逻辑再逐步加持久化、中断、多 Agent 这些能力。一上来就搭大框架很容易在细节里迷失。LangGraph 的学习曲线主要在前面的概念理解上一旦理解了 State、Node、Edge、Conditional Edge 这四个东西后面就是组合的问题了。最后分享一个小技巧把图的结构用graph.get_graph().draw_mermaid()打印出来虽然这里不画图但你可以把它贴到支持 Mermaid 的编辑器里看。可视化之后流程哪里有问题一目了然比盯着代码强。
