04 多工具批量审批 | Knowledge Notes

它是什么

当模型一次生成多个需要审批的工具调用，Deep Agents 会把它们合并进一个 interrupt。人类恢复时必须按 action_requests 的顺序提供同样数量的 decisions。它解决的问题是：一次审一批，避免来回打断。

常用程度

中。模型并行调用多个工具时常见，尤其是“删除文件并发邮件通知”这类组合任务。

最小代码

文件：deepagent_src/human_loop_teach/04_multiple_tool_calls.py

interrupt_on = {
    "delete_record": {"allowed_decisions": ["approve", "reject"]},
    "notify_email": {"allowed_decisions": ["approve", "reject"]},
}

运行

uv run python deepagent_src/human_loop_teach/04_multiple_tool_calls.py

预期输出末尾：

multiple tool calls HITL real agent ok

验证方式

脚本断言 interrupt 里有两个 action：

delete_record
notify_email

然后按顺序 approve 第一个、reject 第二个。

常见误区

decisions 不是按工具名自动匹配，而是按 action_requests 顺序匹配。生产 UI 里要小心保持顺序，不然就会把审批结果套错工具。

相关资源

查看示例代码：deepagent_src/human_loop_teach/04_multiple_tool_calls.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 delete_record(record_id: str) -> str:
    """Delete a record."""
    return f"Deleted record {record_id}"


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


def main() -> None:
    agent = create_deep_agent(
        model=get_real_model(),
        tools=[delete_record, notify_email],
        interrupt_on={
            "delete_record": {"allowed_decisions": ["approve", "reject"]},
            "notify_email": {"allowed_decisions": ["approve", "reject"]},
        },
        checkpointer=MemorySaver(),
        system_prompt=(
            "When asked for the batch approval demo, call delete_record and "
            "notify_email in the same response. Do not call other tools."
        ),
    )
    config = {"configurable": {"thread_id": "hitl-multiple-tools"}}
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "批量审批演示：调用 delete_record(record_id=42)，"
                        "并调用 notify_email(to=admin@example.com, subject=Deleted)"
                    ),
                }
            ]
        },
        config=config,
        version="v2",
    )
    print_graph_output(result)
    interrupt_value = require_interrupt(result)
    actions = interrupt_value["action_requests"]
    assert [action["name"] for action in actions] == ["delete_record", "notify_email"]

    result = agent.invoke(
        Command(
            resume={
                "decisions": [
                    {"type": "approve"},
                    {
                        "type": "reject",
                        "message": "Human rejected sending notification email.",
                    },
                ]
            }
        ),
        config=config,
        version="v2",
    )
    print_graph_output(result)

    assert any("Deleted record 42" in output for output in tool_messages(result, name="delete_record"))
    assert not any(output.startswith("Sent notification") for output in tool_messages(result, name="notify_email"))
    print("multiple tool calls HITL real agent ok")


if __name__ == "__main__":
    main()