← 回到知识图谱
实践持续完善

智能体编排框架横评(2):LangGraph

本文总结LangGraph 入门:先跑通一个最小的图,再逐个讲 State、Node、Edge、检查点和 interrupt,最后一步步写出建单前等员工确认的报销图。

创建 2026-11-05更新 2026-11-06阅读约 33 分钟
Agent框架LangGraph

智能体编排框架横评 第 2 站 · 上一站:LangChain · 下一站:厂商 Agent SDK · 全程:总览

LangGraph 是什么

LangGraph 是 LangChain 团队做的一个 Python 库(也有 JS 版),用来把 Agent 的执行流程写成一张图。

  • 图里的每个节点是一个普通函数,负责做一件事,比如查制度、调模型、建工单。
  • 节点之间的边规定先做什么、后做什么、什么条件下走哪条路。
  • 所有节点读写同一份状态,状态就是一个字典。
  • 配上检查点以后,图每走一步都会把状态存下来,所以可以中途暂停、等人确认、进程重启后接着跑。

它解决的问题是:流程的步骤、分支和审批点由业务决定,不能全交给模型临场发挥,而且流程可能要跑很久、中间要停下来等人。

和第 1 站的关系:LangChain 的 create_agent 给你一个现成的"模型调工具"循环,它自己就是编译成一张 LangGraph 图来跑的。只是模型加几个工具问答,用 create_agent 就够了;要自己决定流程怎么走,才需要直接写 LangGraph。

安装

pip install -U langgraph

langgraph 自带一个内存版检查点。要把检查点存进数据库,另装对应的包:

pip install -U langgraph-checkpoint-sqlite    # SQLite
pip install -U langgraph-checkpoint-postgres  # PostgreSQL

本文的代码按 LangGraph 1.x 的 Python 官方文档写,是示意代码,没有运行过。

第一个图

先写一个只有一个节点的图:输入一段文字,节点把它变成大写。

hello_graph.py 文件代码:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):       # 1. 定义状态里有哪些字段
    text: str

def shout(state: State):      # 2. 定义节点:读状态,返回要更新的字段
    return {"text": state["text"].upper()}

builder = StateGraph(State)   # 3. 用状态类型创建一张图
builder.add_node(shout)       #    加节点,节点名默认就是函数名
builder.add_edge(START, "shout")
builder.add_edge("shout", END)

graph = builder.compile()     # 4. 编译后才能运行
print(graph.invoke({"text": "hello langgraph"}))

输出结果为:

{'text': 'HELLO LANGGRAPH'}

写任何 LangGraph 程序都是这四步:定义状态,写节点函数,加节点和边,编译后调用 invoke。START 和 END 是两个特殊节点,分别表示图的入口和出口。

基本概念

State:节点共享的数据

State 是一个 TypedDict,列出图里要传递的字段。节点函数收到的是当前完整状态,返回的是要更新的那几个字段,没返回的字段保持不变。

class State(TypedDict):
    question: str
    answer: str

def answer_node(state: State):
    return {"answer": f"收到:{state['question']}"}   # 只更新 answer,question 不动

默认情况下,节点返回的新值会覆盖旧值。如果想让多个节点往同一个列表里追加,要给字段加一个 reducer,告诉 LangGraph 怎么合并:

from typing import Annotated
import operator

class State(TypedDict):
    evidence: Annotated[list[str], operator.add]   # 新值和旧值用 + 拼起来

def a(state): return {"evidence": ["制度:打车标准 80 元"]}
def b(state): return {"evidence": ["发票:INV-102 金额 118.5 元"]}

a 和 b 都跑完以后,evidence 里就有两条。对话消息列表有一个专用的 reducer add_messages,用法相同。

Node:干活的函数

节点就是普通的 Python 函数,参数是状态,返回值是要更新的字段字典。用 add_node 加进图里:

builder.add_node(answer_node)              # 节点名是 "answer_node"
builder.add_node("answer", answer_node)    # 也可以自己起名

