lang

1. 本章目标 #

第 19 章下过一个判断：该模型判断的交给模型，该保证执行的交给图。第 20～24 章把图的三大能力（顺序、分支、循环）都学完了，但一直在自己手搓节点，前面十几章积累的 create_agent 好像被丢在了一边。

这一章把两条线接起来：

让确定性的 Python 节点和 create_agent 在同一张图里各干各的。

典型形状就是本章实战：

预处理（确定性）→ Agent（概率性）→ 后处理（确定性）

前面校验、鉴权、脱敏，中间让模型自由发挥，后面审计、复检、落库。中间那段允许模型犯错，前后两段必须每次都执行。

学完你应能：

把 create_agent 直接当节点塞进 StateGraph，并说清父子状态是怎么对接的
在三种对接方式里选对：直接嵌入、包一层转换、隔离消息
知道父图的业务字段（工号、租户、订单号）该走哪条路进子 Agent
避开本章头号坑：键名不对时 Agent 会照跑照烧钱，但不报错、结果被静默丢弃
用 xray=1 和 subgraphs=True 看清嵌套图内部
知道「图边保证执行」的边界在哪（它不是 finally）
产出：一条工单处理流水线：鉴权、脱敏、Agent 应答、审计落库

参考文档：

1.1 为什么「混搭」值得单独讲一章 #

如果只是「把一个对象当节点塞进图」，一行 add_node 就说完了，不值得一章。真正需要讲的是父子两张图之间那道状态边界，它是本章所有内容的源头：

你想做的事	卡在哪	本章的答案
让 Agent 成为流水线的一环	父图状态和子图状态长得不一样	§3.1 先读子图的 schema，再决定怎么接
父图的工号、租户要给工具用	子图默认只认 `messages`，业务字段进不去	§3.4 走 `context=` 或 `state_schema=`
不想让工具调用记录污染对话历史	直接嵌入会把子图全部消息倒进父图	§4.3 包一层，只回传结论
键名写错了，程序却不报错	状态合并「只认声明过的键」，不认识的静默丢弃	§5 三行断言把它变成启动即报错
排查「Agent 在里面到底干了什么」	默认观测手段只能看到「agent 节点跑了」	§6 `xray=1` + `subgraphs=True`
保证「无论如何都要归档」	图边不是 `try/finally`	§7 在节点里把异常转成正常状态更新

这道边界的规则可以概括成一句话，本章会反复碰到：

父图往子图传参时，只传子图声明过的键；子图写回来时，也只合并父图声明过的键。两边都不认识的，一律悄悄丢掉，不报错、不警告。

这条规则本身很简单，麻烦的是它失败时不出声。§5 的头号坑就是这条规则的直接后果。

1.2 本章统一环境 #

# 读取 .env 里的 DEEPSEEK_API_KEY，避免把密钥写进代码
from dotenv import load_dotenv
# tool 装饰器：把普通 Python 函数变成模型可调用的工具
from langchain_core.tools import tool

# override=True 让 .env 的值覆盖系统环境变量里可能存在的同名旧值
load_dotenv(override=True)

# @tool 会读取函数名、类型标注和 docstring，自动生成模型看到的工具描述
@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    # 打一行日志，用来在输出里确认「模型到底有没有真的调工具」
    print(f"   [tool] get_stock({sku})")
    # 用字典模拟库存表；查不到就返回兜底文案，而不是抛异常
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

模型输出天生有波动，本章正文里的自然语言回复每次跑都会略有不同。判断结论对不对，要看的是消息条数、审计条目、键名这类结构性输出，它们是稳定的。

2. 为什么不干脆全交给 Agent #

一个自然的疑问：鉴权、脱敏、审计这些事，为什么不写成工具让 Agent 自己调？

因为第 19 章那个实验的结论在这里同样成立：提示词是请求，图边是保证。 具体到这条流水线，把三件事全塞给 Agent 会有四个问题。

一、该执行的可能不执行。 「回复用户前必须先审计落库」写在提示词里，模型大概率会照做，但没有任何机制保证。而 agent → archive 这条边，只要 agent 跑完就一定会走。

二、不该执行的可能被绕过。 权限校验如果是个工具，模型理论上可以不调它就直接回答。而放在 Agent 之前的确定性节点，是模型够不着的：它既不在工具列表里，也不在上下文里，模型连「有这么一道检查」都不知道。

三、白花钱。 越权请求本来一眼就能拒，交给 Agent 就得先起一轮模型调用。前置校验拦下来的请求，模型费用是零，§8 的实战会把这一点量化到审计日志里（0 次模型调用）。

四、顺序不可控。 「先脱敏再查询」这种强顺序要求，在 Agent 的 model ↔ tools 环里只能靠提示词约束；在图里就是一条边。更关键的是：脱敏一旦作为工具由模型来调，模型在调它之前就已经看到原文了，脱敏本身就失去了意义。

反过来也不要走极端：回复内容怎么组织、要不要追问、调哪个工具查什么，这些交给图就是灾难，写死的分支永远覆盖不完用户的说法。这正是要混搭的原因：两种东西各有各的地盘。

这类事情	交给谁	为什么
参数校验、权限、限流	确定性节点	规则明确，且必须百分百执行
脱敏、格式化、字段映射	确定性节点	有确定答案，让模型做纯属浪费
审计、落库、埋点	确定性节点	漏一条就是事故
理解用户想干什么	Agent	说法千变万化，规则写不完
决定查什么、查几次	Agent	需要看到中间结果才能决定
组织自然语言回复	Agent	模型的主场

判断某件事该归哪边，有个简单的自问：「这一步做错了，是我能提前写出规则拦住的，还是只能靠模型现场理解？」 前者归图，后者归 Agent。

3. 把 create_agent 当节点用 #

3.1 它本来就是一张图 #

第 19 章验证过，create_agent 返回的是 CompiledStateGraph。既然是图，就能当成节点塞进另一张图，LangGraph 里这叫子图（subgraph）。

不过在写 add_node 之前，得先看清它对状态的要求。任何 Runnable 都能报出自己的输入输出结构，CompiledStateGraph 也一样：

from dotenv import load_dotenv
from langchain_core.tools import tool

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

# create_agent：第 9 章的主角，一行创建一个 model ↔ tools 环
from langchain.agents import create_agent

# 建一个最小的库存助手 Agent
agent = create_agent(
    # 模型标识，格式是「提供方:模型名」
    model="deepseek:deepseek-v4-flash",
    # 只挂一个查询工具
    tools=[get_stock],
    # 系统提示：约束回答风格
    system_prompt="你是库存助手，回答尽量简短。",
)

# 运行时类型：一张编译好的图，而不是什么特殊的 Agent 对象
print("类型:", type(agent).__name__)
# 沿继承链往上看前四层，确认它同时是 Pregel 和 Runnable
print("继承链:", [c.__name__ for c in type(agent).__mro__][:4])
# 输入 JSON Schema 的顶层属性 = 子图认识哪些输入键
print("input schema:", list(agent.get_input_jsonschema()["properties"].keys()))
# required 列出哪些键是必须提供的
print("input required:", agent.get_input_jsonschema().get("required"))
# 输出 JSON Schema 的顶层属性 = 子图会往外写哪些键
print("output schema:", list(agent.get_output_jsonschema()["properties"].keys()))

类型: CompiledStateGraph
继承链: ['CompiledStateGraph', 'Pregel', 'PregelProtocol', 'Runnable']
input schema: ['messages']
input required: ['messages']
output schema: ['messages', 'structured_response']

中间那两行 schema 是本章最该记住的东西。 子图默认只认 messages 这一个输入键，输出 messages 和 structured_response。父图的状态和它对不对得上，决定了后面一切。

顺便看一眼 Agent 内部用的状态类型，理解「它认识什么」会更具体：

# AgentState 是 create_agent 内部默认使用的状态类型
from langchain.agents import AgentState

# TypedDict 会把基类的键合并进 __annotations__，所以这里看到的就是全部字段
print("AgentState 自带的键:", list(AgentState.__annotations__.keys()))

AgentState 自带的键: ['messages', 'jump_to', 'structured_response']

三个键各有分工：

messages 是唯一的输入通道，挂了 add_messages reducer（第 21 章），所以是追加语义
jump_to 是中间件跳转用的内部字段（第 10 章的 jump_to="end" 就写这里），不出现在 input schema 里
structured_response 只在配了 response_format 时才有值，见 §3.3

注意 input schema 里只有 messages，这是默认情况。如果你希望子 Agent 也认识业务字段（比如工号），可以用 state_schema= 扩展，§3.4 会演示。所以准确的说法是：

子图认识哪些键，由它自己的 state_schema 决定；不在这个范围内的父图字段，一个都传不进去。

3.2 最简单的情况：父图状态里有 messages #

只要父图状态包含 messages，直接 add_node 就能跑。下面这段是本章后面反复用到的父图骨架：

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

# operator.add 用作 audit 字段的 reducer，实现追加而不是覆盖
import operator
# Annotated 用来给字段挂 reducer
from typing import Annotated

# 图的四件套：起点、终点、自带 messages 的状态基类、图构建器
from langgraph.graph import END, START, MessagesState, StateGraph

# 继承 MessagesState 就自带一个挂好 add_messages 的 messages 字段
class TicketState(MessagesState):
    # 父图自己的业务字段：子图完全不认识它
    user_id: str
    # 审计轨迹；挂 operator.add 让每个节点各写一段而不互相覆盖
    audit: Annotated[list[str], operator.add]

# 确定性前处理节点
def pre(state: TicketState) -> dict:
    # 打印用来确认执行顺序
    print("   [node] pre")
    # 只往 audit 追加一条，不碰 messages
    return {"audit": ["预处理完成"]}

# 确定性后处理节点
def post(state: TicketState) -> dict:
    # 同样打印一行
    print("   [node] post")
    # 同样只写 audit
    return {"audit": ["后处理完成"]}

# 用 TicketState 作为状态模式创建构建器
builder = StateGraph(TicketState)
# 注册前处理节点
builder.add_node("pre", pre)
# 关键的一行：直接把编译好的 agent 当节点注册进来
builder.add_node("agent", agent)
# 注册后处理节点
builder.add_node("post", post)
# 入口进 pre
builder.add_edge(START, "pre")
# pre 之后进 agent
builder.add_edge("pre", "agent")
# agent 之后进 post
builder.add_edge("agent", "post")
# post 之后结束
builder.add_edge("post", END)
# 编译成可执行图
graph = builder.compile()

# 三个键都给初值：messages 是子图要的，另两个是父图自己的
inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}
# dict(inputs) 拷一份，避免后面复用这份输入时被上一次调用改动
out = graph.invoke(dict(inputs))
# 子图不认识 user_id，看它有没有被弄丢
print("user_id 还在吗:", out["user_id"])
# 两个确定性节点各写了一条
print("audit:", out["audit"])
# 子图内部产生的消息有没有进父图
print("消息条数:", len(out["messages"]))
# 最终状态里有哪些键
print("返回的键:", sorted(out.keys()))
# 逐条看消息，确认哪些是子图带出来的
for m in out["messages"]:
    # tool_calls 只有 AIMessage 才有，用 getattr 兜住其他消息类型
    names = [c["name"] for c in getattr(m, "tool_calls", [])]
    # 打印消息类型、内容前 40 字、以及这条消息请求了哪些工具
    print(f"   {type(m).__name__}: content={str(m.content)[:40]!r} tool_calls={names}")

   [node] pre
   [tool] get_stock(A-100)
   [node] post
