跳到主内容

LangGraph 状态图实战:用节点、边与检查点构建可控智能体

100%
LangGraph 状态图实战:用节点、边与检查点构建可控智能体

一句话结论

LangGraph 用「节点 + 边 + 共享状态」把智能体画成一张状态图——每一步做什么、何时分支、何时停下来等人审批全部显式声明,换来的是可恢复、可中断、可时间回溯的可控智能体,代价是要先想清楚图再写代码。

LangChain 入门教程里那个 20 行的智能体跑通后,很多人在真实业务里撞上同一堵墙:循环式智能体(agent loop)走到第十几步就失控——上下文越滚越长、该停不停、出错想回放却无从下手。 LangGraph 的答案是把「控制流」从模型手里拿回来一部分:你定义图的形状,模型只在允许分支的节点里做决定。

三个核心概念

概念是什么要点
State(状态)所有节点可读写的共享内存存原始数据,提示词在节点内现拼
Node(节点)接收状态、执行一步、返回状态增量的函数一个节点只做一件事
Edge(边)固定边直达;条件边按状态路由条件边必须写具名路由函数

最小骨架长这样(Python):

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, HumanMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]  # reducer:累积而非覆盖

def agent_node(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")
app = graph.compile(checkpointer=MemorySaver())

那个 add_messages 是最常被忽略的细节:没有 reducer,列表字段会被节点返回值整体覆盖而不是追加——官方文档称漏写它是 LangGraph 最常见的 bug 。

官方推荐的五步设计法

LangChain 文档用一个客服邮件智能体示范了标准思路:


  1. 把工作流拆成离散步骤,每步一个节点:读邮件 → 分类意图 → 文档检索 → 起草回复 → 人工审核 → 发送。

  2. 给每个节点分类:LLM 步骤(需要静态提示 + 状态里的动态上下文)、数据步骤(检索、缓存、重试策略)、动作步骤(发信等副作用,不缓存)、人工步骤(定义触发条件与输入格式)。

  3. 设计状态:只存跨步骤需要的原始数据(原邮件、分类结果、检索结果、草稿),能推导的不存。

  4. 连边:流程固定的用普通边;「分类后走哪条路」这类决策用条件边 + 具名路由函数。

  5. 编译并选检查点:测试用 MemorySaver,生产换 Postgres/Redis/SQLite——没有检查点就没有恢复、中断与回溯。

检查点:可控性的来源

检查点(checkpointer)按 thread_id 保存每次执行的状态快照,由此得到四个生产级能力:

  • 持久记忆:同一 thread 的多轮对话状态跨运行保留。
  • 人工审批(interrupt):interrupt_before=["tools"] 让图在工具调用前暂停,人批准后用 Command(resume=True) 续跑——审批点应设在「进入有副作用的节点」的边上,而不是事后补救。
  • 时间旅行调试:get_state_history() 拿到历史快照,从任意检查点分叉重放,复现一次糟糕的智能体执行不用重跑整场对话。
  • 流式事件:stream_mode="updates" 把每个节点的状态增量推给前端,用户能实时看到「正在思考 → 正在调用工具 → 已拿到结果」。
官方的 60 秒设计检查:写下节点名(说不出来说明任务还不是「智能体形状」)、声明最小状态(每个列表字段配 reducer)、画出边、先选好检查点、把审批放在副作用节点之前的边上。五条过不了任何一条,就先别动手写代码。

进阶:子图与多智能体

子图(subgraph)拥有自己的状态和检查点,这是搭监督者-工作者(supervisor-worker)架构的正规姿势:监督者图按意图把任务路由给各领域的 worker 子图,每个子图独立演进、独立测试。这也是我们在多智能体编排里讲的 Lead + Subagent 形态在 LangGraph 里的具体落法——区别在于 LangGraph 把「谁报告给谁、状态怎么合并」全部写进了图结构,而不是靠提示词约定。

工程红线同样明确:MemorySaver 只配测试环境,生产必须上持久化存储;条件边的路由函数保持纯函数(只读状态、返回节点名);状态里别塞万物——任务专属字段(计划、预算计数、已检索文档列表)提升到顶层字段,比全挤在 messages 里可调试得多。

选型上,是否用 LangGraph 判断标准只有一条:任务需要「显式控制流 + 中途人工介入 + 断点恢复」中的至少两项。三项都不需要的,普通的 agent loop 更轻;拿不准的可以回到智能体还是工作流的四个判断问题先过一遍,涉及人审批的环节再对照智能体人机协作(HITL)的落位原则。


LangGraph 和直接用 LangChain Agent 的区别是什么?
LangChain Agent 是循环式:模型决定下一步直到收敛,控制流在模型手里。 LangGraph 把控制流画成显式的图:节点固定、分支条件你自己写,模型只在指定节点做决策。前者灵活,后者可控。

messages 忘写 reducer 会有什么症状?
节点返回 {“messages”: [msg]} 后整个历史被替换成这一条,表现为多轮对话「失忆」。解决:状态定义里给列表字段加 Annotated[list, add_messages]。

生产环境能用 MemorySaver 吗?
不能。 MemorySaver 只存内存,进程重启全丢。 Postgres 、 Redis 、 SQLite 的官方 checkpointer 都可以直接换入,compile 时替换对象即可。

人工审批为什么要放在工具节点之前?
工具调用是产生外部副作用的时刻(发邮件、写数据库、删文件)。 interrupt_before=[“tools”] 让图停在这个动作发生前,人看到模型打算调用什么再放行,阻止的就是「批准之前不该发生的副作用」。

这篇有帮助吗?
云上的幻象
云上的幻象查看主页

七彩云博客,分享 WordPress 建站实战与 AI 工具测评,覆盖服务器运维、站长工具、软件资源与电商运营干货,专注原创实用的主题插件、网站加速与安全优化教程。

847文章4评论

相关文章

评论 (0)

欢迎你,新朋友,感谢参与互动!文明发言,理性交流 · 首次评论将在审核后展示