LangGraph 从入门到实战：StateGraph 与 Agent 工具调用循环

1. 从零理解 LangGraph它到底解决什么问题1.1 为什么有了 LangChain 还需要 LangGraph很多人第一次接触 LangGraph 都会问同一个问题我已经在用 LangChain 了为什么还要多学一个框架这个问题我当初也纠结过后来在实际项目里踩了坑才真正想明白。LangChain 的核心抽象是Chain也就是一条线性的调用链输入进来经过提示词模板、模型、输出解析器结束。这条链路是单向的、无环的。你让它做一次问答、一次摘要、一次翻译它很好用。但一旦你要做下面这些事情Chain 就开始力不从心了模型输出需要根据内容走不同的分支比如判断用户意图后决定调用哪个工具需要循环执行直到满足某个条件比如 Agent 反复调用工具直到任务完成需要在多个步骤之间维护和更新状态比如对话历史、中间结果、累积的上下文需要人工介入某个中间环节比如审批、修正这些需求的共同点是流程不再是直线而是带分支和环路的图。LangGraph 就是为这个场景设计的。它把整个流程建模成一个状态图StateGraph节点是计算步骤边是流转逻辑状态在节点之间传递和更新。打个比方LangChain 像一条流水线原料从一头进成品从另一头出LangGraph 像一张地铁线路图有换乘站、有环线、有分支你可以根据当前情况决定下一站去哪。1.2 StateGraph 的三个核心概念LangGraph 的 StateGraph 说复杂也复杂说简单也简单核心就三个东西State状态整个图共享的一份数据。它通常是一个字典或者 TypedDict里面放着消息列表、中间变量、计数器等等。每个节点都能读到它也能返回更新后的部分。Node节点一个函数接收当前 State返回对 State 的更新。节点可以是调用大模型、执行工具、做数据转换任何 Python 函数都行。Edge边定义节点之间的流转关系。普通边是固定的 A 到 B条件边conditional edge则根据一个路由函数的返回值决定下一步走哪。理解了这三个概念你就理解了 LangGraph 的 80%。剩下的都是在这三个基础上的组合和变体。1.3 适合谁来学学完能做什么这篇内容适合三类人一是有 Python 基础、想入门 Agent 开发的工程师二是用过 LangChain 但被复杂流程卡住的人三是想理解 Agent 底层编排原理的学习者。学完之后你能独立搭出一个带工具调用循环的 Agent能根据模型输出做条件路由能维护多轮对话状态。这些能力是当前 Agent 开发的基本功面试里也经常被问到 LangGraph 和 LangChain 的区别、StateGraph 怎么设计这类问题。提示LangGraph 的版本迭代比较快本文基于较稳定的 API 讲解核心概念不会变但具体参数名建议对照你安装的版本确认。2. 环境搭建与第一个 StateGraph2.1 安装与依赖选择先把环境弄干净。我强烈建议用虚拟环境因为 LangGraph 依赖的包不少混装容易出问题。python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install langgraph langchain-openai这里有个选型细节值得说langchain-openai是官方维护的 OpenAI 兼容接口包如果你用的是其他模型服务只要它兼容 OpenAI 的接口格式改一下base_url就能用。这样写代码时不用绑定具体厂商迁移成本低。安装完可以验证一下import langgraph print(langgraph.__version__)版本号能打印出来就说明装好了。如果报错八成是 Python 版本太低LangGraph 一般要求 3.9 以上。2.2 定义 State 的正确姿势State 的定义直接决定了你后面写代码顺不顺手。新手最容易犯的错是把 State 定义得太随意用普通 dict结果多个节点更新同一个字段时互相覆盖。推荐用TypedDict加Annotated来定义尤其是消息列表这种需要追加而不是覆盖的字段from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] tool_calls_count: int关键在add_messages这个 reducer。默认情况下节点返回的新值会覆盖旧值。但对话消息需要的是追加——新一轮的消息要加到历史后面而不是把历史冲掉。add_messages就是告诉 LangGraph这个字段用追加的方式合并。我踩过的坑一开始没用 reducer结果 Agent 每轮对话都失忆只记得最新一条消息。排查了半天才发现是 State 合并策略的问题。所以这个细节一定要在定义阶段就处理好。2.3 构建最小可运行图先搭一个最简单的图两个节点一条边跑通了再往上加复杂度from langgraph.graph import StateGraph, START, END def node_a(state: AgentState): return {messages: [(user, 来自节点A)]} def node_b(state: AgentState): return {messages: [(assistant, 来自节点B)]} builder StateGraph(AgentState) 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: [], tool_calls_count: 0}) print(result[messages])START和END是两个特殊标记分别代表图的入口和出口。compile()会把图冻结成一个可执行对象之后每次invoke都是一次完整的执行。实测下来这个最小例子能帮你快速确认环境没问题、State 合并正常。别小看这一步很多后续的诡异 bug 都是因为最基础的 State 行为没搞对。3. 条件路由让流程学会看情况走3.1 条件边的工作原理条件边是 LangGraph 从流程图进化成智能编排的关键。它的机制是这样的你提供一个路由函数这个函数读当前 State返回一个字符串LangGraph 拿这个字符串去匹配你预先注册的路径映射决定跳到哪个节点。路由函数本身不修改 State它只做判断。这个职责分离很重要——判断逻辑和执行逻辑分开写代码才清晰。3.2 一个意图分流的实战例子假设我们要做一个客服入口根据用户问题分流到查订单或通用问答两条路def route_by_intent(state: AgentState): last_msg state[messages][-1].content if 订单 in last_msg or 物流 in last_msg: return order return general builder.add_conditional_edges( classify, route_by_intent, { order: order_node, general: general_node, } )add_conditional_edges的第三个参数是路径映射表把路由函数的返回值映射到实际节点名。这样路由函数只管返回语义化的标签具体跳哪由映射表决定改起来很方便。3.3 路由设计的几个经验第一路由函数要尽量纯粹。不要在里面调模型、发请求只做基于已有 State 的判断。需要模型判断的单独放一个节点去做把结果写进 State再让路由函数读。第二路径映射要覆盖所有返回值。如果路由函数可能返回三种结果映射表里就得有三种否则运行时会报找不到路径。我建议路由函数最后加一个兜底 return映射表里也配一个默认分支。第三条件边可以形成环。这正是 Agent 工具调用循环的基础——路由函数判断还需要继续调用工具就跳回模型节点形成循环。下一节详细讲。注意循环一定要有终止条件否则会无限执行下去。常见的做法是在 State 里放一个计数器超过阈值就强制走向 END。4. Agent 工具调用循环的完整实现4.1 工具调用循环的本质Agent 之所以叫 Agent核心就在于它能自己决定调用什么工具、调用几次。这个能力的底层就是一个循环把对话历史发给模型模型返回要么是最终回答要么是我要调用某个工具如果是工具调用执行工具把结果加回对话历史回到第 1 步直到模型给出最终回答用 LangGraph 表达就是两个节点加一条条件边model节点和tools节点条件边判断模型输出里有没有工具调用。4.2 定义工具与绑定模型先定义工具。LangGraph 兼容 LangChain 的工具定义方式from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 fake_data {北京: 晴25度, 上海: 多云28度} return fake_data.get(city, 暂无数据) tools [get_weather]工具的 docstring 很重要模型就是靠它来判断什么时候该调用这个工具的。写得越清楚模型选得越准。然后绑定模型from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools)bind_tools会把工具的描述信息一起发给模型模型就知道有哪些工具可用。4.3 构建循环图现在把循环搭起来from langgraph.prebuilt import ToolNode def call_model(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState): last state[messages][-1] if last.tool_calls: return tools return END builder StateGraph(AgentState) builder.add_node(model, call_model) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, model) builder.add_conditional_edges(model, should_continue, {tools: tools, END: END}) builder.add_edge(tools, model) agent builder.compile()注意builder.add_edge(tools, model)这一行——工具执行完无条件回到模型节点。这就是循环的闭合点。而should_continue决定是继续循环还是结束。ToolNode是 LangGraph 提供的预置节点它会自动读取消息里的 tool_calls执行对应工具把结果包装成 ToolMessage 返回。省了自己写解析逻辑。4.4 跑一次完整调用result agent.invoke({ messages: [(user, 北京今天天气怎么样)], tool_calls_count: 0 }) for msg in result[messages]: print(type(msg).__name__, :, getattr(msg, content, ))你会看到消息序列大致是用户消息 → 模型消息带 tool_calls→ 工具消息天气结果→ 模型消息最终回答。这个序列完整展示了工具调用循环的一次迭代。4.5 循环控制与安全边界生产环境里循环必须加护栏。我一般会做两件事一是限制最大迭代次数。在 State 里放计数器每次进 model 节点加一超过比如 10 次就强制结束。防止模型陷入反复调用同一个工具的死循环。二是给 compile 传 recursion_limitagent builder.compile() result agent.invoke(inputs, config{recursion_limit: 25})recursion_limit是 LangGraph 层面的硬限制超过会抛异常。这是最后一道防线即使你的业务逻辑有漏洞也不会把资源跑爆。提示recursion_limit默认是 25对于大多数 Agent 场景够用。如果你的任务确实需要很多步可以调高但要配合业务层的计数器一起用。5. 常见问题排查与避坑清单5.1 消息不追加反而被覆盖这是最高频的问题。症状是 Agent 只记得最后一轮对话。原因几乎都是 State 里的 messages 字段没加add_messagesreducer。检查你的 TypedDict 定义确认写成了Annotated[list, add_messages]。5.2 条件边报找不到路径报错信息通常是找不到某个节点或路径。排查顺序先看路由函数的返回值再看add_conditional_edges的映射表确认两者能对上。特别注意END这个特殊值映射表里要写成{END: END}而不是字符串END。5.3 工具调用后模型不继续有时候工具执行完了模型却不再生成最终回答。常见原因是工具返回的内容格式不对或者 ToolMessage 的tool_call_id没对上。用ToolNode能避免大部分这类问题因为它会自动处理 id 匹配。如果你手写工具节点务必保证返回的 ToolMessage 带上正确的tool_call_id。5.4 排查速查表症状可能原因解决方向对话失忆messages 无 reducer加 add_messages找不到路径路由返回值与映射表不匹配对齐两者加兜底分支无限循环缺少终止条件加计数器 recursion_limit工具结果丢失tool_call_id 不匹配用 ToolNode 或手动对齐 id状态字段互相覆盖多节点写同一字段明确 reducer 或拆分字段5.5 几个我踩过的坑第一个坑是在节点里直接修改 state 字典。LangGraph 期望节点返回更新量而不是原地修改。原地改虽然有时看着能跑但会破坏框架的状态管理导致难以排查的 bug。老老实实 return 一个字典。第二个坑是把路由逻辑写得太重。我一开始在路由函数里调模型做意图判断结果每次路由都要等一次模型响应慢得离谱。正确做法是把模型判断单独做成一个节点结果写进 State路由函数只读不调。第三个坑是忽略异步。如果你的工具涉及网络请求同步执行会阻塞整个流程。LangGraph 支持异步节点用async def定义节点、ainvoke调用能显著提升并发场景下的吞吐。这个在单机测试时感觉不明显一上量就体现出来了。6. 从能跑到好用进阶优化思路6.1 状态设计的分层当 Agent 变复杂State 会越来越臃肿。我的做法是分层核心对话消息放一层任务相关的中间变量放一层控制类字段计数器、标志位放一层。虽然都在同一个 TypedDict 里但命名上区分开比如messages、task_*、ctrl_*读代码时一眼能看出字段用途。6.2 人工介入节点LangGraph 支持在流程中插入人工确认。做法是在需要确认的地方加一个节点把当前状态暴露出去等外部输入后再继续。这在涉及敏感操作比如自动发邮件、自动下单的 Agent 里非常必要。实现上可以用interrupt机制让图暂停外部恢复后再往下走。6.3 多 Agent 协作的雏形单个 Agent 能力有限复杂任务往往需要多个 Agent 分工。LangGraph 的图结构天然适合这个场景把每个 Agent 做成一个子图用一个调度节点根据任务类型路由到不同的子图。子图之间通过共享的 State 传递信息。这就是多 Agent 协作的基本形态比把所有逻辑塞进一个大 Agent 要清晰得多。6.4 可观测性Agent 跑起来之后你得知道它每一步在干什么。LangGraph 可以配合回调机制记录每个节点的输入输出。我习惯在开发阶段把每个节点的 State 变化打日志出问题时能快速定位是哪一步跑偏了。上线后则关注循环次数、工具调用成功率这些指标。最后分享一个我自己的体会LangGraph 的学习曲线主要卡在状态思维上。习惯了写线性代码的人一开始很难适应节点返回更新量、框架负责合并这种模式。但一旦转过这个弯你会发现用它表达复杂的 Agent 逻辑比手写 if-else 和 while 循环清晰太多。建议从最小的两节点图开始一步步加条件边、加循环、加工具每加一层都跑通再往下走比一上来就抄一个完整 Agent 例子要扎实得多。