user_id 还在吗: u-1
audit: ['预处理完成', '后处理完成']
消息条数: 4
返回的键: ['audit', 'messages', 'user_id']
   HumanMessage: content='A-100 有货吗' tool_calls=[]
   AIMessage: content='' tool_calls=['get_stock']
   ToolMessage: content='库存 12 件' tool_calls=[]
   AIMessage: content='A-100 有货，库存 12 件。' tool_calls=[]

三点值得留意：

父图多出来的字段不会被子图吃掉。 user_id 和 audit 子图完全不认识，但它们原封不动地留着。父图只把子图声明过的键传进去，也只合并它写出来的键，其余部分根本不参与这次交互。
子图内部的消息全都进了父图历史。 4 条 = 用户提问 + 模型请求工具（content 为空，只有 tool_calls）+ 工具结果 + 最终回复。这未必是你想要的，§4.3 会处理。
那条 content='' 的 AIMessage 是最容易踩的雷。 前端如果直接渲染 messages，用户会看到一条空气泡；下游如果用 messages[-1] 之外的下标取内容，很容易取到空字符串。

3.3 structured_response 要父图声明才收得到 #

如果 Agent 配了 response_format，它会输出第二个键 structured_response。父图必须显式声明这个字段才拿得到。

DeepSeek 用 response_format 有两个坑，第 6 章和第 23 章都遇到过，这里一次性解决：必须用 ToolStrategy 包一层（直接传 Pydantic 类会报 This response_format type is unavailable now），而且要用 extra_body={"thinking": {"type": "disabled"}} 关掉思考模式（否则报 Thinking mode does not support this tool_choice）。所以模型要用 init_chat_model 手工构造，不能只写模型名字符串：

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from pydantic import BaseModel, Field

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

# ToolStrategy：用「强制调一个工具」的方式实现结构化输出
from langchain.agents.structured_output import ToolStrategy
# init_chat_model：手工构造模型对象，才能传 extra_body 这类厂商私有参数
from langchain.chat_models import init_chat_model
# Pydantic 用来声明结构化输出的形状
from pydantic import BaseModel, Field

# 这个类既是「模型要填的表」，也是父图收到的对象类型
class Stock(BaseModel):
    """结构化的库存结论。"""

    # description 会进 schema，直接影响模型填得准不准
    sku: str = Field(description="商品编号")
    # 布尔字段，方便下游代码直接分支
    available: bool = Field(description="是否有货")
    # 留一个自由文本字段，装模型的补充说明
    note: str = Field(description="一句话说明")

# DeepSeek 必须显式关掉思考模式，否则结构化输出会被拒
model_no_think = init_chat_model(
    # 模型标识
    "deepseek:deepseek-v4-flash",
    # 结构化任务用 0 温度，减少字段填写的随机性
    temperature=0,
    # extra_body 里的内容会原样透传给 DeepSeek 的 HTTP 接口
    extra_body={"thinking": {"type": "disabled"}},
)
# 带结构化输出的 Agent
agent_sr = create_agent(
    # 传模型对象而不是字符串，才能带上 extra_body
    model=model_no_think,
    # 工具照旧
    tools=[get_stock],
    # 提示模型先查证再给结论
    system_prompt="你是库存助手，查证后给出结构化结论。",
    # ToolStrategy 包一层是 DeepSeek 的必需写法
    response_format=ToolStrategy(Stock),
)

# 父图状态：除了 messages，还要显式声明 structured_response
class ParentWithSR(MessagesState):
    # 少了这一行，算出来的结构化结果就会被静默丢弃
    structured_response: Stock | None

# 建一张只有 agent 一个节点的父图
builder = StateGraph(ParentWithSR)
# 直接嵌入带结构化输出的 Agent
builder.add_node("agent", agent_sr)
# 入口直接进 agent
builder.add_edge(START, "agent")
# agent 跑完就结束
builder.add_edge("agent", END)
# 输入里 structured_response 先给 None 占位
sr_inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "structured_response": None}
# 编译成可执行图
graph = builder.compile()
# 执行
result = graph.invoke(sr_inputs)
# 看父图最终有哪些键
print("父图 keys:", sorted(result.keys()))
# 结构化结果本身
print("structured_response:", result["structured_response"])
# 确认它是 Pydantic 对象而不是 dict
print("它的类型:", type(result["structured_response"]).__name__)
# 结构化模式下消息条数会比普通模式多
print("消息条数:", len(result["messages"]))
# 看清多出来的是什么
print("消息类型:", [type(m).__name__ for m in result["messages"]])

   [tool] get_stock(A-100)
父图 keys: ['messages', 'structured_response']
structured_response: sku='A-100' available=True note='当前库存 12 件，有货。'
它的类型: Stock
消息条数: 5
消息类型: ['HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage', 'ToolMessage']

5 条而不是 4 条，因为 ToolStrategy 是靠「再强制调一次工具」来收结构化结果的：最后那对 AIMessage + ToolMessage 就是这次额外的结构化工具调用。这也解释了为什么 ToolStrategy 比原生 json_schema 多花一轮 token，它换来的是对不支持 json_schema 的模型也能用。

父图不声明 structured_response 会怎样：

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from pydantic import BaseModel, Field

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

class Stock(BaseModel):
    """结构化的库存结论。"""

    sku: str = Field(description="商品编号")
    available: bool = Field(description="是否有货")
    note: str = Field(description="一句话说明")

model_no_think = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)
agent_sr = create_agent(
    model=model_no_think,
    tools=[get_stock],
    system_prompt="你是库存助手，查证后给出结构化结论。",
    response_format=ToolStrategy(Stock),
)

# 父图直接用 MessagesState，也就是只有 messages 一个键
builder = StateGraph(MessagesState)
# 嵌入同一个带结构化输出的 Agent
builder.add_node("agent", agent_sr)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 编译
graph = builder.compile()
# 执行：注意这次不传 structured_response
result = graph.invoke({"messages": [{"role": "user", "content": "A-100 有货吗"}]})
# 结构化结果去哪了
print("父图 keys:", sorted(result.keys()), " <- structured_response 被丢了")

   [tool] get_stock(A-100)
父图 keys: ['messages']  <- structured_response 被丢了

注意那行 [tool] get_stock(A-100)：模型和工具全都正常跑了，结构化结果也算出来了，只是在写回父图的最后一步被扔掉。 又是一次静默丢弃，没有任何警告。

这和第 20 章「节点返回状态里没有的键会被悄悄扔掉」是同一条规则，只是这次扔的是子图的输出。这条规则会在 §5 以更贵的形式再来一次。

3.4 父图的业务字段怎么进子 Agent #

§3.2 演示了「父图多出来的字段不会被子图吃掉」，但反过来的需求同样常见：工具需要知道当前是哪个坐席、哪个租户、哪个订单。这些字段在父图状态里，而子图默认只认 messages，怎么送进去？

有人会想把工号拼进用户消息（"我是 agent-7，帮我查……"）。别这么干：模型可能不理它、可能转述错、更可能被用户输入覆盖掉（提示词注入）。正确做法有两条路。

3.4.1 路线一 context #

路线一：context= 运行时上下文（配合包一层的写法）。 上下文不进消息、模型看不见，工具用 runtime.context 读。这是第 9 章讲过的机制：

from dataclasses import dataclass

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import ToolRuntime
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

# dataclass 用来声明上下文的形状
from dataclasses import dataclass

# ToolRuntime：工具签名里声明它，框架就会注入运行时信息
from langchain.tools import ToolRuntime
# AIMessage：包装节点自己造的回复
from langchain_core.messages import AIMessage

# @dataclass 自动生成 __init__，省掉样板代码
@dataclass
class Ctx:
    """上下文类型：只放这次调用需要的身份信息。"""

    # 坐席工号
    operator_id: str

# runtime 参数由框架注入，模型既看不到也填不了
@tool
def whoami(runtime: ToolRuntime[Ctx, None]) -> str:
    """返回当前坐席工号。"""
    # 没传 context 时 runtime.context 是 None，用兜底值避免 AttributeError
    uid = runtime.context.operator_id if runtime.context else "<无>"
    # 打印确认工具真的读到了工号
    print(f"   [tool] whoami -> {uid}")
    # 把结果作为字符串返回给模型
    return f"当前坐席是 {uid}"

# 创建 Agent 时必须声明 context_schema，否则 context= 传不进去
agent_ctx = create_agent(
    # 模型
    model="deepseek:deepseek-v4-flash",
    # 只挂这个查身份的工具
    tools=[whoami],
    # 让模型老实用工具而不是自己编
    system_prompt="用工具查出当前坐席工号后原样回答。",
    # 声明上下文类型
    context_schema=Ctx,
)

# 父图状态：带一个业务字段 operator_id
class OpState(MessagesState):
    # 坐席工号，父图自己维护
    operator_id: str

# 包一层的节点：负责把父图字段翻译成子图的 context
def agent_node_ctx(state: OpState) -> dict:
    # 手工调用子 Agent：messages 走输入，工号走 context
    result = agent_ctx.invoke(
        # 子图只认 messages
        {"messages": state["messages"]},
        # 父图字段从这里进去，不占消息、不进上下文窗口
        context=Ctx(operator_id=state["operator_id"]),
    )
    # 只回传最终结论
    return {"messages": [AIMessage(content=result["messages"][-1].content)]}

# 建父图
builder = StateGraph(OpState)
# 注册的是包装函数
builder.add_node("agent", agent_node_ctx)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 输入里给上工号，但用户消息里完全没提它
op_inputs = {"messages": [{"role": "user", "content": "我是谁"}], "operator_id": "agent-7"}
# 编译
graph = builder.compile()
# 执行
result = graph.invoke(dict(op_inputs))
# 看工号有没有传到工具里
print("回复:", result["messages"][-1].content)

   [tool] whoami -> agent-7
回复: 当前坐席工号是 agent-7。

用户消息里一个字都没提工号，工具却准确读到了 agent-7。 这就是 context 的价值：它走的是「运行时」这条旁路，不经过模型，模型既看不见也改不了。

3.4.2 路线二 state_schema #

路线二：state_schema= 扩展子图认识的键（配合直接嵌入）。 让子 Agent 的状态类型自己长出这个字段，父图直接嵌入就能对上：

from dataclasses import dataclass

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import ToolRuntime
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@dataclass
class Ctx:
    """上下文类型：只放这次调用需要的身份信息。"""

    operator_id: str

@tool
def whoami(runtime: ToolRuntime[Ctx, None]) -> str:
    """返回当前坐席工号。"""
    uid = runtime.context.operator_id if runtime.context else "<无>"
    print(f"   [tool] whoami -> {uid}")
    return f"当前坐席是 {uid}"

class OpState(MessagesState):
    operator_id: str

op_inputs = {"messages": [{"role": "user", "content": "我是谁"}], "operator_id": "agent-7"}

# AgentState 是 Agent 默认状态类型，在它上面扩展
from langchain.agents import AgentState

# 扩展后的 Agent 状态：自带 messages，再加一个业务字段
class MyAgentState(AgentState):
    # 这个键会出现在子图的 input schema 里
    operator_id: str

# 这个工具从 state 而不是 context 读工号
@tool
def whoami2(runtime: ToolRuntime) -> str:
    """返回当前坐席工号。"""
    # runtime.state 是子图当前状态，用 get 兜住键不存在的情况
    uid = runtime.state.get("operator_id", "<无>")
    # 打印确认
    print(f"   [tool] whoami2 -> {uid}")
    # 返回给模型
    return f"当前坐席是 {uid}"