节点里可以调模型、调工具、查数据库,LangGraph 不限制你在里面写什么。

Edge:节点之间怎么走

普通边表示做完 A 一定去 B:

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)

条件边根据状态决定下一步去哪。写一个路由函数,返回下一个节点的名字:

def route(state: State):
    return "manual_review" if "超标" in state["answer"] else END

builder.add_conditional_edges("judge", route)

并行:从同一个节点连出两条普通边,两个节点会同时执行。想等两路都完成再往下走,把它们写成列表:

builder.add_edge(START, "policy")
builder.add_edge(START, "invoice")
builder.add_edge(["policy", "invoice"], "confirm")   # 两路都到了才跑 confirm

并行的两个节点如果写同一个字段,这个字段必须有 reducer,否则运行时会报 INVALID_CONCURRENT_GRAPH_UPDATE。

节点也可以不靠边,直接在返回值里说下一步去哪,用的是 Command:

from typing import Literal
from langgraph.types import Command

def judge(state: State) -> Command[Literal["create", "__end__"]]:
    over = state["amount"] > 80
    return Command(update={"over": over}, goto="create" if over else END)

返回类型注解 Command[Literal[...]] 要写上,LangGraph 靠它知道这个节点可能去哪些节点。

Checkpointer 和 thread_id:记住跑到哪了

前面的图每次 invoke 都从零开始,跑完什么也不留。编译时传一个 checkpointer,图每走一步都会把状态存一份快照,叫检查点。快照按 thread_id 分组,同一个 thread_id 就是同一次会话。

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "t-123"}}

graph.invoke({"text": "hello"}, cfg)
print(graph.get_state(cfg).values)   # 读出这个线程最新的状态

InMemorySaver 存在内存里,进程退出就没了,只适合开发调试。上线时换成 SqliteSaver 或 PostgresSaver(第一次用 PostgresSaver 要调一次 .setup() 建表)。

有了检查点,才谈得上暂停、恢复和查看历史,下面的 interrupt 就建立在它上面。

Interrupt:停下来等人

在节点里调用 interrupt(值),图会在这里暂停:当前状态写进检查点,invoke 返回,你的程序可以把这个值展示给用户。等用户做了决定,用 Command(resume=决定) 再调一次 invoke,图接着往下跑,interrupt() 的返回值就是用户的决定。

from langgraph.types import interrupt, Command

def confirm(state: State):
    ok = interrupt({"question": "确认提交这张报销单吗?"})
    return {"confirmed": ok}

graph = builder.compile(checkpointer=InMemorySaver())   # 必须有 checkpointer
cfg = {"configurable": {"thread_id": "t-123"}}

out = graph.invoke({"text": "..."}, cfg, version="v2")
print(out.interrupts)          # 停在 confirm,这里能看到要问用户的内容

graph.invoke(Command(resume=True), cfg, version="v2")   # 用户点了确认,接着跑

version="v2" 让 invoke 返回一个对象,中断信息放在 .interrupts 属性里。恢复时要用同一个 thread_id,LangGraph 才知道接着哪一次往下跑。

有一点必须记住:恢复时,调用 interrupt 的那个节点会从头重新执行,不是从 interrupt 那一行继续。所以 interrupt 前面的代码会跑两次。查数据跑两次没关系,建单、发通知这类有副作用的操作不要放在 interrupt 前面。

动手:一步步写报销图

例子:员工问"我上周出差打车超标了,还能报吗?能的话帮我提报销工单"。我们要做的流程是:

  1. 同时查差旅制度和发票(INV-102 金额 118.5 元,标准 80 元);
  2. 两路结果都拿到后,停下来请员工确认;
  3. 员工确认就建单,拒绝就结束。

模型判断的部分先省略,重点看流程、并行和中断怎么写。search_policy、get_invoice、create_ticket 是三个已有的工具函数。

第 1 步:定义状态

from typing import Annotated, TypedDict
import operator

class S(TypedDict):
    evidence: Annotated[list[str], operator.add]   # 两个节点会同时往里写,所以要 reducer
    ticket_id: str

