人在回路

了解如何为敏感工具操作配置人工批准

某些工具操作可能很敏感，需要在执行前获得人工批准。Deep Agents 通过 LangGraph 的中断能力支持人在回路工作流。你可以使用 interrupt_on 参数配置哪些工具需要批准。设置 interrupt_on 后，HumanInTheLoopMiddleware 会被添加到 Deep Agents 栈中。如果某次运行在工具返回结果之前被取消或中断，同一栈中的 PatchToolCallsMiddleware 会自动修复消息历史。

graph LR
    Agent[Agent] --> Check{Interrupt?}
    Check --> |no| Execute[Execute]
    Check --> |yes| Human{Human}

    Human --> |approve| Execute
    Human --> |edit| Execute
    Human --> |reject| ToolMessage[ToolMessage]
    Human --> |respond| ToolMessage

    Execute --> Agent
    ToolMessage --> Agent

    classDef trigger fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    classDef decision fill:#FDF3FF,stroke:#7E65AE,stroke-width:2px,color:#504B5F
    classDef alert fill:#F8E8E6,stroke:#B27D75,stroke-width:2px,color:#634643

    class Agent trigger
    class Check,Human decision
    class Execute process
    class ToolMessage process

基本配置

interrupt_on 参数接受一个将工具名称映射到中断配置的字典。每个工具都可以用以下方式配置：

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

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
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,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="openai:gpt-5.5",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="anthropic:claude-sonnet-5",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
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,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
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,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
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,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

python

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

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

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

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

# Checkpointer is REQUIRED for human-in-the-loop
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,  # Default: approve, edit, reject, respond
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

决策类型

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

决策类型	描述	示例使用场景
✅ `approve`	按智能体提出的原始参数执行该工具。	按所写内容原样发送邮件草稿
✏️ `edit`	在执行前修改工具参数。	在发送邮件前更改收件人
❌ `reject`	完全跳过执行此工具调用，并向智能体返回拒绝反馈。	拒绝删除文件并说明原因
💬 `respond`	将人工的消息直接作为合成的工具结果返回，跳过执行，适用于 "ask user" 风格的工具。	用直接回复来回应 `"ask_user"` 提示

当人工拒绝某个提议的操作时使用 reject。仅当人工充当工具时（例如回答 ask_user 提示）使用 respond。不要使用 respond 来拒绝有副作用的工具，因为其消息可能被模型视为成功的工具结果。

TIP

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

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

python

interrupt_on = {
    # Sensitive operations: allow all options
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},

    # Moderate risk: approval or rejection only
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # Must approve (no rejection allowed)
    "critical_operation": {"allowed_decisions": ["approve"]},
}

条件中断

默认情况下，interrupt_on 中列出的每次工具调用都会暂停以等待审查。要仅暂停部分调用，请为工具的 InterruptOnConfig 添加 when 谓词。该谓词接收一个 ToolCallRequest，返回 True 表示中断，返回 False 表示自动批准，因此你可以根据工具的参数进行把关。

NOTE

条件中断需要 langchain>=1.3.3。

python

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:
    """Pause writes to paths outside the workspace directory."""
    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(),
)

python

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:
    """Pause writes to paths outside the workspace directory."""
    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(),
)

python

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:
    """Pause writes to paths outside the workspace directory."""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")

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

python

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:
    """Pause writes to paths outside the workspace directory."""
    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(),
)

python

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:
    """Pause writes to paths outside the workspace directory."""
    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(),
)

python

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:
    """Pause writes to paths outside the workspace directory."""
    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(),
)

python

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:
    """Pause writes to paths outside the workspace directory."""
    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，告诉智能体该工具未被执行以及接下来该做什么。

python

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

# Create config with thread_id for state persistence
config = {"configurable": {"thread_id": str(uuid7())}}

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

# Check if execution was interrupted
if result.get("__interrupt__"):  
    # Extract interrupt information
    interrupt_value = result["__interrupt__"][0].value  
    action_requests = interrupt_value["action_requests"]
    review_configs = interrupt_value["review_configs"]

    # Create a lookup map from tool name to review config
    config_map = {cfg["action_name"]: cfg for cfg in review_configs}

    # Display the pending actions to the user
    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']}")

    # Get user decisions (one per action_request, in order)
    decisions = [
        {
            "type": "reject",
            "message": "User rejected deleting temp.txt. Do not retry deletion.",
        }
    ]

    # Resume execution with decisions
    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config  # Must use the same config!
    )

# Process final result
print(result["messages"][-1].content)

使用流式传输处理中断

对于实时 UI，在处理 HITL 中断时使用 .stream() 而非 .invoke() 来显示 token 和工具调用进度。中断在 updates 流中显示为 __interrupt__ 条目。流完成后，收集人工决策，并通过使用 Command(resume=...) 再次流式输出来恢复。

NOTE

对于典型的 interrupt_on HITL 暂停，使用扁平的 {"decisions": [...]} 负载恢复，其形状与 .invoke() 相同。当同时有多个中断待处理时（例如并行分支），则改为将每个中断 ID 映射到其恢复值。请参阅处理多个中断。

python

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

config = {"configurable": {"thread_id": str(uuid7())}}
stream_input = {"messages": [{"role": "user", "content": "Delete the file temp.txt"}]}