# 用扩展后的状态类型创建 Agent
agent_ss = create_agent(
    # 模型
    model="deepseek:deepseek-v4-flash",
    # 换成读 state 的工具
    tools=[whoami2],
    # 同样的提示
    system_prompt="用工具查出当前坐席工号后原样回答。",
    # 关键参数：告诉 Agent 它的状态里多了一个键
    state_schema=MyAgentState,
)
# 确认 input schema 真的变宽了
print("扩展后的 input schema:", list(agent_ss.get_input_jsonschema()["properties"].keys()))

# 父图沿用 OpState，它正好有 messages + operator_id
builder = StateGraph(OpState)
# 这次直接嵌入 Agent，不包装
builder.add_node("agent", agent_ss)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 编译
graph = builder.compile()
# 执行：这次不传 context，工号靠状态传下去
result = graph.invoke(dict(op_inputs))
# 工号是通过状态传下去的
print("回复:", result["messages"][-1].content)

扩展后的 input schema: ['messages', 'operator_id']
   [tool] whoami2 -> agent-7
回复: 您的工号是 agent-7。

input schema 从 ['messages'] 变成了 ['messages', 'operator_id']：子图认识的键真的变宽了，父图这个字段于是能直接对上。

两条路怎么选：

	路线一 `context=`	路线二 `state_schema=`
对接方式	必须包一层（要手工 `invoke`）	直接嵌入即可
字段能不能被子图改	不能，只读	能，工具可以用 `Command` 写回
会不会出现在子图 checkpoint 里	不会	会
适合装什么	身份、租户、请求级配置	需要在 Agent 内部演进的业务状态

默认选路线一。 身份这类东西本来就该是只读的，context 天然表达了这个约束；而路线二把父图字段暴露给子图的所有工具，等于扩大了可写面。

4. 三种对接方式 #

直接 add_node(agent) 只是其中一种，而且往往不是最合适的那种。三种方式的差别只有一个：父子状态之间那层翻译，是让 LangGraph 自动做，还是你自己写。

4.1 方式一：直接嵌入 #

也就是 §3.2 那种写法。为了后面几节反复对比，先把父图骨架抽成一个函数：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

# 这个工厂函数只有一个参数：agent 这个位置放什么
def build(node):
    # 沿用 §3.2 的 TicketState
    builder = StateGraph(TicketState)
    # 前处理
    builder.add_node("pre", pre)
    # 中间这个节点由调用方决定：可以是 Agent 本体，也可以是包装函数
    builder.add_node("agent", node)
    # 后处理
    builder.add_node("post", post)
    # 入口进 pre
    builder.add_edge(START, "pre")
    # pre 之后进 agent
    builder.add_edge("pre", "agent")
    # agent 之后进 post
    builder.add_edge("agent", "post")
    # post 之后结束
    builder.add_edge("post", END)
    # 编译返回
    return builder.compile()

# 方式一：把 Agent 本体交给 build（inputs 沿用 §3.2 那份）
graph = build(agent)
# 执行
result_direct = graph.invoke(dict(inputs))
# 看父图最终留下多少条消息
print(f"直接嵌入（{len(result_direct['messages'])} 条）：")
# 逐条打印，观察工具噪音
for m in result_direct["messages"]:
    # 只打印类型和内容前 40 字
    print(f"   {type(m).__name__}: {str(m.content)[:40]!r}")
# 审计里只有两个确定性节点写的内容
print("audit:", result_direct["audit"])

   [node] pre
   [tool] get_stock(A-100)
   [node] post
直接嵌入（4 条）：
   HumanMessage: 'A-100 有货吗'
   AIMessage: ''
   ToolMessage: '库存 12 件'
   AIMessage: 'A-100 有货，库存 12 件。'
audit: ['预处理完成', '后处理完成']

适用： 父图状态就是围绕 messages 组织的，而且你希望子 Agent 的完整对话历史成为父图历史的一部分（比如就是一个客服对话，工具调用记录也要留痕）。

代价： 父图状态必须有 messages 键（否则就是 §5 的头号坑），子图的工具噪音全部外泄，而且 Agent 节点在审计里什么都没留下：你只知道它跑了，不知道它干了什么。

4.2 方式二：包一层做状态转换 #

父图状态如果不是消息式的（比如就是普通的业务字段），直接嵌入会出大问题（§5 详说）。这时候要自己包一个普通节点，在里面完成「父图字段 → 子图输入 → 父图字段」的往返翻译：

from typing import TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

# TypedDict：声明一个纯业务字段的状态，完全不含 messages
from typing import TypedDict

# 父图状态：一进一出两个字段
class QAState(TypedDict):
    # 用户问题
    question: str
    # 最终答案
    answer: str

# 包装节点：它是普通函数，不是子图
def agent_node_qa(state: QAState) -> dict:
    """父图字段 → 子图输入 → 父图字段。"""
    # 把业务字段现场拼成子图要的 messages
    result = agent.invoke({"messages": [{"role": "user", "content": state["question"]}]})
    # 只取最后一条消息的文本写回业务字段
    return {"answer": result["messages"][-1].content}

# 建一张纯业务字段的图
builder = StateGraph(QAState)
# 注意注册的是 agent_node_qa，不是 agent
builder.add_node("agent", agent_node_qa)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 编译
graph = builder.compile()
# 执行并打印整个状态
print(graph.invoke({"question": "A-100 有货吗", "answer": ""}))

   [tool] get_stock(A-100)
{'question': 'A-100 有货吗', 'answer': 'A-100 有货，库存 12 件。'}

适用： 父图状态和 messages 无关，或者你想精确控制喂给 Agent 的上下文：比如只给最近三轮（state["messages"][-6:]）、只给摘要、或者把检索到的资料拼进去。这些都是「翻译」的一部分，直接嵌入没有插手的余地。

这也是最容易读懂的写法：转换逻辑明明白白写在函数里，不用去猜 LangGraph 帮你做了什么。代价是多写几行，以及你要自己负责把结果写回正确的键。

4.3 方式三：包一层并隔离消息 #

父图状态有 messages，但你不希望 Agent 的内部过程污染对外的对话历史。做法是只把结论回传，中间过程折进审计字段：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()

# 隔离写法的包装节点
def agent_node_iso(state: TicketState) -> dict:
    """子 Agent 的内部消息不外泄，只回一条结论。"""
    # 把父图现有历史整体喂给子 Agent
    result = agent.invoke({"messages": state["messages"]})
    # 用切片取出「子图新增的那几条」：父图原有的部分不算
    inner = result["messages"][len(state["messages"]):]
    # 结论就是最后一条消息的文本
    answer = result["messages"][-1].content
    # 数一下 ToolMessage，就知道工具被执行了几次
    tool_calls = sum(1 for m in inner if type(m).__name__ == "ToolMessage")
    # 结论进 messages，过程进 audit
    return {
        # 对外只留一条干净的 AIMessage
        "messages": [AIMessage(content=answer)],
        # 内部过程写进 audit：给人看的信息不占模型上下文
        "audit": [f"agent: 内部 {len(inner)} 条消息、{tool_calls} 次工具调用，仅回传结论"],
    }

# 同一个父图骨架，只把中间节点换成包装函数
graph = build(agent_node_iso)
# 执行
result_iso = graph.invoke(dict(inputs))
# 对比消息条数
print(f"隔离写法（{len(result_iso['messages'])} 条）：")
# 逐条打印，确认没有工具噪音
for m in result_iso["messages"]:
    # 同样只打印类型和内容前 40 字
    print(f"   {type(m).__name__}: {str(m.content)[:40]!r}")
# 审计里多了一条 Agent 自己写的记录
print("audit:", result_iso["audit"])

   [node] pre
   [tool] get_stock(A-100)
   [node] post
隔离写法（2 条）：
   HumanMessage: 'A-100 有货吗'
   AIMessage: 'A-100 有货，库存 12 件。'
audit: ['预处理完成', 'agent: 内部 3 条消息、1 次工具调用，仅回传结论', '后处理完成']

和 §4.1 的 4 条对比：同样一次对话，父图历史从 4 条降到 2 条，而且丢掉的恰好是那条 content='' 的空消息和工具流水。

适用： 绝大多数生产场景。理由有四个：

对外的对话历史干净，前端直接渲染不用过滤空消息
下一轮对话带的上下文短，省 token，工具调用记录往往比结论长得多，而且它对下一轮的价值很低
内部过程写进 audit 而不是 messages，排查时反而更好查（模型看不到，人看得到）
想加防御性检查（比如「送进模型前再确认一次已脱敏」）有地方可放，§8 的实战就这么做

那条切片 result["messages"][len(state["messages"]):] 有个前提值得点明：子 Agent 是把父图历史原样带出来再追加的，所以前缀长度不变。 如果你在子 Agent 上挂了会删消息的中间件（比如第 11 章 §7.3 的 SummarizationMiddleware），这个切片就不再准确，得改成按消息 id 做差集。

4.4 怎么选 #

方式	父图状态要求	消息噪音	能否控制喂给模型的上下文	适用
直接嵌入	必须有 `messages`	全部外泄	不能	纯对话场景，且要完整留痕
包一层转换	任意	你说了算	能	父图不是消息式的
包一层隔离	有 `messages`	只留结论	能	生产默认选这个

一句话：能包一层就包一层。 多写五行代码，换来状态边界清晰、消息可控、出问题好查。直接嵌入省下的那点代码，很容易在 §5 的坑上加倍还回去。

4.5 包一层会不会丢掉可观测性 #

包装写法最常见的顾虑是：手工 invoke 的子 Agent，还算不算这张图的子图？ 直觉会说「不算了，只是节点里一次普通函数调用」。实测结论相反：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()

def agent_node_iso(state: TicketState) -> dict:
    result = agent.invoke({"messages": state["messages"]})
    inner = result["messages"][len(state["messages"]):]
    tool_calls = sum(1 for m in inner if type(m).__name__ == "ToolMessage")
    return {
        "messages": [AIMessage(content=result["messages"][-1].content)],
        "audit": [f"agent: 内部 {len(inner)} 条消息、{tool_calls} 次工具调用，仅回传结论"],
    }

# 一个把 Agent 写在函数内部临时创建的反例
def agent_node_local(state: TicketState) -> dict:
    # 每次执行都新建一个 Agent（这本身就浪费，这里只为演示探测边界）
    local_agent = create_agent(model="deepseek:deepseek-v4-flash", tools=[get_stock])
    # 手工调用
    result = local_agent.invoke({"messages": state["messages"]})
    # 只回传结论
    return {"messages": [AIMessage(content=result["messages"][-1].content)], "audit": ["agent"]}

# 模块级的 agent + 包一层：xray 能否展开
print("模块级 agent + 包一层:", list(build(agent_node_iso).get_graph(xray=1).nodes.keys()))
# 函数内部临时创建：xray 能否展开
print("函数内临时创建:      ", list(build(agent_node_local).get_graph(xray=1).nodes.keys()))

模块级 agent + 包一层: ['__start__', 'pre', 'post', '__end__', 'agent:__start__', 'agent:model', 'agent:tools', 'agent:__end__']
函数内临时创建:       ['__start__', 'pre', 'agent', 'post', '__end__']