第 2 步:查制度和查发票两个节点

def policy(s: S):
    return {"evidence": [search_policy("差旅 打车 标准")]}

def invoice(s: S):
    return {"evidence": [get_invoice("INV-102")]}

第 3 步:确认节点

确认节点里只放 interrupt 和路由,不放任何有副作用的代码,因为恢复时它会从头再跑一遍。

from typing import Literal
from langgraph.graph import END
from langgraph.types import interrupt, Command

def confirm(s: S) -> Command[Literal["create", "__end__"]]:
    ok = interrupt({"tool": "create_ticket", "invoice": "INV-102", "over": 38.5})
    return Command(goto="create" if ok else END)

over: 38.5 是 118.5 减去 80 得到的超标额,这里直接写出来做示意。

第 4 步:建单节点,带幂等键

建单是有副作用的操作。进程如果在调用 create_ticket 之后、检查点写入之前崩溃,恢复时这个节点会再跑一次。给它一个固定的幂等键,工单系统看到同一个键就返回原来那张单,不会建两张。

from langgraph.runtime import Runtime

def create(s: S, runtime: Runtime):
    key = f"{runtime.execution_info.thread_id}:INV-102"
    return {"ticket_id": create_ticket("INV-102", idempotency_key=key)}

键由 thread_id 和发票号拼成,每次重跑算出来都一样。不要在节点里用 uuid4() 现生成,那样每次重跑都是一个新键。

第 5 步:连边并编译

from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver

g = StateGraph(S)
for fn in (policy, invoice, confirm, create):
    g.add_node(fn)
g.add_edge(START, "policy")
g.add_edge(START, "invoice")                 # policy 和 invoice 并行
g.add_edge(["policy", "invoice"], "confirm") # 两路都到才确认
app = g.compile(checkpointer=InMemorySaver())  # 上线换 PostgresSaver

confirm 去 create 还是 END 由它返回的 Command 决定,所以这里不用再加边。

第 6 步:运行,然后恢复

cfg = {"configurable": {"thread_id": "t-123"}}

out = app.invoke({"evidence": []}, cfg, version="v2")
print(out.interrupts)   # 停在 confirm,把要确认的内容给员工看

# 员工点了确认(可以是几分钟后,也可以是另一个进程)
app.invoke(Command(resume=True), cfg, version="v2")

下面这张图按步骤演示了这次运行:第 1 步两个查询并行,第 2 步停在 confirm,恢复后 confirm 从头重跑,第 3 步建单。

报销图按超步推进,停在 confirm可缩放的动画图,启用 JavaScript 后可以播放和放大。
  1. 第 1 超步:policy 和 invoice 同时跑,结果都写进 evidence
  2. 两路都完成才进 confirm,节点里调用 interrupt 停下
  3. 员工确认,confirm 跳到 create 建单
  4. 拿到单号,回复员工
  5. 员工拒绝时 confirm 直接去 END,不建单

第 1 超步的细节:

  1. Plan:第一步选订阅输入通道的节点,这里是 policy 和 invoice
  2. Execution:两个节点并行,互相看不到对方写了什么
  3. Update:两条结果用 operator.add 合并,没挂 reducer 会报错
  4. 这一步写一份检查点;有一个节点失败,整步都不生效

第 2 超步:confirm的细节:

  1. confirm 节点调用 interrupt,把要确认的内容交出去
  2. 暂停靠抛异常实现,当前状态写进检查点
  3. invoke 返回,out.interrupts 里是待确认的内容
  4. 员工确认后,用同一个 thread_id 发 Command(resume=True)
  5. confirm 从第一行重新执行,interrupt 这次直接返回 True
  6. 节点返回 Command(goto=create),下一个超步去建单

第 3 超步:create的细节:

  1. 幂等键用 thread_id 拼发票号,两样都已经落盘
  2. 调建单接口时带上这个键
  3. 接口调完进程崩了,恢复时 create 节点会再跑一次
  4. 重跑时算出来的还是同一个键
  5. 接口按键去重,最后只有一张单;用 UUID 就要先单独落盘

