1. 从零理解 LangGraph 到底在解决什么问题1.1 为什么单纯的链式调用不够用了如果你之前用过 LangChain 的 Chain应该会有个直观感受一条链从头跑到尾中间很难根据实际情况拐弯。用户问“帮我查下明天北京的天气”链式结构能处理但如果用户问“帮我看看明天适不适合去北京出差顺便把航班也查一下”这就不是一条直线能搞定的事了——你得先判断意图再决定调天气接口还是航班接口甚至两个都要调最后还要把结果汇总起来做决策。这种“根据中间结果动态决定下一步”的需求就是 LangGraph 要解决的核心问题。它把整个流程建模成一张有向图节点是具体的处理单元比如调用模型、执行工具、做判断边是流转逻辑。跟传统 Chain 最大的区别在于图可以有环可以条件跳转可以在任意节点暂停等待人工介入也可以把状态持久化下来随时恢复。我自己的理解是LangChain 解决的是“怎么把组件拼起来”LangGraph 解决的是“怎么让组件按逻辑跑起来”。两者不是替代关系LangGraph 底层依然大量复用 LangChain 的模型封装和工具抽象只是在上层加了一套状态机和图调度的机制。1.2 StateGraph 的核心心智模型StateGraph 这个名字拆开看就很好理解State Graph。State 是整个图共享的一份数据你可以把它想象成一个公共的白板每个节点都能往上写东西、擦东西Graph 则规定了这些节点谁先谁后、什么条件下走哪条路。关键点在于State 不是随便一个字典就完事它需要用 TypedDict 或者 Pydantic 模型定义清楚结构并且每个字段要指定归约方式reducer。默认情况下节点返回的新值会直接覆盖旧值但如果你用Annotated[list, add]这种写法新值就会追加到旧列表后面。这个设计非常关键因为 Agent 执行过程中消息列表是不断累加的如果每次都被覆盖历史对话就丢了。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_step: str上面这段代码定义了一个最基础的状态messages用add_messages做归约保证每轮对话都追加而不是覆盖next_step用来记录路由决策。实际项目里我还会加上user_id、session_id、retry_count这类字段方便做多用户隔离和重试控制。1.3 LangGraph 和 LangChain 的边界在哪网上搜“langchain和langgraph的区别”的人特别多我用一句话概括LangChain 是工具箱LangGraph 是流水线控制器。你完全可以在 LangGraph 的节点里调用 LangChain 的 LLMChain、Retriever、Tool也可以在纯 LangGraph 项目里只用最基础的模型接口。面试里经常被问到的一个点是什么时候该用 LangGraph什么时候用 LangChain 的 AgentExecutor 就够了我的判断标准是——如果你的流程能用一棵决策树描述清楚且不需要回退和循环AgentExecutor 够用一旦出现“工具调用失败要重试”“需要人工审批后再继续”“多个 Agent 互相协作”这类需求直接上 LangGraph别在 Chain 上硬凑。2. 条件路由让图学会自己判断下一步2.1 条件边的本质是一个纯函数条件路由在 LangGraph 里通过add_conditional_edges实现它的核心是一个路由函数输入当前 State输出下一个节点的名字。这个函数必须是纯函数不能有副作用因为它可能被调用多次比如做 checkpoint 恢复时。def route_after_llm(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return tools return end这段路由逻辑很直白如果模型返回的消息里带了工具调用请求就去执行工具否则直接结束。实际项目里路由函数会复杂得多可能要判断重试次数、判断用户意图分类、判断是否需要走人工审核分支。我踩过的一个坑是路由函数里千万不要去调 LLM 或者外部 API。一方面会拖慢整个图的执行速度另一方面在 checkpoint 恢复时会产生不可预期的重复调用。路由判断需要的信息应该在之前的节点里就算好写进 State。2.2 多分支路由的组织方式当分支超过两个时推荐用映射表的方式组织而不是写一长串 if-elsedef route_by_intent(state: AgentState) - str: intent state.get(intent, unknown) return intent graph.add_conditional_edges( classify, route_by_intent, { weather: weather_node, flight: flight_node, both: parallel_dispatch, unknown: fallback_node, } )第三个参数是路径映射把路由函数的返回值映射到具体节点名。这样做的好处是路由逻辑和节点注册解耦加新分支只需要改映射表。另外 LangGraph 会在编译时校验映射表里的节点是否都存在写错了会直接报错比运行时才发现问题要好得多。2.3 循环与终止条件的控制Agent 的工具调用循环天然是个环模型决定调工具 → 执行工具 → 结果回给模型 → 模型再决定。这个环必须有明确的终止条件否则就是死循环烧 token。常见的终止策略有三种组合使用最大轮次限制State 里维护step_count超过阈值强制走 end 分支无工具调用即终止模型返回的消息里没有tool_calls就结束显式结束信号模型输出特定标记比如FINAL_ANSWER:时终止def should_continue(state: AgentState) - str: if state.get(step_count, 0) 10: return force_end last state[messages][-1] if not getattr(last, tool_calls, None): return end return tools提示最大轮次不要设太大我一般设 8 到 12 之间。设 20 以上基本等于没有保护因为正常任务很少需要超过 10 轮工具调用超过往往意味着模型陷入了某种循环。3. Agent 工具调用循环的完整实现3.1 工具定义与绑定工具调用循环的第一步是让模型知道有哪些工具可用。LangChain 的tool装饰器是最省事的方式它会自动从函数签名和 docstring 生成 JSON Schemafrom langchain_core.tools import tool tool def get_weather(city: str, date: str) - str: 查询指定城市指定日期的天气情况。 Args: city: 城市名称如北京 date: 日期格式 YYYY-MM-DD # 实际实现省略 return f{city} {date} 晴18-26度docstring 的质量直接决定模型能不能正确调用工具。我见过太多人写个查询天气就完事结果模型不知道该传什么参数、参数格式是什么调用失败率极高。把 docstring 当成给模型看的 API 文档来写参数类型、格式、示例都写清楚。绑定工具用model.bind_tools(tools)返回的模型对象在调用时会自动带上工具定义。注意不是所有模型都支持工具调用选型时要确认。3.2 循环节点的标准写法一个完整的工具调用循环通常包含两个核心节点agent调模型和tools执行工具。from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode def call_model(state: AgentState): response model_with_tools.invoke(state[messages]) return { messages: [response], step_count: state.get(step_count, 0) 1 } tool_node ToolNode(tools) 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, force_end: END, }) builder.add_edge(tools, agent) graph builder.compile()这里有个细节值得说tools节点执行完后直接连回agent形成闭环。LangGraph 的ToolNode会自动处理ToolMessage的构造包括tool_call_id的对应关系手写的话很容易漏掉这个字段导致模型报错。3.3 状态持久化与断点恢复生产环境的 Agent 必须能持久化状态否则用户刷新页面或者服务重启整个对话就丢了。LangGraph 通过 checkpointer 机制实现from langgraph.checkpoint.memory import MemorySaver # 生产环境用 SqliteSaver 或 PostgresSaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user-123-session-456}} result graph.invoke({messages: [(user, 北京明天天气)]}, config)thread_id是状态隔离的关键同一个 thread_id 的多次 invoke 会共享状态。我一般用用户ID 会话ID组合这样既能恢复单次会话又不会串号。用 SqliteSaver 做本地持久化时数据库文件路径要放在持久化卷上别放容器临时目录。这个坑我在部署时踩过容器一重启数据全没了。4. 常见问题与排查技巧实录4.1 工具调用循环停不下来怎么办这是最高频的问题。表现是 Agent 反复调用同一个工具或者在不同工具之间来回横跳。排查顺序如下现象可能原因解决方向反复调同一工具工具返回结果模型无法理解检查工具返回值格式加清晰的成功/失败标识工具间来回跳路由逻辑有歧义在 prompt 里明确任务完成条件达到最大轮次才停终止条件太宽松收紧 should_continue 判断报 tool_call_id 错误消息历史里工具调用和结果不配对用 ToolNode 而非手写我遇到过一次特别隐蔽的工具返回了超长的 JSON模型每次都要花大量 token 解析解析失败就重试。后来把工具返回值改成精简的摘要文本问题立刻消失。工具返回值要面向模型设计不是面向程序这点很多人会忽略。4.2 条件路由走错分支的调试方法路由走错通常有两个原因路由函数的判断依据在 State 里不存在或者判断逻辑写错了。调试时最有效的办法是在路由函数里打日志def route_after_llm(state: AgentState) - str: last state[messages][-1] has_tools bool(getattr(last, tool_calls, None)) print(f[ROUTE] step{state.get(step_count)} has_tools{has_tools}) return tools if has_tools else end配合 LangGraph 的 stream 模式能看到每个节点执行后的 State 快照定位问题很快。另外记得检查add_messages归约是否生效如果 messages 被覆盖了路由函数拿到的永远是最后一条判断自然出错。4.3 生产部署的几个实操心得第一给每个节点加超时。模型调用和工具执行都可能卡住LangGraph 本身不提供超时机制需要在节点函数内部用asyncio.wait_for或者信号量控制。第二checkpoint 存储要定期清理。用 Postgres 做 checkpointer 时状态表会随对话量线性增长我一般按 thread 的最后更新时间做归档超过 30 天没活动的直接删。第三工具执行要幂等。因为 checkpoint 恢复可能导致节点重跑如果工具是“下单”“发消息”这类有副作用的操作必须用业务 ID 做去重。这个在测试环境不容易发现上线后一旦触发恢复就是生产事故。第四日志要带 thread_id 和 step_count。排查线上问题时没有这两个字段基本等于盲人摸象。我习惯在每个节点入口打一条结构化日志包含节点名、thread_id、当前步数、State 关键字段摘要。4.4 关于 Agent 记忆体系的补充热词里“agent 记忆体系中短期、长期、永久记忆如何实现”问得很多。在 LangGraph 里短期记忆就是 State 里的 messages随 thread 存在长期记忆需要外挂向量库在节点里做检索和写入永久记忆一般是结构化的用户画像存关系库。我的做法是在 agent 节点调用模型前先从长期记忆里检索相关片段拼进 prompt模型回复后再异步写入新的记忆。注意写入要异步否则会拖慢响应。另外记忆检索的 top_k 不要设太大3 到 5 条足够多了反而干扰模型判断。5. 从入门到能用的进阶路径5.1 建议的学习顺序如果你是完全的新手我建议按这个顺序推进先跑通一个最简单的两节点图START → agent → END理解 State 和节点返回值的关系然后加上 ToolNode 和条件路由跑通完整的工具调用循环接着接入 checkpointer理解 thread_id 和状态恢复最后再考虑多 Agent 协作、人工介入、并行分支这些高级特性。网上“langgraph 菜鸟教程”类的资料不少但很多只讲 API 不讲为什么这么设计看完还是不会自己搭。我的建议是以官方文档为主配合一个真实的小项目练手比如做一个能查天气、查汇率、算日期的个人助手把循环、路由、持久化都覆盖到。5.2 多 Agent 协作的切入时机单 Agent 能搞定的事不要上多 Agent。多 Agent 的复杂度不是线性增长而是指数级的——状态怎么共享、消息怎么传递、冲突怎么解决每个都是坑。真正需要多 Agent 的场景通常是任务可以明确拆分成几个专业领域且各领域之间耦合度低。比如一个“研究报告生成”系统可以拆成检索 Agent、分析 Agent、写作 Agent各自有独立的工具集和 prompt。LangGraph 里可以用子图subgraph的方式组织每个子图是一个独立的 StateGraph通过父图调度。5.3 性能优化的几个方向Agent 的响应速度是用户体验的关键。我实测下来最有效的优化有三个一是并行化无依赖的工具调用模型一次返回多个 tool_calls 时ToolNode 默认是并行执行的别改成串行二是缓存模型调用相同输入直接返回缓存结果LangChain 有现成的 Cache 接口三是精简 prompt系统提示词每多 100 token每轮调用就多花 100 token 的钱和时间把不必要的历史消息裁剪掉。流式输出也是必做的LangGraph 的astream_events能把每个节点的中间结果推出来前端可以做到“模型边想边显示”体感速度提升非常明显。5.4 安全方面的注意事项Agent 能调工具就意味着能产生副作用安全边界必须提前划好。我的做法是所有写操作类工具下单、发邮件、改数据都加人工确认节点用 LangGraph 的interrupt机制暂停图执行等用户确认后再 resume。读操作类工具可以放开但也要做参数校验防止模型被诱导去查不该查的数据。另外工具的参数要做白名单校验尤其是涉及文件路径、SQL 语句、URL 的工具绝不能把模型输出的字符串直接拼进去执行。这个原则跟传统 Web 安全里的输入校验是一回事只是现在输入变成了模型生成的。6. 一个可直接复用的最小完整示例把前面所有内容串起来下面是一个能直接跑的最小 Agent 实现包含状态定义、工具、路由、循环和持久化from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import MemorySaver from langchain_core.tools import tool from langchain_openai import ChatOpenAI class State(TypedDict): messages: Annotated[list, add_messages] step_count: int tool def get_weather(city: str, date: str) - str: 查询指定城市指定日期的天气。city 为城市名date 格式 YYYY-MM-DD。 return f{city} {date} 晴18-26度 tools [get_weather] model ChatOpenAI(modelgpt-4o-mini).bind_tools(tools) def call_model(state: State): resp model.invoke(state[messages]) return {messages: [resp], step_count: state.get(step_count, 0) 1} def should_continue(state: State) - str: if state.get(step_count, 0) 10: return end last state[messages][-1] return tools if getattr(last, tool_calls, None) else end builder StateGraph(State) 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) graph builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: demo-1}} for event in graph.stream( {messages: [(user, 北京明天天气怎么样)], step_count: 0}, config, stream_modevalues ): event[messages][-1].pretty_print()这段代码跑起来后你会看到模型先返回一个带 tool_calls 的消息然后 ToolNode 执行工具结果回给模型模型再生成最终回答。整个循环由should_continue控制最多 10 轮。把 MemorySaver 换成 SqliteSaver 就能持久化换个 thread_id 就是新会话。我个人在实际项目中的体会是LangGraph 的学习曲线主要卡在“状态设计”和“路由逻辑”这两块API 本身并不复杂。把这两个想清楚了剩下的就是工程细节的堆砌。建议一开始别追求大而全先让一个最小闭环跑起来再逐步往上加能力这样每一步都有正反馈也不容易在复杂配置里迷失。
