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-else：

def 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 Schema：

from 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(checkpointer=checkpointer) 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 反复调用同一个工具，或者在不同工具之间来回横跳。排查顺序如下：

我遇到过一次特别隐蔽的：工具返回了超长的 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(model="gpt-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(checkpointer=MemorySaver()) config = {"configurable": {"thread_id": "demo-1"}} for event in graph.stream( {"messages": [("user", "北京明天天气怎么样")], "step_count": 0}, config, stream_mode="values" ): event["messages"][-1].pretty_print()

这段代码跑起来后，你会看到模型先返回一个带 tool_calls 的消息，然后 ToolNode 执行工具，结果回给模型，模型再生成最终回答。整个循环由should_continue控制，最多 10 轮。把 MemorySaver 换成 SqliteSaver 就能持久化，换个 thread_id 就是新会话。

我个人在实际项目中的体会是，LangGraph 的学习曲线主要卡在“状态设计”和“路由逻辑”这两块，API 本身并不复杂。把这两个想清楚了，剩下的就是工程细节的堆砌。建议一开始别追求大而全，先让一个最小闭环跑起来，再逐步往上加能力，这样每一步都有正反馈，也不容易在复杂配置里迷失。