人机交互

某些工具操作可能很敏感，需要在执行前获得人工批准。Deep Agents 通过 LangGraph 的中断功能支持人机交互工作流。你可以使用 interrupt_on 参数配置哪些工具需要审批。

基本配置

interrupt_on 参数接受一个字典，将工具名称映射到中断配置。每个工具可以配置为：

True：启用中断，使用默认行为（允许批准、编辑、拒绝、回复）
False：禁用此工具的中断
{"allowed_decisions": [...]}：自定义配置，指定允许的决策

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

@tool
def delete_file(path: str) -> str:
    """从文件系统中删除文件。"""
    return f"已删除 {path}"

@tool
def read_file(path: str) -> str:
    """从文件系统中读取文件。"""
    return f"{path} 的内容"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件。"""
    return f"已向 {to} 发送电子邮件"

# Human in the Loop功能必须使用检查点保存器
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="google_genai:gemini-3.1-pro-preview",
    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"：完全跳过执行此工具调用
"respond"：将人工的消息直接作为工具结果返回，跳过执行——适用于“询问用户”类型的工具

你可以为每个工具自定义可用的决策：

interrupt_on = {
    # 敏感操作：允许所有选项
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},

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

    # 必须批准（不允许拒绝）
    "critical_operation": {"allowed_decisions": ["approve"]},
}

处理中断

当中断被触发时，代理会暂停执行并交还控制权。检查结果中的中断并相应处理。

from langchain_core.utils.uuid import uuid7
from langgraph.types import Command

# 创建带有 thread_id 的配置以实现状态持久化
config = {"configurable": {"thread_id": str(uuid7())}}

# 调用代理
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Delete the file temp.txt"}]},
    config=config,
    version="v2",
)

# 检查执行是否被中断
if result.interrupts:
    # 提取中断信息
    interrupt_value = result.interrupts[0].value  
    action_requests = interrupt_value["action_requests"]
    review_configs = interrupt_value["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_request 一个，按顺序）
    decisions = [
        {"type": "approve"}  # 用户批准了删除操作
    ]

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

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

多个工具调用

当代理调用多个需要审批的工具时，所有中断会批量包含在单个中断中。你必须按顺序为每个中断提供决策。

config = {"configurable": {"thread_id": str(uuid7())}}

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

if result.interrupts:
    interrupt_value = result.interrupts[0].value  
    action_requests = interrupt_value["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,
        version="v2",
    )

编辑工具参数

当 "edit" 在允许的决策中时，你可以在执行前修改工具参数：

if result.interrupts:
    interrupt_value = result.interrupts[0].value  
    action_request = interrupt_value["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,
        version="v2",
    )

子代理中断

使用子代理时，你可以使用工具调用上的中断和工具调用内的中断。

工具调用上的中断

每个子代理可以有自己的 interrupt_on 配置，覆盖主代理的设置：

agent = create_deep_agent(
    model="google_genai:gemini-3.1-pro-preview",
    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
)

当子代理触发中断时，处理方式相同——检查结果中的 interrupts 并使用 Command 恢复。

工具调用内的中断

子代理工具可以直接调用 interrupt() 来暂停执行并等待审批：

from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt

from deepagents.graph import create_deep_agent
from deepagents.middleware.subagents import CompiledSubAgent


@tool(description="Request human approval before proceeding with an action.")
def request_approval(action_description: str) -> str:
    """使用 interrupt() 原语请求人工批准。"""
    # interrupt() 暂停执行并返回传递给 Command(resume=...) 的值
    approval = interrupt({
        "type": "approval_request",
        "action": action_description,
        "message": f"Please approve or reject: {action_description}",
    })

    if approval.get("approved"):
        return f"Action '{action_description}' was APPROVED. Proceeding..."
    else:
        return f"Action '{action_description}' was REJECTED. Reason: {approval.get('reason', 'No reason provided')}"


def main():
    checkpointer = InMemorySaver()
    model = ChatAnthropic(
        model_name="claude-sonnet-4-6",
        max_tokens=4096,
    )

    compiled_subagent = create_agent(
        model=model,
        tools=[request_approval],
        name="approval-agent",
    )

    parent_agent = create_deep_agent(
        model="google_genai:gemini-3.1-pro-preview",
        checkpointer=checkpointer,
        subagents=[
            CompiledSubAgent(
                name="approval-agent",
                description="An agent that can request approvals",
                runnable=compiled_subagent,
            )
        ],
    )

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

    print("Invoking agent - sub-agent will use request_approval tool...")

    result = parent_agent.invoke(
        {
            "messages": [
                HumanMessage(
                    content="Use the task tool to launch the approval-agent sub-agent. "
                    "Tell it to use the request_approval tool to request approval for 'deploying to production'."
                )
            ]
        },
        config=config,
        version="v2",
    )

    # 检查中断
    if result.interrupts:
        interrupt_value = result.interrupts[0].value  
        print(f"\nInterrupt received!")
        print(f"  Type: {interrupt_value.get('type')}")
        print(f"  Action: {interrupt_value.get('action')}")
        print(f"  Message: {interrupt_value.get('message')}")

        print("\nResuming with Command(resume={'approved': True})...")
        result2 = parent_agent.invoke(
            Command(resume={"approved": True}),
            config=config,
            version="v2",
        )

        if not result2.interrupts:
            print("\nExecution completed!")
            # 查找工具响应
            tool_msgs = [m for m in result2.value.get("messages", []) if m.type == "tool"]
            if tool_msgs:
                print(f"  Tool result: {tool_msgs[-1].content}")
        else:
            print("\nAnother interrupt occurred")
    else:
        print("\n  No interrupt - the model may not have called request_approval")


if __name__ == "__main__":
    main()

运行时，这会产生以下输出：

Invoking agent - sub-agent will use request_approval tool...

Interrupt received!
  Type: approval_request
  Action: deploying to production
  Message: Please approve or reject: deploying to production

Resuming with Command(resume={'approved': True})...

Execution completed!
  Tool result: Great! The approval request has been processed. The action **"deploying to production"** was **APPROVED**. You can now proceed with the production deployment.

最佳实践

始终使用检查点存储器

人机交互需要检查点存储器来在中断和恢复之间持久化代理状态：

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    model="google_genai:gemini-3.1-pro-preview",
    tools=[...],
    interrupt_on={...},
    checkpointer=checkpointer  # HITL 所必需
)

使用相同的线程 ID

恢复时，你必须使用具有相同 thread_id 的相同配置：

# 第一次调用
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config, version="v2")

# 恢复（使用相同的配置）
result = agent.invoke(Command(resume={...}), config=config, version="v2")

匹配决策顺序与操作

决策列表必须与 action_requests 的顺序匹配：

if result.interrupts:
    interrupt_value = result.interrupts[0].value  
    action_requests = interrupt_value["action_requests"]

    # 为每个操作创建一个决策，按顺序
    decisions = []
    for action in action_requests:
        decision = get_user_decision(action)  # 你的逻辑
        decisions.append(decision)

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

根据风险定制配置

根据风险级别配置不同的工具：

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,
}

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交问题。

Get started

Deployment

Core capabilities

Frontend

Protocols

Command line interface

基本配置

决策类型

处理中断

多个工具调用

编辑工具参数

子代理中断

工具调用上的中断

工具调用内的中断

最佳实践

始终使用检查点存储器

使用相同的线程 ID

匹配决策顺序与操作

根据风险定制配置

​基本配置

​决策类型

​处理中断

​多个工具调用

​编辑工具参数

​子代理中断

​工具调用上的中断

​工具调用内的中断

​最佳实践

​始终使用检查点存储器

​使用相同的线程 ID

​匹配决策顺序与操作

​根据风险定制配置

基本配置

决策类型

处理中断

多个工具调用

编辑工具参数

子代理中断

工具调用上的中断

工具调用内的中断

最佳实践

始终使用检查点存储器

使用相同的线程 ID

匹配决策顺序与操作

根据风险定制配置