CMA：实时监控子 Agent
实时观察子 Agent 的执行过程
🌐 查看英文原文 | 源码 Notebook
多 Agent：实时观看课程团队工作
委托给子 Agent 的协调器有一个可观测性差距。session 级别的流在模型生成时预览主线程的文本，但子 Agent 的输出只有在整轮缓冲后才变得可见。如果研究员运行两分钟，你两分钟什么都看不到。
本 notebook 使用四个 Managed Agents API 功能来消除这个差距：
- 每线程增量流式传输。 每个 session 线程的流接受与 session 级别流相同的
event_deltas参数，因此子 Agent 的文本在其自己的流上实时预览。 - session 创建时的初始事件。
sessions.create接受initial_events，因此 session 在创建它的同一次调用中开始工作。 - Agent 模型上的 effort。
model.effort设置 Claude 在每次推理调用上工作的努力程度，按 Agent 设置。在团队中，这是一个按角色的成本杠杆。 - agent 更新时的可选版本。
agents.update将当前版本视为可选的并发键而不是必需的。
你将构建的团队规划一个为期一周的七年级科学单元：协调器委托给标准研究员（网络搜索、高 effort）和课程编写者（无网络访问），然后组装单元计划。如果 multiagent 协调器模式对你来说是新的，请从 CMA_coordinate_specialist_team.ipynb 开始。本 notebook 基于这些形状构建。
1. 设置客户端
这些功能搭载标准的 managed-agents-2026-04-01 beta header。本 notebook 中的 SDK 调用（sessions.create 上的 initial_events、线程流上的 event_deltas、model.effort）需要 anthropic>=0.118.0。
%%capture
%pip install -qU "anthropic>=0.118.0" python-dotenvimport os
import anthropic
from dotenv import load_dotenv
load_dotenv()
BETAS = ["managed-agents-2026-04-01"]
MODEL = os.environ.get("COOKBOOK_MODEL", "claude-sonnet-5")
client = anthropic.Anthropic()2. 创建两个专家
研究员获得网络搜索和 high effort 级别，因为标准对齐是判断密集型工作：它必须权衡哪些表现期望真正适合该单元，而不是列出它找到的所有内容。effort 位于 model 对象内部，接受 low、medium、high、xhigh 或 max（作为纯字符串或 {"type": "high"}）。并非每个模型都接受每个级别，无效的组合会在创建时被拒绝。
Effort 也是团队的主要成本杠杆：更高级别让 Claude 在每次推理调用上花费更多 token，更低级别限制该花费。因为它是按 agent 设置的，你只为需要深度的角色购买深度。在更大的团队中，格式化或分诊角色可以降到 low 而不影响研究员的预算。
在 agent 上设置 effort，而不是按 session：按 session 的 model 覆盖内的 effort 级别不会被应用。
Web 工具的配置还接受允许和阻止的域名列表，因此生产部署可以将该研究员限制在它应该信任的标准网站上。有关配置形状，请参阅配置工具集。
RESEARCHER_SYSTEM = """You research US middle school science standards.
Given a unit topic and grade level, use web search to find:
- The two or three NGSS performance expectations the unit should target
- What students are expected to already know, and where the topic leads next
- Two or three documented student misconceptions for the topic
Return via send_to_parent:
{"standards": [...], "prior_knowledge": ..., "leads_to": ...,
"misconceptions": [...], "sources": [...]}"""
standards_researcher = client.beta.agents.create(
name="standards_researcher",
description="Finds the NGSS standards and documented misconceptions for a science topic.",
model={"id": MODEL, "effort": "high"},
system=RESEARCHER_SYSTEM,
tools=[
{
"type": "agent_toolset_20260401",
"configs": [{"name": "web_search"}, {"name": "web_fetch"}],
}
],
betas=BETAS,
)课程编写者只根据协调器交给它的内容工作。禁用 Web 工具使其课程基于研究员审查过的发现，而不是新搜索返回的内容。
WRITER_SYSTEM = """You write middle school lesson plans.
You will be given a topic, grade level, target standards, and known misconceptions.
Draft five 45-minute lessons, one per day. For each day give:
- Objective (tied to a target standard)
- Warm-up (5 min)
- Main activity (30 min)
- Exit ticket (one question that probes a misconception where possible)
Return via send_to_parent: {"days": [{"day": 1, "objective": ..., "warm_up": ...,
"main_activity": ..., "exit_ticket": ...}, ...]}"""
lesson_writer = client.beta.agents.create(
name="lesson_writer",
description="Drafts a day-by-day lesson sequence from standards and misconceptions.",
model={"id": MODEL},
system=WRITER_SYSTEM,
tools=[
{
"type": "agent_toolset_20260401",
"configs": [
{"name": "web_search", "enabled": False},
{"name": "web_fetch", "enabled": False},
],
}
],
betas=BETAS,
)创建响应回显解析后的模型配置，包括你省略的字段。研究员显示你设置的 high，编写者显示模型的默认 effort。如果 effort 返回 None，你的组织的 beta header 还没有携带该功能：该字段被丢弃而不是被拒绝，因此此回显是捕获它的地方。
for agent in (standards_researcher, lesson_writer):
effort = getattr(agent.model, "effort", None)
print(f"{agent.name}: {agent.id} v{agent.version} effort={effort and effort.type}")3. 创建协调器
协调器携带 multiagent 名册。生成和委托工具从名册中自动注入，名册将每个子 Agent 固定到当前的版本。当稍后更新研究员时，该固定很重要。
COORDINATOR_SYSTEM = """You plan one-week teaching units.
Given a topic and grade level:
1. Send the topic and grade to standards_researcher.
2. When the researcher reports back, send the topic, grade, target standards, and
misconceptions to lesson_writer.
3. Write /mnt/session/outputs/unit_plan.md with sections: Overview, Standards alignment,
Day-by-day plan (from the writer), Misconceptions to watch for, Materials.
Keep it under three pages. To revise the plan, rewrite the whole file with the write
tool rather than patching it with edit. While a subagent is working, wait silently: no
status commentary until its report arrives."""
coordinator = client.beta.agents.create(
name="unit_planner",
description="Plans one-week teaching units by delegating research and drafting.",
model={"id": MODEL},
system=COORDINATOR_SYSTEM,
tools=[{"type": "agent_toolset_20260401"}],
multiagent={
"type": "coordinator",
"agents": [standards_researcher.id, lesson_writer.id],
},
betas=BETAS,
)
print(f"{coordinator.name}: {coordinator.id} v{coordinator.version}")4. 内联第一条消息启动 session
没有 initial_events，启动 session 需要两次调用：创建它，然后发送 user.message。在创建时种子化第一条消息将其合并为一次。该数组接受最多 50 个 user.message 和 user.define_outcome 事件，按顺序处理，验证是全有或全无的：如果任何事件失败，则不创建 session。非空列表在同一次调用中启动 Agent 循环，因此 session 返回时已经在向 running 移动。
如果你想根据评分标准对单元计划评分，请在此处添加一个 user.define_outcome 事件（它必须包含一个 rubric，每次创建只允许一个）。CMA_verify_with_outcome_grader.ipynb 涵盖了该循环。
import time
env = client.beta.environments.create(
name="curriculum-studio",
config={"type": "anthropic_cloud", "networking": {"type": "unrestricted"}},
betas=BETAS,
)
session = client.beta.sessions.create(
agent=coordinator.id,
environment_id=env.id,
title="Unit plan: water cycle, grade 7",
initial_events=[
{
"type": "user.message",
"content": [
{
"type": "text",
"text": "Plan a one-week unit on the water cycle for 7th grade science. "
"Classes are 45 minutes. Write the plan to /mnt/session/outputs/unit_plan.md.",
}
],
}
],
betas=BETAS,
)
deadline = time.monotonic() + 60
while session.status == "idle" and time.monotonic() < deadline:
time.sleep(1)
session = client.beta.sessions.retrieve(session.id, betas=BETAS)
print(f"{session.id}: {session.status}")
if session.status == "idle":
events = client.beta.sessions.events.list(session.id, limit=1, betas=BETAS)
if next(iter(events), None) is None:
raise RuntimeError(
"The session has no events: your organization's beta header does not carry "
"initial_events yet. The field is dropped at the gateway, not rejected."
)
raise RuntimeError(
"The seeded message is in the event log but the session is still idle after 60s. "
"Re-run this cell to keep waiting."
)5. 实时观看整个团队
预览在设计上是线程范围的。一个连接只预览它读取的线程：session 级别的流预览主线程，子 Agent 的预览只出现在该子 Agent 自己的流上 GET /v1/sessions/{session_id}/threads/{thread_id}/stream。因此模式是每个线程一个流：读取 session 级别的流获取协调器信息，每次 session.thread_created 事件宣布一个子 Agent 时（它携带 session_thread_id 和 agent_name），在后台线程中将一个观察器附加到该线程的流上。
两条规则塑造了代码：
- 预览是临时缓冲区，缓冲事件是记录。 增量是尽力而为的，可能在负载下停止。随后的缓冲
agent.message携带完整内容。SDK 的accumulate_managed_agents_event辅助函数将event_start、event_delta和缓冲事件折叠为一个快照，在记录到达时替换预览。每个流连接运行一个累加器。 - 无重放。 在模型请求开始后打开的连接不会收到该进行中请求的增量，重新连接永远不会重放错过的增量。在
session.thread_created到达时立即附加观察器可以从子 Agent 的第一个响应开始捕获其工作。
每个观察器在 session.thread_status_idle 上退出——这是子 Agent 轮次完成时发射的事件。
import threading
from anthropic.lib.sessions import accumulate_managed_agents_event
print_lock = threading.Lock()
current_speaker = None
reconciliations = []
def say(speaker, text):
"""Print streamed text, labeling it whenever the speaker changes."""
global current_speaker
with print_lock:
if speaker != current_speaker:
print(f"\n\n=== {speaker} ===")
current_speaker = speaker
print(text, end="", flush=True)
def text_of(event):
if event is None:
return ""
return "".join(block.text for block in event.content if block.type == "text")
def fold_preview(snapshot, ev, speaker):
"""Fold one preview event into the speaker's snapshot and return the new snapshot.
Prints delta text as it arrives. When the buffered record lands, logs a
reconciliation row comparing it against the accumulated preview. The id
check guards the case where every delta for a message was shed: the
snapshot then still holds the previous message, not a preview of this one.
"""
if ev.type == "event_delta":
say(speaker, ev.delta.content.text)
elif ev.type == "agent.message":
preview = text_of(snapshot) if snapshot and snapshot.id == ev.id else ""
final = text_of(ev)
reconciliations.append((speaker, len(preview), len(final), final.startswith(preview)))
return accumulate_managed_agents_event(snapshot, ev)
def watch_child(thread_id, agent_name):
"""Stream one child thread, previewing its text as the model generates it."""
snapshot = None
with client.beta.sessions.threads.events.stream(
thread_id,
session_id=session.id,
event_deltas=["agent.message"],
betas=BETAS,
) as thread_stream:
for ev in thread_stream:
if ev.type in ("event_start", "event_delta", "agent.message"):
snapshot = fold_preview(snapshot, ev, agent_name)
elif ev.type == "agent.tool_use":
detail = ev.input.get("query", "") if ev.name == "web_search" else ""
say(agent_name, f"\n[{ev.name}] {detail}")
elif ev.type in ("session.thread_status_idle", "session.thread_status_terminated"):
break主循环为协调器的文本提供相同的 fold_preview，加上只出现在主线程上的协调事件：子 Agent 生成时的 session.thread_created、协调器交接任务时的 agent.thread_message_sent 以及子 Agent 报告回来时的 agent.thread_message_received。从子线程交叉发布的工具调用携带 session_thread_id，因此循环跳过这些：子 Agent 自己的观察器会显示它们。
循环在任何 session.status_idle 上结束，当停止原因不是 end_turn 时打印停止原因：等待工具确认或重试用完的 session 应该结束单元格而不是挂起它。两个循环也在终止状态事件上中断，因此不可恢复的 session 错误会结束单元格而不是让流保持打开。
watchers = []
snapshot = None
with client.beta.sessions.events.stream(
session.id, event_deltas=["agent.message"], betas=BETAS
) as stream:
for ev in stream:
if ev.type in ("event_start", "event_delta", "agent.message"):
snapshot = fold_preview(snapshot, ev, "unit_planner")
elif ev.type == "session.thread_created":
say("unit_planner", f"\n[spawned {ev.agent_name}]")
watcher = threading.Thread(
target=watch_child,
args=(ev.session_thread_id, ev.agent_name),
daemon=True,
)
watcher.start()
watchers.append(watcher)
elif ev.type == "agent.thread_message_sent":
say("unit_planner", f"\n[task sent to {ev.to_agent_name}]")
elif ev.type == "agent.thread_message_received":
say("unit_planner", f"\n[{ev.from_agent_name} reported back]")
elif ev.type == "agent.tool_use" and not ev.session_thread_id:
# session_thread_id marks a call cross-posted from a child thread;
# that child's watcher already shows it
say("unit_planner", f"\n[{ev.name}]")
elif ev.type == "session.status_terminated":
break
elif ev.type == "session.status_idle":
if ev.stop_reason.type != "end_turn":
say("unit_planner", f"\n[idle without end_turn: {ev.stop_reason.type}]")
break
for watcher in watchers:
watcher.join(timeout=60)
if any(watcher.is_alive() for watcher in watchers):
print("\n[a child stream is still open after the join timeout]")如果流连接但只传递缓冲事件，说明 event_deltas 参数被剥离而不是被拒绝：这是 beta header 不携带该功能的标志。线程流 URL 上的 404 表示路径错误或缺少 managed-agents beta header（线程端点是 beta 门控的）。路径是 /threads/{thread_id}/stream，不是 /threads/{thread_id}/events/stream。
完整的流式传输契约（增量形状、对账保证和故障排除表）在事件和流式传输中。
6. 对照记录检查预览
连接预览的增量给出缓冲事件文本的前缀：一个前缀，不一定是整个文本，因为增量可能在负载下被丢弃。该保证使对账成为一个单一的替换。下面的每一行是一个缓冲的 agent.message，与累加器在记录到达时构建的预览进行比较。
for agent_name, preview_chars, final_chars, is_prefix in reconciliations:
print(
f"{agent_name:22s} preview {preview_chars:5d} chars | "
f"final {final_chars:5d} chars | preview is prefix: {is_prefix}"
)7. 读取单元计划
协调器的提示说用整文件 write 调用编写计划，因此完整文档在事件日志中，对该文件的最后一次 write 是最终版本。如果你自己的 Agent 用 edit 修补文件，请读取文件本身而不是重放写入。
unit_plan = ""
for ev in client.beta.sessions.events.list(session.id, limit=1000, betas=BETAS):
if (
ev.type == "agent.tool_use"
and ev.name == "write"
and ev.input.get("file_path", "").endswith("unit_plan.md")
):
unit_plan = ev.input["content"] # keep the last write, in case of revisions
# Show the structure rather than the full plan. Print unit_plan to read it all.
for line in unit_plan.splitlines():
if line.startswith("#"):
print(line)8. 无需版本往返即可发布 prompt 更改
假设审查反馈说每个对齐声明都需要旁边有标准代码。agents.update 将 agent 的当前 version 视为可选的并发键：省略它，更新无条件应用，服务器为你递增版本。配置脚本不必为了写回而读取 agent。
在可能有并发写入者时提供 version（不匹配返回 409，因此你始终从已知状态更新）。在单个流程拥有 agent 时省略它，例如同步签入 agent 定义的 CI 作业。
updated = client.beta.agents.update(
standards_researcher.id,
system=RESEARCHER_SYSTEM
+ "\nCite the performance expectation code (for example MS-ESS2-4) next to every "
"alignment claim.",
betas=BETAS,
)
print(f"standards_researcher v{standards_researcher.version} -> v{updated.version}")多 Agent 设置的一个注意事项：协调器的名册在协调器创建时固定了研究员的版本，因此现有协调器继续委托给旧版本。更新协调器（其 multiagent 字段或任何字段）以根据最新的子版本重新解析名册。
9. 清理
from utilities import wait_for_idle_status
wait_for_idle_status(client, session.id)
client.beta.sessions.archive(session.id, betas=BETAS)
client.beta.environments.archive(env.id, betas=BETAS)
print("archived")中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。