📚 AI 工程与智能体教程站

简体中文

English

简体中文

English

第 10 章 · 人工审批与条件中断 ​

interrupt_on 让指定工具调用前暂停，checkpointer 是必需的。还可以按条件触发，例如只在写 /workspace 之外的路径时暂停。

机制与设计选择 ​

恢复用 Command(resume=...) 并沿用同一 thread_id。被中断的节点从头重跑，副作用可能重复，外部写操作要幂等。

示例 ​

python
import os
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model=os.environ["MODEL_ID"],
    api_key=os.environ["DEEPAGENT_API_KEY"],
    base_url=os.environ.get("DEEPAGENT_BASE_URL", "https://api.openai.com/v1"),
)

from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from deepagents import create_deep_agent

@tool
def send_email(to: str, subject: str) -> str:
    """Send an email (mock)."""
    return f"sent to {to}: {subject}"

agent = create_deep_agent(
    model=model,
    tools=[send_email],
    interrupt_on={"send_email": True},
    checkpointer=MemorySaver(),
)
config = {"configurable": {"thread_id": "ch10"}}
paused = agent.invoke(
    {"messages": [{"role": "user", "content": "Send an email to ops@example.com with subject 'Deploy OK'."}]},
    config=config,
)
print("Interrupted:", bool(paused["__interrupt__"]))

from langgraph.types import Command
result = agent.invoke(Command(resume=True), config=config, version="v2")
print(result.value["messages"][-1].content)

运行与验收 ​

需要 API Key。验收：第一次 invoke 返回中断（interrupt 非空）；resume=True 后 send_email 被执行。

容易踩的坑 ​

不要把 interrupt 当“事务”：节点重跑意味着外部副作用可能重复。敏感工具建议用条件中断（when 谓词）而不是全量开启，成本更低。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 中断需要什么？

2. 如何恢复？

3. 副作用如何防重复？

4. 全量开启审批划算吗？

动手实践 ​

  1. 把中断条件改成“仅当收件人是外部域名”。
  2. 为一次两个敏感工具调用准备批量审批。
  3. 写一个 reject 并附说明的恢复路径。

官方参考 ​

核验日期：2026-09-17，运行环境：Deep Agents 0.7.15。