包一层照样能被 xray=1 展开。 机制在 langgraph/pregel/_utils.py 里：注册节点时，LangGraph 会用 inspect.getsource() 取出这个函数的源码、用 AST 找出它引用了哪些全局变量和闭包变量，再逐个检查里面有没有编译好的图；找到就登记为这个节点的子图。同理，stream(subgraphs=True) 也照样能看到子图内部（§6.2），挂了 checkpointer 之后子图状态也照样独立存一份（§6.3）。

所以「包一层」的代价里不包含可观测性。但这个机制有两个前提，上面那个反例正好踩中第一个：

子 Agent 必须是函数外部的对象（模块级全局或闭包捕获）。 在节点函数内部临时 create_agent，它只是个局部变量，不在扫描范围里，xray=1 就只能看到一个 agent 方块。
必须取得到源码。 用 exec 动态定义的节点函数、或在纯交互式 REPL 里定义的函数，inspect.getsource() 会抛 OSError，探测直接放弃（这时功能一切正常，只是观测不到子图）。写在 .py 文件或 Jupyter 里都不受影响。

顺便说，「每次调用都新建 Agent」本身就该避免，它会重复初始化模型客户端。把 Agent 建在模块级，或者用 functools.lru_cache 缓存（第 24 章实战的做法），既省开销又保住了可观测性。

5. 头号坑：Agent 空跑，还照样烧钱 #

这是本章最危险的一个，因为它同时满足四个条件：编译通过、运行不报错、结果看起来只是「没生效」、但账单照走。

5.1 现象：什么都对，就是没用 #

把 §4.2 的 QAState 直接配上 add_node(agent)，也就是父图没有 messages 键：

from typing import TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

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

# 父图状态还是那两个业务字段，没有 messages
builder = StateGraph(QAState)
# 这次直接嵌入 Agent 本体，不包装
builder.add_node("agent", agent)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 编译：注意这一步不会报错
graph_no_messages = builder.compile()
# 明确打印一行，强调编译是通过的
print("编译通过了")
# 执行并打印完整状态
print("invoke 结果:", graph_no_messages.invoke({"question": "A-100 有货吗", "answer": ""}))

编译通过了
invoke 结果: {'question': 'A-100 有货吗', 'answer': ''}

answer 是空的。多数人的第一反应是「节点没跑」，其实不是。

5.2 展开看内部：钱是怎么花掉的 #

用 subgraphs=True 展开看看里面到底发生了什么。这个参数会让每个 chunk 多带一个命名空间元组，标明这条更新来自哪一层：

from typing import TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

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

builder = StateGraph(QAState)
builder.add_node("agent", agent)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph_no_messages = builder.compile()

# subgraphs=True 让流式输出钻进子图
for ns, upd in graph_no_messages.stream({"question": "A-100 有货吗", "answer": ""}, subgraphs=True):
    # ns 是命名空间元组：() 表示父图，('agent:xxx',) 表示子图内部
    print(f"ns={ns}")
    # upd 是 {节点名: 更新} 的字典
    for node, val in (upd or {}).items():
        # 父图那一步的更新可能整个是 None，先兜住
        if val is None:
            # 打印出来，这正是「输出被丢弃」的证据
            print(f"   node={node} update=None")
            # 跳过后面的消息解析
            continue
        # 子图返回的更新里带 messages
        for m in val.get("messages", []):
            # 打印模型说了什么
            print(f"   node={node} {type(m).__name__} content={str(m.content)[:60]!r}")
            # usage_metadata 是这次调用的实际 token 消耗
            print(f"       usage={getattr(m, 'usage_metadata', None)}")

ns=('agent:ce90360e-9d62-68f3-ef70-427bed46b754',)
   node=model AIMessage content='请问您需要查询哪个 SKU 的库存？'
       usage={'input_tokens': 362, 'output_tokens': 128, 'total_tokens': 490, 'input_token_details': {'cache_read': 256}, 'output_token_details': {'reasoning': 116}}
ns=()
   node=agent update=None

看清发生了什么：

父图把状态过滤成子图声明的键，messages 在父图里不存在，于是子图拿到一份空历史
子图照常执行，真的调用了模型：490 个 token 已经花出去了（数字每次略有波动）
模型只看到系统提示，什么用户问题都没看到，礼貌地回了句「请问您需要查询哪个 SKU」
子图返回 {"messages": [...]}，父图状态里没有这个键，静默丢弃
父图这一步的更新是 None

「拿到一份空历史」还可以说得更精确一点。拿一个只会打印输入的假子图替掉 Agent，就能看到子图节点到底拿到了什么：

from typing import TypedDict

from langgraph.graph import END, START, MessagesState, StateGraph

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

# 一个只负责打印输入的假子图，用来看清父图到底传了什么进来
def spy_node(state):
    # 打印子图节点实际收到的键
    print("   子图节点收到 state keys:", sorted(state.keys()))
    # 用 get 兜住，避免 KeyError 掩盖真相
    print("   messages =", state.get("messages", "<键不存在>"))
    # 返回空更新
    return {"messages": []}

# 假子图用 MessagesState，和真 Agent 的输入要求一致
spy_builder = StateGraph(MessagesState)
# 注册探针节点
spy_builder.add_node("m", spy_node)
# 入口
spy_builder.add_edge(START, "m")
# 出口
spy_builder.add_edge("m", END)
# 编译成子图
spy_graph = spy_builder.compile()

# 父图仍然是没有 messages 的 QAState
builder = StateGraph(QAState)
# 嵌入假子图
builder.add_node("agent", spy_graph)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 编译
graph = builder.compile()
# 执行，只为看打印
graph.invoke({"question": "问题", "answer": ""})

   子图节点收到 state keys: ['messages']
   messages = []

子图收到的 messages 是一个空列表，而不是「键不存在」。这个 [] 不是父图给的，是子图自己的 reducer 给的：messages 挂了 add_messages，带 reducer 的通道会用「空值」初始化（第 21 章讲过 list 型 reducer 的空值就是 []）。所以整条链是：

父图没有这个键 → 传进来的更新是空的 → 子图通道退回 reducer 的空值 [] → Agent 拿着空历史去问模型 → 模型只看到系统提示。

正是这个「空值兜底」把一个本该报错的场景变成了一次正常的付费调用。 反过来，如果某个字段没挂 reducer，它在子图里就是彻底不存在的（探针换成 messages: list 就会打印 <键不存在>），那样反倒容易在读取时抛 KeyError，早早暴露问题。有 reducer 更安全，但也更安静。

全程零报错，零警告，只有账单知道。 而且这个 bug 在测试里极难发现：它不崩溃，只是「效果不太好」，很容易被误判成提示词写得不够好，然后你就开始调提示词，越调越困惑。

5.3 两个「碰巧能跑」的相邻情况 #

顺便看两个相邻的写法，它们同样不报错，但结果不同。

5.3.1 一、键名对了但类型标错 #

一、键名对了但类型标错（messages: str）。

from typing import TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

# 键名是对的，但类型标注写成了 str
class WrongType(TypedDict):
    # 实际上流转的是消息列表，这里的标注在骗人
    messages: str

# 建图
builder = StateGraph(WrongType)
# 直接嵌入 Agent
builder.add_node("agent", agent)
# 入口
builder.add_edge(START, "agent")
# 出口
builder.add_edge("agent", END)
# 编译
graph = builder.compile()
# 输入也按标注给一个裸字符串
result = graph.invoke({"messages": "A-100 有货吗"})
# 看返回值到底是什么类型
print("返回 messages 的类型:", type(result["messages"]).__name__)
# 数一下条数
print("条数:", len(result["messages"]))
# 首条被转成了什么
print("首条类型:", type(result["messages"][0]).__name__)
# 末条是正常回答吗
print("末条:", str(result["messages"][-1].content)[:50])

   [tool] get_stock(A-100)
返回 messages 的类型: list
条数: 4
首条类型: HumanMessage
末条: A-100 有货，库存 12 件。

居然正常工作：LangChain 会把裸字符串强制转成 HumanMessage。但你的类型标注是个谎言：写着 str，实际拿到 list。下游任何按 str 处理（比如 state["messages"].strip()）的地方都会炸在别处，而且看起来和这里毫无关系。

5.3.2 二、有messages但没配add_messages reducer #

二、有 messages 但没配 add_messages reducer（messages: list）。 如果图里只有 Agent 一个节点，它也能跑对，因为子图返回的是完整消息列表，覆盖恰好等于正确结果。但只要父图别处还有节点往 messages 写，后写的就会把先写的整段盖掉：

import operator
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

# messages 用裸 list，也就是「覆盖」语义
class NoReducer(TypedDict):
    # 少了 Annotated[list, add_messages]
    messages: list
    # audit 照旧挂 reducer，用来对比
    audit: Annotated[list[str], operator.add]

# 一个普通节点，往 messages 里追加一条工单备注
def extra(state):
    # 它以为自己是在追加
    return {"messages": [HumanMessage(content="补充一句：客户比较着急")], "audit": ["extra"]}

# 建图：先跑 extra，再跑 Agent
builder = StateGraph(NoReducer)
# 那个「以为自己在追加」的节点
builder.add_node("extra", extra)
# 直接嵌入的 Agent
builder.add_node("agent", agent)
# 入口进 extra
builder.add_edge(START, "extra")
# extra 之后进 agent
builder.add_edge("extra", "agent")
# agent 之后结束
builder.add_edge("agent", END)
# 编译
graph = builder.compile()
# 执行
result = graph.invoke({"messages": [{"role": "user", "content": "A-100 有货吗"}], "audit": []})
# 消息条数远少于预期
print("消息条数:", len(result["messages"]))
# 逐条看，找找用户那句话去哪了
for m in result["messages"]:
    # 打印类型和内容
    print(f"   {type(m).__name__}: {str(m.content)[:40]!r}")
# 用生成式检查用户原话是否还在历史里
kept = any("A-100 有货吗" in str(m.content) for m in result["messages"])
# 明确打印结论
print("用户那句「A-100 有货吗」还在吗:", kept)

消息条数: 2
   HumanMessage: '补充一句：客户比较着急'
   AIMessage: '明白，客户比较急。请问需要查询哪个 SKU 的库存？我马上帮您查。'
用户那句「A-100 有货吗」还在吗: False

用户的问题被 extra 节点整个覆盖掉了，于是 Agent 只看到那句备注，反过来问用户要查哪个 SKU。注意这里连工具都没调（输出里没有 [tool]），它根本不知道要查什么。

这是「碰巧对了」和「写对了」的分界线：只有一个节点写 messages 时覆盖等于追加，多一个节点，后写的就会把先写的盖掉，而且盖掉的往往是最关键的那条用户输入。

5.4 三行断言把静默 bug 变成启动报错 #

上面三种情况有个共同点：它们都能在跑之前用键名对比查出来。 子图自己会报出它需要什么，父图的 TypedDict 也能报出它有什么，对一下就行：

import operator
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import MessagesState

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

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

# typing.get_type_hints 比直接读 __annotations__ 更稳，能处理字符串形式的标注
import typing

# 子图需要的键：直接从它的 input schema 拿，不用手写
required = set(agent.get_input_jsonschema()["properties"])
# 打印出来确认
print("子图需要的键:", required)

# 拿两个父图状态做对照实验
for name, state_cls in (("继承 MessagesState 的状态", TicketState), ("只有业务字段的状态", QAState)):
    # 分组标题
    print(f"\n--- {name} ---")
    # TypedDict 会把基类的键合并进 __annotations__，所以这里能看到继承来的 messages
    print("  __annotations__:", set(state_cls.__annotations__))
    # 换成 get_type_hints 结果一致，拿不准时用这个更稳妥
    print("  get_type_hints:", set(typing.get_type_hints(state_cls)))
    # 子集判断：子图需要的键必须全在父图状态里
    print("  断言通过:", required <= set(state_cls.__annotations__))

