某些工具操作具有敏感性，需要在执行前获得人工批准。深度智能体通过 LangGraph 的中断能力支持人在回路的工作流。您可以使用 interrupt_on 参数配置哪些工具需要审批。

​
基本配置

interrupt_on 参数接收一个字典，用于映射工具名称到中断配置。每个工具可以配置为：
  • True：启用默认中断行为（允许批准、编辑、拒绝）
  • False：禁用该工具的中断
  • {"allowed_decisions": [...]}：自定义允许的决策
from langchain_core.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

@tool
def delete_file(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"

@tool
def read_file(path: str) -> str:
    """Read a file from the filesystem."""
    return f"Contents of {path}"

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

# 人在回路必须配置检查点
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[delete_file, read_file, send_email],
    interrupt_on={
        "delete_file": True,  # 默认：允许批准、编辑、拒绝
        "read_file": False,   # 不需要中断
        "send_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer  # 必须！
)

​
决策类型

allowed_decisions 列表控制人工在审查工具调用时可执行的操作：
  • "approve"：按智能体原样执行工具
  • "edit"：在执行前修改工具参数
  • "reject"：跳过此次工具调用
您可以为每个工具自定义可用的决策： 
interrupt_on = {
    # 高风险操作：允许所有选项
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},

    # 中等风险：仅允许批准或拒绝
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # 必须执行：仅允许批准
    "critical_operation": {"allowed_decisions": ["approve"]},
}

​
处理中断

当触发中断时，智能体会暂停执行并将控制权交还给您。检查返回结果中的中断信息，并按需处理。 
import uuid
from langgraph.types import Command

# 配置 thread_id 以持久化状态
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 调用智能体
result = agent.invoke({
    "messages": [{"role": "user", "content": "Delete the file temp.txt"}]
}, config=config)

# 检查是否发生中断
if result.get("__interrupt__"):
    # 提取中断信息
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]
    review_configs = interrupts["review_configs"]

    # 构建工具名称到审批配置的映射
    config_map = {cfg["action_name"]: cfg for cfg in review_configs}

    # 向用户展示待审批的操作
    for action in action_requests:
        review_config = config_map[action["name"]]
        print(f"Tool: {action['name']}")
        print(f"Arguments: {action['args']}")
        print(f"Allowed decisions: {review_config['allowed_decisions']}")

    # 获取用户决策（与 action_requests 顺序一致）
    decisions = [
        {"type": "approve"}  # 用户批准删除操作
    ]

    # 携带决策恢复执行
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config  # 必须使用相同的 config！
    )

# 处理最终结果
print(result["messages"][-1]["content"])

​
多个工具调用

当智能体调用多个需要审批的工具时，所有中断会合并到同一次暂停中。必须按顺序为每个工具提供决策。 
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "Delete temp.txt and send an email to admin@example.com"
    }]
}, config=config)

if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]

    # 此处有两个工具需要审批
    assert len(action_requests) == 2

    # 按照 action_requests 的顺序提供决策
    decisions = [
        {"type": "approve"},  # 第一个工具：delete_file
        {"type": "reject"}    # 第二个工具：send_email
    ]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

​
编辑工具参数

当允许 "edit" 决策时，可以在执行前修改工具参数： 
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_request = interrupts["action_requests"][0]

    # 智能体给出的原始参数
    print(action_request["args"])  # {"to": "everyone@company.com", ...}

    # 用户决定修改收件人
    decisions = [{
        "type": "edit",
        "edited_action": {
            "name": action_request["name"],  # 必须包含工具名称
            "args": {"to": "team@company.com", "subject": "...", "body": "..."}
        }
    }]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

​
子智能体中的中断

每个子智能体都可以拥有自己的 interrupt_on 配置，用于覆盖主智能体的设置： 
agent = create_deep_agent(
    tools=[delete_file, read_file],
    interrupt_on={
        "delete_file": True,
        "read_file": False,
    },
    subagents=[{
        "name": "file-manager",
        "description": "Manages file operations",
        "system_prompt": "You are a file management assistant.",
        "tools": [delete_file, read_file],
        "interrupt_on": {
            # 重写配置：在子智能体中要求读取操作审批
            "delete_file": True,
            "read_file": True,  # 与主智能体配置不同！
        }
    }],
    checkpointer=checkpointer
)
当子智能体触发中断时，处理方式相同——检查 __interrupt__，并使用 Command 携带决策恢复执行。

​
最佳实践

​
始终使用检查点

人在回路场景需要检查点来在中断与恢复之间持久化状态： 
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    tools=[...],
    interrupt_on={...},
    checkpointer=checkpointer  # HITL 必须配置检查点
)

​
使用相同的线程 ID

恢复执行时必须使用包含相同 thread_id 的配置： 
# 第一次调用
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config)

# 恢复执行（仍然使用相同 config）
result = agent.invoke(Command(resume={...}), config=config)

​
决策顺序要与动作一致

decisions 列表必须与 action_requests 的顺序一一对应： 
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"][0].value
    action_requests = interrupts["action_requests"]

    # 为每个 action 创建一个决策，保持顺序一致
    decisions = []
    for action in action_requests:
        decision = get_user_decision(action)  # 自定义逻辑
        decisions.append(decision)

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

​
根据风险分级配置

根据风险等级为不同工具配置不同的中断策略： 
interrupt_on = {
    # 高风险：提供完整控制（批准、编辑、拒绝）
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
    "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},

    # 中风险：不允许编辑
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # 低风险：无需中断
    "read_file": False,
    "list_files": False,
}
Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.