LangGraph:构建有状态 Agent 的低层编排框架
Agent 正从"一次调用一次回答"进化为长时运行、有状态的复杂工作流——多步推理、调用工具、等待人类审批、跨会话记忆。用裸 while 循环写这类 agent,很快会撞上一堆工程难题:进程崩了从哪恢复?中间某步需要人确认怎么办?多轮对话的状态存哪?LangChain 团队出品的 LangGraph 正是为此而生:一个低层编排框架,把 agent 建模成一张"状态图"——节点是步骤、边是控制流,底层配齐持久化执行、人机协同、双层记忆。Klarna、Replit、Elastic 等公司都在用它构建生产级 agent。参考 GitHub 仓库。
1、概述:LangGraph 是什么
官方定位:"Low-level orchestration framework for building stateful agents"(构建有状态 agent 的低层编排框架)。注意两个关键词:低层(给你的是编排原语而非现成 agent,控制力拉满)和有状态(状态的持久、恢复、演进是一等公民)。
定位 低层编排框架:构建/管理/部署长时有状态 agent
出品 LangChain Inc(但可脱离 LangChain 独立使用)
信任方 Klarna · Replit · Elastic 等生产环境在用
核心模型 状态图(StateGraph):节点=步骤,边=控制流
五大能力 持久执行 / 人机协同 / 双层记忆 / LangSmith 调试 / 生产部署
形态 Python(pip install langgraph)+ JS/TS 版 LangGraph.js
灵感来源 Google Pregel 图计算 + Apache Beam 流处理
License MIT核心要点:
1. 它是"编排原语",不是现成 agent——复杂控制逻辑自己搭,框架管基础设施
2. 与 LangChain 生态无缝,但完全可独立使用
3. 设计灵感来自 Pregel/Beam——把"图执行 + 容错"的分布式系统经验带到 agent
4. 想要"开箱即用的深度 agent"可用其上的 Deep Agents 高层包
2、核心思想:把 agent 建模成状态图
写 agent 的直觉方式是 while 循环:调模型→看结果→决定下一步→循环。简单场景够用,但复杂工作流会变成面条代码。LangGraph 的答案是把流程显式建模为图:
// 状态图三要素
State(状态) 一个共享的、可演进的字典/模式,贯穿整个执行
Node(节点) 一个函数:读状态 → 干活(LLM 调用/工具/逻辑)→ 返回状态更新
Edge(边) 控制流:固定边(A 完成必到 B)或条件边(按状态路由)
// while 循环 vs 状态图
while 循环写法(隐式流程,难恢复难审查):
while True:
action = llm.decide(state)
state = execute(action) # 崩了?从头再来
状态图写法(显式流程,每步可 checkpoint):
START → agent 节点 → 条件边路由 → tools 节点 → 回 agent → ... → END
# 每一步的状态转换都被记录,可暂停/恢复/审查/回放
// 条件边示例(agent 的"思考-行动"循环)
graph.add_conditional_edges(
"agent", # 从 agent 节点出发
should_continue, # 函数:看状态决定去哪
{"continue": "tools", "end": END} # 路由表
)核心要点:
1. State 是核心:全图共享一份状态,节点只读它、更新它
2. 条件边取代 if-else 面条——路由逻辑显式、可视化
3. 图结构带来"每步可观测、可暂停、可恢复"的工程能力
4. 循环(agent→tools→agent)天然支持——这是与 DAG 工作流的本质区别
3、五大核心能力总览
LangGraph 为"长时运行的有状态工作流"提供了五件基础设施,每件都对着一个裸写 agent 的痛点:
能力 解决的痛点
------------------------------------------------------------------------
Durable execution 长任务中途崩溃 → 自动从断点恢复,不从头再跑
(持久执行)
Human-in-the-loop 关键步骤需人审 → 任意节点暂停,人检查/修改状态后继续
(人机协同)
Comprehensive memory 只活在单次对话 → 短期工作记忆 + 跨会话长期记忆
(综合记忆)
LangSmith 调试 agent 是黑盒 → 可视化执行路径、状态转换、运行时指标
(观测)
Production deployment 自建部署太重 → LangSmith Deployment 专为长时有状态工作流
(生产部署) 而建的可扩展基础设施核心要点:
1. 五大能力环环相扣:图模型是骨架,持久化是地基
2. 人机协同建立在持久化之上——暂停后状态不丢,人审完接着跑
3. LangSmith 补齐"可观测",Deployment 补齐"可托管"
4. 后面三节分别展开最硬核的三个:持久执行、人机协同、记忆
4、快速上手:最小可运行的图
装包后几行代码就能跑起来一个最简单的图——单节点聊天机器人,再进阶到"agent + 工具循环":
pip install -U langgraph# 最小示例:单节点聊天机器人
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
def chatbot(state: MessagesState):
return {"messages": [llm.invoke(state["messages"])]}
graph = (
StateGraph(MessagesState) # 状态:消息列表(自动累积)
.add_node("chatbot", chatbot) # 节点:调 LLM
.add_edge(START, "chatbot") # 入口边
.add_edge("chatbot", END) # 出口边
)
app = graph.compile(checkpointer=...) # 挂上 checkpointer 即持久化
# 进阶:agent + 工具循环(LangGraph 最经典的形态)
# START → agent ⇄ tools(条件边循环)→ END
# agent 节点:LLM 决定"回答"还是"调工具"
# tools 节点:执行工具,结果回传 agent
# should_continue:有工具调用 → tools;没有 → END核心要点:
1. 建图四步:定义状态 → 加节点 → 连边 → compile
2. MessagesState 内置消息累积语义,开箱即用
3. compile 时挂 checkpointer,同一张图立即获得持久化能力
4. "agent⇄tools 循环 + 条件边"是构建 tool-calling agent 的标准范式
5、持久执行:崩了从断点继续
长时任务的生死题:一个跑几十分钟的 agent,第 18 步崩了怎么办?LangGraph 的 Durable execution 让 agent扛得住故障、跑得了很久,并从精确的断点自动恢复:
// 原理:checkpointer 给每一步状态快照
每个"超步"(superstep)执行完 → 状态持久化(内存/SQLite/Postgres...)
故障恢复 → 加载最近 checkpoint → 从断点继续,已完成步骤不重跑
// 价值场景
长时间任务 深度研究、批量处理跑几十分钟~几小时,中途崩溃代价不可接受
过程留痕 每步状态都在,随时回放"agent 当时看到了什么、决定了什么"
时间旅行 从历史任一 checkpoint 分叉重跑,调试与 A/B 神器
并发扩展 checkpoint 落库后,恢复可在另一台机器上进行
// 用法核心就是 compile 时传入 checkpointer:
# from langgraph.checkpoint.memory import MemorySaver # 开发
# from langgraph.checkpoint.sqlite import SqliteSaver # 单机
# from langgraph.checkpoint.postgres import PostgresSaver # 生产
app = graph.compile(checkpointer=MemorySaver())
# 每个 thread_id 一条独立的状态线(一个会话/一个任务实例)核心要点:
1. checkpoint 每 superstep 落一次,恢复从最近快照续跑
2. checkpointer 可插拔:内存(开发)→ SQLite → Postgres(生产)
3. thread_id 隔离不同会话/任务实例的状态线
4. 快照还带来时间旅行:从任意历史点分叉重跑
6、人机协同:interrupt 机制
高风险动作(发邮件、转账、删数据、提交 PR)不该让 agent 单方面决定。LangGraph 的 human-in-the-loop 让你在执行中的任意点暂停、检查甚至修改 agent 状态,人确认后再继续:
// 核心原语:在需要人审的节点里调用 interrupt()
def human_approval(state: AgentState):
answer = interrupt({ # ← 执行到此暂停,状态已持久化
"question": "确认执行以下操作吗?",
"pending_action": state["pending_action"],
})
# 人回复后从此处恢复,answer 即人工输入
if answer == "yes":
return Command(goto="execute")
return Command(goto="revise")
// 典型流程
agent 规划 → interrupt 暂停(状态入 checkpoint)
→ 人检查/修改状态(如编辑即将发送的邮件内容)
→ 恢复执行 → 完成
// 为什么能"暂停后继续"?
// 因为持久化(第5节)先把状态存好了——人审一小时、一天都不怕
// 人机协同正是建立在 durable execution 地基之上的能力核心要点:
1. interrupt() 在节点内触发暂停,状态已落 checkpoint
2. 人工不止"批准/拒绝"——还能直接修改状态再继续
3. 暂停可跨天:人审的等待期由持久化兜底
4. 与"工具调用前强制确认"组合,构成生产级安全网
7、记忆:短期与长期双层
真正的"有状态"不止活过崩溃,还要活过会话。LangGraph 的记忆体系分两层:
层级 范围 载体
------------------------------------------------------------------------
短期记忆 单次会话内 即 State 本身:
(working memory) 多轮推理的工作区 消息历史、中间结果
(checkpointer 持久化)
长期记忆 跨会话、跨天/月 独立的 store:
(persistent memory) "记住用户偏好/事实" 按命名空间读写的记忆条目
跨 thread 存活
// 组合效果
会话内:agent 能记住本轮上下文 → 短期记忆(State)
新会话:agent 还记得"你是谁、你喜欢什么" → 长期记忆(store)
跨线程:同用户多个任务共享画像 → 长期记忆按用户命名空间组织核心要点:
1. 短期记忆 = State + checkpointer:会话内工作区的持久化
2. 长期记忆 = 独立 store:按命名空间跨会话读写
3. 两层配合:工作区想得起来,长期偏好忘不掉
4. 这是"聊天机器人"与"私人助理"的分水岭能力
8、总结
LangGraph 把"agent 编排"从手写 while 循环升级为显式的状态图 + 完整的工程基础设施:图模型给你控制力,checkpoint 给你可靠性,interrupt 给你安全性,双层记忆给连续性,LangSmith 给可观测性。
关键要点:
- 定位:低层编排框架,构建长时运行的有状态 agent(MIT)
- 模型:StateGraph——节点是步骤、边是控制流、条件边做路由
- 持久执行:checkpoint 每 superstep 快照,崩溃从断点恢复
- 人机协同:interrupt 任意点暂停,人可检查/修改状态后继续
- 记忆:短期(State+checkpoint)+ 长期(跨会话 store)双层
- 生态:可独立用;上有 Deep Agents 高层包,旁有 LangSmith 调试/部署
- 血统:灵感源自 Google Pregel 与 Apache Beam,API 风格学 NetworkX
对于要构建生产级 agent——尤其是多步骤、长耗时、需要人审或跨会话记忆——的团队而言,LangGraph 已经是事实上的主流选择之一(Klarna/Replit/Elastic 均在生产使用)。它的价值不在"帮你调模型",而在把分布式系统几十年沉淀的"状态、快照、恢复、协调"经验搬进了 agent 世界:你用图描述业务逻辑,它保证这条逻辑跑得久、断得起、停得下、记得住、看得清。如果只想快速起步,先从 StateGraph 单节点跑通,再逐步加条件边、checkpointer 与 interrupt——每加一层,你的 agent 就离"生产级"更近一步。