子图需要的键: {'messages'}

--- 继承 MessagesState 的状态 ---
  __annotations__: {'messages', 'user_id', 'audit'}
  get_type_hints: {'messages', 'user_id', 'audit'}
  断言通过: True

--- 只有业务字段的状态 ---
  __annotations__: {'question', 'answer'}
  get_type_hints: {'question', 'answer'}
  断言通过: False

落到工程里就是一行 assert，放在 add_node 旁边：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

# 从子图 schema 现取需要的键，Agent 换了 state_schema 也不用改这行
required = set(agent.get_input_jsonschema()["properties"])
# 父图状态声明的键
provided = set(TicketState.__annotations__)
# 对不上就直接启动失败，并把缺哪个键打出来
assert required <= provided, f"父图状态缺少子图需要的键: {required - provided}"

这里能直接用 __annotations__ 是因为 TypedDict 会把基类的键合并进子类的 __annotations__（普通 Python 类不会，只有自己声明的）。所以继承 MessagesState 之后，messages 也在里面。

三行代码，把一个静默烧钱的 bug 变成启动就报错。注意这个断言是从 schema 现取的，所以哪天你给子 Agent 加了 state_schema（§3.4），它会自动跟着变严，不需要维护一份手写清单。

6. 看清嵌套图内部 #

嵌套之后，默认的观测手段都只能看到「agent 这个节点跑了」，看不到里面。有三个工具能钻进去：画图用 xray，实时看用 subgraphs=True，事后查用 task.state。

6.1 xray=1：把子图画开 #

默认画图，子图就是一个方块：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()

# 用 §4.1 的骨架建一张直接嵌入 Agent 的图
graph = build(agent)
# 默认画法：子图内部完全不展开
print(graph.get_graph().draw_mermaid())

---
config:
  flowchart:
    curve: linear
---
graph TD;
    __start__([<p>__start__</p>]):::first
    pre(pre)
    agent(agent)
    post(post)
    __end__([<p>__end__</p>]):::last
    __start__ --> pre;
    agent --> post;
    pre --> agent;
    post --> __end__;
    classDef default fill:#f2f0ff,line-height:1.2
    classDef first fill-opacity:0
    classDef last fill:#bfb6fc

加上 xray=1 就展开了：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()
graph = build(agent)

# xray=1 表示往下展开一层子图；给更大的数字可以展开更深的嵌套
print(graph.get_graph(xray=1).draw_mermaid())

---
config:
  flowchart:
    curve: linear
---
graph TD;
    __start__([<p>__start__</p>]):::first
    pre(pre)
    post(post)
    __end__([<p>__end__</p>]):::last
    __start__ --> pre;
    agent\3a__end__ --> post;
    pre --> agent\3a__start__;
    post --> __end__;
    subgraph agent
    agent\3a__start__(<p>__start__</p>)
    agent\3amodel(model)
    agent\3atools(tools)
    agent\3a__end__(<p>__end__</p>)
    agent\3a__start__ --> agent\3amodel;
    agent\3amodel -.-> agent\3a__end__;
    agent\3amodel -.-> agent\3atools;
    agent\3atools -.-> agent\3amodel;
    end
    classDef default fill:#f2f0ff,line-height:1.2
    classDef first fill-opacity:0
    classDef last fill:#bfb6fc

三处细节值得留意：

原来那个 agent(agent) 方块消失了，取而代之的是一个 subgraph agent ... end 区块，父图的边改成直接连到子图的 __start__ / __end__
agent\3amodel 里的 \3a 是 Mermaid 对冒号的转义，实际节点名是 agent:model
子图内部 model -.-> tools 和 tools -.-> model 都是虚线，说明两个方向都是条件边，这正是第 24 章 §4.4 对比过的：create_agent 的 tools → model 也走条件边，比手写版多一层判断

如果只想拿节点清单而不想读 Mermaid，直接看 nodes 更方便：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()
graph = build(agent)

# 取展开后的图对象
drawable = graph.get_graph(xray=1)
# 节点名里带冒号的就是子图内部节点
print("nodes:", list(drawable.nodes.keys()))
# 边也一并展开了，conditional 标出哪些是条件边
print("edges:")
# 遍历所有边
for e in drawable.edges:
    # 打印起点、终点、是否条件边
    print(f"   {e.source} -> {e.target} conditional={e.conditional}")

nodes: ['__start__', 'pre', 'post', '__end__', 'agent:__start__', 'agent:model', 'agent:tools', 'agent:__end__']
edges:
   __start__ -> pre conditional=False
   agent:__end__ -> post conditional=False
   pre -> agent:__start__ conditional=False
   post -> __end__ conditional=False
   agent:__start__ -> agent:model conditional=False
   agent:model -> agent:__end__ conditional=True
   agent:model -> agent:tools conditional=True
   agent:tools -> agent:model conditional=True

第 24 章手写的那个 model ↔ tools 环，现在作为 subgraph agent 嵌在流水线中间，两章的内容在这张图里合上了。

6.2 subgraphs=True：流式钻进子图 #

默认的 stream 把子图当黑盒，整个 Agent 只报一次：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()
graph = build(agent)

# 不加参数：子图内部的每一步都被压成父图的一次更新
for chunk in graph.stream(dict(inputs)):
    # chunk 是 {节点名: 更新} 的字典
    for node, upd in chunk.items():
        # 只打印节点名和它写了哪些键
        print(f"[{node}] {list(upd.keys())}")

   [node] pre
[pre] ['audit']
   [tool] get_stock(A-100)
[agent] ['messages']
   [node] post
[post] ['audit']

加上 subgraphs=True，每个 chunk 会多一个命名空间元组：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def build(node):
    builder = StateGraph(TicketState)
    builder.add_node("pre", pre)
    builder.add_node("agent", node)
    builder.add_node("post", post)
    builder.add_edge(START, "pre")
    builder.add_edge("pre", "agent")
    builder.add_edge("agent", "post")
    builder.add_edge("post", END)
    return builder.compile()
graph = build(agent)

# 解包成 (命名空间, 更新) 两部分
for ns, chunk in graph.stream(dict(inputs), subgraphs=True):
    # 同一个命名空间下可能有多个节点更新
    for node, upd in chunk.items():
        # 只打印键名，避免消息内容把输出淹掉
        print(f"ns={ns} node={node} keys={list(upd.keys())}")

   [node] pre
ns=() node=pre keys=['audit']
   [tool] get_stock(A-100)
ns=('agent:efc1b04a-aecf-239f-77e9-1cc02cab651a',) node=model keys=['messages']
ns=('agent:efc1b04a-aecf-239f-77e9-1cc02cab651a',) node=tools keys=['messages']
ns=('agent:efc1b04a-aecf-239f-77e9-1cc02cab651a',) node=model keys=['messages']
ns=() node=agent keys=['messages']
   [node] post
ns=() node=post keys=['audit']

ns=() 是父图，ns=('agent:...',) 是子图内部，冒号后面那串是这次子图任务的 id（每次运行都不同）。现在能清楚看到 Agent 转了几圈：model → tools → model，正好是第 24 章那个环的两轮。排查「Agent 在里面干了什么」，这是第一手段。

两个提醒：

父图那条 node=agent 的更新是最后才出现的，它是子图整体的输出。同一批消息你会看到两遍：一遍来自子图内部的 model/tools，一遍来自父图的 agent。做前端增量渲染时要按命名空间过滤，否则会重复。
§4.5 验证过，包一层的写法同样能被 subgraphs=True 看到内部，不用为了观测而放弃隔离。

6.3 子图的历史状态 #

挂了 checkpointer 之后，子图的状态也被单独存了一份，存在自己的 checkpoint_ns 下。从父图的历史里找到子图任务，用它的 config 反查：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def agent_node_iso(state: TicketState) -> dict:
    result = agent.invoke({"messages": state["messages"]})
    inner = result["messages"][len(state["messages"]):]
    tool_calls = sum(1 for m in inner if type(m).__name__ == "ToolMessage")
    return {
        "messages": [AIMessage(content=result["messages"][-1].content)],
        "audit": [f"agent: 内部 {len(inner)} 条消息、{tool_calls} 次工具调用，仅回传结论"],
    }

# InMemorySaver：进程内的 checkpointer，学习和调试够用
from langgraph.checkpoint.memory import InMemorySaver

# 重建一张一样的图，这次多接收一个 saver 参数
def build_ck(node, saver):
    # 状态模式不变
    builder = StateGraph(TicketState)
    # 三个节点照旧
    builder.add_node("pre", pre)
    # 中间节点由调用方决定
    builder.add_node("agent", node)
    # 后处理
    builder.add_node("post", post)
    # 入口进 pre
    builder.add_edge(START, "pre")
    # pre 之后进 agent
    builder.add_edge("pre", "agent")
    # agent 之后进 post
    builder.add_edge("agent", "post")
    # post 之后结束
    builder.add_edge("post", END)
    # compile 时传 checkpointer，图就有了持久化能力
    return builder.compile(checkpointer=saver)

# 这次用隔离写法，验证「父图只留 2 条，子图仍存 4 条」
graph = build_ck(agent_node_iso, InMemorySaver())
# 一个 thread_id 代表一条会话线
cfg = {"configurable": {"thread_id": "t-1"}}
# 跑一次，把状态写进 checkpoint
graph.invoke(dict(inputs), cfg)
# 父图当前状态里只有隔离后的 2 条
print("父图消息条数:", len(graph.get_state(cfg).values["messages"]))

# 遍历父图的所有历史快照
for s in graph.get_state_history(cfg):
    # 每个快照里可能有若干待执行/已执行的任务
    for t in s.tasks:
        # 只有子图任务才带 state，普通节点这里是 None
        if t.state:
            # 用子图任务的 config 反查子图快照
            sub = graph.get_state(t.state)
            # 子图的命名空间，形如 agent:<uuid>
            print(f"子图 ns: {t.state['configurable']['checkpoint_ns']}")
            # 子图状态里只有它自己声明的键
            print(f"子图 values keys: {list(sub.values.keys())}")
            # 子图完整保留了内部过程
            print(f"子图消息条数: {len(sub.values['messages'])}")
            # 逐个类型看，工具调用记录都在
            print(f"子图消息类型: {[type(m).__name__ for m in sub.values['messages']]}")

   [node] pre
   [tool] get_stock(A-100)
   [node] post
父图消息条数: 2
子图 ns: agent:bdab6520-88c5-8aa6-5b43-d023cefdb0e2
子图 values keys: ['messages']
子图消息条数: 4
子图消息类型: ['HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage']

即使用了 §4.3 的隔离写法、父图只留了 2 条消息，子图那 4 条完整记录依然存在 checkpoint 里。 隔离的是父图的工作状态，不是审计记录，这一点对合规场景很重要：你既拿到了干净的对话历史，又没有丢掉「模型当时看到了什么、调了什么」的证据。

之所以包一层也能存下来，是因为子 Agent 自己没挂 checkpointer 时会继承父图的（配置通过运行时上下文往下传），并被分配一个独立的 checkpoint_ns。反过来说：

想让子 Agent 的记录进另一个库，就给它单独 create_agent(..., checkpointer=另一个saver)，并在包装节点里传自己的 thread_id
get_state(cfg) 只看父图，不会顺便把子图状态带出来；要么走上面的 task.state，要么用 get_state(cfg, subgraphs=True)，后者在图暂停在子图内部时（第 26 章的 interrupt）才有内容，正常结束后 tasks 是空的

