人机协作

某些工具操作可能比较敏感，需要在执行前获得人工批准。Deep Agents 通过 LangGraph 的中断功能支持人机协作工作流。您可以使用 interrupt_on 参数来配置哪些工具需要审批。当设置了 interrupt_on 时，HumanInTheLoopMiddleware 会被添加到默认中间件栈中。如果在工具返回结果之前运行被取消或中断，同一栈中的 PatchToolCallsMiddleware 会自动修复消息历史。

基本配置

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

True：启用中断并使用默认行为（允许批准、编辑、拒绝、回复）
False：禁用该工具的中断
InterruptOnConfig：自定义配置。设置 allowed_decisions 以控制审查选项。在 Python 中，可以添加一个可选的 when 谓词来仅中断特定调用（参见条件中断）。

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="openai:gpt-5.5",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

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


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


@tool
def fetch_file(path: str) -> str:
    """从文件系统中读取一个文件。"""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """发送一封电子邮件。"""
    return f"Sent email to {to}"


# 检查点器是实现人机交互所必需的
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # 默认：批准、编辑、拒绝、回复
        "fetch_file": False,  # 无需中断
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
    },
    checkpointer=checkpointer,  # 必需！
)

决策类型

allowed_decisions 列表控制人工审查工具调用时可以采取的操作：

决策类型	描述	示例用例
✅ `approve`	使用代理提出的原始参数执行工具。	完全按照所写内容发送电子邮件草稿
✏️ `edit`	在执行前修改工具参数。	发送前更改收件人
❌ `reject`	完全跳过执行此工具调用，并向代理返回拒绝反馈。	拒绝文件删除并说明原因
💬 `respond`	将人类的消息直接作为合成工具结果返回，跳过执行，适用于“询问用户”类工具。	直接回复 `"ask_user"` 提示

当人工拒绝提议的操作时使用 reject。仅当人工扮演工具角色时使用 respond，例如回答 ask_user 提示。不要使用 respond 来拒绝具有副作用的工具，因为模型可能会将其消息视为成功的工具结果。

编辑工具参数时，请保守地进行更改。对原始参数进行重大修改可能会导致模型重新评估其方法，并可能多次执行工具或采取意外操作。

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

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

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

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

条件中断

默认情况下，interrupt_on 中列出的每个工具调用都会暂停以进行审查。要仅暂停某些调用，请在工具的 InterruptOnConfig 中添加 when 谓词。该谓词接收一个 ToolCallRequest 并返回 True 以中断或 False 以自动批准，因此您可以根据工具的参数进行控制。

条件中断需要 langchain>=1.3.3。

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="openai:gpt-5.5",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """暂停对工作空间目录外路径的写入操作。"""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,
        },
    },
    checkpointer=MemorySaver(),
)

当 when 谓词返回 False 时，调用将运行而不中断。当它返回 True 时，或者当您省略 when 时，调用将照常暂停。计算结果为 False 的调用永远不会被添加到中断批次中，因此审查者只会看到需要决策的操作。更多配置选项和示例，请参见 LangChain 人机协作文档。

处理中断

当触发中断时，代理会暂停执行并返回控制权。检查结果中的中断并相应处理。如果用户拒绝某个操作，请包含一个明确的 message，告知代理该工具未执行以及下一步该怎么做。

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": "删除文件 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"工具: {action['name']}")
        print(f"参数: {action['args']}")
        print(f"允许的决策: {review_config['allowed_decisions']}")

    # 获取用户决策（每个 action_request 一个，按顺序）
    decisions = [
        {
            "type": "reject",
            "message": "用户拒绝删除 temp.txt。请勿重试删除操作。",
        }
    ]

    # 使用决策恢复执行
    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": "删除 temp.txt 并向 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",
            "message": "用户拒绝此操作。请勿重试此工具调用。",
        }  # 第二个工具: send_email
    ]

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

拒绝消息

当审查者返回 reject 决策时，Deep Agents 会跳过工具调用并将拒绝反馈发送回代理。如果您省略 message，默认反馈会告知模型该工具未执行，除非用户要求，否则不要重试相同的工具调用。对于敏感或具有副作用的工具，请在决策中传递特定领域的 message。明确说明代理是应该放弃该操作、提出后续问题还是尝试更安全的替代方案。

decisions = [
    {
        "type": "reject",
        "message": "用户拒绝删除此文件。请勿重试删除。改为询问要归档哪个文件。",
    }
]

编辑工具参数

