07 综合案例 | Knowledge Notes

它是什么

综合案例在同一个 Agent 配置里同时启用自定义工具审批和文件系统权限审批。脚本遇到中断后会在终端展示工具名、参数和可选决策，由你手动输入 approve、edit 或 reject。它解决的问题是：真实项目通常会同时有多类风险动作，但最终决策来自人，不是代码写死。

最小代码

文件：deepagent_src/human_loop_teach/07_comprehensive_case.py

interrupt_on={
    "notify_email": {"allowed_decisions": ["approve", "edit", "reject"]},
}
permissions=[
    FilesystemPermission(
        operations=["write"],
        paths=["/secrets/**"],
        mode="interrupt",
    )
]

运行

uv run python deepagent_src/human_loop_teach/07_comprehensive_case.py

预期输出末尾：

interactive comprehensive HITL demo completed

运行过程中会出现类似提示：

需要人工审核：
tool: notify_email
args: {'to': 'wrong@example.com', 'subject': 'Deploy', 'body': 'Ready'}
可选: approve, edit, reject
请输入决策:

串联了哪些知识点

interrupt_on：拦截 notify_email
FilesystemPermission(mode="interrupt")：拦截 /secrets/** 写入
approve：按原参数执行工具
edit：在终端逐个字段修改工具参数
reject：在终端输入拒绝原因，跳过工具执行
同一个 thread_id：每次暂停后都必须用对应的同一份 config 恢复

常见误区

不要把“同一个 Agent 同时配置多种审批规则”和“模型同一轮生成多个工具调用”混成一件事。第 04 章演示批量审批；本章演示同一套 Agent 配置如何覆盖不同风险动作。

相关资源

查看示例代码：deepagent_src/human_loop_teach/07_comprehensive_case.py

from __future__ import annotations

from deepagents import FilesystemPermission, 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
from _model import get_real_model


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


def ask_terminal_decisions(interrupt_value: dict) -> list[dict]:
    review_configs = {
        config["action_name"]: config
        for config in interrupt_value.get("review_configs", [])
    }
    decisions = []
    for action in interrupt_value["action_requests"]:
        allowed = review_configs[action["name"]]["allowed_decisions"]
        choices = [choice for choice in ("approve", "edit", "reject") if choice in allowed]

        print("\n需要人工审核：")
        print(f"tool: {action['name']}")
        print(f"args: {action['args']}")
        print(f"可选: {', '.join(choices)}")

        while True:
            decision_type = input("请输入决策: ").strip()
            if decision_type in choices:
                break
            print("输入不合法，重新输。")

        if decision_type == "approve":
            decisions.append({"type": "approve"})
            continue

        if decision_type == "reject":
            message = input("拒绝原因（回车用默认）: ").strip()
            decisions.append(
                {
                    "type": "reject",
                    "message": message or f"Human rejected {action['name']}. Do not retry.",
                }
            )
            continue

        edited_args = {}
        for key, value in action["args"].items():
            new_value = input(f"{key} [{value}]: ").strip()
            edited_args[key] = new_value or value
        decisions.append(
            {
                "type": "edit",
                "edited_action": {"name": action["name"], "args": edited_args},
            }
        )

    return decisions


def main() -> None:
    agent = create_deep_agent(
        model=get_real_model(),
        tools=[notify_email],
        interrupt_on={
            "notify_email": {"allowed_decisions": ["approve", "edit", "reject"]},
        },
        permissions=[
            FilesystemPermission(
                operations=["write"],
                paths=["/secrets/**"],
                mode="interrupt",
            )
        ],
        checkpointer=MemorySaver(),
        system_prompt=(
            "For the comprehensive HITL demo, call exactly the requested tool "
            "once. Do not call other tools."
        ),
    )

    email_config = {"configurable": {"thread_id": "hitl-comprehensive-email"}}
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "调用 notify_email(to=wrong@example.com, subject=Deploy, body=Ready)",
                }
            ]
        },
        config=email_config,
        version="v2",
    )
    print_graph_output(result)
    interrupt_value = require_interrupt(result)
    action = interrupt_value["action_requests"][0]
    assert action["name"] == "notify_email"

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

    file_config = {"configurable": {"thread_id": "hitl-comprehensive-file"}}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "把内容 demo 写入 /secrets/deploy.txt"}]},
        config=file_config,
        version="v2",
    )
    print_graph_output(result)
    interrupt_value = require_interrupt(result)
    action = interrupt_value["action_requests"][0]
    assert action["name"] == "write_file"

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

    print("interactive comprehensive HITL demo completed")


if __name__ == "__main__":
    main()