CMA：人工审核关卡
在人机协作中加入人工审批环节
🌐 查看英文原文 | 源码 Notebook
门控：使用自定义工具的人工参与
许多工作流处于”完全自动化”和”总是询问人工”之间的灰色地带。费用审批就是一个经典例子：Agent 可以自行处理明确的情况，但它应该知道何时将模糊的情况升级为人工审查。校准在这里很重要——一个什么都升级的 Agent 让人筋疲力尽，而一个什么都不升级的 Agent 是危险的。
本 notebook 围绕两个自定义工具构建一个费用审批器：decide() 用于明确的情况，escalate() 用于模糊的情况。两者都通过你的应用程序往返——在这里你要么记录结果（decide），要么将其放在审查者面前（escalate）。
什么是自定义工具
到目前为止，cookbook 一直使用内置的 agent_toolset（bash、read、write 等），它们都在沙盒容器内运行。自定义工具不同：当 Agent 调用一个时，session 暂停并发射一个 agent.custom_tool_use 事件，你的应用程序看到调用，运行你想要的任何代码，然后 POST 回一个 user.custom_tool_result 事件。Session 使用该结果在 Agent 的上下文中恢复。
这在两种情况下是正确的形状：
- 数据存在于沙盒无法到达的地方。 你自己网络边界后面的任何东西。Agent 通过往返调用回到你的应用程序。
- 你想要人工参与，或者你自己的审计和批准层在每次调用之前。 这就是本 notebook 所做的：
decide和escalate不仅是抽象意义上的”工具”，它们是你的业务逻辑和人工审查者从 Agent 接管的接缝。
（其他扩展模式，MCP 工具集和 resources= 仓库挂载，分别在 operate notebook 和 orchestrate notebook 中介绍。）
本 notebook 有两部分。A 部分通过本地流式传输事件并在每个自定义工具调用到达时响应来驱动 session——在开发期间很方便，因为一切都在一个进程中发生，你可以实时看到行为。B 部分是对生产 webhook 模式的简短指引，在 operate notebook 中端到端介绍。
fixture 位于 example_data/gate/ 中，包含一个 policy.yaml 和十二张收据，涵盖了策略的每个分支。
import json
import os
from collections import Counter
from pathlib import Path
from anthropic import Anthropic
from utilities import wait_for_idle_status
MODEL = os.environ.get("COOKBOOK_MODEL", "claude-sonnet-4-6")
client = Anthropic()
FIXTURE = Path("example_data") / "gate"1. 上传策略和收据
policy = client.beta.files.upload(
file=("policy.yaml", (FIXTURE / "policy.yaml").read_bytes(), "text/yaml")
)
receipts = client.beta.files.upload(
file=(
"receipts.jsonl",
(FIXTURE / "inbox" / "receipts.jsonl").read_bytes(),
"application/jsonl",
)
)2. 定义带有两个自定义工具的 agent
自定义工具在与内置工具集相同的 tools= 数组中声明，带有 "type": "custom" 和输入的 JSON schema。每个声明告诉模型工具的用途（description）、用什么来调用它（input_schema）以及它的名称是什么。Agent 决定何时调用它们；你的代码决定它们在被调用时做什么。
这里我们也保持内置的 agent_toolset_20260401 启用，以便 Agent 可以内联读取策略文件和收据。decide 和 escalate 是使每个决策都成为往返的两个自定义工具。
agent = client.beta.agents.create(
name="cookbook-gate",
model=MODEL,
system=(
"You are an expense approver. Read each receipt in "
"receipts.jsonl against the policy in policy.yaml and make "
"exactly ONE tool call per receipt. Call decide(receipt_id, "
"action, reason) for clear cases, or escalate(receipt_id, "
"question) for ambiguous ones (near thresholds, unclear "
"categories, suspicious notes). Once you've called decide "
"or escalate for a given receipt, that receipt is finalized "
"— do not call either tool for it again. After processing "
"all receipts exactly once, stop."
),
tools=[
{
"type": "agent_toolset_20260401",
"default_config": {
"enabled": True,
"permission_policy": {"type": "always_allow"},
},
},
{
"type": "custom",
"name": "decide",
"description": "Record a final approve/reject for a clear-cut receipt.",
"input_schema": {
"type": "object",
"properties": {
"receipt_id": {"type": "string"},
"action": {"type": "string", "enum": ["approve", "reject"]},
"reason": {"type": "string"},
},
"required": ["receipt_id", "action", "reason"],
},
},
{
"type": "custom",
"name": "escalate",
"description": "Surface an ambiguous receipt for human review.",
"input_schema": {
"type": "object",
"properties": {
"receipt_id": {"type": "string"},
"question": {"type": "string"},
},
"required": ["receipt_id", "question"],
},
},
],
)
env = client.beta.environments.create(
name="cookbook-gate-env",
config={"type": "cloud", "networking": {"type": "limited"}},
)
session = client.beta.sessions.create(
environment_id=env.id,
agent={"type": "agent", "id": agent.id, "version": agent.version},
resources=[
{"type": "file", "file_id": policy.id, "mount_path": "policy.yaml"},
{"type": "file", "file_id": receipts.id, "mount_path": "receipts.jsonl"},
],
title="Expense gate",
)
print(f"session: {session.id}")A 部分：开发期间的本地流式传输
驱动自定义工具 Agent 的最简单方式是流式传输 session 的事件并在每个工具调用到达时做出反应。decide 调用被记录，escalate 调用获得一个模拟的内联人工决策。在生产中，你会将升级排队并让真正的审查者稍后回来处理，这就是 operate notebook 所涵盖的内容。
def simulate_human_review(receipt_id: str, question: str) -> str:
# Real implementation would show this in a UI and await input.
# Here: reject anything the agent flags as suspicious.
return "reject" if "suspicious" in question.lower() else "approve"
# The iterate notebook factored its streaming loop out into
# `stream_until_end_turn`, and most other notebooks just import it.
# This one doesn't, because every decision the agent makes is a
# custom tool call, which means the session keeps going idle with
# `stop_reason.type == "requires_action"` and
# `stop_reason.event_ids` pointing at the `agent.custom_tool_use`
# events waiting for a response. We POST a `user.custom_tool_result`
# for each, let the session resume, and eventually break on a
# `session.status_idle` that arrives with `end_turn`. The helper
# only knows how to exit on `end_turn`, so we need the full loop
# here.
decisions = {} # receipt_id -> final decision record
tool_use_events = {}
responded_to = set() # event_ids we've already replied to
print("=== Part A: streaming ===")
with client.beta.sessions.events.stream(session.id) as stream:
client.beta.sessions.events.send(
session_id=session.id,
events=[
{
"type": "user.message",
"content": [
{
"type": "text",
"text": (
"Read /mnt/session/uploads/policy.yaml and "
"/mnt/session/uploads/receipts.jsonl. Process "
"all 12 receipts. For each receipt, make "
"exactly one decide() or escalate() call and "
"then move on to the next. When every receipt "
"has been processed once, stop."
),
}
],
}
],
)
# Note on the responded_to set: when an agent emits more than 5
# parallel custom tool calls, the server returns
# `stop_reason.event_ids` as a sliding window of the next 5
# pending. Each status_idle we observe in the stream has that
# window pinned at the moment the event was emitted, but by the
# time we iterate to the next status_idle event, the server has
# already advanced past the events we just responded to. So we
# need to dedupe across status_idle events to avoid double-
# responding to the same custom tool call (which 400s).
for ev in stream:
if ev.type == "agent.custom_tool_use":
tool_use_events[ev.id] = ev
elif ev.type == "session.status_idle" and ev.stop_reason:
if ev.stop_reason.type == "requires_action":
for event_id in ev.stop_reason.event_ids:
if event_id in responded_to:
continue
tool_ev = tool_use_events[event_id]
name, args = tool_ev.name, tool_ev.input
receipt_id = args["receipt_id"]
if name == "decide":
decisions[receipt_id] = {"lane": args["action"], **args}
result = {"recorded": True}
elif name == "escalate":
human = simulate_human_review(receipt_id, args["question"])
decisions[receipt_id] = {
"lane": "escalated",
"human_decision": human,
**args,
}
result = {"human_decision": human}
else:
result = {"error": f"unknown tool {name}"}
client.beta.sessions.events.send(
session_id=session.id,
events=[
{
"type": "user.custom_tool_result",
"custom_tool_use_id": event_id,
"content": [{"type": "text", "text": json.dumps(result)}],
}
],
)
responded_to.add(event_id)
elif ev.stop_reason.type == "end_turn":
break
elif ev.type == "session.status_terminated":
break
wait_for_idle_status(client, session.id)
lanes = Counter(d["lane"] for d in decisions.values())
print(f"\n{len(decisions)} decisions: {dict(lanes)}")
client.beta.sessions.archive(session.id)
client.beta.environments.archive(env.id)
client.beta.agents.archive(agent.id)
print("archived")B 部分：生产环境的 webhook
本地流式传输模式在开发期间运行良好，但它在人类思考时保持 HTTP 连接打开，这不容易扩展。生产模式改为在 Console 中注册一个在 session.status_idled 上触发的 webhook——这是 Agent 已完成或正在等待工具结果的信号。你的服务器检查事件，将任何待处理的升级放在审查者面前，并在人工完成后 POST 回 user.custom_tool_result——你的这边没有长期连接。
operate notebook 端到端介绍了完整的 webhook 设置：Console 注册、HMAC 签名验证、FastAPI 处理器以及回到 events.send 的往返。响应 Agent 的代码与上面的 A 部分相同；只是触发方式变了（webhook 推送代替流式拉取）。
中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。