进阶

上面的内容够写出一个能用的图了。下面几节讲图运行时的细节,遇到问题时再回来看。

图是怎么一步步跑的:超步

LangGraph 把一次运行分成若干个超步(super-step)。每个超步有三个阶段:

  1. Plan:选出这一步要执行的节点。第一步选连在 START 后面的节点,之后选上一步有更新、并且连到它们的节点。
  2. Execution:并行执行这些节点。执行过程中,各节点的写入互相看不到。
  3. Update:把这一步所有节点的写入合并到状态里,有 reducer 的字段按 reducer 合并。

报销图里,policy 和 invoice 在第 1 个超步,confirm 在第 2 个,create 在第 3 个。检查点在每个超步结束时写入。

由此可以理解两条规则:

  • 并行节点写同一个字段必须有 reducer,因为它们的写入在同一个 Update 阶段合并。并行写入的先后顺序不固定,reducer 最好不依赖顺序。
  • 同一个超步里有一个节点抛异常,这一步的所有写入都不生效。已经成功的节点结果会单独存下,恢复时只重跑失败的那几个。

recursion_limit 限制的是超步数,超过会抛 GraphRecursionError。它要放在 config 的顶层,比如 {"recursion_limit": 50, "configurable": {...}},放进 configurable 里不生效。默认值很大,模型陷入循环时会产生大量调用,建议自己设一个上限。

检查点什么时候写:durability

invoke 和 stream 可以传 durability 参数,控制检查点写入的时机:

模式 什么时候写 代价
"async"(默认) 执行下一步的同时在后台写 进程崩溃可能丢掉最近几步,那几步会重跑
"sync" 下一步开始前写完 每步多一次同步写
"exit" 只在图结束、出错或中断时写 中途崩溃无法从中间恢复

不管哪种模式,它只管 LangGraph 自己的状态,不保证外部系统只被调用一次。节点重试和恢复都会把整个节点再跑一遍,所以工单、支付这类副作用一律要带幂等键,见 可靠执行:状态、幂等与恢复。

查看和分叉历史

检查点按线程串成历史,可以回到过去的某一步:

history = list(app.get_state_history(cfg))   # 新的在前
snap = history[2]                            # 找到想回去的那一步

app.invoke(None, snap.config)                # 回放:从那一步之后重新执行

fork = app.update_state(snap.config, {"evidence": ["改过的证据"]})
app.invoke(None, fork)                       # 分叉:改了状态再往下跑,原历史不动

回放时模型调用、API 请求、interrupt 都会再来一遍。

Store:跨线程的数据

State 和检查点只属于一个线程。员工的报销偏好、过去的判定这类跨会话的数据,放进 Store:

from langgraph.store.memory import InMemoryStore

app = g.compile(checkpointer=InMemorySaver(), store=InMemoryStore())

def create(s: S, runtime: Runtime):
    runtime.store.put(("user", "u-001"), "last_invoice", {"id": "INV-102"})
    ...

Store 按 (namespace 元组, key) 存字典。线程内的放 State,跨线程的放 Store,这和 记忆 里短期记忆与长期记忆的划分一致。

多个 Agent 怎么组合

LangGraph 提供三种组合手段:

  • 子图:一个编译好的图可以直接作为另一个图的节点。
  • Send:条件边返回一组 Send(节点名, 参数),运行时按数量动态扇出,结果用 reducer 汇合。适合"拆成 N 个子任务分别处理再汇总"。
  • Command(goto=..., graph=Command.PARENT):子图里的节点直接跳回父图的另一个节点,handoff 就是这么实现的。

官方现在推荐的多 Agent 写法是把子 Agent 包成工具让主 Agent 调用,只有流程完全自定义时才直接写图。多 Agent 框架的横向比较见 智能体编排框架横评(4):多智能体框架。

流式输出和调试

stream 可以边跑边拿结果,stream_mode 决定拿什么:

for chunk in app.stream({"evidence": []}, cfg, stream_mode="updates"):
    print(chunk)   # 每个节点跑完,打印它更新了哪些字段

常用的几种:values 是每步完整状态,updates 是每步增量,messages 是模型逐 token 输出,custom 是节点里用 get_stream_writer() 自己发的内容。

调试可以接 LangSmith:设环境变量 LANGSMITH_TRACING=true 和 API key,每次运行的节点、输入输出都能在网页上看到。本地运行 langgraph dev 会起一个开发服务,在 LangSmith Studio 里单步看图。也可以通过 OpenTelemetry 接到自己的链路追踪系统。更多见 可观测性:追踪一条任务的代价。

另一种写法:Functional API

除了画图,LangGraph 还支持用普通函数写流程:

from langgraph.func import entrypoint, task

@task
def check_invoice(no: str):
    return get_invoice(no)

@entrypoint(checkpointer=InMemorySaver())
def reimburse(no: str):
    inv = check_invoice(no).result()
    ok = interrupt({"invoice": inv})
    return create_ticket(no) if ok else None

@entrypoint 标记入口函数,里面用 if、for 写流程;有副作用或结果不固定的调用包进 @task。恢复时整个函数从头执行,已经完成的 task 直接从检查点取结果。所以 task 之外不能有副作用,也不要在部署新版本时调换 task 的调用顺序,否则正在等待中的会话恢复时可能对不上。

两种写法底层是同一个引擎,可以混用。流程要画出来给别人看、分支多,用 Graph API;改造已有代码、流程是线性的,用 Functional API 更省事。

和 MCP 的关系

LangGraph 本身不处理 MCP。MCP 工具先通过 LangChain 的适配层变成普通工具,再在节点里调用,这部分在 智能体编排框架横评(1):LangChain 1.x,MCP 本身见 技能、MCP 与集成。图部署到 LangSmith 的 Agent Server 以后,可以通过 /mcp 端点把整个 Agent 暴露成一个 MCP 工具。

常见问题

恢复后建了两张单。 最常见的原因是在同一个节点里先建单再 interrupt。恢复时节点从头跑,建单又执行一次。把建单放到 interrupt 之后的单独节点里,并带幂等键。

加了一条并行边就报 INVALID_CONCURRENT_GRAPH_UPDATE。 两个节点在同一步写了同一个字段,而这个字段没有 reducer。按"State"一节给它加 Annotated[..., reducer]。

interrupt 不生效,图没停下来。 检查两点:编译时有没有传 checkpointer;interrupt 有没有被 try/except 包住。interrupt 是靠抛一个特殊异常实现暂停的,裸的 except 会把它吞掉。

一个节点里调了多次 interrupt,恢复的值对错位了。 多次 interrupt 按出现顺序和 resume 值配对。不要有条件地跳过其中一个,也不要在 while 循环里反复 interrupt。需要多轮审批时,用条件边回到这个节点。

设了 recursion_limit 没用。 它要放在 config 顶层,不能放进 configurable。

状态里的对象存不进检查点。 默认序列化器支持 LangChain 消息、datetime、enum 等常见类型,自定义类和 DataFrame 不行。状态里尽量只放能转成 JSON 的数据。

检查点数据库要当敏感数据保护。 加载检查点时会反序列化,能写检查点库的人就可能让服务执行任意对象构造。LangGraph 出过这类安全公告,升级到修复版本后,还要按官方说明打开严格反序列化模式(环境变量 LANGGRAPH_STRICT_MSGPACK=true)。见 安全边界:让权限落到工具层。

版本说明

LangGraph 1.0 在 2025-10-22 发布,官方承诺 2.0 之前不做破坏性变更。本文按 2026-11 时的 1.2 系列文档写。和旧教程对照时注意三点:

  • invoke(..., version="v2") 是 1.1 加的,返回带 .interrupts 的对象。旧教程里从结果字典的 __interrupt__ 键取中断,仍然能用,但官方计划以后移除。
  • 检查点时机以前用 checkpoint_during=True/False 控制,现在换成了 durability。
  • langgraph-supervisor 包不再积极维护,新项目按官方迁移文档,把子 Agent 包成工具。

