LangChain系列之LangGraph 实战:把 Agent 循环亲手画出来
LangGraph:把 Agent 循环画出来
LangChain 系列 · 第 4 篇
一、type(agent) 打出的秘密
先跑一段代码。这段代码和第一篇里几乎一样,只多了一行打印:
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
model = init_chat_model("gpt-5.4-mini")
agent = create_agent(model=model)
print(type(agent))
# <class 'langgraph.graph.state.CompiledStateGraph'>
我盯着这行输出看了很久。
我用 create_agent 建出来的是一个"编译好的状态图",不是我以为的某种 Agent 对象。也就是说,官方文档里那句"LangChain 在 1.0 后把所有 Agent 统一到 create_agent",落地的真相是:它把 LangGraph 画好的一张图,预置到了你面前。
这篇我把这张图拆开,带你把同一个 Agent 亲手画一遍。
画完你会得到两样东西。一是 create_agent 不再神秘,它每个参数对应图的哪个环节,清清楚楚。二是你从此有了改图的能力,循环想怎么转,由你说了算。
二、先认识 Agent:大脑加四件套
拆图之前,先把 Agent 认清楚。官方定义很长,我拆成人话:
Agent 是以大语言模型为推理与决策核心的软件系统。它结合工具调用、记忆与环境交互能力,规划并执行复杂任务来达成目标。
一句话版本:Agent 是会动脑也会动手的程序。大脑是 LLM,手脚是工具,规划是思路,记忆是档案。四类组件的关系看这张图:
四类组件的出现频率差得很远,选型时心里要有数:
| 组件 | 出现频率 | 说明 |
|---|---|---|
| 行动 Action | 必须 | 没有执行动作,就谈不上 Agent |
| 工具 Tool | 几乎总是 | 大多数 Agent 靠工具拿到外部能力 |
| 规划 Planning | 有条件 | 简单问答不需要,复杂任务才拆解 |
| 记忆 Memory | 最易省略 | 无状态最省事,有状态才更聪明 |
判断一个系统是不是真 Agent,业界常看 5 个能力:理解用户问题、拆解任务、判断何时需要工具、选出正确工具、利用工具结果生成回答并推进任务。
前面 4 个能力,一次模型调用就能完成大半。真正的分水岭在第 5 个,推进任务。
模型回答一次就停了。想推进任务,就得让组件协作起来,一圈一圈转。Agent 为什么必须是一张循环的图,下一节给你答案。
三、Agent 为什么必须是一张图
组件摆在那里是静态的。让大脑、工具、记忆协作起来的,是循环。
单次模型调用只能做一件事。问它天气,它要么直接给答案,要么返回一个工具调用请求。想完成"查天气再决定要不要带伞",一次推理做不到,模型必须来回多轮。
这个模式有个名字,叫 ReAct。思考,行动,观察结果,再思考,直到得出最终答案:
问题在于,多轮推理会产生中间状态。上一轮的工具结果要传给下一轮,调了几次工具要数得清,什么时候该停要有个判断。这些状态散落在代码里,就是一团乱麻。
LangGraph 的答案是把状态显式建模。消息列表、工具结果、循环次数,全部放进一个 State 对象里,图的每个节点读写它,状态流转一目了然。
在 LangChain 1.x 的体系里,LangGraph 是专门的执行与编排层。LangChain 负责提供组件(模型、工具、检索器),LangGraph 负责让它们按图跑起来,LangSmith 负责观察运行过程。三兄弟各管一段,画图用的 StateGraph 正是 LangGraph 的核心 API。
这就是为什么 Agent 底层是一张图。循环是表象,状态机才是本质。
再往深问一层:用 while 循环也能实现同样的多轮推理,为什么要用图?
写一个 while 循环,你只有两个控制点:循环条件和循环体。中间过程全部封在局部变量里,跑完了就丢了。
想在某个中间步骤停下来插一句人工确认,做不到。进程崩溃后想从断点恢复,更做不到。
图的解法是把每个中间步骤都变成显式节点。任何两步之间都可以暂停、检查、插入新逻辑。
搜索系统里调度任务从来不用裸循环,用的都是状态机加转移表,道理一模一样。状态机让过程可暂停、可恢复、可观测,这三条正是 Agent 上产线最需要的能力。
四、三件套:State、Node、Edge
LangGraph 的概念少得惊人,一共三个。我按出现顺序讲。
4.1 State:全图共享的状态
State 是一个类型化字典,定义图上所有节点能读能写的数据结构。最常见的 Agent 状态就是消息列表:
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
这里有个关键细节:Annotated[list, add_messages]。
如果只写 list,每个节点返回新列表会直接覆盖旧值。加上 add_messages 这个 reducer,LangGraph 会把新消息追加到旧列表后面,而不是替换。消息历史能跨节点累积,靠的就是这个注解。
State 里的字段可以自己加。加一个 step_count 记录循环轮数,加一个 retry_count 记录工具重试次数,加一个 user_approved 标记人工审核是否通过。字段越多,图的治理能力越强,条件边里就能按这些字段做路由,后面画 Agent 时会看到 tool_calls 这个字段怎么当路由依据。
4.2 Node:干活的普通函数
节点就是一个 Python 函数。输入整个 State,输出要更新的那部分字段:
def call_model(state: AgentState) -> dict:
response = model.invoke(state["messages"])
return {"messages": [response]}
函数返回的 dict 会按 reducer 规则合并进 State。没有魔法,没有类继承,一个纯函数就是一个节点。
节点内部可以写任何代码。调模型、查数据库、调外部 API、跑一段 Python 逻辑,都行。节点抛异常,图的执行就在这个节点停下,错误会带上节点名和状态快照一起抛出,定位问题比隐式循环容易得多。
4.3 Edge:决定怎么走
边把节点连起来。普通边表示必然走向,条件边表示按 State 内容分叉:
graph.add_edge(START, "model") # 普通边:入口必到 model
graph.add_conditional_edges("model", judge) # 条件边:model 之后看 judge 结果
把三件套放进一个最小例子,一个自增计数器:
from langgraph.graph import END, START, StateGraph
class CounterState(TypedDict):
count: int
def increment(state: CounterState) -> dict:
return {"count": state["count"] + 1}
def should_continue(state: CounterState) -> str:
return "increment" if state["count"] < 3 else END
graph = StateGraph(CounterState)
graph.add_node("increment", increment)
graph.add_edge(START, "increment")
graph.add_conditional_edges("increment", should_continue,
{"increment": "increment", END: END})
一个循环就画出来了。count 从 0 加到 3,条件边放行,图走到 END。
这个例子看着小,它把 while 循环换成图的收益全展示出来了:
| 维度 | 手工 while 循环 | StateGraph |
|---|---|---|
| 中间状态 | 封在局部变量里,跑完即丢 | 显式存在 State 里,随时可查 |
| 暂停 | 只能打断点,改不了状态 | 节点之间可插入 interrupt,改完继续 |
| 崩溃恢复 | 进程没了状态全丢 | 接上 checkpointer,断点续跑 |
| 复用 | 逻辑写在循环里,搬不走 | 节点、子图都能单独复用 |
| 概念 | 一句话作用 | 对应代码 |
|---|---|---|
| State | 全图共享的数据结构,节点读写它 | class AgentState(TypedDict) |
| Node | 一个纯函数,收 State 吐部分 State | def call_model(state): ... |
| Edge | 普通边定走向,条件边定分叉 | add_edge / add_conditional_edges |
五、亲手画一张会跑的 Agent 循环图
概念齐了,现在画真正的 Agent。目标:一个能自己决定何时调工具的循环。
我准备一个天气工具,模型负责决策,工具负责执行:
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""查询指定城市的当前天气。"""
return f"{city}:晴,26 度"
tools = [get_weather]
model = init_chat_model("gpt-5.4-mini", temperature=0).bind_tools(tools)
模型绑定工具后,它的输出里就可能出现 tool_calls,这是循环的开关。
接下来画图,一共 3 步。
第 1 步:模型节点
模型节点做的事情很纯粹,把消息列表交给模型,把返回结果追加进 State:
def call_model(state: AgentState) -> dict:
response = model.invoke(state["messages"])
return {"messages": [response]}
模型这次返回的,可能是普通回答,也可能是带 tool_calls 的请求。它自己不知道下一步该干嘛,决定权在边。
这里有个机理值得说透。bind_tools 做的事,是把工具的名称、描述、参数结构塞进模型调用里的 tools 字段,模型据此学会"遇到需要外部信息的问题时,先输出一个调用请求"。模型本身不会执行任何函数,它只是表达意图。
所以循环的第一轮常常是这样:模型读完用户问题,判断自己缺信息,输出一个 tool_calls,内容是一份完整的调用申请单,工具名、参数都填好了。执行不执行,由图的边和工具节点决定。
第 2 步:条件边
看最后一条消息有没有 tool_calls。有,就去工具节点;没有,说明模型认为不需要外部能力,直接结束:
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
return "tools" if last_message.tool_calls else END
graph.add_conditional_edges(
"model",
should_continue,
{"tools": "tools", END: END},
)
第 3 步:工具节点
工具的执行逻辑完全不用自己写。LangGraph 预置了 ToolNode,你只需要告诉它有哪些工具:
from langgraph.prebuilt import ToolNode
graph.add_node("model", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "model")
graph.add_edge("tools", "model")
工具节点跑完后,把 ToolMessage 追加进消息列表,边又把控制权交回 model。一个闭环成型。
完整代码
from typing import Annotated, TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
@tool
def get_weather(city: str) -> str:
"""查询指定城市的当前天气。"""
return f"{city}:晴,26 度"
tools = [get_weather]
model = init_chat_model("gpt-4o-mini", temperature=0).bind_tools(tools)
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
def call_model(state: AgentState) -> dict:
response = model.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
return "tools" if last_message.tool_calls else END
graph = StateGraph(AgentState)
graph.add_node("model", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue,
{"tools": "tools", END: END})
graph.add_edge("tools", "model")
agent = graph.compile()
编译后 invoke 一次,看消息链怎么演变:
result = agent.invoke({
"messages": [{"role": "user", "content": "上海天气怎么样?"}]
})
for m in result["messages"]:
print(f"{m.type}: {m.content or m.tool_calls}")
你会看到 4 步演变:
human: 上海天气怎么样?
ai: [{'name': 'get_weather', 'args': {'city': '上海'}, ...}] # 模型请求调工具
tool: 上海:晴,26 度 # 工具执行结果
ai: 上海今天晴,气温 26 度。 # 模型给出最终答案
Human 消息进图,AI 请求调工具,Tool 结果回流,AI 给最终答案。这张图完整跑通了一次思考、行动、观察的循环:
ToolNode 内部做的事值得拆开看。它拿到 AIMessage 里的 tool_calls,按工具名找到函数,把 args 传进去执行,再把返回结果包成 ToolMessage 追加回消息列表。你之前手工写的那段"调函数、拼结果、回传"的代码,ToolNode 全替你干了。
一个工具往往不够用。一次复杂请求里,模型可能同时要查上海和北京的天气。这时 AIMessage 里会带两个 tool_calls,ToolNode 会逐个执行,把两条 ToolMessage 一次性追加回去:
ai: 调 get_weather(city=上海) + get_weather(city=北京) # 一个请求,两个工具调用
tool: 上海:晴,26 度
tool: 北京:多云,18 度
ai: 上海晴,北京多云,两地的天气都查好了。 # 综合两条结果回答
你的工具节点代码一行不用改,图天然支持并行工具调用。这就是把工具执行收编进框架的价值,手写循环时你得自己维护工具结果列表,容易漏配。
画这张图常见的坑有三个,写代码时提前避开。
第一个坑是忘了 bind_tools。模型没绑定工具,永远不会输出 tool_calls,条件边永远走 END,Agent 退化成普通对话,工具白挂。
第二个坑是条件边映射表写错。add_conditional_edges 的第二个参数返回的是节点名,映射表要把每个返回值对应到真实节点。返回了图里不存在的名字,运行时报错直接抛出来。
第三个坑是工具节点忘了连回模型。tools 执行完没有边回 model,图走到工具节点就断了,结果没人消费。
多轮任务会在这张图上转好几圈。模型先查 A 再查 B,条件边就带着它走两轮。转的每一圈都意味着一轮模型调用,都要计费。
这意味着一个隐患:如果模型陷入反复请求工具的循环,图的终止条件永远不满足,token 就会一直烧。create_agent 里你能用中间件限制最大轮数,自己画图时这个保险丝要自己加。计数器例子里的 count < 3 条件,就是你给循环加的第一道护栏,思路完全一样。
这就是 create_agent 内部每天在做的事。循环终止的条件只有一条,某次模型输出不再携带 tool_calls。
六、回头看 create_agent:同一张图的预置版
现在回到开头的那个 type()。create_agent 到底做了什么?
它替你完成了两件事。一是按我们上面画的拓扑把图建好,模型节点、工具节点、条件边一个不少。二是帮你接好了配套能力,结构化输出、中间件、记忆,这些在裸图上都需要自己接线。
| create_agent 参数 | 手工画图时对应的环节 |
|---|---|
model |
模型节点里 bind_tools 后的那个模型 |
tools |
ToolNode 的工具列表 |
system_prompt |
注入 State 的初始 SystemMessage |
response_format |
模型输出侧的结构化约束 |
middleware |
挂在图上的拦截器,改消息流或控制循环 |
checkpointer |
compile 时传入的记忆后端 |
你可以把 create_agent 当成一张出厂预置的图。它覆盖 80% 的常规场景,且经过官方打磨,出错概率低。
那什么时候值得自己画图?我的判断标准只有一条:预置图的固定拓扑满足不了你的控制需求。
| 场景 | 用 create_agent | 自己画 StateGraph |
|---|---|---|
| 标准 ReAct 问答 | 直接上 | 没必要 |
| 自定义终止条件(如按轮数、按预算) | 受限于预置行为 | 条件边自己写 |
| 固定流程编排(先检索再生成) | 不合适 | 线性边即可,简单清晰 |
| 并行扇出(同时查多个源再汇总) | 做不到 | 条件边 + 多个工具节点 |
| 人工审核分支(高危操作先停) | 靠 middleware 间接实现 | interrupt 直接插入节点 |
| 子图复用(同一流程多处用) | 不能拆 | 子图节点随意嵌套 |
我自己在产线的经验是:单 Agent 场景 create_agent 够用,图一旦需要定制控制流,马上切到 StateGraph。两者是互补关系,不存在谁取代谁。
这里补一句中间件的位置。create_agent 的 middleware 参数,本质就是挂在图上的拦截器,能看能改每轮消息流。比如 ModelCallLimitMiddleware 限制最大调用轮数,等于给循环上了保险丝。
这张图的每一层都能被治理,这正是状态机相对隐式循环的工程优势。
调试也有现成工具。编译好的图能直接导出成 mermaid,粘到任意渲染器里看拓扑:
print(agent.get_graph().draw_mermaid())
LangSmith 的 Studio 更进一步,能看到每轮 State 的完整快照,还能手动改 State 后从断点继续跑。搜索系统里查链路日志排查问题的习惯,在 Agent 上同样成立,工具还更顺手。
七、给图加一行记忆
自己画图后,LangGraph 的底层能力开始对你敞开。第一个最实用的是记忆。
对话 Agent 天然无状态,第二轮提问时它已经忘了第一轮。解决办法在 compile 时传入 checkpointer:
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = graph.compile(checkpointer=checkpointer)
调用时带上 thread_id,同一个会话的消息历史就会自动累积:
config = {"configurable": {"thread_id": "session-001"}}
agent.invoke({"messages": [{"role": "user", "content": "我叫阿伦"}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "我叫什么?"}]}, config)
# 第二问能答出:你叫阿伦
加记忆只花了一行,换来的是跨轮对话能力。
记忆后端可以换。InMemorySaver 只把状态存内存,进程重启就丢,适合开发调试。想持久化,LangGraph 提供 SqliteSaver、PostgresSaver 等实现,一个文件或一张表就能把会话状态落盘:
| 后端 | 存储 | 适合 |
|---|---|---|
| InMemorySaver | 内存 | 开发调试、单机原型 |
| SqliteSaver | SQLite 文件 | 单机生产,轻量持久化 |
| PostgresSaver | PostgreSQL | 多实例共享,产线标配 |
换后端只改一行 import,图的代码完全不用动。这是 checkpointer 抽象设计的功劳,也是状态机显式建模的又一体现。
同类的底层能力还有 interrupt 人工断点。图跑到指定节点前暂停,等人工确认再继续,高危操作审核就靠它。
八、会画图的人,才谈得上改图
回到开头那个 type()。现在你应该理解,那行输出是 LangGraph 的身份证,没有任何玄学成分。
create_agent 是官方预置的图,好用、安全、出厂带治理能力。但预置图是固定的,想加一个人工审核步骤,想改成先检索再回答的固定流程,想在多节点之间并行扇出,你就需要自己画图。
StateGraph 的能力不止这篇画的这些。interrupt 人工断点、checkpoint 持久化、子图复用、时间旅行调试,每一个都是产线级武器。
关注我的个人公众号后台回复「LangGraph」,可以领取LangGraph实战教程
