【LangGraph 基础】状态、流式、中断、时间旅行与多 Agent

本文以 9 个示例代码来学习LangGraph基础。它们不是零散 API 演示，而是一条完整的进阶路线：

flowchart LR
    A["01 最小 Agent<br/>sync_async / base_agent"] --> B["02 手写 ReAct 图<br/>state_graph_agent"]
    B --> C["03 流式输出<br/>stream_modes"]
    C --> D["04 执行监控<br/>tasks_debug"]
    B --> E["05 人机协同<br/>interrupt_agent"]
    B --> F["06 时间旅行<br/>time_travel"]
    A --> G["07 思考模型<br/>thinking_agent"]
    B --> H["08 多 Agent<br/>supervisor_agent"]

先用高层封装跑通一个 Agent，再拆开手写 ReAct 循环，理解 State、Node、Edge 的本质；随后补齐流式输出、人工审批、历史回溯、思考模型和多 Agent 编排这些工程能力。

---

一、最小可用 Agent：模型 + 工具 + 循环

1.1 高层封装

langchain.agents.create_agent 是最短路径：给模型、工具和系统提示词，它自动生成「模型调用 → 工具执行 → 再调模型」的循环。

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model(model="deepseek-chat", model_provider="deepseek")

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="You are a helpful assistant",
)

result = agent.invoke(
    {"messages": [("user", "What's the weather in London?")]}
)
print(result["messages"][-1].content)

工具就是一个普通函数，docstring 会被转成工具描述，参数签名会被转成 JSON Schema：

def get_weather(city: str) -> str:
    """Get the weather for a given city."""
    return f"The weather in {city} is sunny."

langgraph.prebuilt.create_react_agent 是另一种更贴近底层的封装，能力基本一致，还支持 response_format、pre_model_hook、post_model_hook 等参数。在新版 LangGraph 中它已逐步让位给 create_agent，新项目优先用后者。

1.2 四种调用方式

API	是否阻塞	返回形态	适用场景
invoke	是	最终完整 state	脚本、后台任务
stream	是	逐步产出 chunk	CLI、同步服务
ainvoke	否	最终完整 state	Web 服务、并发
astream	否	异步逐步产出	SSE / WebSocket 推送

关键细节：异步调用要放在同一个事件循环里。如果先 asyncio.run(agent.ainvoke(...))，再 asyncio.run(agent.astream(...))，底层 HTTP client 可能跨 loop 复用而报错。正确做法是把所有异步逻辑合并进一个 async def run_async()：

async def run_async():
    result = await agent.ainvoke(question)
    async for chunk in agent.astream(question, stream_mode="values"):
        ...

asyncio.run(run_async())

1.3 会话记忆：checkpointer 与 thread_id

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(..., checkpointer=checkpointer)

config = {"configurable": {"thread_id": "123"}}
agent.invoke({"messages": [("user", "你好")]}, config)
agent.invoke({"messages": [("user", "我叫什么？")]}, config)  # 记得上一轮

• checkpointer 负责持久化每一步的状态快照。
• thread_id 是会话身份，同一 thread_id 共享历史。
• InMemorySaver 只适合学习和测试，进程退出即丢失；生产环境应换 SqliteSaver、PostgresSaver 等持久化实现。
• 后续的 interrupt、get_state_history、debug 流模式都依赖 checkpointer。

1.4 结构化输出与钩子

from pydantic import BaseModel

class WeatherResponse(BaseModel):
    conditions: str

agent = create_react_agent(
    model=model,
    tools=[get_weather],
    prompt="You are a helpful assistant",
    checkpointer=checkpointer,
    response_format=WeatherResponse,
    pre_model_hook=lambda state: print(state["messages"][-1]),
    post_model_hook=lambda state: print(state["messages"][-1]),
)

result = agent.invoke(..., config)
print(result["structured_response"])   # WeatherResponse 实例

• response_format 让模型最终产出一个符合 Pydantic schema 的结构化结果，放在 structured_response。
• pre_model_hook / post_model_hook 分别在每次模型调用前后执行，适合做日志、裁剪上下文、注入状态。

---

二、拆开黑盒：手写 ReAct 图

高层封装帮你隐藏了循环。state_graph_agent.py 把这个循环完全摊开，这是理解 LangGraph 的核心。

2.1 四个概念

概念	作用
State	图里所有节点共享、流转的数据
Node	一个普通函数：接收 state，返回「部分更新」
Edge	节点之间的固定连线
Conditional Edge	根据 state 动态决定下一步，是循环的关键

