编译与运行

前面六篇讲完了图怎么搭。本篇讲怎么把图跑起来——compile() 的细节、invoke / stream / batch 的不同、config 的用法、状态保持、异常处理。掌握本篇，你的图就真正能"动"起来。

一、compile() 详解

回顾 StateGraph 状态图：compile() 把声明式图变成可执行版本 CompiledGraph。

app = graph.compile(
    checkpointer=...,            # 检查点器，持久化 + 记忆
    interrupt_before=["node"],   # 在指定节点之前暂停
    interrupt_after=["node"],    # 在指定节点之后暂停
    # 0.2+ 也支持在节点内用 interrupt() 主动暂停，更推荐
)

compile 做的事：

1. 校验图结构（孤立/不可达节点警告）。
2. 把边编译成可调度结构。
3. 打包 checkpointer / interrupt 等运行时配置。
4. 返回不可变的 CompiledGraph，可并发复用。

编译产物是 Runnable，可以用 LangChain 的 Runnable 协议：invoke / stream / batch / ainvoke / astream。

二、运行方式一览

方法	模式	适用
invoke(input, config)	同步、一次性返回最终状态	大部分场景
stream(input, config)	同步、流式逐步返回事件	看中间步骤、UI 渐进显示
ainvoke(input, config)	异步一次性	异步代码环境
astream(input, config)	异步流式	异步 + 流式
batch(inputs, config)	批量并发	多输入并行处理

下面分别演示。

1. invoke

result = app.invoke(initial_state)   # 返回最终完整 state（dict）

最直接。一次调用拿到最终状态。

2. stream / astream

for event in app.stream(initial_state):
    print(event)   # 每次节点产出更新就 yield 一个事件

# 异步版本
async for event in app.astream(initial_state):
    print(event)

事件结构取决于 stream_mode（见下）。在 Web 服务（FastAPI）里推荐用异步。

3. batch

results = app.batch([input1, input2, input3])  # 并发跑多输入

返回顺序与输入一致；节点有副作用时注意并发安全。

三、stream_mode 详解

stream(stream_mode=...) 控制事件粒度：

stream_mode	含义	事件结构
"values"	每次状态变更后输出完整 state	dict 完整 state
"updates"（默认）	输出每个节点的更新	{"node_name": update_dict}
"messages"	输出 LLM 流式 token	(AIMessageChunk, metadata)
"debug"	极详细调试信息	任务/节点/执行细节

# updates：看每个节点更新了什么 → {'node_a': {'field': '...'}}
for event in app.stream(s, stream_mode="updates"): ...
# values：看每次完整状态 → {'field': '...', ...}
for event in app.stream(s, stream_mode="values"): ...
# messages：LLM 流式 token（逐字输出）
for chunk, meta in app.stream(s, stream_mode="messages"):
    print(chunk.content, end="", flush=True)

可以多个模式叠加，事件会带 mode 标签。

for event in app.stream(initial_state, stream_mode=["updates", "values"]):
    print(event)
# 每个事件带 mode 标签

四、config 参数

invoke / stream 的第二参数是 config，传运行时配置：

config = {
    "configurable": {
        "thread_id": "user-123",     # 线程 ID，配合 checkpointer 实现记忆
        "user_id": "u123",            # 自定义业务参数
    },
    "recursion_limit": 50,            # 限制迭代次数防死循环
}
result = app.invoke(input, config=config)

thread_id 与记忆

配了 checkpointer 后，thread_id 决定状态保存在哪：

from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(checkpointer=MemorySaver())

# 第一次对话
config = {"configurable": {"thread_id": "thread-1"}}
app.invoke({"messages": [HumanMessage("我叫张三")]}, config=config)

# 第二次对话，同 thread_id
app.invoke({"messages": [HumanMessage("我叫什么")]}, config=config)
# AI 能记得"张三"，因为状态被持久化了

不同 thread_id 互不影响——相当于不同会话。

recursion_limit

防止图死循环。默认 25。超过会抛 RecursionError：

RecursionError: Recursion limit of 25 reached without hitting a stop condition

需要更多迭代时显式调高：

config = {"recursion_limit": 100}

但调高前要想清楚：是真的需要这么多步，还是图逻辑有 bug 让它不退出？

自定义参数

config 里的 configurable 字段可以放任何东西，节点通过第二参数读取：

def my_node(state, config):
    user_id = config["configurable"]["user_id"]
    return {...}

app.invoke(input, config={"configurable": {"user_id": "u123"}})

五、输入输出格式

输入：dict，State 的字段子集（可选字段可省略）。

app.invoke({
    "user_query": "什么是 LangGraph",
    # 其它字段用默认值或不设
})

输出：dict，最终完整 state（不是最后一步的更新）。

result = app.invoke(...)
# result 包含 State 里所有字段（被节点更新后的最终值）

新手常误解：以为 invoke 返回最后一步节点的返回值。错——它返回的是合并所有更新后的完整 state。

六、多次调用与状态保持

没配 checkpointer：每次 invoke 都从初始状态开始，互不影响。

app.invoke({"counter": 0})   # 返回 {counter: 2}
app.invoke({"counter": 0})   # 还是返回 {counter: 2}，不记得上次

配了 checkpointer + 同 thread_id：状态在多次 invoke 间保持。

app = graph.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "t1"}}

# 第一次
app.invoke({"messages": [HumanMessage("你好")]}, config=config)
# 第二次（注意只传新消息，旧消息在 checkpointer 里）
app.invoke({"messages": [HumanMessage("再说一次你好")]}, config=config)

注意：第二次的输入会被合并到已有状态上（按 reducer），不是覆盖整个 state。这是 checkpointer 的核心价值——让图"有记忆"。

