03 条件中断 | Knowledge Notes

它是什么

条件中断用 when predicate 根据工具参数决定是否暂停。比如写 /workspace/ 里可以自动通过，写 /secrets/ 就必须审批。它解决的问题是：不是每次调用同一个工具都同样危险。

常用程度

中。只要同一个工具既能做低风险操作，又能做高风险操作，就应该考虑条件中断。

最小代码

文件：deepagent_src/human_loop_teach/03_conditional_interrupt.py

def writes_outside_workspace(request: ToolCallRequest) -> bool:
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")

运行

uv run python deepagent_src/human_loop_teach/03_conditional_interrupt.py

预期输出末尾：

conditional interrupt HITL real agent ok

验证方式

脚本跑两次：

/workspace/a.txt：when 返回 False，不 interrupt，工具直接执行。
/secrets/a.txt：when 返回 True，触发 interrupt，脚本用 reject 恢复。

常见误区

when 只决定是否暂停，不负责权限执行本身。真正的访问控制还应该在工具内部、后端或文件系统权限层做。

相关资源

查看示例代码：deepagent_src/human_loop_teach/03_conditional_interrupt.py

from __future__ import annotations

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
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 write_note(file_path: str, content: str) -> str:
    """Write a note to a path."""
    return f"Wrote {content} to {file_path}"


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


def main() -> None:
    agent = create_deep_agent(
        model=get_real_model(),
        tools=[write_note],
        interrupt_on={
            "write_note": {
                "allowed_decisions": ["approve", "reject"],
                "when": writes_outside_workspace,
            }
        },
        checkpointer=MemorySaver(),
        system_prompt="When asked to write a note, call write_note exactly once.",
    )

    safe_config = {"configurable": {"thread_id": "hitl-conditional-safe"}}
    safe = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "调用 write_note，file_path=/workspace/a.txt，content=safe",
                }
            ]
        },
        config=safe_config,
        version="v2",
    )
    print_graph_output(safe)
    assert not safe.interrupts
    assert any("/workspace/a.txt" in output for output in tool_messages(safe, name="write_note"))

    risky_config = {"configurable": {"thread_id": "hitl-conditional-risky"}}
    risky = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "调用 write_note，file_path=/secrets/a.txt，content=risky",
                }
            ]
        },
        config=risky_config,
        version="v2",
    )
    print_graph_output(risky)
    require_interrupt(risky)
    risky = agent.invoke(
        Command(resume={"decisions": [{"type": "reject", "message": "Outside workspace."}]}),
        config=risky_config,
        version="v2",
    )
    print_graph_output(risky)
    assert not any("/secrets/a.txt" in output and output.startswith("Wrote") for output in tool_messages(risky, name="write_note"))
    print("conditional interrupt HITL real agent ok")


if __name__ == "__main__":
    main()