搜索

Resource / Python

LangGraph
Cheatsheet

整理 StateGraph、State、Reducer、Command、Checkpoint、Interrupt 和 Streaming 等常用能力，并说明各 API 的使用场景。示例基于当前官方稳定文档整理。

参考版本
LangGraph 1.2.11
语言范围
Python · Graph API
STARTinput
NODEupdate
ROUTECommand
SAVEcheckpoint
ENDoutput
一次 Graph 运行过程：State 进入 Node，Node 返回更新并决定后续路由，系统在每个 superstep 结束后保存 Checkpoint。
一句话理解

LangGraph 是构建有状态 Agent 的底层编排框架，负责管理 State、执行流程、持久化和恢复。Prompt 与具体 Agent 结构仍由应用自己设计。

01 / BOOT

安装与最小图

基础库要求 Python 3.10+，只需要安装 langgraph。使用模型和工具集成时，通常还要安装 langchain 及对应的模型厂商集成包；LangGraph 本身也可以独立使用。

terminalBash
pip install -U langgraph

如需本地 Agent Server 与 Studio，使用单独的 CLI 安装方式，并保证 Python 3.11+：

terminal · local serverBash
pip install -U "langgraph-cli[inmem]"
minimal_graph.pyPython
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
 
class State(TypedDict):
    question: str
    answer: str
 
def answer(state: State):
    return {"answer": f"收到：{state['question']}"}
 
builder = StateGraph(State)
builder.add_node("answer", answer)
builder.add_edge(START, "answer")
builder.add_edge("answer", END)
 
graph = builder.compile()
result = graph.invoke(
    {"question": "LangGraph 是什么？"},
    version="v2",
)
print(result.value["answer"])

compile() 会检查 Graph 配置，并生成可以 invoke 或 stream 的可执行 Graph；checkpointer、breakpoint 等运行配置也在这里接入。

02 / MODEL

四个核心概念

State

图的共享数据结构。节点读取当前状态，只返回需要更新的字段。

Node

一个可执行步骤：普通函数、异步函数或 runnable。职责越单一，恢复与测试越容易。

Edge

决定下一步去哪。固定边表达确定流程，条件边表达纯路由。

Reducer

定义同一状态字段收到多个更新时如何合并；未声明时默认覆盖。

执行按 superstep 推进：同一 superstep 的多个节点可以并行运行，完成后将更新写回状态，再进入下一步。只要并行节点可能更新同一 key，就必须先定义可合并的 reducer。

03 / STATE

State 与 Reducer

State 应优先保存实际业务数据，不要保存已经拼好的 prompt。能够由其他字段计算得到的值通常不必重复存储。

State schema 选择
方案适用场景注意事项
TypedDict默认首选；类型清晰、运行开销低不做运行时验证
dataclass需要默认值或更强的数据对象语义写法比 TypedDict 更重
Pydantic确实需要递归运行时校验校验有性能成本
MessagesState主要状态就是消息列表内置 messages 与 add_messages
state.pyPython
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
 
class State(TypedDict):
    # 默认：后一次更新覆盖前一次值
    status: str
 
    # 并行分支的结果需要 reducer 合并
    results: Annotated[list[str], operator.add]
 
    # 按消息 ID 更新或追加，并反序列化为 Message 对象
    messages: Annotated[list[AnyMessage], add_messages]

输入 State、输出 State 与内部 State

Graph 默认使用同一份 State 作为输入和输出。可以通过 input_schema 与 output_schema 限制对外暴露的输入和输出字段；Node 之间的中间结果放在 Overall State 或单独的 Private State 中，无需返回给调用方。

state_boundaries.pyPython
builder = StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState,
)

State、Context 与 Store 不要混用

当前运行

State

保存运行中会变化、需要写入 Checkpoint 的业务数据。

单次调用

Context

提供用户 ID、模型选择、数据库连接等本次调用需要的依赖。

跨 Thread

Store