七、超时与异常处理

节点异常

节点抛异常会终止图执行，异常向上抛给调用方。如果想"某节点失败不影响整体"，节点内部 try/except 自己处理：

try:
    app.invoke(input)
except SomeException as e:
    raise  # 或处理

recursion_limit 异常

超过递归限制抛 GraphRecursionError，多半是死循环：

from langgraph.errors import GraphRecursionError
try:
    app.invoke(input, config={"recursion_limit": 25})
except GraphRecursionError:
    print("超过递归限制，可能死循环")

超时

invoke 没有内建 timeout 参数。需要超时用 asyncio.wait_for（异步）：

import asyncio
async def run():
    try:
        return await asyncio.wait_for(app.ainvoke(input), timeout=30)
    except asyncio.TimeoutError:
        print("超时")
asyncio.run(run())

八、完整示例：invoke vs stream 对比

# run_demo.py
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


class State(TypedDict):
    counter: Annotated[int, add]
    log: Annotated[list, add]


def step_a(state: State) -> dict:
    return {"counter": 1, "log": ["A"]}

def step_b(state: State) -> dict:
    return {"counter": 1, "log": ["B"]}

def step_c(state: State) -> dict:
    return {"counter": 1, "log": ["C"]}


g = StateGraph(State)
g.add_node("a", step_a)
g.add_node("b", step_b)
g.add_node("c", step_c)
g.add_edge(START, "a")
g.add_edge("a", "b")
g.add_edge("b", "c")
g.add_edge("c", END)
app = g.compile(checkpointer=MemorySaver())

# 方式1：invoke
print("=== invoke ===")
result = app.invoke(
    {"counter": 0, "log": []},
    config={"configurable": {"thread_id": "t1"}},
)
print("最终状态：", result)

# 方式2：stream（updates 模式）
print("\n=== stream updates ===")
for event in app.stream(
    {"counter": 0, "log": []},
    config={"configurable": {"thread_id": "t2"}},
    stream_mode="updates",
):
    print("事件：", event)
# 方式3：把 stream_mode 改成 "values" 即可看到每次完整状态快照

输出：

=== invoke ===
最终状态： {'counter': 3, 'log': ['A', 'B', 'C']}

=== stream updates ===
事件： {'a': {'counter': 1, 'log': ['A']}}
事件： {'b': {'counter': 1, 'log': ['B']}}
事件： {'c': {'counter': 1, 'log': ['C']}}

对比：

• invoke：只给最终结果。
• updates：每次节点更新给一个事件（增量）。
• values：每次给完整状态（快照）。

九、断点续跑

配 checkpointer 后，图执行到一半被打断（如异常、人为中断）也能恢复：

config = {"configurable": {"thread_id": "long-task"}}

# 假设这次跑到一半挂了
try:
    app.invoke(long_input, config=config)
except Exception:
    pass

# 查看当前状态
state = app.get_state(config)
print("当前状态：", state.values)
print("下一步执行：", state.next)  # ('step_b',) 表示下次从 step_b 继续

# 恢复执行（None 表示继续原流程）
app.invoke(None, config=config)

这是 checkpointer 最强大的能力——长任务可恢复。配合 interrupt() 能做"暂停等用户审核，审核完继续"。

十、与 LangGraph4j 对比

维度	Python	Java
编译	graph.compile(checkpointer=...)	graph.compile(checkpointer, ...)
运行	app.invoke(input)	app.invoke(input)
流式	app.stream / astream	app.stream
批量	app.batch	app.batch (如有)
config	dict 字典	Map
thread_id	config["configurable"]["thread_id"]	同
recursion_limit	默认 25	默认 25
断点续跑	app.get_state + app.invoke(None)	同

API 高度一致。

十一、常见踩坑

1. recursion_limit 默认 25 不够用：复杂循环（如 ReAct 多轮工具调用）容易超。显式调高 config={"recursion_limit": 100}。
2. 忘了 thread_id 导致无记忆：配了 checkpointer 但 invoke 不传 thread_id——状态不持久化，"图失忆"。
3. 传 None 恢复但状态不对：app.invoke(None, config=config) 恢复时，必须用同一个 thread_id，否则拿不到旧状态。
4. invoke 返回的误解：以为返回最后一步节点输出，其实是完整 state。
5. stream 模式不对：默认 updates 给的是增量，想要完整状态用 values。
6. batch 顺序错乱：batch 返回顺序与输入顺序一致，但如果节点有副作用要注意并发安全。
7. checkpointer 状态累积变大：长对话 state 会越来越大（消息历史）。用 RemoveMessage 或定期清理。
8. 异步同步混用：ainvoke 和 invoke 不要在同一应用里混用；要真异步节点用 async def。
9. get_state 在没配 checkpointer 时调用：会报错或返回空。checkpointer 是状态持久化的前提。
10. 节点抛异常后状态不一致：异常发生时部分节点已更新 state，但 checkpointer 可能没保存最后一帧。用 get_state 看实际状态再决定是否恢复。

十二、小结

• compile() 生成 CompiledGraph，可注入 checkpointer / interrupt 配置。
• 运行方式：invoke（同步一次性）/ stream（流式）/ ainvoke / astream / batch。
• stream_mode：values（完整快照）/ updates（增量）/ messages（token 流）/ debug。
• config 传 thread_id（记忆）和 recursion_limit（防死循环）。
• invoke 返回最终完整 state，不是最后一步输出。
• 配 checkpointer + thread_id：状态在多次调用间保持、长任务可恢复。
• 调试技巧：用 stream_mode="updates" 看每步更新、用 get_state 看当前状态。

核心概念模块到这里全部讲完。接下来进入 基础教程 模块，开始把这些概念拼成常见模式。