Claude Managed Agents 多 Agent 协调大模型做规划、小模型做执行的经济学设计【免费下载链接】claude-cookbooksA collection of notebooks/recipes showcasing some fun and effective ways of using Claude.项目地址: https://gitcode.com/GitHub_Trending/an/claude-cookbooks大多数 agent 工作负载内部都包含两种性质完全不同的活少量规划与判断以及大量机械性的阅读和执行。网页研究是极端情况——核对二十个事实意味着让模型读入数十万 token 的网页按前沿模型的价格计费这笔阅读费会占掉账单的大头。Claude Managed Agents 的 coordinator 模式把这两类工作拆开前沿模型负责规划和综合但它从不直接碰原始网页廉价 worker 在各自并行的上下文里完成所有阅读只把蒸馏后的发现汇报回来。这篇文章基于 claude-cookbooks 仓库中的 cookbook managed_agents/CMA_plan_big_execute_small.ipynb带你完成一条完整路径用multiagent字段配置前沿 coordinator 廉价 worker的双模型团队跑一个带预算护栏的研究任务再运行一个验证标准对齐的单前沿模型对照组最后用每线程的usage.list_cost对比两边真实账单。运行环境是 Python 3.11 及以上、一个有 Managed Agents beta 权限的 Anthropic API Key。准备条件按 managed_agents/README.md 的 Getting started 说明设置好ANTHROPIC_API_KEY环境变量在 Jupyter 中打开该 notebook 并从第一个 cell 开始顺序执行每个 notebook 会自己安装依赖。第一个 cell 安装依赖%%capture %pip install -qU anthropic0.121.0 python-dotenvbudget参数和session.usage上的成本字段需要anthropic0.121.0。然后是客户端与模型常量import os import time import anthropic from dotenv import load_dotenv load_dotenv() BETAS [managed-agents-2026-04-01] # The frontier model plans and synthesizes; the cheap model reads the web. COORDINATOR_MODEL os.environ.get(COOKBOOK_COORDINATOR_MODEL, claude-fable-5) WORKER_MODEL os.environ.get(COOKBOOK_WORKER_MODEL, claude-sonnet-5) client anthropic.Anthropic()两个模型都可以用环境变量覆盖不设置时 coordinator 用claude-fable-5worker 用claude-sonnet-5。如果你之前没接触过 Managed Agents 的 agent / environment / session / 流式事件循环先看入口 notebook managed_agents/CMA_iterate_fix_failing_tests.ipynb如果没见过multiagent字段managed_agents/CMA_coordinate_specialist_team.ipynb 用一个异构专家团队介绍了它。本文刻意只用一种 worker 类型因为重点是成本结构而不是团队设计。定义 worker小模型只负责读worker 是一个普通 agent模型、一套收窄到web_searchweb_fetch的工具、一个 system prompt。工具收窄有两个目的一是防止廉价模型跑到 bash 或文件系统上去二是安全边界——worker 会读取任意不可信的网页只能搜索、抓取、汇报就是这类输入想要的爆炸半径。worker client.beta.agents.create( namesearch-worker, modelWORKER_MODEL, # Everything off except the two web tools: the workers job is # reading, and scoping keeps the cheap model from wandering into # bash or the filesystem. tools[ { type: agent_toolset_20260401, default_config: {enabled: False}, configs: [ {name: web_search, enabled: True}, {name: web_fetch, enabled: True}, ], } ], system( You are a search worker researching one focused sub-question for a coordinator. Use web_search and web_fetch to find the answer. Be thorough: try multiple query phrasings, follow promising links, and cross-check facts across sources. Report back with the specific answer you found and the evidence (URLs, quotes) that supports it. If you could not find a definitive answer, say exactly what you did find and what remains uncertain. Always finish by calling submit_result. ), betasBETAS, )每个 worker 实例在自己的 session thread 里研究一个聚焦的子问题它读到的超大网页不会进入任何其他 agent 的上下文。worker 会自动获得submit_result和send_to_parent两个工具无需你定义。定义 coordinator大模型只拿 roster不拿工具coordinator 没有任何自己的工具只有一个multiagentroster 指向 worker。正是这个字段让它成为 coordinator服务端会自动给它create_agent、send_to_agent、wait_for_agents、list_agents这些工具你一行都不用写。coordinator client.beta.agents.create( namesearch-coordinator, modelCOORDINATOR_MODEL, multiagent{ type: coordinator, agents: [{type: agent, id: worker.id}], }, system( You are coordinating a team of search workers to answer a hard web-research question. Your workers have web_search and web_fetch; you do not. Break the question into focused sub-questions and delegate each to a worker via create_agent. Run several workers in parallel on independent sub-questions, and ALWAYS call wait_for_agents after spawning before drawing any conclusion. When a worker reports, decide whether its findings answer the sub-question or whether to send a follow-up with send_to_agent. If a worker returns an infrastructure error (rate limit, timeout) instead of findings, re-assign the same sub-question to a fresh worker. Once you have enough evidence, synthesize the workers findings into a single final answer to the original question. ), betasBETAS, ) print(fworker {worker.id}) print(fcoordinator {coordinator.id})这个委派关系有两点需要写进你的操作习惯roster 在 coordinator 创建或更新时被快照。如果之后修改了 worker 的定义必须更新或重建 coordinator否则 coordinator 手里还是旧的名单。coordinator 看不到 roster 里 agent 的任何信息——不是 prompt、不是名字、不是描述。它的create_agent工具只接受一个 agent 名和一个任务字符串。coordinator 对 worker 的一切认知都来自它自己的 system prompt服务端不做校验所以 coordinator prompt 里对 worker 的描述必须和 worker 实际 prompt 保持一致。单一 worker 类型时任何名字都会解析到那一个 worker多种 worker 类型时要在 coordinator prompt 里显式命名。创建带预算的会话并运行研究任务worker 数量和 fan-out 数量是数据依赖的——coordinator 自己决定派几个 workerprompt 里没有硬性上限。所以会话要带一个budget作为整个团队跨所有线程的强制花费上限。这里设$10amount是货币次单位的整数字符串1000即 $10.00正常跑不会碰到但如果某个坏问题让 coordinator 无限派 worker会话会在触顶时暂停而不是继续烧钱。几个与budget相关的规则来自 managed_agents/CMA_cap_session_spend.ipynbamount是整数字符串拒绝小数美分只接受USDAPI 返回的所有成本金额包括usage.list_cost都用同一编码上限按公开 list price 计算覆盖会话内每个线程含子 agent 线程的 token 成本与实际折扣无关budget只能在sessions.create时附上——不带 budget 创建的会话之后永远无法补加触顶后 stop reason 是budget_reached会话进入idle但文件、工具状态、对话都保留用sessions.update把max_list_cost提到已消费成本之上会话自己恢复执行不需要补发消息把上限改到已消费成本以下会收到 400上限在模型请求之间强制所以记录的list_cost可能略微越过max_list_cost——定上限时留一个请求的余量。env client.beta.environments.create( nameresearch-fanout, config{type: anthropic_cloud, networking: {type: unrestricted}}, ) session client.beta.sessions.create( agentcoordinator.id, environment_idenv.id, # guardrail on>def text_of(content): return .join(b.text for b in content or [] if b.type text) def clip(s, n160): return s[:n] (... if len(s) n else ) final_answer with client.beta.sessions.events.stream(session.id, betasBETAS) as stream: for ev in stream: match ev.type: case agent.message: if text : text_of(ev.content).strip(): final_answer text print(f[coordinator] {clip(text, 200)}) case session.thread_created: print(f[spawn] {ev.agent_name} ({ev.session_thread_id})) case agent.thread_message_sent: print(f[delegate - {ev.to_agent_name}] {clip(text_of(ev.content))}) case agent.thread_message_received: print(f[report - {ev.from_agent_name}] {clip(text_of(ev.content))}) case session.status_idle: break print(f\n[team finished in {time.monotonic() - t_start:.0f}s]) print( * 70) print(final_answer)判断委派是否真正发生看三类事件session.thread_createdworker 被 spawn、agent.thread_message_sentcoordinator 把子问题交出去、agent.thread_message_received发现被汇报回来。值得注意的形态是每条[delegate -]都是一条短消息每条[report -]都是一份蒸馏摘要——产生这些报告的那些网页原文从未穿过 coordinator 的上下文。这个隔离就是全部的成本故事。反过来如果一次运行里没有任何[spawn]行说明 coordinator 用自己的知识直接作答、没有委派你只多付了一次前沿模型往返。对照组单个前沿模型按同样的验证标准执行没有这个模式要花多少钱的现实替代是一个带同样两个 web 工具的单个前沿 agent。比较成立有一个前提solo agent 必须被压在同样的验证标准上。放任不管的话前沿模型会自行省着读——每个事实只查一个来源账单确实更低但那是验证标准更低的产品不是同一份工作换了个价格。所以下面的 solo prompt 显式要求每个事实至少两个独立 fetch 交叉验证与团队臂对齐。solo client.beta.agents.create( namesolo-researcher, modelCOORDINATOR_MODEL, tools[ { type: agent_toolset_20260401, default_config: {enabled: False}, configs: [ {name: web_search, enabled: True}, {name: web_fetch, enabled: True}, ], } ], system( You research hard web questions with audit-grade rigor. Use web_search and web_fetch. For EVERY fact you report, verify it from at least two independent fetches (the authoritative page plus one corroborating source), and re-fetch when two sources disagree. Never carry a fact forward on one source or from memory. In your answer, give each fact with both source URLs, and explicitly flag any fact where sources conflicted. Before finishing, audit your own answer: list each claim and check it has two cited sources. ), betasBETAS, ) t_solo time.monotonic() solo_session client.beta.sessions.create(agentsolo.id, environment_idenv.id, betasBETAS) client.beta.sessions.events.send( solo_session.id, betasBETAS, events[{type: user.message, content: [{type: text, text: QUESTION}]}], ) solo_answer with client.beta.sessions.events.stream(solo_session.id, betasBETAS) as stream: for ev in stream: match ev.type: case agent.message: if text : text_of(ev.content).strip(): solo_answer text case session.status_idle: break print(f[solo finished in {time.monotonic() - t_solo:.0f}s]) print(clip(solo_answer, 300))注意 solo 会话没有budget——它是单次对照运行如果你的负载会失控按前面一节补上上限。用每线程 usage 与 list_cost 计量两边成本归属是 API 内建的每个 session thread 都带累计usage会话和每个线程都上报usage.list_cost服务端按公开 list price 计算的 token 成本。列出所有线程parent_thread_id is None的主线程就是 coordinator其余子线程是 workersolo 会话则没有子线程。会话自己的list_cost就是总额主数字不需要自己维护价目表。API 给不了的是一个反事实数字这次团队负载如果全部按前沿价格计费是多少。这需要一张价目表对 token 数重算所以代码里只为这一个 what-if 保留 input/output 价格5 分钟缓存写入按 input 价 1.25x、1 小时写入 2x、缓存读取 0.1x其余全部用真实的list_cost。claude-sonnet-5这里用的是 introductory 价2026-08-31 前 $2/$10之后标准价 $3/$15价目过期后要自行更新。# $ / MTok input and output from the pricing page, used only for the # all-frontier counterfactual below; the real bills come from the APIs # server-side list_cost. PRICES { claude-fable-5: {input: 10.0, output: 50.0}, claude-sonnet-5: {input: 2.0, output: 10.0}, } def total_input(u): cache u.cache_creation # None on threads with no cache activity return ( u.input_tokens u.cache_read_input_tokens (cache.ephemeral_5m_input_tokens if cache else 0) (cache.ephemeral_1h_input_tokens if cache else 0) ) def counterfactual_cost(u, model): Re-price a threads tokens at another models rates (the what-if). p PRICES[model] cache u.cache_creation return ( u.input_tokens * p[input] (cache.ephemeral_5m_input_tokens if cache else 0) * p[input] * 1.25 (cache.ephemeral_1h_input_tokens if cache else 0) * p[input] * 2.0 u.cache_read_input_tokens * p[input] * 0.1 u.output_tokens * p[output] ) / 1e6 def dollars(list_cost): return float(list_cost.amount) / 100 # amount is an integer string in cents def report(session_id): session_usage client.beta.sessions.retrieve(session_id, betasBETAS).usage threads list(client.beta.sessions.threads.list(session_id, betasBETAS)) primary next(t for t in threads if t.parent_thread_id is None) workers [t for t in threads if t.parent_thread_id is not None] workers_in sum(total_input(t.usage) for t in workers) print( f primary thread ({primary.agent.model.id}): f{total_input(primary.usage):9,} in / {primary.usage.output_tokens:6,} out f - ${dollars(primary.usage.list_cost):.2f} ) if workers: print( f {len(workers)} worker(s): {workers_in:9,} in / f{sum(t.usage.output_tokens for t in workers):6,} out f - ${sum(dollars(t.usage.list_cost) for t in workers):.2f} ) print( f workers share of input: {workers_in / (workers_in total_input(primary.usage)):.0%} ) total dollars(session_usage.list_cost) print(f total cost (session usage.list_cost): ${total:.2f}) return total, threads print(split team (fable coordinator sonnet workers):) split_cost, split_threads report(session.id) print(\nsolo frontier agent:) solo_cost, _ report(solo_session.id) # The counterfactual that isolates the rate split: this runs team # workload with every token billed at the frontier rate. frontier_team_cost sum(counterfactual_cost(t.usage, COORDINATOR_MODEL) for t in split_threads) print(f\nsolo / split cost ratio on this pair of runs: {solo_cost / split_cost:.1f}x) print(fthe split teams workload at all-frontier rates: ${frontier_team_cost:.2f})读输出时抓住三行每臂的total cost会话级usage.list_cost、workers share of input团队输入 token 中走 worker 价的比例、最后的solo / split cost ratio。两边做的是几乎相同的阅读量——这正是对齐验证标准的目的。差别在于阅读按什么价格计费、以及工作的形状团队的二十次查询以廉价价格并行跑在 worker 线程里solo agent 在同一个前沿价格上下文里串行磨完。以下数字是 notebook 作者的运行示例不是你会得到的固定值作者两次实验中团队大约便宜 2.5x、快 3x团队 84-98% 的输入 token 按 worker 价计费。token 量每次运行都会变任何单次打印出来的比值都是一个样本稳定的是结构本身。什么时候这个拆分不划算notebook 在构建过程中实际观察到四条限制判断你的负载是否适用时逐条对照比较必须锁死验证标准。放任不管的 solo 前沿 agent 读得少得多每个事实一个来源比团队便宜——但那是另一个、验证标准更低的产品。验证标准固定时拆分的成本优势才成立。委派有地板成本。每个 worker 线程都有固定启动开销。把同一份工作拆成更多、更窄的 brief 反而让作者的账单上升——brief 粒度存在一个最优点不要无限细分。验证标准只覆盖你写进它的东西。作者提交的一次运行里两臂都把 20 个事实对照 nps.gov 核实了但公园名单本身是模型记忆拼出来的把第 10 名的位置给了 Kings Canyon按面积实际第 12 名该位置属于 Great Smoky Mountains。事实被审计了问题分解没有。如果前提本身重要多花一次委派让 worker 去核实它。coordinator 只知道你告诉它的东西。服务端不会把 worker 的 prompt 展示给它见前文 roster 快照一节所以经济学也依赖你在 coordinator prompt 里准确描述 worker 的行为。明确不划算的情形问题太窄没有足够的阅读量可套利coordinator 直接用自己的知识作答而没有委派运行里没有任何[spawn]行等于白付一次前沿往返以及任务需要对原始材料本身做前沿级判断——微妙的文档分析而不是事实核查时廉价阅读者可能恰好把关键内容摘要掉了。另外如果会话撞上budget_reached先确认是预算太小还是 coordinator 在空转前者用sessions.update抬高上限即可恢复后者回到 coordinator prompt 检查委派逻辑。下一步notebook 给出的延伸方向都对应仓库里的现成路径给团队加专用 worker 类型并做 per-role 工具收窄——managed_agents/CMA_coordinate_specialist_team.ipynb 展示了一个三角色团队以及为什么每个角色要有自己的工具范围把每线程计量接入生产遥测——线程级usage是按每次委派而不只是按会话归因花费的方式把同样的拆分套到你自己的 token 密集负载上文档审查和日志分诊与网页研究具有同一种读得多、coverage 型的轮廓。预算暂停、抬升与移除的完整机制在 managed_agents/CMA_cap_session_spend.ipynb 中。【免费下载链接】claude-cookbooksA collection of notebooks/recipes showcasing some fun and effective ways of using Claude.项目地址: https://gitcode.com/GitHub_Trending/an/claude-cookbooks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考