保存跨 Thread 的用户偏好、长期 Memory 和共享数据。

runtime_context.pyPython
from dataclasses import dataclass
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
 
@dataclass
class Context:
    user_id: str
    model_name: str
 
def call_model(state: State, runtime: Runtime[Context]):
    model = load_model(runtime.context.model_name)
    return {"answer": model.invoke(state["question"])}
 
builder = StateGraph(State, context_schema=Context)
graph = builder.compile()
graph.invoke(
    {"question": "..."},
    context={"user_id": "u-1", "model_name": "provider:model"},
)

04 / ROUTE

Edge、Command 与 Send

路由 API 选择
需求使用关键规则
固定下一步add_edge()显式连接节点或 START / END
只做条件路由add_conditional_edges()路由函数返回目的节点或路径映射 key
更新状态并路由Command用返回类型列出可能目的节点，便于图渲染
动态并行分发Send每个分支可携带独立输入，聚合字段需要 reducer
command.pyPython
from typing import Literal
from langgraph.types import Command
 
def review(state: State) -> Command[Literal["publish", "revise"]]:
    approved = state["score"] >= 0.8
    return Command(
        update={"approved": approved},
        goto="publish" if approved else "revise",
    )

Command 的主要字段是 update、goto、graph 和 resume。节点或工具通常返回前三者；调用 invoke() / stream() 恢复中断时才传 Command(resume=...)。

Map-reduce：用 Send 动态 fan-out

map_reduce.pyPython
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.types import Send
 
class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]
 
def fan_out(state: MapState):
    return [Send("worker", {"item": item}) for item in state["items"]]
 
builder.add_conditional_edges("planner", fan_out)

05 / TOOLS

ToolNode：处理标准工具调用流程

标准 Agent 优先使用 LangChain 的 create_agent。需要自己控制 StateGraph、工具执行和返回路径时，使用预构建的 ToolNode 与 tools_condition，而不是重复实现并行调用、错误处理和状态注入。

tool_loop.pyPython
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.prebuilt import ToolNode, tools_condition
 
model_with_tools = model.bind_tools(tools)
 