while True:
    pending_interrupt = None

    for mode, data in agent.stream(  
        stream_input,
        stream_mode=["messages", "updates"],  
        config=config,
    ):
        if mode == "messages":
            token, metadata = data
            # Display tokens in real time
            if token.content:
                print(token.content, end="", flush=True)

        elif mode == "updates" and isinstance(data, dict):
            if "__interrupt__" in data:  
                pending_interrupt = data["__interrupt__"][0]  

    # If no interrupt occurred, the agent finished
    if pending_interrupt is None:
        break

    interrupt_value = pending_interrupt.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_map[action["name"]]
        print(f"\nTool: {action['name']}")
        print(f"Arguments: {action['args']}")
        print(f"Allowed: {review['allowed_decisions']}")

    # Collect one decision per action_request, in order
    decisions = [{"type": "approve"} for _ in action_requests]
    stream_input = Command(resume={"decisions": decisions})  

print()

TIP

结合 "messages" 和 "updates" 流模式，在向用户显示实时 LLM token 的同时仍能捕获 HITL 中断。"messages" 模式流式输出 token；"updates" 模式传递中断负载。

多次工具调用

当智能体调用多个需要批准的工具时，所有中断会被合并在单个中断中。你必须按顺序为每一个提供决策。

python

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
)

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

    # Two tools need approval
    assert len(action_requests) == 2

    # Provide decisions in the same order as action_requests
    decisions = [
        {"type": "approve"},  # First tool: delete_file
        {
            "type": "reject",
            "message": "User rejected this action. Do not retry this tool call.",
        }  # Second tool: send_email
    ]

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

拒绝消息

当审查者返回 reject 决策时，Deep Agents 会跳过该工具调用，并将拒绝反馈发送回智能体。如果你省略 message，默认反馈会告诉模型该工具未被执行，并且除非用户要求，否则不要重试同一工具调用。

对于敏感或有副作用的工具，请在决策中传入特定领域的 message。明确说明智能体应该放弃该操作、提出后续问题，还是尝试更安全的替代方案。

python

decisions = [
    {
        "type": "reject",
        "message": "User rejected deleting this file. Do not retry deletion. Ask which file to archive instead.",
    }
]

编辑工具参数

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

python

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

    # Original args from the agent
    print(action_request["args"])  # {"to": "everyone@company.com", ...}

    # User decides to edit the recipient
    decisions = [{
        "type": "edit",
        "edited_action": {
            "name": action_request["name"],  # Must include the tool name
            "args": {"to": "team@company.com", "subject": "...", "body": "..."}
        }
    }]

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

子智能体中断

使用子智能体时，你可以使用对工具调用的中断和工具调用内的中断。

对工具调用的中断

每个子智能体都可以有自己的 interrupt_on 配置，覆盖主智能体的设置：

python

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": "Manages file operations",
        "system_prompt": "You are a file management assistant.",
        "tools": [delete_file, read_file],
        "interrupt_on": {
            # Override: require approval for reads in this subagent
            "delete_file": True,
            "read_file": True,  # Different from main agent!
        }
    }],
    checkpointer=checkpointer
)

当子智能体触发中断时，处理方式相同——检查结果上的 __interrupt__ 并使用 Command 恢复。

进行流式传输时，子智能体中断也会在父 updates 流中显示为 __interrupt__ 条目。中断传递不需要 subgraphs=True。如果你还想流式输出子智能体 token 和进度事件，请启用 subgraphs=True。完整模式请参阅使用流式传输处理中断。

工具调用内的中断

子智能体工具可以直接调用 interrupt() 来暂停执行并等待批准：

python

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:
    """Request human approval using the interrupt() primitive."""
    # interrupt() pauses execution and returns the value passed to 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.6-flash",
        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
    )

    # Check for interrupt
    if result.get("__interrupt__"):  
        interrupt_value = result["__interrupt__"][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
        )

        if "__interrupt__" not in result2:  
            print("\nExecution completed!")
            # Find the tool response
            tool_msgs = [m for m in result2.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()

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

text

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.

文件系统权限中断

NOTE

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

除了 interrupt_on，你还可以通过为权限规则标记 mode="interrupt" 来暂停内置文件系统工具。当智能体在匹配中断模式规则的路径上调用 write_file 或 edit_file 时，create_deep_agent 会像配置的工具一样抛出相同的人在回路中断，并使用文件系统工具的名称作为操作名称。

python

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(),  # Required to pause and resume
)

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

python

from langgraph.types import Command

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

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Save the API key to /secrets/key.txt"}]},
    config=config,
)

if result.get("__interrupt__"):
    action = result["__interrupt__"][0].value["action_requests"][0]
    print(f"Approve {action['name']} on {action['args']}?")

    # Resume with the human decision (approve, edit, or reject).
    result = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,  # Same thread ID
    )

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

最佳实践

始终使用检查点器

人在回路需要检查点器来在中断与恢复之间持久化智能体状态：

python

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    tools=[...],
    interrupt_on={...},
    checkpointer=checkpointer  # Required for HITL
)

使用相同的线程 ID

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

python

# First call
config = {"configurable": {"thread_id": "my-thread"}}
result = agent.invoke(input, config=config)

# Resume (use same config)
result = agent.invoke(Command(resume={...}), config=config)

使决策顺序与操作一致

决策列表必须与 action_requests 的顺序一致：

python

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

    # Create one decision per action, in order
    decisions = []
    for action in action_requests:
        decision = get_user_decision(action)  # Your logic
        decisions.append(decision)

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

按风险定制配置

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

python

interrupt_on = {
    # High risk: full control (approve, edit, reject)
    "delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
    "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},

    # Medium risk: no editing allowed
    "write_file": {"allowed_decisions": ["approve", "reject"]},

    # Low risk: no interrupts
    "read_file": False,
    "ls": False,
}

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织