ReAct 循环长这样：

flowchart LR
    START([START]) --> agent["agent 节点<br/>调用模型"]
    agent -->|"AIMessage 带 tool_calls"| tools["tools 节点<br/>执行工具"]
    agent -->|"无 tool_calls"| END([END])
    tools -->|"ToolMessage"| agent

2.2 State 与 reducer

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

add_messages 是 reducer。没有它，节点返回 {"messages": [新消息]} 会覆盖整个列表；有了它，新消息会追加到列表，并按消息 ID 去重、更新。

这就是为什么多个节点都可以放心返回 {"messages": [...]}——状态会自动累积。

2.3 节点：返回部分更新

tools_by_name = {"get_weather": get_weather}
model_with_tools = model.bind_tools([get_weather])

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

def call_tools(state: State) -> dict:
    last = state["messages"][-1]
    tool_messages = []
    for call in last.tool_calls:          # 一次可能要求多个工具
        result = tools_by_name[call["name"]](**call["args"])
        tool_messages.append(
            ToolMessage(content=result, tool_call_id=call["id"])
        )
    return {"messages": tool_messages}

节点只返回自己改动的字段，不需要手动合并 state。

2.4 条件边与组装

def should_continue(state: State):
    if state["messages"][-1].tool_calls:
        return "tools"
    return "end"

from langgraph.graph import StateGraph, START, END

