05 subagent interrupts | Knowledge Notes

它是什么

Subagent 可以定义自己的 interrupt_on，覆盖或补充父 Agent 的审批策略。子 Agent 触发的中断会冒泡到父 Agent 的 GraphOutput.interrupts，恢复方式仍然是 Command(resume=...)。它解决的问题是：专家子 Agent 有自己的风险边界。

常用程度

中。只要 subagent 有敏感工具，就应该考虑给它单独配置。

最小代码

文件：deepagent_src/human_loop_teach/05_subagent_interrupt.py

secret_reader_subagent = {
    "name": "secret-reader",
    "tools": [read_secret],
    "interrupt_on": {
        "read_secret": {"allowed_decisions": ["approve", "reject"]},
    },
}

运行

uv run python deepagent_src/human_loop_teach/05_subagent_interrupt.py

预期输出末尾：

subagent interrupt HITL real agent ok

验证方式

主 Agent 先通过 task 委派给 secret-reader，子 Agent 调 read_secret 时触发 interrupt。脚本 approve 后，父 Agent 的 task 工具结果包含子 Agent 返回值。

常见误区

父 Agent 的 interrupt_on 不一定适合每个 subagent。自定义 subagent 指定 interrupt_on 后，会以子 Agent 自己的配置为准。

相关资源

查看示例代码：deepagent_src/human_loop_teach/05_subagent_interrupt.py

from __future__ import annotations

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

from _hitl_output import print_graph_output, require_interrupt, tool_messages
from _model import get_real_model


@tool
def read_secret(path: str) -> str:
    """Read a sensitive secret path."""
    return f"secret-from:{path}"


secret_reader_subagent = {
    "name": "secret-reader",
    "description": "Reads sensitive paths after human approval.",
    "system_prompt": (
        "You are secret-reader. Always call read_secret exactly once and return "
        "only the tool result."
    ),
    "tools": [read_secret],
    "interrupt_on": {
        "read_secret": {"allowed_decisions": ["approve", "reject"]},
    },
}


def main() -> None:
    agent = create_deep_agent(
        model=get_real_model(),
        subagents=[secret_reader_subagent],
        checkpointer=MemorySaver(),
        system_prompt="Delegate secret reads to secret-reader.",
    )
    config = {"configurable": {"thread_id": "hitl-subagent-interrupt"}}
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "委派 secret-reader 读取 /secrets/token.txt",
                }
            ]
        },
        config=config,
        version="v2",
    )
    print_graph_output(result)
    interrupt_value = require_interrupt(result)
    assert interrupt_value["action_requests"][0]["name"] == "read_secret"

    result = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,
        version="v2",
    )
    print_graph_output(result)

    assert any("secret-from:/secrets/token.txt" in output for output in tool_messages(result, name="task"))
    print("subagent interrupt HITL real agent ok")


if __name__ == "__main__":
    main()