当 "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.6-flash",
    tools=[delete_file, read_file],
    interrupt_on={
        "delete_file": True,
        "read_file": False,
    },
    subagents=[{
        "name": "file-manager",
        "description": "管理文件操作",
        "system_prompt": "你是一个文件管理助手。",
        "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="在继续执行操作前请求人工批准。")
def request_approval(action_description: str) -> str:
    """使用 interrupt() 原语请求人工批准。"""
    # interrupt() 暂停执行并返回传递给 Command(resume=...) 的值
    approval = interrupt({
        "type": "approval_request",
        "action": action_description,
        "message": f"请批准或拒绝：{action_description}",
    })

    if approval.get("approved"):
        return f"操作 '{action_description}' 已被批准。继续执行..."
    else:
        return f"操作 '{action_description}' 已被拒绝。原因：{approval.get('reason', '未提供原因')}"


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.6-flash",
        checkpointer=checkpointer,
        subagents=[
            CompiledSubAgent(
                name="approval-agent",
                description="一个可以请求批准的代理",
                runnable=compiled_subagent,
            )
        ],
    )

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

    print("调用代理 - 子代理将使用 request_approval 工具...")

    result = parent_agent.invoke(
        {
            "messages": [
                HumanMessage(
                    content="使用任务工具启动 approval-agent 子代理。"
                    "告诉它使用 request_approval 工具为'部署到生产环境'请求批准。"
                )
            ]
        },
        config=config,
        version="v2",
    )

    # 检查中断
    if result.interrupts:
        interrupt_value = result.interrupts[0].value  
        print(f"\n收到中断！")
        print(f"  类型: {interrupt_value.get('type')}")
        print(f"  操作: {interrupt_value.get('action')}")
        print(f"  消息: {interrupt_value.get('message')}")

        print("\n使用 Command(resume={'approved': True}) 恢复...")
        result2 = parent_agent.invoke(
            Command(resume={"approved": True}),
            config=config,
            version="v2",
        )

        if not result2.interrupts:
            print("\n执行完成！")
            # 查找工具响应
            tool_msgs = [m for m in result2.value.get("messages", []) if m.type == "tool"]
            if tool_msgs:
                print(f"  工具结果: {tool_msgs[-1].content}")
        else:
            print("\n发生了另一个中断")
    else:
        print("\n  没有中断 - 模型可能没有调用 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.

文件系统权限中断

文件系统权限中断需要 deepagents>=0.6.8。

除了 interrupt_on，您还可以通过将权限规则标记为 mode="interrupt" 来暂停内置的文件系统工具。当代理在与中断模式规则匹配的路径上调用 write_file 或 edit_file 时，create_deep_agent 会引发与已配置工具相同的人机协作中断，并使用文件系统工具的名称作为操作名称。

from deepagents import FilesystemPermission, create_deep_agent
from langgraph.checkpoint.memory import MemorySaver


agent = create_deep_agent(
    model=model,
    permissions=[
        FilesystemPermission(
            operations=["write"],
            paths=["/secrets/**"],
            mode="interrupt",
        ),
    ],
    checkpointer=MemorySaver(),  # 需要用于暂停和恢复
)

处理并恢复中断的方式与工具调用中断相同：运行直到暂停，检查请求，然后使用决策恢复。

from langgraph.types import Command

config = {"configurable": {"thread_id": "fs-thread-1"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "将 API 密钥保存到 /secrets/key.txt"}]},
    config=config,
    version="v2",
)

if result.interrupts:
    action = result.interrupts[0].value["action_requests"][0]
    print(f"批准 {action['name']} 于 {action['args']}？")

    # 使用人工决策（批准、编辑或拒绝）恢复。
    result = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,  # 相同的线程 ID
        version="v2",
    )

文件系统权限中断会与您传递的任何 interrupt_on 合并，因此单个审查步骤可以同时覆盖自定义工具和受保护的文件系统路径。

最佳实践

始终使用检查点

人机协作需要一个检查点来在中断和恢复之间持久化代理状态：

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    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,
    "ls": False,
}

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Deployment

Execution environment

Context management

Delegation

Steering

Frontend

Protocols

基本配置

决策类型

条件中断

处理中断

多个工具调用

拒绝消息

编辑工具参数

子代理中断

工具调用时的中断

工具调用内部的中断

文件系统权限中断

最佳实践

始终使用检查点

使用相同的线程 ID

匹配决策顺序与操作

根据风险定制配置

​基本配置

​决策类型

​条件中断

​处理中断

​多个工具调用

​拒绝消息

​编辑工具参数

​子代理中断

​工具调用时的中断

​工具调用内部的中断

​文件系统权限中断

​最佳实践

​始终使用检查点

​使用相同的线程 ID

​匹配决策顺序与操作

​根据风险定制配置

基本配置

决策类型

条件中断

处理中断

多个工具调用

拒绝消息

编辑工具参数

子代理中断

工具调用时的中断

工具调用内部的中断

文件系统权限中断

最佳实践

始终使用检查点

使用相同的线程 ID

匹配决策顺序与操作

根据风险定制配置