7. 「边保证执行」的边界在哪 #

§2 说「agent → archive 这条边保证 archive 一定执行」。这句话需要加一个限定，否则容易误导。

试试让 Agent 节点抛异常：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

# 一个必然失败的 Agent 节点，模拟模型服务挂了
def boom_agent(state: TicketState) -> dict:
    # 直接抛，不做任何处理
    raise RuntimeError("模型服务 503")

# 归档节点：本节要反复确认它到底有没有执行
def archive(state: TicketState) -> dict:
    # 打印是最直接的执行证据
    print("   [node] archive 执行了")
    # 往审计追加一条落库记录
    return {"audit": ["archive: 落库"]}

# 建一张 pre → agent → archive 的直线图
builder = StateGraph(TicketState)
# 前处理
builder.add_node("pre", pre)
# 换成必然抛错的 Agent 节点
builder.add_node("agent", boom_agent)
# 归档
builder.add_node("archive", archive)
# 入口进 pre
builder.add_edge(START, "pre")
# pre 之后进 agent
builder.add_edge("pre", "agent")
# 这条边看起来「保证」了归档
builder.add_edge("agent", "archive")
# 归档完结束
builder.add_edge("archive", END)
# 执行并捕获异常，看异常是原样抛出还是被包装
try:
    # messages 给空列表就够了，boom_agent 根本不看它
    builder.compile().invoke({"messages": [], "user_id": "u-1", "audit": []})
    # 走到这里说明没抛错，那才反常
    print("没抛错？")
except Exception as e:
    # 异常类型和消息都没被改写
    print("抛错:", type(e).__name__, e)
    # 上面没有 archive 的打印，说明它没执行
    print("archive 没有执行")

   [node] pre
抛错: RuntimeError 模型服务 503
archive 没有执行

边保证的是「上游成功完成，就一定走这条边」，不是「无论如何都执行」。 图边是控制流，不是 try/finally。第 22 章讲过节点异常不会被包装，整张图直接停在那里。

挂了 checkpointer 的话现场还在，可以修好再续跑：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def boom_agent(state: TicketState) -> dict:
    raise RuntimeError("模型服务 503")

def archive(state: TicketState) -> dict:
    print("   [node] archive 执行了")
    return {"audit": ["archive: 落库"]}

builder = StateGraph(TicketState)
builder.add_node("pre", pre)
builder.add_node("agent", boom_agent)
builder.add_node("archive", archive)
builder.add_edge(START, "pre")
builder.add_edge("pre", "agent")
builder.add_edge("agent", "archive")
builder.add_edge("archive", END)

# 同一张图，这次挂 checkpointer
graph = builder.compile(checkpointer=InMemorySaver())
# 新的会话线
cfg2 = {"configurable": {"thread_id": "t-2"}}
# 照样会抛错，先接住
try:
    # 注意这次多传了 cfg2，状态才会写进 checkpoint
    graph.invoke({"messages": [], "user_id": "u-1", "audit": []}, cfg2)
except Exception as e:
    # 只打类型，重点在后面的快照
    print("抛错:", type(e).__name__)
# 取出崩溃时的快照
snap = graph.get_state(cfg2)
# next 指向「下一步该跑谁」，也就是失败的那个节点
print("next:", snap.next)
# 失败之前写进去的状态都还在
print("audit:", snap.values["audit"])
# 待执行任务里带着错误信息（注意存的是字符串形式）
print("tasks:", [(t.name, type(t.error).__name__ if t.error else None) for t in snap.tasks])

   [node] pre
抛错: RuntimeError
next: ('agent',)
audit: ['预处理完成']
tasks: [('agent', 'str')]

next=('agent',) 说明这次失败没有推进 checkpoint：pre 的成果留下了，agent 仍是待执行状态。修好代码后用同一个 thread_id 传 None 续跑，就会从 agent 重来，而不是从头开始。注意 t.error 里存的是错误的字符串形式（类型是 str），不是异常对象，它是要序列化进 checkpoint 的，所以拿不到原始的 traceback 对象。

但如果你要的是「不管 Agent 成没成功，都必须归档」，就得让 Agent 节点自己把异常兜住，把失败变成一个正常的状态更新：

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv(override=True)

@tool
def get_stock(sku: str) -> str:
    """查询某个 SKU 的库存数量。"""
    print(f"   [tool] get_stock({sku})")
    return {"A-100": "库存 12 件", "B-200": "库存 0 件"}.get(sku, "查无此 SKU")

agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[get_stock],
    system_prompt="你是库存助手，回答尽量简短。",
)

class TicketState(MessagesState):
    user_id: str
    audit: Annotated[list[str], operator.add]

def pre(state: TicketState) -> dict:
    print("   [node] pre")
    return {"audit": ["预处理完成"]}

def post(state: TicketState) -> dict:
    print("   [node] post")
    return {"audit": ["后处理完成"]}

inputs = {"messages": [{"role": "user", "content": "A-100 有货吗"}], "user_id": "u-1", "audit": []}

def archive(state: TicketState) -> dict:
    print("   [node] archive 执行了")
    return {"audit": ["archive: 落库"]}

# 带降级的 Agent 节点
def safe_agent(state: TicketState) -> dict:
    try:
        # 真实代码这里是 agent.invoke(...)；为了稳定复现，直接抛一个模拟异常
        raise RuntimeError("模型服务 503")
    except Exception as e:
        # 把「异常」变成「一种正常的结果」：给用户一句兜底话术
        return {
            # 对外回复照旧是一条 AIMessage，下游不用特殊处理
            "messages": [AIMessage(content="抱歉，服务暂时不可用，请稍后重试。")],
            # 真实原因写进审计，供排查用
            "audit": [f"agent: 调用失败({e})，已降级"],
        }

# 换成降级版本再建一张图
builder = StateGraph(TicketState)
# 前处理
builder.add_node("pre", pre)
# 这次用兜住异常的版本
builder.add_node("agent", safe_agent)
# 归档
builder.add_node("archive", archive)
# 入口进 pre
builder.add_edge(START, "pre")
# pre 之后进 agent
builder.add_edge("pre", "agent")
# agent 之后进 archive
builder.add_edge("agent", "archive")
# 归档完结束
builder.add_edge("archive", END)
# 编译
graph = builder.compile()
# 这次不需要 try/except，图能正常跑完
result = graph.invoke({"messages": [], "user_id": "u-1", "audit": []})
# 三个节点各写了一段，归档也执行了
print("audit:", result["audit"])
# 用户拿到的是兜底话术而不是 500
print("对外回复:", result["messages"][-1].content)

   [node] pre
   [node] archive 执行了
audit: ['预处理完成', 'agent: 调用失败(模型服务 503)，已降级', 'archive: 落库']
对外回复: 抱歉，服务暂时不可用，请稍后重试。

这和第 24 章 §6 的降级思路是同一个：把「异常」转成「一种正常的结果」，控制流就能继续往下走。 概率性组件（模型调用、外部 API）放进图里时，这几乎是标配。

顺便说清两种做法的分工，别以为兜住异常就万事大吉：

	不兜（让它抛）	兜住并降级
下游节点	不执行	正常执行
用户看到	报错 / 超时	一句兜底话术
现场	靠 checkpointer 保留，可续跑	已经写进正常状态，续跑无意义
适合	内部批处理、可重试的离线任务	面向用户的在线请求

在线请求兜住、离线任务别兜：因为兜住之后你就失去了「原地重试」的机会，而在线请求本来也不该让用户等重试。

8. 实战：工单处理流水线 #

8.1 设计 #

一条客服工单流水线，把本章和前几章的要点都用上：

START → validate ──（条件边）─┬─ redact → agent → archive → END
                              └─ reject ─────────↗

节点	类型	干什么	为什么必须是这个类型
`validate`	确定性	解析订单号、校验存在性、鉴权	规则明确，且不能被绕过
`redact`	确定性	手机号脱敏后才给模型	有确定答案，且必须在 Agent 之前
`agent`	Agent	查订单、组织回复	用户说法千变万化
`reject`	确定性	标准拒绝话术	一分钱模型费都不花
`archive`	确定性	出站复检 + 审计落库	漏一条就是事故

三条设计上的要点：

reject 和 agent 都汇入 archive。 不管走哪条路都要归档，这是用图结构表达的，不靠任何人记得。
redact 在 agent 之前。 模型从头到尾看不到完整手机号，这是结构性保证，不是提示词约束。
审计里记录「几次模型调用」。 让「前置拦截省钱」这件事从口号变成日志里的 0 次模型调用，可以直接拿去做回归断言。

Agent 节点用 §4.3 的隔离写法，并在里面加一道兜底断言：万一以后有人往图里加节点、或者把 redact 挪走，它会大声报错而不是静默泄露。

8.2 ticket_pipeline.py #

"""工单处理流水线：确定性前处理 → Agent → 确定性后处理。"""

# 让类型标注延迟求值，这样 Annotated[...] 里的写法不受定义顺序限制
from __future__ import annotations

# operator.add 用作 audit 字段的 reducer，实现「追加而不是覆盖」
import operator
# 正则库：解析订单号、匹配手机号
import re
# Annotated 用来给字段挂 reducer；Literal 用来声明路由函数的返回值范围
from typing import Annotated, Literal

# 从 .env 读取 DEEPSEEK_API_KEY
from dotenv import load_dotenv
# create_agent：第 9 章的 Agent 工厂，返回一张编译好的图
from langchain.agents import create_agent
# AIMessage：确定性节点自己伪造回复时用它包装
from langchain_core.messages import AIMessage
# tool 装饰器：把普通函数变成模型可调用的工具
from langchain_core.tools import tool
# 图的四件套：起点、终点、自带 messages 的状态基类、图构建器
from langgraph.graph import END, START, MessagesState, StateGraph

# override=True 让 .env 里的值覆盖系统环境变量中的同名旧值
load_dotenv(override=True)

# 模拟订单库：键是订单号，值里带 dept 字段用于权限判断
ORDERS = {
    # 零售类订单，已发货
    "20260813001": {"sku": "A-100", "status": "已发货", "amount": 39, "dept": "零售"},
    # 零售类订单，待发货
    "20260813002": {"sku": "B-200", "status": "待发货", "amount": 58, "dept": "零售"},
    # 企业类订单，只有高权限坐席能看
    "20260813003": {"sku": "C-900", "status": "已签收", "amount": 1280, "dept": "企业"},
}
# 模拟权限表：每个坐席工号能看哪些部门的订单
ALLOWED_DEPTS = {"agent-1": {"零售"}, "agent-2": {"零售", "企业"}}
# 中国大陆手机号的粗匹配正则，用于脱敏和出站复检
PHONE = re.compile(r"1[3-9]\d{9}")

# 唯一给模型的工具：只读查询，不涉及权限判断
@tool
def query_order(order_id: str) -> str:
    """按订单号查询订单状态、金额和商品编号。"""
    # 打印一行，用来在输出里确认模型到底有没有真的调工具
    print(f"   [tool] query_order({order_id})")
    # 查订单库
    order = ORDERS.get(order_id)
    # 查不到就走文字说明这条路，而不是抛异常打断整张图
    if not order:
        # 让模型自己把「查不到」组织成话术
        return f"订单 {order_id} 不存在"
    # 查到了就返回一句结构清晰的自然语言，方便模型直接引用
    return f"订单 {order_id}：商品 {order['sku']}，状态 {order['status']}，金额 {order['amount']} 元"

