LangGraph interrupt() 新增 response_schema：让人工审批输入在恢复前过一道结构化校验

pip install -U "langgraph>=1.2.12"
python -c "from importlib.metadata import version; print(version('langgraph'))"

确认版本不低于 1.2.12 后，搭一个最小审批图作为基线：一个节点在发布前调用 interrupt() 停车等人工审批，人工的恢复值直接写进图状态。旧版行为是恢复值原样返回——前端把 approved 传成 "yes"（字符串）或干脆漏传，节点代码要么静默拿到脏数据、要么在下游某个比较判断里炸出难看的类型错误。基线先跑一遍感受这个痛点，下一步用 schema 治它。

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt

class State(TypedDict):
    answer: object

def review_node(state: State) -> State:
    # 旧写法：interrupt({"question": "批准发布吗？"})——恢复值不校验
    decision = interrupt({"question": "批准发布这份周报吗？"})
    return {"answer": decision}

builder = StateGraph(State)
builder.add_node("review", review_node)
builder.add_edge(START, "review")
builder.add_edge("review", END)
graph = builder.compile(checkpointer=MemorySaver())

from typing import Optional
from pydantic import BaseModel
from langgraph.types import Command, interrupt

class Approval(BaseModel):
    approved: bool
    note: Optional[str] = None

def review_node(state: State) -> State:
    decision = interrupt(
        {"question": "批准发布这份周报吗？"},
        response_schema=Approval,          # 关键：声明恢复值结构
    )
    # 走到这里说明校验已通过，decision 是 Approval 实例
    return {"answer": decision}

config = {"configurable": {"thread_id": "t1"}}
graph.invoke({"answer": None}, config)                      # 跑到 interrupt 停车

result = graph.invoke(
    Command(resume={"approved": True, "note": "数据口径已核对"}),
    config,
)
print(result["answer"])   # Approval(approved=True, note='数据口径已核对')

变化全在恢复环节：Command(resume={...}) 给出的字典先被 TypeAdapter 按 Approval 校验，通过后校验后的对象作为 interrupt() 的返回值交给节点——你的下游代码从「防御式检查字典字段」变成「直接用强类型实例」，多余字段被剔除、缺字段当场报错。这是 1.2.12 对审批流最大的体验改善：把「人工输入合不合格」从下游节点的责任，上移到了恢复边界。

from dataclasses import dataclass

class ApprovalDict(TypedDict):
    approved: bool

@dataclass
class ApprovalData:
    approved: bool

# 三种强类型写法等价：转 JSON Schema 透出 + 恢复时校验
interrupt("approve?", response_schema=ApprovalDict)
interrupt("approve?", response_schema=ApprovalData)

# 裸 JSON Schema dict：只透出、不校验，返回原始值
RAW = {"type": "object",
       "properties": {"approved": {"type": "boolean"}}}
interrupt("approve?", response_schema=RAW)

官方测试覆盖了五种形态（无 schema、裸字典、Pydantic、TypedDict、dataclass），行为边界很清晰：前一种无约束，裸字典透出但不校验，后三种「透出 + 校验 + 返回实例」三合一。团队落地时建议固化一条规范——审批类 interrupt 一律用 Pydantic 模型（字段约束与自定义校验器能力最全），裸字典只留给「纯展示、下游自己兜底」的场景。另外注意返回值类型差异：Pydantic 形态恢复后拿到的是模型实例，裸字典形态拿到的还是原样字典，下游代码别写混。

# 服务端：把暂停现场整理给前端
pending = graph.get_state(config).tasks[0].interrupts[0]
payload = {
    "id": pending.id,                   # interrupt 的稳定 id
    "value": pending.value,             # 暂停时给的问题上下文
    "schema": pending.response_schema,  # JSON Schema，渲染表单用
}

# 前端提交后：按 id 恢复（比裸值恢复更适合多 interrupt 场景）
graph.invoke(
    Command(resume={payload["id"]: {"approved": True}}),
    config,
)

Interrupt 对象新增的 response_schema 字段让客户端自此可以「自渲染表单」：SDK 侧的线程模型里 interrupts 同样带这个字段（JSON Schema 字典），前端拿它就能生成带类型与必填约束的输入控件，不必前后端各维护一份字段定义。恢复方式也支持两种：Command(resume=值) 与 Command(resume={interrupt_id: 值})——前者适合单 interrupt 的简单流，后者按 id 精确对位，当一个节点里有多个暂停点、或同一线程里可能有多个待恢复项时，按 id 恢复更稳。

def two_step_node(state: State) -> State:
    topic = interrupt("先确认主题")                       # 无 schema：拿原始值
    decision = interrupt("再走审批", response_schema=Approval)  # 有 schema
    return {"answer": [topic, decision]}

# 恢复值按「暂停顺序」逐个消费：invoke 一轮给一个
graph.invoke(Command(resume="Q3 复盘"), config)   # 喂给 topic
graph.invoke(Command(resume={"approved": True}), config)  # 喂给 decision
# 结果: {"answer": ["Q3 复盘", Approval(approved=True)]}

LangGraph 的 interrupt 恢复机制按调用顺序索引对位：同一节点里第 N 次调用的 interrupt()，消费恢复列表里的第 N 个值——这个规则在混搭 schema 时依然成立，每个暂停点各自按自己的 schema 校验自己那份恢复值（官方测试专门验证了「前一个无 schema、后一个带 schema」的组合：前一个返回原始值，后一个校验失败时抛 ValidationError、修正后重试成功）。写多步审批流时给每个 interrupt 的 value 里带上下文标记（如 {"step": "topic"}），前端能准确知道当前在等哪一个输入。

from pydantic import ValidationError

def resume_with_retry(graph, config, max_retry: int = 2):
    pending = graph.get_state(config).tasks[0].interrupts[0]
    for _ in range(max_retry):
        raw = collect_human_input(pending)      # 你的表单采集逻辑
        try:
            return graph.invoke(Command(resume=raw), config)
        except ValidationError as e:
            show_form_error(e)                  # 提示哪里没填对，重新采集
    return escalate_to_admin(pending)           # 超限转人工坐席

生产落地的三件套：错误即反馈——捕获 ValidationError 后把 Pydantic 的错误定位翻译成表单字段红字，让填写者当轮修正，线程状态全程无损；重试有上限——超过次数走升级通道（转人工坐席或管理员代批），别让填错的人无限循环；审计留痕——把每次恢复的原始值、校验结果、操作人记进日志，审批流的可追溯性要求在这里和传统工作流完全一致。升级到 1.2.12 后，把这三件事包成统一的 resume 工具函数，全项目的 interrupt 就都长一个样了。