LangChain系列之LangGraph 实战:把 Agent 循环亲手画出来

2026-09-10 02:540 条评论154 次阅读

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,手脚是工具,规划是思路,记忆是档案。四类组件的关系看这张图:

需要外部能力

观察结果

信息已齐

读写

用户提问

LLM 大脑
理解 · 决策

规划 Planning
拆任务 · 选工具

行动 Action
调工具执行

工具 Tool
搜索 / API / 数据库

最终答案

记忆 Memory
短期会话 / 长期知识

四类组件的出现频率差得很远,选型时心里要有数:

组件 出现频率 说明
行动 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 给最终答案。这张图完整跑通了一次思考、行动、观察的循环:

有 tool_calls

无 tool_calls

START

model 节点
bind_tools 的模型

tools 节点
ToolNode 执行

END

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实战教程