# 子 Agent：只负责「看懂用户问什么 + 查证 + 组织回复」
support_agent = create_agent(
    # 模型标识，格式是「提供方:模型名」
    model="deepseek:deepseek-v4-flash",
    # 只挂查询工具，鉴权和脱敏一律不给模型碰
    tools=[query_order],
    # 系统提示约束风格；注意这是「请求」，真正的保证靠图边
    system_prompt="你是电商客服。用工具查证后再回答，回答控制在两句话内，不要编造信息。",
)

# 父图状态：继承 MessagesState 拿到带 add_messages reducer 的 messages
class TicketState(MessagesState):
    # 坐席工号，鉴权用；模型看不到这个字段
    operator_id: str
    # 从用户文本里解析出来的订单号，落库时要写进审计
    order_id: str
    # 非空表示被前置拦截，同时充当路由依据
    rejected: str
    # 全程审计轨迹；挂 operator.add 让每个节点各写一段而不互相覆盖
    audit: Annotated[list[str], operator.add]

def validate(state: TicketState) -> dict:
    """确定性前处理：解析订单号 + 权限校验。不调模型。"""
    # 取最后一条消息的文本，也就是本次工单内容
    text = state["messages"][-1].content
    # 用正则找 11 位数字当订单号
    m = re.search(r"\b(\d{11})\b", text)
    # 第一关：文本里根本没有订单号
    if not m:
        # 直接判拦截，连模型都不用起
        return {"rejected": "未识别到订单号", "audit": ["validate: 缺订单号，拦截"]}

    # 取出捕获组里的订单号
    order_id = m.group(1)
    # 查订单是否真实存在
    order = ORDERS.get(order_id)
    # 第二关：订单号格式对但查无此单
    if not order:
        # 拦截，同时把 order_id 记下来方便审计追溯
        return {"order_id": order_id, "rejected": f"订单 {order_id} 不存在",
                # 审计里写清拦截原因
                "audit": [f"validate: 订单 {order_id} 不存在，拦截"]}

    # 查这个坐席能看哪些部门；工号不在表里就得到空集合
    allowed = ALLOWED_DEPTS.get(state["operator_id"], set())
    # 第三关：订单所属部门不在允许范围内
    if order["dept"] not in allowed:
        # 越权，拦截
        return {"order_id": order_id, "rejected": f"无权访问「{order['dept']}」类订单",
                # 越权尝试必须留痕，这是合规硬要求
                "audit": [f"validate: {state['operator_id']} 越权访问 {order_id}，拦截"]}

    # 三关都过了：只写 order_id 和审计，rejected 保持空字符串
    return {"order_id": order_id, "audit": [f"validate: {order_id} 校验通过"]}

def redact(state: TicketState) -> dict:
    """确定性前处理：手机号脱敏后再交给模型。"""
    # 待脱敏的就是最后这条用户消息
    last = state["messages"][-1]
    # 先把所有命中的手机号找出来，用于计数和写审计
    hits = PHONE.findall(last.content)
    # 没有敏感信息就什么都不改
    if not hits:
        # 只留一条审计，方便确认这一步确实执行过
        return {"audit": ["redact: 无敏感信息"]}

    # 把每个手机号替换成「前 3 位 + **** + 后 4 位」
    masked = PHONE.sub(lambda m: m.group()[:3] + "****" + m.group()[-4:], last.content)
    # 取第一个命中号码的掩码形式，作为「模型看到的是什么」的证据
    sample = hits[0][:3] + "****" + hits[0][-4:]
    # 同时改写消息和写审计
    return {
        # 用同一个 id 覆盖原消息（第 21 章：add_messages 按 id 替换而不是追加）
        "messages": [type(last)(content=masked, id=last.id)],
        # 审计里写清脱敏了几处、模型将看到什么形态
        "audit": [f"redact: 已脱敏 {len(hits)} 处手机号，模型看到的是 {sample}"],
    }

def agent_node(state: TicketState) -> dict:
    """Agent 节点：只把结论回传父图，内部工具噪音不外泄。"""
    # 兜底断言：万一以后有人把 redact 挪走或加了绕过分支，这里会大声报错而不是静默泄露
    assert not any(PHONE.search(str(m.content)) for m in state["messages"]), "脱敏未生效，拒绝调用模型"
    # 手动调用子 Agent，输入只喂 messages（子图也只认这一个键）
    result = support_agent.invoke({"messages": state["messages"]})
    # 用切片取出「子 Agent 新增的那几条」，父图原有的部分不算
    inner = result["messages"][len(state["messages"]):]
    # 最终结论是最后一条消息的文本
    answer = result["messages"][-1].content
    # 新增消息里的 AIMessage 条数 = 模型被调用了几次（花了几次钱）
    model_calls = sum(1 for m in inner if type(m).__name__ == "AIMessage")
    # ToolMessage 条数 = 工具被执行了几次
    tool_calls = sum(1 for m in inner if type(m).__name__ == "ToolMessage")
    # 回传结论 + 审计
    return {
        # 只回传一条干净的结论，中间的 tool_calls / ToolMessage 都留在子图里
        "messages": [AIMessage(content=answer)],
        # 内部过程写进 audit：给人看的信息不占模型上下文
        "audit": [f"agent: {model_calls} 次模型调用、{tool_calls} 次工具调用，内部 {len(inner)} 条消息只回传结论"],
    }

def reject(state: TicketState) -> dict:
    """确定性兜底：被拦截时给一句标准话术，不调模型。"""
    # 回复 + 审计一起写回
    return {
        # 拒绝话术由代码拼出来，措辞完全可控、可回归测试
        "messages": [AIMessage(content=f"抱歉，本次请求无法处理：{state['rejected']}。")],
        # 明确记下这条路径的模型成本是零
        "audit": ["reject: 已返回标准拒绝话术（0 次模型调用）"],
    }

def archive(state: TicketState) -> dict:
    """确定性后处理：出站复检 + 审计落库。"""
    # 即将发给用户的文本
    reply = state["messages"][-1].content
    # 出站复检：回复里是否残留完整手机号
    leaked = PHONE.findall(reply)
    # 没残留就记通过，有残留就把泄露内容记进审计等人处理
    notes = ["archive: 出站复检通过"] if not leaked else [f"archive: 警告，回复中残留手机号 {leaked}"]
    # 再追加一条落库记录；order_id 为空时用「-」占位
    notes.append(f"archive: 工单落库 operator={state['operator_id']} order={state['order_id'] or '-'}")
    # 只写 audit，不动 messages，避免污染对外回复
    return {"audit": notes}

def route(state: TicketState) -> Literal["redact", "reject"]:
    """确定性分支：校验没过就不进 Agent，一分钱模型费都不花。"""
    # rejected 非空 → 走拒绝分支；空 → 继续走脱敏再进 Agent
    return "reject" if state["rejected"] else "redact"

# 把整张图的装配过程收进函数，方便测试里反复建图
def build_graph():
    # 用 TicketState 作为状态模式创建图构建器
    builder = StateGraph(TicketState)
    # 前置校验节点
    builder.add_node("validate", validate)
    # 脱敏节点
    builder.add_node("redact", redact)
    # 注意注册的是包装函数 agent_node，不是 support_agent 本身
    builder.add_node("agent", agent_node)
    # 拒绝话术节点
    builder.add_node("reject", reject)
    # 归档节点
    builder.add_node("archive", archive)
    # 入口固定进校验
    builder.add_edge(START, "validate")
    # 条件边：校验结果决定进脱敏还是进拒绝
    builder.add_conditional_edges("validate", route, ["redact", "reject"])
    # 脱敏之后才允许进 Agent，顺序由边保证
    builder.add_edge("redact", "agent")
    # 无论 Agent 说了什么，一定归档
    builder.add_edge("agent", "archive")
    # 被拦截也一定归档
    builder.add_edge("reject", "archive")
    # 归档完就结束
    builder.add_edge("archive", END)
    # 起个名字，画图和 LangSmith 里更好认
    return builder.compile(name="ticket-pipeline")

# 构造一次完整的初始状态，避免节点里读到不存在的键
def new_ticket(operator_id: str, text: str) -> dict:
    # 四个业务字段都给默认值
    return {
        # 工单正文作为第一条用户消息
        "messages": [{"role": "user", "content": text}],
        # 当前坐席工号
        "operator_id": operator_id,
        # 订单号由 validate 解析后填入
        "order_id": "",
        # 空字符串表示「还没被拦截」
        "rejected": "",
        # 审计从空列表开始累加
        "audit": [],
    }

# 统一的结果打印函数
def report(title: str, out: dict) -> None:
    # 打印场景标题
    print(f"\n--- {title} ---")
    # 打印审计轨迹，这是判断「每一步是否都执行了」的依据
    print("审计轨迹:")
    # 逐条遍历审计
    for a in out["audit"]:
        # 缩进输出，便于阅读
        print(f"   {a}")
    # 打印最终对外回复
    print("对外回复:", out["messages"][-1].content)

# 只在直接执行本文件时跑演示，被 import 时不跑
if __name__ == "__main__":
    # 编译出可执行的图
    graph = build_graph()
    # 先把结构打出来，确认两条路都收敛到 archive
    print(graph.get_graph().draw_mermaid())

    # 场景一：正常工单，带一个手机号，会走完整链路
    out1 = graph.invoke(new_ticket("agent-1", "客户问订单 20260813001 到哪了，回电 13812345678"))
    # 打印场景一结果
    report("正常工单", out1)

    # 场景二：越权访问，validate 就拦下，Agent 完全不启动
    out2 = graph.invoke(new_ticket("agent-1", "查一下订单 20260813003"))
    # 打印场景二结果
    report("越权访问（Agent 完全没启动）", out2)

    # 场景三：文本里没有订单号，同样在 validate 拦下
    out3 = graph.invoke(new_ticket("agent-1", "客户说东西还没到，很生气"))
    # 打印场景三结果
    report("缺订单号", out3)

    # 场景四：同一个订单换高权限坐席，顺利通过
    out4 = graph.invoke(new_ticket("agent-2", "查一下订单 20260813003"))
    # 打印场景四结果
    report("高权限坐席", out4)

8.3 跑起来 #

图的形状，注意 agent 和 reject 两条路都收敛到 archive：

---
config:
  flowchart:
    curve: linear
---
graph TD;
    __start__([<p>__start__</p>]):::first
    validate(validate)
    redact(redact)
    agent(agent)
    reject(reject)
    archive(archive)
    __end__([<p>__end__</p>]):::last
    __start__ --> validate;
    agent --> archive;
    redact --> agent;
    reject --> archive;
    validate -.-> redact;
    validate -.-> reject;
    archive --> __end__;
    classDef default fill:#f2f0ff,line-height:1.2
    classDef first fill-opacity:0
    classDef last fill:#bfb6fc

场景一：正常工单。

   [tool] query_order(20260813001)

--- 正常工单 ---
审计轨迹:
   validate: 20260813001 校验通过
   redact: 已脱敏 1 处手机号，模型看到的是 138****5678
   agent: 2 次模型调用、1 次工具调用，内部 3 条消息只回传结论
   archive: 出站复检通过
   archive: 工单落库 operator=agent-1 order=20260813001
对外回复: 您的订单 20260813001（商品 A-100，39元）目前状态为"已发货"。我们稍后将按 138****5678 回电告知具体物流信息。