def call_model(state: MessagesState):
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node("llm", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", tools_condition)
builder.add_edge("tools", "llm")
每个 Tool Call 都要有对应的 ToolMessage

消息历史中的 AI Tool Call 后面必须有对应 ToolMessage，否则模型服务可能拒绝这段消息序列。

工具返回 Command

ToolNode 会传播工具返回的 Command。自定义工具执行节点必须自行把 Command 作为节点更新返回。

路由机制只选一种

工具驱动的动态 goto 与节点静态边同时存在时，两条路径都会运行。

06 / PERSIST

Checkpoint 与 Memory

Checkpointer 在每个 superstep 保存 Checkpoint，并按 thread_id 组织 State。有了 Checkpoint，Graph 才能在 Interrupt、进程重启或后续请求中从之前的 State 继续运行，同时也支持短期 Memory 和 Time Travel。

persistence.pyPython
from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()  # 仅用于本地例子和测试
graph = builder.compile(checkpointer=checkpointer)
 
config = {"configurable": {"thread_id": "demo-1"}}
result = graph.invoke(
    {"question": "记住这次运行"},
    config=config,
    version="v2",
)
 
snapshot = graph.get_state(config)
history = list(graph.get_state_history(config))
THREAD

Checkpointer

保存一次会话线程的图状态。每次可恢复运行都应传稳定、唯一的 thread_id。

CROSS-THREAD

Store

保存跨线程、跨会话数据，例如用户偏好。通过 builder.compile(store=store) 注入。

InMemorySaver 与 InMemoryStore 适合示例和测试，不适合作为生产持久层。官方提供独立的 SQLite 与 PostgreSQL checkpoint 包；生产示例使用 PostgreSQL：

terminal · PostgreSQLBash
pip install -U "psycopg[binary,pool]" \
  langgraph langgraph-checkpoint-postgres

首次使用 PostgreSQL saver 时调用 checkpointer.setup() 创建所需表。部署在 Agent Server 上时，服务端会管理 checkpointer，不应在图定义里再手动初始化。

Time Travel：Replay 与 Fork

Replay 使用旧 checkpoint 的配置重新执行其后的节点；Fork 先调用 update_state() 创建分支，再从分支继续。两者都不会删除或回滚原执行历史。

time_travel.pyPython
history = list(graph.get_state_history(config))
before_review = next(
    snapshot for snapshot in history
    if snapshot.next == ("review",)
)
 
# 创建新 checkpoint 分支，不会修改原历史
fork_config = graph.update_state(
    before_review.config,
    values={"approved": False},
)
 
# 从新分支继续；该 checkpoint 之后的节点重新执行
fork_result = graph.invoke(None, fork_config, version="v2")

长对话要控制历史消息的保留方式

Checkpoint 会保存消息，但不需要每次都把全部历史消息交给模型。可以根据应用选择 token-aware trim、摘要、过滤，或使用 RemoveMessage 永久删除；删除操作要求消息字段使用 add_messages Reducer。

message_lifecycle.pyPython
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
 
# 删除指定消息
return {"messages": [RemoveMessage(id=message.id)]}
 
# 或清空整个消息 channel
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

序列化与加密

默认 JsonPlusSerializer 能处理常见 LangGraph、LangChain 和 Python 类型。只有明确接受反序列化风险时才启用 pickle fallback。如果生产环境的 Checkpoint 中包含身份信息、工具结果或业务数据，应考虑加密存储。

encrypted_checkpoint.pyPython
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
 
# 从 LANGGRAPH_AES_KEY 读取密钥
serde = EncryptedSerializer.from_pycryptodome_aes()
checkpointer = PostgresSaver.from_conn_string(DB_URI, serde=serde)
checkpointer.setup()

07 / RESUME

Interrupt：在任意节点暂停并恢复

interrupt(payload) 保存当前状态并暂停图；之后使用同一个 thread_id 和 Command(resume=value) 恢复。payload 与 resume value 应保持 JSON 可序列化。

human_in_the_loop.pyPython
from langgraph.types import Command, interrupt
 
def approve(state: State):
    approved = interrupt({
        "question": "是否发布当前结果？",
        "draft": state["answer"],
    })
    return {"approved": bool(approved)}
 
# 第一次调用：运行到 interrupt 并保存状态
pending = graph.invoke(input, config=config, version="v2")
print(pending.interrupts)
 
# 同一个 thread_id：把人工答案送回 interrupt()
completed = graph.invoke(
    Command(resume=True),
    config=config,
    version="v2",
)

并行 Interrupt 要按 ID 恢复

多个并行分支可以在同一轮同时暂停。一次恢复全部等待项时，应把每个 Interrupt ID 映射到对应答案，避免把同一个值错误地送入多个审批节点。

resume_parallel_interrupts.pyPython
pending = graph.invoke(input, config=config, version="v2")
 
resume_map = {
    item.id: answer_for(item)
    for item in pending.interrupts
}
 
completed = graph.invoke(
    Command(resume=resume_map),
    config=config,
    version="v2",
)

08 / STREAM

Streaming v2

当前文档推荐使用统一的 v2 流格式。每个 part 都包含 type、ns 和 data；invoke(..., version="v2") 返回的 GraphOutput 通过 .value 与 .interrupts 读取结果。

stream.pyPython
for part in graph.stream(
    input,
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if part["type"] == "updates":
        print("state update:", part["data"])
    elif part["type"] == "messages":
        message_chunk, metadata = part["data"]
        print(message_chunk.content, end="")
常用 stream_mode
模式输出用途
updates每步产生的状态增量默认 UI 与调试视图
values每步后的完整状态需要完整快照时使用
messagesLLM token / message chunk + metadata聊天打字流
custom节点通过 get_stream_writer() 发出的自定义数据进度、检索状态
checkpoints / tasks / debug持久化、任务与调试事件观测与诊断

需要子图事件时传 subgraphs=True。异步应用使用 astream()；不要为了流式输出把整个页面改为客户端渲染。

09 / LIMITS

循环终止与递归限制

循环图必须同时具备业务退出条件和运行上限。超过 recursion_limit 会抛出 GraphRecursionError；它通常是在提示条件边没有收敛，而不只是配置值太小。

recursion_limit.pyPython
from langgraph.errors import GraphRecursionError
 
try:
    result = graph.invoke(
        input,
        config={"recursion_limit": 100},
    )
except GraphRecursionError:
    handle_incomplete_run()
配置位置

recursion_limit 是 config 顶层字段，不能放进 configurable。

接近上限时退出

把 RemainingSteps 加入 State，可在步数耗尽前转到总结或返回部分结果的 Node。

先查循环

如果业务本不应运行很多轮，优先检查条件边、结束节点与重试回路，不要直接提高上限。

10 / COMPOSE

子图与 Functional API

子图适合复用一段独立流程或隔离 State。父图启用持久化后，子图是否继承或单独使用 Checkpointer，决定其 State 如何保存：

子图 checkpointer 选择
编译方式语义适用
checkpointer=None从父图继承；每次调用隔离大多数子图，官方默认建议
checkpointer=True按 thread 持久化子图状态子图自身需要多轮记忆
checkpointer=False无持久化纯函数式步骤；不能依赖 interrupt / durable execution

从子图跳转到父图节点时可返回 Command(graph=Command.PARENT, goto="...")。相关状态字段必须在父子图之间共享，并定义兼容 reducer。

什么时候用 Functional API

已有过程式控制流，只想为关键步骤增加持久化、Task 和恢复能力时，可以使用 @entrypoint 和 @task。如果流程分支较多，或者希望清楚看到 Graph 结构，Graph API 更合适。可重试的外部操作应封装成支持幂等执行的 Task。

Durable Execution：把外部操作封装成 Task

恢复执行时，Node 或 entrypoint 可能重新运行。把 API 请求、发送消息、写文件等外部操作封装为 @task，已经成功的 Task 可以从 Checkpoint 读取结果。Task 本身仍要使用幂等键，因为进程可能在操作完成后、结果持久化前失败。

durable_side_effect.pyPython
from langgraph.func import task
 
@task
def send_notification(event_id: str, message: str):
    # 用 event_id 作为幂等键，避免恢复时重复发送
    return notification_api.send_once(event_id, message)
 
def notify(state: State):
    receipt = send_notification(
        state["event_id"], state["message"]
    ).result()
    return {"receipt": receipt}

11 / OPERATE

可靠性、错误策略与测试

瞬时错误网络抖动、限流RetryPolicy
模型可修复错误工具参数或格式不合法写入状态并回环
用户可修复错误缺少审批或关键信息interrupt()
未知错误代码缺陷、未预期异常向上抛出
retry.pyPython
from langgraph.types import RetryPolicy
 
builder.add_node(
    "call_api",
    call_api,
    retry_policy=RetryPolicy(max_attempts=3),
)

测试时为每个用例创建新图，并使用全新的内存 checkpointer，避免线程状态互相污染。可以直接调用 graph.nodes["node_name"].invoke(...) 做节点单测，但这种方式会绕过图级 checkpointer；恢复、分支与中断必须通过完整图测试。

Node Cache 只缓存可安全复用的结果

node_cache.pyPython
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy
 
builder.add_node(
    "expensive_lookup",
    expensive_lookup,
    cache_policy=CachePolicy(ttl=300),
)
graph = builder.compile(cache=InMemoryCache())

Cache key 默认由节点输入生成。只缓存输入相同即可安全复用的昂贵计算；不要缓存发送消息、扣款、读取实时权限或依赖隐含用户身份的节点。

修改已经上线的 Graph 时要注意什么

对于已经结束的 Thread，可以调整 Node 和 Edge。对于停在 Interrupt 或下一步即将进入某个 Node 的 Thread，不要删除或重命名该 Node。State 新增或删除字段通常可以兼容旧 Checkpoint；重命名字段会丢失旧 Checkpoint 中对应的数据，修改为不兼容的类型也可能导致恢复失败。

12 / IMPORTS

常用导入路径速查

当前 Python API 导入路径
导入路径常用对象
langgraph.graphStateGraph, MessagesState, START, END
langgraph.graph.messageadd_messages
langgraph.typesCommand, Send, interrupt, RetryPolicy, CachePolicy
langgraph.prebuiltToolNode, tools_condition
langgraph.functask, entrypoint
langgraph.errorsGraphRecursionError
langgraph.managedRemainingSteps
langgraph.checkpoint.memoryInMemorySaver
langgraph.store.memoryInMemoryStore
langgraph.runtimeRuntime
langgraph.configget_stream_writer

13 / DEBUG

常见错误与排查方法

症状、原因与第一检查点
症状常见原因先检查
INVALID_GRAPH_NODE_RETURN_VALUE节点返回 list、字符串或完整对象，而不是状态更新 dict每条代码路径是否都返回包含 State key 的 dict
INVALID_CONCURRENT_GRAPH_UPDATE并行节点写入同一字段，但没有 reducer字段的业务合并规则
GRAPH_RECURSION_LIMIT条件路由没有收敛或循环缺少退出条件END 路径和重试计数，而非先调大限制
模型拒绝消息历史AI Tool Call 后缺少对应 ToolMessage每个 tool_call_id 是否都有对应的 ToolMessage
Interrupt 无法恢复换了 thread_id、没有 checkpointer 或 resume 目标错误原 config 与 pending interrupt ID
节点意外执行两条路径静态 Edge 与 Command / 条件路由同时存在同一出口是否配置了两套路由
恢复后重复写入或发送外部操作未封装为 Task，或外部接口没有幂等键失败发生在操作完成前，还是 Task 结果持久化前
旧 Thread 部署后失败删除、重命名暂停节点，或不兼容地修改 State 类型失败 checkpoint 的 next 与 state schema

14 / SHIP

生产检查清单

  • State 只保存实际业务数据；Node 返回 dict 更新，不原地修改 State。
  • Context 承载本次调用依赖，Store 承载跨 Thread 数据；连接对象和客户端不进入 State。
  • 每个 Node 只处理一项明确任务；为 I/O Node 配置重试条件，写操作使用幂等键。
  • 并行分支写同一字段前，先定义 reducer 和业务合并规则。
  • 循环同时设置业务退出条件和 recursion_limit，接近上限时转到总结或返回部分结果的 Node。
  • 每个 AI Tool Call 都有对应 ToolMessage；自定义工具节点正确传播 Command。
  • 需要恢复或 Interrupt 的图配置持久化 checkpointer，并稳定传递唯一 thread_id。
  • 跨线程记忆放 Store，不把用户长期偏好塞进单个 thread checkpoint。
  • interrupt() 前的写数据库、外部 API 调用等操作支持幂等执行；payload 和 resume value 可以序列化。
  • 生产环境不用内存 saver/store；首次部署正确执行持久层 setup / migration。
  • 包含敏感信息的 checkpoint 已评估加密、密钥轮换、保留周期和访问权限。
  • 流式 UI 处理断连、重复事件和恢复；观测至少覆盖节点耗时、异常与状态迁移。
  • 测试包含节点单测、分支路径、并行 reducer、checkpoint 恢复和中断恢复。
  • 上线前用真实后端做故障注入：限流、超时、进程重启与重复 resume。

15 / SOURCES

官方资料与参考范围

本文内容基于当前 LangGraph 官方稳定文档整理。版本号来自 PyPI，API 用法与推荐实践来自 LangChain / LangGraph 官方文档。