如何用 Claude Agent SDK 编排子 Agent：动态工作流并行核验一份报告

如何用 Claude Agent SDK 编排子 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 做这件事很快会遇到瓶颈它把每份源文档都读进上下文到第七条声明时已经开始走马观花而“再检查一遍你的结论”只是一句指令在上下文压力下会被跳过。claude-cookbooks 仓库中的 claude_agent_sdk/08_Dynamic_workflows.ipynb 示范了另一条路径使用动态工作流dynamic workflows。Claude 为任务编写一段 JavaScript 编排脚本传给Workflow工具运行时在后台执行脚本、并行或分阶段地派生子 Agent中间结果存放在脚本变量里核验流程由脚本本身强制执行。本文沿这个 Notebook 走一遍完整流程从 Python Agent SDK 触发工作流、流式查看运行进度、把事实核验的判定结果与答案对照表核对并读懂 Claude 生成的编排脚本。先判断任务是否适合动态工作流Agent SDK 已经有两种跑多步任务的方式动态工作流是第三种。三者的区别在于一个问题谁持有计划who holds the plan单 Agent子 AgentSubagents动态工作流它是什么一个 Agent 逐步工作主 Agent 派生的工作者运行时执行的脚本谁决定下一步跑什么Claude逐轮决定Claude逐轮决定脚本中间结果存在哪Claude 的上下文窗口主 Agent 的上下文窗口脚本变量可复用的部分Prompt工作者定义编排本身规模一个上下文窗口每轮几个委派任务单次运行数十到数百个 Agent中断后重跑当前轮重跑当前轮同一会话内可恢复在子 Agent 模式下Claude 是编排者它逐轮决定委派什么没有任何机制保证它委派了每一片、汇总了结果、或做了核验每个工作者的结果都落回主 Agent 的上下文委派 4 次没问题400 次就不行。动态工作流把这些属性反过来脚本决定什么运行中间结果存在脚本变量里脚本本身是可以读、保存、编辑、重跑的文件。代价是 token——工作流会派生大量子 Agent比单 Agent 会话消耗明显更多。Notebook 给出的使用判据是任务超出一个上下文窗口、需要由结构强制执行的核验、或者编排本身值得保留时才用动态工作流。文档中的决策参考表适合用条件单 Agent任务装得下一个上下文窗口且不需要独立核验。多数任务属于这一档子 Agent需要几个专业工作者如研究员、评审者且希望主 Agent 根据结果动态调整计划动态工作流任务超出一个上下文窗口条目多、来源多需要不能省略的核验或是一个会反复执行的过程准备环境安装 SDK 并验证版本Notebook 声明的前置条件Python 3.11 或更高版本了解async/await在.env文件中配置ANTHROPIC_API_KEYAnthropic API key了解 Agent SDK 基础query()和ClaudeAgentOptions如果没接触过先做 claude_agent_sdk/00_The_one_liner_research_agent.ipynb如果你还没初始化教程系列的环境安装 uv、配置.env、把虚拟环境注册为 Jupyter kernel先按 claude_agent_sdk/README.md 的 Getting Started 操作。动态工作流对版本有硬性要求它运行在 Agent SDK 内置驱动的 Claude Code CLI 中要求 CLIv2.1.154 及以上而claude-agent-sdk0.2.90 及以上版本内置了兼容的 CLI。Notebook 的安装单元格用版本下限来保证这一点%%capture %pip install -U claude-agent-sdk0.2.90 python-dotenv安装后Notebook 的下一个单元格做导入并断言版本防止在旧版 SDK 上继续import re import shutil import time from pathlib import Path from dotenv import load_dotenv from IPython.display import Markdown, display import claude_agent_sdk from claude_agent_sdk import ( AssistantMessage, ClaudeAgentOptions, ResultMessage, TaskNotificationMessage, TaskProgressMessage, TextBlock, ToolResultBlock, ToolUseBlock, UserMessage, query, ) load_dotenv() MODEL claude-sonnet-5 # Dynamic workflows require Claude Code v2.1.154; SDK 0.2.90 and later bundle a compatible CLI. version_match re.match(r(\d)\.(\d)\.(\d), claude_agent_sdk.__version__) sdk_version tuple(int(g) for g in version_match.groups()) if version_match else (0, 0, 0) assert sdk_version (0, 2, 90), ( fclaude-agent-sdk {claude_agent_sdk.__version__} is too old for dynamic workflows - re-run the install cell, then restart the kernel ) print(fclaude-agent-sdk {claude_agent_sdk.__version__})输出示例Notebook 运行记录claude-agent-sdk 0.2.125两个注意点如果断言失败重跑安装单元格并重启 kernel——在已经导入过旧版的 kernel 里升级 SDK 时必须重启让 Python 加载新版本。另外动态工作流在所有付费计划、Anthropic API 访问以及 Amazon Bedrock、Google Cloud 的 Agent Platform、Microsoft Foundry 上可用。成本方面Notebook 明确提示完整跑一遍大约消耗 24 美元 API 用量若有阶段失败并重试会更多工作流运行结束时会在输出中打印本次花费。创建核验工作区OrbitCart 投资者报告Notebook 的下一组单元格把一个小工作区写到磁盘OrbitCart 是虚构公司文中所有公司名、指标和客户数据均为演示用虚构数据。运行该单元格后工作目录中会出现orbitcart_data/ ├── investor_update.md # 草稿投资者报告含 10 条编号声明 └── sources/ # 声明应追溯回的“地面真值” ├── monthly_sales.csv # 月度销售数据 ├── customer_survey.txt # 客户调研结果 ├── uptime_report.txt # 平台可用性报告 └── press_coverage.md # 媒体报道该单元格的定义和输出Notebook 示例输出Workspace: orbitcart_data/ Investor update claims: 10 | Source documents: 4草稿里的 10 条声明是故意植入了问题的这样工作流才有真实的东西可抓。Notebook 给出的答案对照表如下声明植入的真相1. “Q2 营收 $4.8M环比 42%”与来源冲突。CSV 合计 Q2 为 $4.61M相对 Q1 的 $3.6M 增长约 28%4. “NPS 为 71”与来源冲突。调研结果是 625. “99.99% 正常运行时间”与来源冲突。可靠性报告季度平均 99.71%5 月有一次 6 小时事故6. TechPedal 引语与来源冲突。原文是“one ofthe fastest-growingonlinecycling retailers”不是“the fastest-growing”7. 六月闪购、8. 客服响应时间无法核验。没有任何来源覆盖这两项2、3、9、10有来源支持跑完后工作流的判定结果将与这张对照表核对。Agent 处理的是文件所以数据放在磁盘上但在 Notebook 里创建它是为了让你看清 Agent 究竟看到了什么。配置 SDK 选项放行 Workflow 工具从 Agent SDK 触发工作流需要两件事在allowed_tools中允许Workflow工具——这是 Claude 用来启动工作流的工具SDK 会自动批准该列表上的工具在 prompt 里用平实的话直接要求工作流“use a workflow to…”。Claude 会把这样的直接请求当作对“编排即脚本”模式的确认。交互式 CLI 还有关键词触发和/effort ultracode模式但从 SDK 调用时一句明确要求就够了。Notebook 用三个辅助函数承载运行逻辑run_agent()流式执行 query 并打印活动日志、抓取脚本路径和成本show_script()展示 Claude 生成的编排脚本workflow_options()构建ClaudeAgentOptions。完整代码如下# The Workflow tool result mentions the persisted script path and task ID in plain # text; these regexes scrape it. A convenience for this demo, not a stable API. SCRIPT_PATH_RE re.compile(rScript file: (.?\.js)) TASK_ID_RE re.compile(rTask ID: (\S)) async def run_agent(prompt: str, options: ClaudeAgentOptions) - dict: Stream a query through the Agent SDK and print a compact activity log. Returns a dict with the final result text, the total cost, and - when a workflow ran - the path of the orchestration script Claude generated. info {result: None, cost_usd: None, script_path: None} last_tool_count 0 # throttle progress lines started time.monotonic() async for msg in query(promptprompt, optionsoptions): if isinstance(msg, AssistantMessage): for block in msg.content: if isinstance(block, TextBlock): # Show the first line of Claudes narration if narration : block.text.strip(): print(f\nClaude: {narration.splitlines()[0]}) elif isinstance(block, ToolUseBlock): if block.name Workflow: script block.input.get(script, ) n_lines len(script.splitlines()) print( f\n[Workflow tool] Claude wrote a {n_lines}-line orchestration script ) else: print(f - {block.name}) elif isinstance(msg, UserMessage): blocks msg.content if isinstance(msg.content, list) else [] for block in blocks: if not isinstance(block, ToolResultBlock): continue text block.content if isinstance(block.content, str) else str(block.content) if match : SCRIPT_PATH_RE.search(text): info[script_path] match.group(1) task TASK_ID_RE.search(text) task_id task.group(1) if task else ? print(f[Launched] Workflow running in the background (task {task_id})) elif isinstance(msg, TaskProgressMessage): # Progress arrives every few seconds; print a line every ~8 new agent tool calls tools msg.usage.get(tool_uses, 0) if tools - last_tool_count 8: last_tool_count tools tokens msg.usage.get(total_tokens, 0) seconds msg.usage.get(duration_ms, 0) / 1000 print( f ... {tools:3} agent tool calls | {tokens:9,} tokens | {seconds:4.0f}s ) elif isinstance(msg, TaskNotificationMessage): print(f[{msg.status.capitalize()}] {msg.summary}) elif isinstance(msg, ResultMessage): # A background workflow yields one result per turn: the launch turn, then the # final turn after the workflow completes. Keep the latest and report once. info[result] msg.result info[cost_usd] msg.total_cost_usd cost f | ${info[cost_usd]:.2f} if info[cost_usd] is not None else minutes (time.monotonic() - started) / 60 print(f\n Run complete in {minutes:.1f} min{cost} ) return info def show_script(info: dict) - None: Render the orchestration script Claude generated during a run. path info.get(script_path) if not path or not Path(path).exists(): print(No workflow script was generated in this run.) return display(Markdown(fjavascript\n{Path(path).read_text()}\n)) def workflow_options() - ClaudeAgentOptions: ClaudeAgentOptions for a workflow-enabled agent working in the OrbitCart workspace. return ClaudeAgentOptions( cwdstr(WORKSPACE), modelMODEL, allowed_tools[Read, Write, Edit, Glob, Grep, Workflow], permission_modeacceptEdits, max_turns40, )其中WORKSPACE即工作区单元格里定义的Path(orbitcart_data).resolve()。几个与ClaudeAgentOptions相关的行为值得先知道SDK 中没有交互式权限提示工具调用遵循你配置的权限规则所以要显式给出allowed_tools列表子 Agent 以acceptEdits模式运行并继承该允许列表。你可能仍会看到Bash等只读调用在允许列表之外成功——非交互运行会自动放行只读命令、拒绝其余命令每个子 Agent 都是完整的 Claude Code agent干净上下文只看到脚本给它的 prompt在会话工作目录中按允许列表工作。默认都跑在会话模型上但脚本可以把某个阶段路由到别的模型机械性抽取用便宜模型、困难判断用最强模型你可以在 prompt 里引导也可以用CLAUDE_CODE_SUBAGENT_MODEL环境变量一次性覆盖所有 agent脚本是真实产物运行时把它写到~/.claude/projects/下会话目录中的一个文件并把路径包含在工具结果里run_agent()靠正则把它抓下来。运行事实核验工作流任务设定法务在每条声明都能追溯到源文档之前不会放行investor_update.md共 10 条声明、4 份源文件。Notebook 的 prompt 把核验组织成四个阶段——这正是fan-out 对抗式核验adversarial verification模式EXTRACT一个 agent 读草稿把每条事实性声明提取为结构化输出VERIFY每条声明一个 agent、并行运行各自在干净上下文里重新读源文档SKEPTIC每个 “confirmed” 判定都由一个质疑者 agent 复核后才生效REPORT最后一个 agent 把判定汇编成 go/no-go 报告。注意 prompt 描述的是工作的“形状”而不是任务本身——用工作流时prompt 描述的是你想要的运行架座harnessFACT_CHECK_PROMPT \ Use a workflow to fact-check the draft investor update at investor_update.md against the source documents in sources/. Structure the workflow exactly like this: 1. EXTRACT: one agent reads investor_update.md and extracts its ten numbered highlights as structured output (claim number, claim text). Keep the drafts own numbering 1-10 and treat each numbered highlight as exactly one claim, even when it bundles two figures. 2. VERIFY: one agent per claim, running in parallel. Each verifier reads the source documents in sources/ and returns a verdict as structured output: - confirmed if a source directly supports the claim (quote the supporting line) - contradicted if a source conflicts with it (quote the conflicting line and state the correct figure) - unverifiable if no source covers it Verifiers must quote the exact lines they relied on. Pay attention to subtle differences between what a source says and what the draft claims it says. 3. SKEPTIC: for every confirmed verdict, one skeptic agent re-reads the cited source and tries to refute the confirmation. If the skeptic finds the citation does not actually support the claim, the verdict changes to contradicted. 4. REPORT: one final agent compiles a markdown fact-check report: a table of every claim with its verdict and evidence, then a summary of what must be fixed before the update can be sent. Return this report as the workflow result. The working directory is already set to the folder containing these files. Refer to every file by relative path (e.g. investor_update.md, sources/monthly_sales.csv) in the script and in agent prompts; do not embed absolute paths. factcheck_info await run_agent(FACT_CHECK_PROMPT, workflow_options())启动后query()流保持活动工作流在后台运行进度事件持续回流。Notebook 记录的一次运行输出如下示例输出实际数值会不同- Bash - Read - Read - Read - Read [Workflow tool] Claude wrote a 152-line orchestration script [Launched] Workflow running in the background (task wr1292rh8) Claude: The fact-check workflow is running in the background (task wr1292rh8, run wf_1072dc96-6a2). Its fanning out: extract the 10 claims → verify each in parallel against sources/ → adversarially re-check every confirmed verdict → compile the final markdown report. Ill let you know as soon as it completes. ... 8 agent tool calls | 304,709 tokens | 13s ... 56 agent tool calls | 549,412 tokens | 42s [Completed] Dynamic workflow Fact-check the OrbitCart investor update against source documents completed Run complete in 2.5 min | $3.29 进度行大约每新增 8 次 agent 工具调用打印一次格式为 agent 工具调用数、累计 token 数、已耗时。由于启动和完成分属两个回合流会为每个回合产出一个ResultMessage最后一个携带最终结果run_agent()保留的就是它。核对判定结果用下面的单元格展示最终报告display(Markdown(factcheck_info[result] or _(no result returned)_))报告是逐声明的判定表加“发送前必须修改”的汇总。Notebook 示例运行输出的节选示例结果#ClaimVerdictEvidence1Revenue $4.8M, up 42% from Q1Contradictedmonthly_sales.csvQ2 $1.48M$1.52M$1.61M $4.61MQ1 $3.6M → 增长约28%不是 42%6TechPedal: “the fastest-growing cycling retailer in North America”Contradictedpress_coverage.md实际引语“one ofthe fastest-growingonlinecycling retailers in North America”——引用失实、程度夸大7June flash sale biggest sales day in company historyUnverifiable没有任何来源提供按日/闪购数据然后把判定与上文的答案对照表核对。特别值得留意声明 6单个仓促的 agent 经常把 “the fastest-growing” 和 “one of the fastest-growing” 混为一谈而一个上下文里只有这一条声明的专职核验者、外加一个质疑者的挑战就难以蒙混过关。这种精确性来自工作流的结构而不是更聪明的模型。另外Notebook 提醒工作流返回的结果可能比答案对照表更严。声明 3、9、10 正是质疑者阶段容易挑战的类型——调研确认了 12,400 这个数字但没有来源建立“grew”的基线声明 3媒体报道的是“recently expanded”进入三个品类并未确认发布发生在 Q2 之内声明 95 月没有丢数据但把 6 小时宕机称为 “brief” 中断质疑者可能提出异议声明 10。你的运行无论确认还是标记它们都是对抗式核验在起作用。偏严的标记只花一分钟人工复核偏松的放行损害的是在投资人面前的信誉。读懂 Claude 生成的编排脚本用show_script(factcheck_info)可以渲染运行期间生成的编排脚本。脚本路径在Workflow工具结果里给出run_agent()已将其存入info[script_path]。每个工作流脚本都由同一组构件组成原语作用export const meta {name, description, phases}声明工作流名称和阶段进度界面把 agent 归组在这些阶段下agent(prompt, options)派生一个干净上下文的子 Agent它只看到给它的 prompt 字符串。options 可设置label、phase、用于结构化输出的 JSONschema、以及modelparallel([...])并发运行一批 agent并等待全部完成屏障后才继续pipeline(items, stage1, stage2, ...)让每个条目独立地依次通过各阶段条目 A 可以处于阶段 2 而条目 B 还在阶段 1phase(...)标记后续 agent 属于哪个阶段调用之间的普通 JavaScript过滤、去重、合并、循环——精确、即时、零 token 成本return {...}脚本返回什么什么就会回到你的会话读 OrbitCart 这份脚本时Notebook 特别指出两处schema选项强制核验者返回结构化判定阶段之间的普通 JS 逻辑例如只把 “confirmed” 判定路由给质疑者不花任何 token。这种设计带来三个性质计划由代码强制执行。质疑者阶段之所以在核验者之后运行是因为脚本的控制流如此安排不依赖 Claude 记住某条指令Claude 的上下文保持干净。每个判定和引用的源文件行都存在于脚本变量里只有最后汇编的报告回到发起工作流的会话脚本属于你。它是磁盘上的一个文件可以读、可以编辑、可以存进项目的.claude/workflows/目录、按名称重跑。编排就变成了团队可复用的资产。运行限制与常见现象并发与规模上限运行时最多 16 个 agent 并发运行CPU 核心少的机器更少单次运行最多 1,000 个 agent规划出的工作量超过并发上限时会排队直到有槽位空出。脚本本身的权限编排脚本不能碰文件系统和 shell只有它派生的 agent 能。失败与恢复如果某阶段中途失败例如某个 agent 耗尽结构化输出重试次数日志里会出现[Failed]通知Claude 通常会重新启动工作流。已完成的 agent 返回缓存结果重试主要重做失败的那个阶段相应地成本会上升。这是可恢复性在起作用不是 Notebook 坏了。停止的运行在同一会话内可恢复新会话则从头开始。结果的回传方式工作流 agent 把发现以返回值交还给脚本这是结果到达脚本变量的方式。Notebook 的运行中曾有一个 agent 试图写SUMMARY.md这类报告文件被要求改为直接返回内容。真正的产出物转换后的文档、代码、数据文件照旧正常写盘。抓取脚本路径的正则不是稳定 APIrun_agent()里用正则从工具结果的纯文本中解析Script file: ...和Task ID: ...Notebook 明确标注这是演示便利设施不是稳定接口。清理Notebook 末尾的清理单元格执行shutil.rmtree(WORKSPACE, ignore_errorsTrue)会删除orbitcart_data/目录及其中的全部文件。想先查看事实核验证据的话跳过它即可重跑核验前先重跑工作区创建单元格。成本控制与下一步Notebook 总结了三个值得保留的成本控制习惯在 prompt 里显式描述结构。你描述的架座就是你会得到的架座包括核验跑多深、哪些阶段可以用更便宜的模型。盯住单次运行成本。运行结束会打印成本核验一切的工作流比不核验的贵这个旋钮就设在 prompt 里。保持数据规模的诚实。示例用 10 条声明和 4 份来源几分钟内跑完同样的结构可以扩展到数百个条目——脚本不介意规模介意的是你的 token 预算。如果任务还没超过一条上下文窗口、也不需要结构强制的核验单 Agent 或子 Agent 仍是更合适的工具子 Agent 模式的深入内容见 claude_agent_sdk/01_The_chief_of_staff_agent.ipynb。Notebook 给出的延伸方向把同一结构提取、并行核验、对抗质疑、汇编指向你自己的“声明对来源”类问题——报告、营销文案、文档、分诊队列把脚本存进项目的.claude/workflows/并提交让团队按名称重跑同一编排或在 Claude Code CLI 里用/workflows交互式地观察和保存运行。【免费下载链接】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），仅供参考