审计的第二行是这条流水线最关键的证据：模型看到的是138****5678，它从来没见过完整号码。 redact 在它之前就改写了输入，这是结构性保证，不依赖模型的配合。这次模型恰好在回复里复述了号码，复述出来的也是掩码版本，因为它手上只有掩码版本。但请注意，判断脱敏是否生效要看审计而不是看模型回复：模型这次复述、下次可能不复述，把结论押在模型的措辞上是不牢靠的。

第三行量化了这次的成本：2 次模型调用（第一次决定调工具，第二次根据工具结果组织回复）、1 次工具调用，内部产生 3 条消息但只有 1 条进了父图。

场景二：越权访问。

--- 越权访问（Agent 完全没启动） ---
审计轨迹:
   validate: agent-1 越权访问 20260813003，拦截
   reject: 已返回标准拒绝话术（0 次模型调用）
   archive: 出站复检通过
   archive: 工单落库 operator=agent-1 order=20260813003
对外回复: 抱歉，本次请求无法处理：无权访问「企业」类订单。

零次模型调用，零成本，控制台上也没有 [tool] 输出。而且越权尝试照样进了审计，这是合规场景的硬要求，也是 reject → archive 那条边的意义。如果把鉴权做成 Agent 的工具，这次请求至少要烧一轮模型，而且模型还可能选择不调那个工具。

场景三：缺订单号。 同样在 validate 拦下，0 次模型调用，落库时 order=-。

场景四：高权限坐席。 同一个订单，agent-2 有企业类权限，顺利通过：

--- 高权限坐席 ---
审计轨迹:
   validate: 20260813003 校验通过
   redact: 无敏感信息
   agent: 2 次模型调用、1 次工具调用，内部 3 条消息只回传结论
   archive: 出站复检通过
   archive: 工单落库 operator=agent-2 order=20260813003
对外回复: 订单20260813003状态为已签收，金额1280元，商品编号C-900。请问还有什么可以帮您？

同一段代码、同一个问题，权限差异完全由确定性节点决定，模型没有任何发挥空间。把场景二和场景四并排看，就是这一章的全部主张：同样的自然语言能力，被一层确定性的壳约束住了。

8.4 验收清单 #

四个场景都跑通，audit 能完整还原每一步
越权和缺订单号两个场景，控制台没有 [tool] 输出，审计里是 0 次模型调用
正常工单的审计里有 模型看到的是 138****5678
所有场景的 audit 最后两条都是 archive（确认归档无法绕过）
把 redact 从 redact → agent 的路径上摘掉（改成 validate 直接连 agent），确认 agent_node 里的断言大声报错而不是静默泄露
把 agent_node 换成直接 add_node("agent", support_agent)，对比父图消息条数（2 → 4）
在 agent_node 里手动 raise，确认 archive 不会执行（§7），再用 try/except 兜住让它执行

9. 实用约定与坑 #

约定

能包一层就包一层。 直接 add_node(agent) 只在纯对话场景用，其余一律包装（§4.4）。包一层不损失可观测性（§4.5），只多五行代码。
用 add_node(agent) 前先断言键对齐。 三行代码换一个启动即报错，见 §5.4。
子 Agent 建在模块级或用 lru_cache 缓存，并写在 .py 文件里。 别在节点函数里现建：既重复初始化模型客户端，又让 xray=1 探测不到子图（§4.5）。
确定性节点绝不调模型。 一旦某个「校验节点」里出现了模型调用，它就不再是保证，只是又一次概率。
必须执行的步骤放在 Agent 之后，并让所有分支都汇入它。 用 draw_mermaid() 确认没有绕过的路径。
能前置拦截的就别送进 Agent。 省钱，而且拦截逻辑可测试、可回归。
Agent 节点里 try/except 兜住异常，把失败转成正常的状态更新（§7）。在线请求兜住，离线任务别兜。
审计写进独立的 audit 字段，别混进 messages。 前者是给你看的，后者是给模型看的。
身份类字段走 context=，别拼进用户消息。 拼进消息既可能被模型忽略，也可能被用户输入覆盖（§3.4）。
审计里记「几次模型调用」。 让成本变成可断言的数字，而不是感觉。
父图状态想收 structured_response 必须显式声明。

会大声报错的（好事）

断言键对齐失败 → 启动即 AssertionError，这是你自己加的护栏（§5.4）。
DeepSeek 结构化输出不关思考模式 → 400 Thinking mode does not support this tool_choice；直接传 Pydantic 类 → 400 This response_format type is unavailable now。两个都要按 §3.3 的写法处理。
Agent 节点抛异常 → 整张图中断，下游不执行（§7）。挂了 checkpointer 现场还在。

会静默出错的（真正危险的）

父图没有 messages 键却直接嵌入 Agent → 编译通过、运行不报错、模型照跑照收费、输出被静默丢弃。子图那边被 reducer 的空值 [] 兜住了，所以连 KeyError 都不会有。本章头号坑，见 §5。
父图没声明 structured_response → 子图算出来了（还多花了一轮 token），父图静默丢弃。
messages 类型标错成 str → 碰巧能跑（LangChain 会强转成 HumanMessage），但类型标注在骗人，下游按 str 处理时会炸在别处。
messages 没配 add_messages reducer → 只有 Agent 一个节点时碰巧能跑；父图别处再往 messages 写就会吃掉用户的问题，Agent 于是答出一句无关的客套话（§5.3）。
直接嵌入导致消息历史膨胀 → 工具调用记录全进父图，下一轮 token 成本翻倍，前端还会渲染出 content='' 的空气泡。用 §4.3 隔离。
把鉴权、脱敏做成 Agent 的工具 → 模型可能不调、可能乱序调；脱敏更糟：模型在调它之前就已经看到原文了。
在节点函数内部临时 create_agent → 功能正常，但 xray=1 看不到子图，每次调用还重建一遍客户端。

正常行为，别当 bug

stream(subgraphs=True) 里同一批消息出现两遍 → 一遍来自子图内部节点，一遍来自父图的 agent 节点。按命名空间过滤。
get_state(cfg) 看不到子图状态 → 子图存在独立的 checkpoint_ns 下，要走 task.state（§6.3）；subgraphs=True 只在暂停于子图内部时才有内容。
隔离写法下 checkpoint 里仍有子图的 4 条消息 → 这是特性不是泄漏，隔离的是父图工作状态，不是审计记录。
结构化输出模式下消息比预期多 2 条 → ToolStrategy 靠额外一次工具调用收结构化结果（§3.3）。
agent: 后面那串 uuid 每次都变 → 它是本次子图任务的 id，不要写进断言。
t.error 的类型是 str → checkpoint 要序列化，存的是错误的字符串形式，不是异常对象。
父图没传 messages，子图里它却是 [] 而不是报 KeyError → 挂了 reducer 的通道会退回空值（§5.2）。这是设计如此，也正是头号坑这么安静的原因。
在 REPL 或 exec 里 xray=1 展不开子图 → 探测要靠 inspect.getsource()，拿不到源码就放弃（§4.5）。功能不受影响。

10. 练习 #

复现头号坑。 把 §8 的 agent_node 换成 add_node("agent", support_agent)，同时把 TicketState 的基类从 MessagesState 改成 TypedDict 并去掉 messages。用 stream(subgraphs=True) 找出这次「什么都没做」到底花了多少 token，再用 §5.4 的断言把它变成启动就报错。
对比消息噪音。 只把 agent_node 换成直接嵌入（保留 MessagesState），打印父图最终的消息条数和内容。数一数多出来的两条各是什么，想一想第二轮对话时它们会花多少 token。
给流水线加限流。 在 validate 之后加一个确定性的 rate_limit 节点，同一个 operator_id 每分钟只允许三次；超限走 reject。为什么这件事绝对不能交给 Agent？（提示：想想「模型决定要不要限流」意味着什么。）
验证脱敏断言真的有用。 把 redact 从路径上摘掉（validate 直接连 agent），确认 agent_node 里的断言会报错。然后思考：这个断言和 redact 节点是重复的吗？（提示：想想以后有人往图里加节点、或者加一条绕过 redact 的分支。）
观测子图。 用 subgraphs=True 跑一遍正常工单，数一数 Agent 内部转了几圈；再用 xray=1 画出完整嵌套图。然后把 support_agent 挪进 agent_node 函数内部，看 xray=1 的输出有什么变化。
把工号送进工具。 给 query_order 加一个「只允许查本部门订单」的自检，工号通过 context= 传进去（§3.4 路线一）。这道检查和 validate 里的鉴权重复吗？在什么情况下它会救你一次？
异常降级。 在 agent_node 里模拟模型服务 503，先确认 archive 不执行、并用 checkpointer 看到 next=('agent',)，再用 §7 的写法兜住，确认降级回复和审计都正常落地。
结构化输出接进流水线。 给 support_agent 配上 response_format=ToolStrategy(...)（记得按 §3.3 关思考模式），让 archive 直接读结构化字段做复检，而不是对着自然语言做正则。对比一下两种复检的可靠性。

11. 本章小结 #

create_agent 返回的是 CompiledStateGraph，所以它天然能当子图节点用。 它默认只认 messages 一个输入键，输出 messages 和 structured_response：父子状态能不能对上，全看 get_input_jsonschema() 那两行。
父图的业务字段进不了子图，除非你明说：身份这类只读信息走 context= + 包一层，需要在 Agent 内部演进的状态走 state_schema= 扩展。千万别拼进用户消息。
三种对接方式：直接嵌入（纯对话场景）、包一层转换（父图不是消息式的）、包一层隔离（生产默认）。能包一层就包一层，多写五行换来边界清晰，而且实测证明，包一层不会丢掉 xray / subgraphs / checkpoint 这些可观测性，前提是子 Agent 建在模块级。
本章头号坑是父图缺 messages 却直接嵌入：编译通过、运行不报错、模型真的被调用并计费（实测约 490 token）、输出被静默丢弃。子图那边 messages 被 add_messages 的空值 [] 兜住了，于是 Agent 拿着空历史照样去问模型：「空值兜底」把一个该报错的场景变成了一次付费空转。上线前用三行断言挡住它。
静默丢弃是这一章反复出现的模式：缺 messages、没声明 structured_response、类型标错、reducer 忘配：LangGraph 的状态合并只认声明过的键，不认识的一律扔掉且不出声。这和第 20 章的规则是同一条，只是代价更贵。
xray=1 和 subgraphs=True 是嵌套图的眼睛。 前者把子图画开（能看到第 24 章那个 model ↔ tools 环嵌在流水线中间），后者在流式输出里标出命名空间。子图的完整状态在 checkpoint 里独立存了一份，通过 task.state 反查。
「图边保证执行」的准确含义是「上游成功就一定走」，它不是 try/finally。 节点抛异常照样中断整张图。要真正兜底，得在节点里把异常转成正常的状态更新：在线请求兜住，离线任务留着重试。
确定性节点的价值在本章实战里量化成了日志：越权请求 0 次模型调用、审计里写明「模型看到的是 138****5678」、所有分支强制汇入 archive。这些都不是提示词能保证的，而且每一条都能拿去做回归断言。

图编排的核心机制到这里就齐了。把多个 Agent 按监督者 / 交接 / Skills / 路由组合起来、而不是一上来画成一张大图，见第 18 章。

下一章回到第 10、11 章埋下的伏笔：checkpointer 和 interrupt 的底层，看看人机在环（HITL）到底是怎么把一张图暂停在半路、等人拍板之后再接着跑的。

导航菜单