graph = StateGraph(State)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)
graph.add_edge(START, "agent")
graph.add_conditional_edges(
    "agent", should_continue, {"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")       # 工具执行完回到模型，形成循环

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

add_conditional_edges 的第三个参数是「路由函数返回值 → 目标节点」的映射。tools → agent 这条回边让图真正循环起来。

编译后可以直接导出图结构，方便文档化：

print(app.get_graph().draw_mermaid())

---

三、六种流式模式：看清图在干什么

同一个 Agent，用不同 stream_mode 能看到不同粒度的执行过程。

模式	产出	典型用途
values	每一步后的完整 state	UI 状态同步、整图回放
updates	{节点名: 增量更新}	节点级进度、轻量日志
messages	(消息 chunk, metadata)	打字机效果、逐 token 渲染
tasks	任务生命周期事件	进度条、耗时统计、错误/中断告警
debug	{step, timestamp, type, payload}	深度排查「状态在哪一步出错」
custom	节点内 writer() 主动推送	业务进度、中间产物

3.1 values / updates / messages

# 完整状态快照
for chunk in agent.stream(q, stream_mode="values"):
    print(len(chunk["messages"]))

# 只看每个节点改了什么
for chunk in agent.stream(q, stream_mode="updates"):
    for node, update in chunk.items():
        print(node, update)

# 逐 token
for msg, meta in agent.stream(q, stream_mode="messages"):
    print(msg.content, meta)

updates 的输出形如 {"agent": {"messages": [...]}}，只包含当前节点返回的那部分更新。

3.2 tasks：执行监控器

tasks 会为每个任务发出开始和结束事件，用 id 配对：

• 开始事件：带 input
• 结束事件：带 result，同时还有 error、interrupts

started = {}
for chunk in agent.stream(q, stream_mode="tasks"):
    if "input" in chunk:
        started[chunk["id"]] = time.time()
        if chunk["name"] == "tools":
            names = [c["name"] for c in chunk["input"]]
            print(f"▶ 调用工具: {', '.join(names)}")
    elif "result" in chunk:
        cost = time.time() - started.pop(chunk["id"], 0.0)
        if chunk["error"]:
            print(f"✗ [{chunk['name']}] 报错: {chunk['error']}")
        elif chunk["interrupts"]:
            print(f"⏸ [{chunk['name']}] 等待人工")
        else:
            print(f"✓ [{chunk['name']}] 完成，耗时 {cost:.2f}s")

这套事件非常适合前端进度条、性能分析、错误告警和中断提示。

3.3 debug：完整时间线

agent = create_agent(
    ..., checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": "1"}}

for chunk in agent.stream(q, config=config, stream_mode="debug"):
    print(chunk["step"], chunk["type"], chunk["payload"])

debug 会把任务事件和 checkpoint 事件统一包装成 {step, timestamp, type, payload}。checkpoint 事件依赖 checkpointer，否则只能看到任务事件。

3.4 custom：节点主动推送

from langgraph.types import StreamWriter

def add_one(state: State, writer: StreamWriter) -> dict:
    writer({"progress": f"收到 {state['value']}，准备 +1"})
    result = state["value"] + 1
    writer({"result": result})
    return {"value": result}

节点签名多一个 writer 参数，就可以在任意时刻向客户端推送业务数据，和图的内部状态解耦。

3.5 组合多种模式

需要同时拿节点更新和 token 流时，可以传列表：

for mode, chunk in agent.stream(q, stream_mode=["updates", "messages"]):
    if mode == "messages":
        msg, meta = chunk
    else:
        node, update = next(iter(chunk.items()))

---

四、人机协同：interrupt 中断与恢复

高风险工具调用前，往往需要人工审批。LangGraph 用 interrupt() 把图暂停，把审批请求抛给客户端。

sequenceDiagram
    participant U as 客户端
    participant G as 图 + checkpointer
    U->>G: invoke(input, thread_id=1)
    G->>G: agent 节点产出 AIMessage(tool_calls)
    G->>G: tools 节点执行到 interrupt()
    G-->>U: 返回结果包含 __interrupt__
    Note over U: 人工审批
    U->>G: invoke(Command(resume="yes"), thread_id=1)
    G->>G: 从 checkpoint 恢复，重新进入 tools 节点
    G->>G: 执行工具 → ToolMessage
    G->>G: 回到 agent → 最终回答
    G-->>U: 最终 state

核心代码：

from langgraph.types import interrupt, Command

def call_tools(state: State) -> dict:
    last = state["messages"][-1]

    # 暂停：把审批请求抛给客户端，等待 resume 值
    approval = interrupt(f"是否允许调用工具？{last.tool_calls}")

    if approval != "yes":
        return {
            "messages": [
                ToolMessage(
                    content="用户拒绝了这次工具调用",
                    tool_call_id=last.tool_calls[0]["id"],
                )
            ]
        }

    return {
        "messages": [
            ToolMessage(
                content=tools_by_name[c["name"]](**c["args"]),
                tool_call_id=c["id"],
            )
            for c in last.tool_calls
        ]
    }

运行方式：

result = app.invoke(input, config)
print(result["__interrupt__"])          # 中断信息

result = app.invoke(Command(resume="yes"), config)   # 恢复

必须记住的几点：

1. interrupt 依赖 checkpointer，否则暂停后状态无法恢复。
2. Command(resume=value) 里的 value 会成为 interrupt() 的返回值。
3. 恢复时节点会从头重新执行，interrupt() 之前如果有副作用（写库、发请求），必须保证幂等。
4. 拒绝时返回一条 ToolMessage，让模型知道工具没执行，从而正常收尾，而不是让图直接崩掉。

---

五、时间旅行：历史快照与分支

每次超级步（super-step）执行后，checkpointer 都会留下一个 checkpoint。于是可以回看历史，甚至从任意历史点「改道」重跑。

5.1 读取历史

history = list(app.get_state_history(config))   # 从最新到最旧

for snap in history:
    print(
        snap.metadata["step"],
        snap.metadata["source"],
        snap.values.get("value"),
        snap.config["configurable"]["checkpoint_id"],
    )

get_state_history 只读快照，不会重新执行任何节点。

5.2 从历史点 fork

以确定性图 value -> add_one -> multiply_two -> END 为例，原始轨迹是 1 → 2 → 4：

flowchart LR
    s0["checkpoint 0<br/>value=1"] --> s1["checkpoint 1<br/>add_one 后 value=2"]
    s1 --> s2["checkpoint 2<br/>multiply_two 后 value=4"]
    s1 -.->|"update_state<br/>as_node=add_one<br/>value=10"| f1["fork checkpoint<br/>value=10"]
    f1 -->|"invoke(None)"| f2["multiply_two 后<br/>value=20"]

target = history[1]          # add_one 之后的快照（value=2）

new_config = app.update_state(
    target.config,
    {"value": 10},           # 想改成的值
    as_node="add_one",       # 当作 add_one 节点产生的更新
)

result = app.invoke(None, new_config)   # 从 fork 点继续
print(result["value"])                  # 20

要点：

• update_state 不会删除原分支，而是创建一个新的 checkpoint 分支。
• as_node 决定「这个更新来自哪个节点」，从而决定接下来走哪些出边。这里声明为 add_one，所以会继续执行 multiply_two。
• invoke(None, new_config) 表示「从给定 checkpoint 继续，不注入新输入」。
• 原分支 1 → 2 → 4 仍然保留在历史中，可随时回看。

这套机制适合做：调试回放、A/B 分支实验、人工修正中间状态后重跑。

---

六、思考模型：reasoning_content 的坑

DeepSeek 的思考模型会把推理过程放在 reasoning_content。ChatDeepSeek 能把它解析进 AIMessage.additional_kwargs，但序列化回请求时不会带上它。当「思考模式 + 工具调用」进行多轮对话时，API 会因缺少 reasoning_content 返回 400。

两个解决方向：

方案 A：在工具循环里关闭 thinking

model = init_chat_model(
    model="deepseek-flash",
    model_provider="deepseek",
    extra_body={"thinking": {"type": "disabled"}},
)

简单直接。代价是失去推理能力，适合对工具调用稳定性要求更高的场景。

方案 B：自定义模型类，把 reasoning_content 补回去

class ChatDeepSeekThinking(ChatDeepSeek):
    def _get_request_payload(self, input_, *, stop=None, **kwargs):
        # super 之前记录每个 AIMessage 的 reasoning_content（含空串）
        reasoning_by_index = {}
        for i, m in enumerate(self._convert_input(input_).to_messages()):
            if isinstance(m, AIMessage):
                rc = m.additional_kwargs.get("reasoning_content")
                if rc is not None:
                    reasoning_by_index[i] = rc

        payload = super()._get_request_payload(input_, stop=stop, **kwargs)

        # 序列化后按相同索引补回顶层字段
        for i, rc in reasoning_by_index.items():
            payload["messages"][i]["reasoning_content"] = rc
        return payload

使用：

model = ChatDeepSeekThinking(model="deepseek-flash")   # 默认开启 thinking

agent = create_react_agent(
    model=model,
    tools=[get_weather],
    prompt="You are a helpful assistant",
    checkpointer=checkpointer,
)

for m in result["messages"]:
    if isinstance(m, AIMessage):
        reasoning = m.additional_kwargs.get("reasoning_content")
        if reasoning:
            print("[思考]", reasoning)
        if m.content:
            print("[回答]", m.content)

这个例子说明：框架的模型适配层不一定覆盖所有厂商特性，必要时可以在子类里改请求序列化逻辑。

---

七、多 Agent：supervisor + 子图

复杂任务可以拆成多个专精 Agent，再由一个 supervisor 路由。LangGraph 的做法很直接：编译好的子图本身就是一个 Runnable，可以直接作为父图的节点。

flowchart TB
    subgraph weather_agent [weather_agent 子图]
        wa[agent] <--> wat["tools<br/>get_weather"]
    end

    subgraph math_agent [math_agent 子图]
        ma[agent] <--> mat["tools<br/>multiply"]
    end

    START([START]) --> route{"route()<br/>LLM 判断路由"}
    route -->|weather| weather_agent
    route -->|math| math_agent
    weather_agent --> END([END])
    math_agent --> END

7.1 复用 ReAct 子图构建函数

def build_agent(tools: list, tools_by_name: dict):
    model_with_tools = model.bind_tools(tools)

    def call_model(state: State) -> dict:
        return {"messages": [model_with_tools.invoke(state["messages"])]}

    def call_tools(state: State) -> dict:
        last = state["messages"][-1]
        return {
            "messages": [
                ToolMessage(
                    content=tools_by_name[c["name"]](**c["args"]),
                    tool_call_id=c["id"],
                )
                for c in last.tool_calls
            ]
        }

    def should_continue(state: State):
        return "tools" if state["messages"][-1].tool_calls else "end"

    graph = StateGraph(State)
    graph.add_node("agent", call_model)
    graph.add_node("tools", call_tools)
    graph.add_edge(START, "agent")
    graph.add_conditional_edges(
        "agent", should_continue, {"tools": "tools", "end": END}
    )
    graph.add_edge("tools", "agent")
    return graph.compile()

7.2 父图只负责路由

weather_agent = build_agent([get_weather], {"get_weather": get_weather})
math_agent = build_agent([multiply], {"multiply": multiply})

def route(state: State):
    response = model.invoke([
        SystemMessage(
            "你是路由器。判断用户问题该交给 weather 还是 math 子代理，"
            "只回答 weather 或 math。"
        ),
        *state["messages"],
    ])
    answer = response.content.strip().lower()
    return "weather" if "weather" in answer else "math"

supervisor = StateGraph(State)
supervisor.add_node("weather_agent", weather_agent)
supervisor.add_node("math_agent", math_agent)
supervisor.add_conditional_edges(
    START, route, {"weather": "weather_agent", "math": "math_agent"}
)
supervisor.add_edge("weather_agent", END)
supervisor.add_edge("math_agent", END)

app = supervisor.compile(checkpointer=InMemorySaver())

关键点：

• 父子图共用同一个 State schema，消息才能自然流转。如果 schema 不同，需要在子图边界做输入/输出转换。
• 编译后的子图可以直接 add_node，父图不关心子图内部有几个节点。
• 这里的 supervisor 是简化版：只路由一次就结束。真实场景可以让子 Agent 执行完回到 supervisor，由 supervisor 决定「继续派活还是结束」，形成多轮协作。

---

八、工程实践与踩坑清单

8.1 API Key 不要写进代码

示例里为了省事直接 os.environ["DEEPSEEK_API_KEY"] = "sk-..."。实际项目应该：

export DEEPSEEK_API_KEY="sk-..."

或使用 .env + python-dotenv，并把 .env 加入 .gitignore。密钥一旦提交到 Git 历史，就应该立即轮换。

8.2 checkpointer 是很多能力的前提

能力	是否依赖 checkpointer
多轮会话记忆	是
interrupt 中断/恢复	是
get_state_history / 时间旅行	是
stream_mode="debug" 的 checkpoint 事件	是
普通单次 invoke	否

8.3 thread_id 必须显式提供

config = {"configurable": {"thread_id": "1"}}

没有 thread_id，checkpointer 无法定位会话，会直接报错或行为不符合预期。

8.4 InMemorySaver 只适合开发

进程重启就丢。生产环境按部署形态选择：

• 单机持久化：SqliteSaver
• 多实例/服务化：PostgresSaver、RedisSaver
• 云托管：LangGraph Platform 的持久化后端

8.5 interrupt 节点会重跑

恢复时从节点开头重新执行，interrupt() 返回 resume 值。所以：

• interrupt() 之前的数据库写入、外部 API 调用要做幂等。
• 不要在 interrupt() 之前做不可逆操作。

8.6 异步 client 不要跨事件循环复用

把 ainvoke / astream 放进同一个 asyncio.run()，避免 HTTP client 绑定到已关闭的 loop。

8.7 stream_mode 可以组合

stream_mode=["updates", "messages"]

输出会变成 (mode, chunk) 二元组，注意解包方式。

8.8 工具函数要有清晰的 docstring

模型靠 docstring 和参数签名决定是否调用、怎么传参。描述含糊会直接导致工具选择错误。

8.9 高层封装 vs 手写图

方式	优点	缺点
create_agent / create_react_agent	代码少、开箱即用、内置能力多	流程固定，深改需要看源码
StateGraph 手写	完全可控、易插入自定义节点	样板代码多，需要自己维护循环

建议：先用高层封装验证需求；当流程需要审批、分支、多 Agent、自定义状态时，再下沉到手写图。

---

九、速查表

# 模型
init_chat_model(model="deepseek-chat", model_provider="deepseek")

# 高层 Agent
from langchain.agents import create_agent
from langgraph.prebuilt import create_react_agent

# 状态图
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 持久化
from langgraph.checkpoint.memory import InMemorySaver

# 人机协同
from langgraph.types import interrupt, Command, StreamWriter

# 历史
app.get_state_history(config)
app.update_state(target_config, values, as_node="node_name")

# 流模式
agent.stream(q, stream_mode="values")
agent.stream(q, stream_mode="updates")
agent.stream(q, stream_mode="messages")
agent.stream(q, stream_mode="tasks")
agent.stream(q, stream_mode="debug")
agent.stream(q, stream_mode="custom")

---

总结

这组示例的主线可以概括为一句话：

LangGraph 用「共享 State + 节点 + 边」描述 Agent 的执行流，用 checkpointer 把每一步变成可持久化、可回放、可中断、可分叉的 checkpoint。

• 最小 Agent 让你快速跑通「模型调用工具」的闭环。
• 手写 ReAct 图让你理解循环、reducer 和条件边的本质。
• 流式模式让你能看到图的每一步，而不是只拿到最终答案。
• interrupt 把人类审批变成图里的一个可暂停节点。
• 时间旅行让历史状态成为可编辑、可分支的一等公民。
• supervisor + 子图让多 Agent 编排退化成「图里套图」。
• 思考模型的适配问题提醒我们：框架抽象之外，仍要理解厂商协议和序列化细节。

把这几个点串起来，基本就具备了用 LangGraph 构建可控、可观测、可恢复的 Agent 应用的能力。