和手写循环的对照

跟读 learn-claude-code:17 个 harness 机制怎么挂到一个循环上 系列从零写了一个 harness,跟读 learn-claude-code(1):循环内核 里的 while 循环、工具调用配对、权限检查、上下文压缩都是自己写的。LangGraph 接管的是另一部分:步与步之间的状态怎么保存、怎么恢复。

机制 手写 harness LangGraph
保存什么 s10 每个任务一个 JSON 文件,带 blockedBy 依赖 每个超步一份检查点,按线程串成历史
崩溃后从哪继续 读任务文件,重新认领没做完的任务 读最新检查点,重跑没完成的节点
暂停等人 循环外的权限确认,或把任务状态挂起 interrupt() 暂停,Command(resume) 恢复
恢复后怎么重放 s16 脚本从头重跑,journal 里有的结果直接返回 节点从头重跑;Functional API 是入口函数从头重跑,已完成的 task 取缓存
外部副作用 先落盘再执行,承认至少一次 同样是至少一次,去重靠你自己的幂等键

两边都没有保存调用栈,恢复都是"从头重跑,已完成的部分取缓存"。固定编排与目标闸门:Workflow 运行时和 Goal Loop 的 Workflow 运行时和 LangGraph 的 Functional API 几乎是同一个思路,代价也一样:缓存之外的代码会再执行一次。

任务图、后台与定时:让工作撑过一轮对话 的任务文件是一份工作清单,可以由多个 worker 认领。LangGraph 的检查点是一次执行的快照,绑定在一个线程上。想要多个 worker 抢任务,要用 Agent Server 的任务队列,或者在图外自己维护。

LangGraph 没管的部分:工具调用和结果的配对、消息裁剪和摘要、工具权限,由 LangChain 的 ToolNode、create_agent 和中间件负责;上下文压缩(对应 计划、子代理与上下文预算)要自己写节点或用第 1 站的摘要中间件。

自测

一个 LangGraph 程序最少要写哪几步?

参考:定义 State(TypedDict),写节点函数(读状态,返回要更新的字段),用 StateGraph 加节点和边(至少有一条从 START 出发、一条到 END),compile() 后调用 invoke。

两个并行节点都往 evidence 里写一条结果,为什么默认会报错,怎么修?

参考:两个节点在同一个超步执行,写入在同一个 Update 阶段合并。默认字段一步只接受一个值,第二个写入触发 INVALID_CONCURRENT_GRAPH_UPDATE。给字段加 reducer,例如 Annotated[list[str], operator.add]。

不传 checkpointer 能用 interrupt 吗?恢复时为什么要用同一个 thread_id?

参考:不能。暂停时要把状态写进检查点,恢复时按 thread_id 找到那一份检查点接着跑。换一个 thread_id 就是一次新的运行。

员工在 confirm 中断时点了确认,confirm 里 interrupt 前面有一行发通知给财务的代码,会发生什么?

参考:恢复时整个节点从头重跑,通知会再发一次。把通知挪到 interrupt 之后,或者拆成单独的节点,或者给通知带幂等键。

durability 的三种模式分别在什么时候写检查点?哪种模式下进程中途崩溃没法从中间恢复?

参考:async(默认)在执行下一步的同时后台写;sync 在下一步开始前写完;exit 只在图结束、出错或中断时写,中途崩溃没法从中间恢复。三种都不保证外部系统只被调用一次。

想回到查完发票、还没确认的那一步,改一下证据再试一次,用什么 API?原来的历史会被覆盖吗?

参考:用 get_state_history 找到那一步的快照,update_state(snapshot.config, values) 分叉出新检查点,再 invoke(None, fork_config)。原历史保留,分叉是一条新分支。

参考资料

以下链接均查阅于 2026-11。

连接

这是一篇可以继续生长的笔记。沿着关系继续阅读 ↗