使用动态工作流大规模编排子代理

本文介绍Claude Code的动态工作流功能，用于编排大规模子代理任务。通过JavaScript脚本定义工作流，由Workflow工具在后台执行，支持并行扇出、对抗性验证等模式。相比单代理和子代理，动态工作流将计划固化在代码中，确保验证强制运行，且脚本可复用。文章以事实核查投资者更新为例，展示从Python Agent SDK触发工作流、流式获取进度、阅读生成脚本的完整流程，并讨论适用场景、成本控制及常见问题。

技术背景

在AI工程实践中，许多任务（如文档审计、内容迁移、研究综合）需要处理大量独立子任务，每个子任务都需要真正的判断和验证。单个代理受限于上下文窗口，无法同时处理数十个声明；子代理模式虽可委派，但计划仍由模型逐轮决定，缺乏结构保证，且中间结果堆积在主代理上下文中，难以扩展。动态工作流应运而生，它将编排逻辑从模型上下文转移到代码中，通过脚本精确控制子代理的生成、并行执行、结果汇总和验证流程，确保每个项目得到一致处理，验证步骤不可跳过，且编排本身成为可保存、可复用的资产。

核心原理

动态工作流的核心原理是：用户通过提示词请求使用工作流，Claude模型生成一个JavaScript编排脚本，该脚本调用Workflow工具，由运行时在后台执行。脚本使用agent()函数创建具有干净上下文的子代理，每个子代理只看到脚本提供的提示，并在会话工作目录中运行，继承配置的工具允许列表。脚本通过parallel()实现并行屏障，pipeline()实现流水线处理，普通JavaScript代码处理过滤、循环等逻辑。子代理的结果通过返回值传回脚本变量，最终脚本return的内容传回会话。运行时限制并发16个代理，单次运行上限1000个代理，超出的任务排队。工作流可恢复，失败阶段重试时已完成代理返回缓存结果。

架构分析

系统架构分为三层：用户层通过Python Agent SDK的query()发起请求，设置allowed_tools包含Workflow工具，并描述工作流结构；编排层由Claude模型生成JavaScript脚本，脚本定义meta（名称、阶段）、agent()（子代理生成）、parallel()（并行屏障）、pipeline()（流水线）等原语，以及阶段间的纯JavaScript逻辑；执行层由Workflow工具运行时在后台执行脚本，管理子代理生命周期、并发控制、进度事件流式返回。设计思想是：将计划固化在代码中，确保验证强制运行；子代理上下文干净，避免上下文污染；脚本作为可复用资产，支持保存、编辑、重新运行。

完整实战教程

简介

对一份投资者更新草稿进行事实核查意味着要检查许多独立的声明，对每一项进行真正的判断，并确保验证确实运行：更多的项目、判断和强制验证，超出了单个上下文窗口的舒适容纳范围。动态工作流是Claude Code的一项功能，适用于此类任务，即需要比单个对话能协调的更多代理的任务。Claude不是逐步进行编排，而是为任务编写一个JavaScript编排脚本，并将其传递给Workflow工具，该工具的运行时在后台执行它。脚本并行或分阶段生成子代理，将结果保存在变量中，并以纯代码应用连接逻辑（过滤、循环、验证传递）。由于Agent SDK驱动Claude Code运行时，你可以直接从Python启动工作流。

每个子代理都是一个完整的Claude Code代理。它从干净的上下文开始，只看到脚本给它的提示，并在你的会话工作目录中使用你配置的工具允许列表。每个代理默认在你的会话模型上运行，但脚本可以将某个阶段路由到不同的模型：对于机械提取使用更便宜的模型，对于困难的判断调用使用最强的模型。你可以从提示中控制这一点，或者使用CLAUDE_CODE_SUBAGENT_MODEL环境变量一次性覆盖所有代理。运行时保持最多16个代理并发运行（在CPU核心有限的机器上更少），并将一次运行的上限设为1,000个代理；计划的工作超过并发限制的工作流将排队，直到有槽位释放。

因为计划存在于代码中而不是模型的上下文窗口中，每个项目都得到相同的处理，验证因为脚本规定而运行，单次运行可以协调数十到数百个代理，并且脚本是一个你可以阅读、编辑、保存和重新运行的文件。代价是令牌：许多子代理会使用大量令牌。当任务超出单个上下文窗口、验证必须通过结构强制执行，或者编排本身值得保留时，工作流是值得的。

通过本教程，你将能够：

工作示例对一份投资者更新草稿与其源文档进行事实核查。相同的形状（许多部分、真正的判断、验证步骤）出现在文档审计、内容迁移、研究综合和分类队列中。

注意： 本笔记本中的所有公司名称、指标和客户数据均为虚构，仅用于演示目的。

前置条件

必备知识：

所需工具：

Agent SDK 捆绑了其驱动的 Claude Code CLI，动态工作流就运行在该运行时中。它们需要 v2.1.154 或更高版本 的 CLI，claude-agent-sdk 0.2.90 及更高版本已捆绑该版本。下面的安装单元格设置了最低版本，因此无需其他设置。

动态工作流在所有付费计划中均可用，支持 Anthropic API 访问，以及 Amazon Bedrock、Google Cloud 的 Agent Platform 和 Microsoft Foundry；有关当前详细信息，请参阅动态工作流文档。

成本说明： 工作流会生成许多子代理，因此它们比单代理会话消耗更多 token。完整运行此 notebook 大约需要 $2 到 $4 的 API 使用费用，如果某个阶段失败并重试，费用会更高。工作流运行完成时会打印其成本。

设置

安装 SDK 并加载你的 API 密钥。下面的版本下限保证了捆绑的 Claude Code CLI 支持动态工作流的 SDK 版本，下一个单元格会检查这一点。如果你在已经导入旧版本 SDK 的内核中升级了 SDK，请重启内核，以便 Python 使用新版本。

%%capture
%pip install -U "claude-agent-sdk>=0.2.90" python-dotenv

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), (
    f"claude-agent-sdk {claude_agent_sdk.__version__} is too old for dynamic workflows - "
    "re-run the install cell, then restart the kernel"
)
print(f"claude-agent-sdk {claude_agent_sdk.__version__}")

创建 OrbitCart 工作区

我们虚构的公司是 OrbitCart，一家在线骑行装备商店。下一个单元格将一个小型工作区写入磁盘：investor_update.md，一份包含 10 项事实性声明的投资者更新草稿，以及一个 sources/ 目录，其中存放这些声明应追溯到的真实数据（月度销售 CSV、客户调查结果、正常运行时间报告和新闻提及文件）。

草稿中故意设置了一些问题，以便工作流有实际内容可捕捉：

智能体处理文件，因此数据保存在磁盘上，但我们在笔记本中创建它，以便你确切看到智能体将看到的内容。运行后，我们将把工作流的判定与此答案键进行比较。

WORKSPACE = Path("orbitcart_data").resolve()
SOURCES = WORKSPACE / "sources"
SOURCES.mkdir(parents=True, exist_ok=True)

# --- The draft investor update: 10 claims for the fact-check workflow ---
(WORKSPACE / "investor_update.md").write_text("""\
# OrbitCart - Q2 2026 Investor Update (DRAFT)

*Prepared by the founding team. Not yet reviewed.*

Dear investors,

Q2 was OrbitCart's strongest quarter yet. Highlights:

1. **Revenue:** Q2 revenue reached **$4.8M, up 42% from Q1**.
2. **Order volume:** We processed **over 38,000 orders** in the quarter.
3. **Customers:** Our active customer base grew to **12,400**.
4. **Customer love:** We earned a **Net Promoter Score of 71**.
5. **Reliability:** The platform maintained **99.99% uptime** across the quarter.
6. **Press:** TechPedal called us **"the fastest-growing cycling retailer in North America."**
7. **Growth:** Our June flash sale was **the biggest single sales day in company history**.
8. **Support:** Customer support resolution time improved to **under 4 hours**.
9. **Catalog:** We launched **3 new product categories** this quarter.
10. **Trust:** **Zero customer data was lost** during the brief May service interruption.

We look forward to discussing these results at the board meeting.

- The OrbitCart team
""")

# --- Source documents: the ground truth the claims should trace back to ---
(SOURCES / "monthly_sales.csv").write_text("""\
month,revenue_usd,orders,new_customers
2026-01,1150000,9800,1020
2026-02,1180000,10100,990
2026-03,1270000,9900,1100
2026-04,1480000,12800,1410
2026-05,1520000,12400,1380
2026-06,1610000,13200,1640
""")

(SOURCES / "customer_survey.txt").write_text("""\
OrbitCart Customer Survey - Q2 2026 Results
Survey window: June 10-17, 2026 | Responses: 2,140

Net Promoter Score (NPS): 62  (Q1: 58)
Active customers at end of Q2: 12,400
Support satisfaction (CSAT): 4.4 / 5

Selected verbatim feedback:
- "Ordering was painless and the bike arrived two days early." (Pro customer, Denver)
- "Great selection, but I wish there were more filter options." (Free customer, Toronto)
- "The team plan made it easy to outfit our whole shop." (Team customer, Portland)
""")

(SOURCES / "uptime_report.txt").write_text("""\
OrbitCart Platform Reliability Report - Q2 2026

April 2026:  99.97% uptime, no incidents
May 2026:    99.18% uptime
  - May 14: checkout degraded for 5h 58m during a database failover.
    No customer data was lost. Root cause: connection pool exhaustion.
June 2026:   99.98% uptime, no incidents

Quarter average: 99.71%
""")

(SOURCES / "press_coverage.md").write_text("""\
# Press mentions - Q2 2026

## TechPedal, May 3, 2026
"OrbitCart has quietly become one of the fastest-growing online cycling retailers
in North America, on the strength of a no-friction checkout and a generous
returns window."

## Gear & Grease Weekly, June 21, 2026
"The Portland-based startup recently expanded into bike maintenance tools, indoor
training equipment, and cycling apparel - three categories that traditional bike
shops have struggled to move online."
""")

print(f"Workspace: {WORKSPACE.name}/")
print(f"Investor update claims: 10 | Source documents: {len(list(SOURCES.iterdir()))}")

子代理与工作流

Agent SDK 已经提供了两种运行多步骤任务的方式，而动态工作流则增加了第三种。它们之间的区别归结为一个问题：谁掌握计划？

| | 单一代理 | 子代理 | 动态工作流 |

|---|---|---|---|

| 是什么 | 一个代理逐步工作 | 主代理生成的工人 | 运行时执行的脚本 |

| 谁决定下一步做什么 | Claude，逐轮决定 | Claude，逐轮决定 | 脚本 |

| 中间结果存储在哪里 | Claude 的上下文窗口 | 主代理的上下文窗口 | 脚本变量 |

| 什么可重复 | 提示词 | 工人定义 | 编排本身 |

| 规模 | 一个上下文窗口 | 每轮委派几个任务 | 每次运行数十到数百个代理 |

| 中断 | 重新开始该轮 | 重新开始该轮 | 在同一会话内可恢复 |

使用子代理（来自 notebook 01 的模式），Claude 是编排者。它逐轮决定委派什么，并且没有任何保证它会委派所有部分、在最后合并结果或验证任何内容。每个工人的结果都会回到主代理的上下文中，这对于四次委派来说没问题，但对于四百次则不行。计划也只存在于 Claude 的上下文中：中断运行后，工作将从头开始。

动态工作流则翻转了这些属性。脚本决定运行什么，中间结果存储在脚本变量中，脚本本身是一个你可以阅读、保存、编辑和重新运行的人工制品。代价是令牌，因为工作流会生成许多子代理。当任务大于一个上下文窗口、当你希望通过结构强制验证、或者当编排本身值得保留时，请使用动态工作流。

动态工作流的剖析

从 Agent SDK 触发工作流需要两件事：

  1. 在 allowed_tools 中允许 Workflow 工具。这是 Claude 用来启动工作流的工具；SDK 会自动批准此列表中的工具。
  2. 在提示中用简单的语言请求工作流：“使用工作流来……”。Claude 会将这样的直接请求视为选择将编排作为脚本执行，而不是逐轮工作。（交互式 CLI 添加了关键字触发器和 /effort ultracode 模式；从 SDK 来看，一个措辞简单的请求就足够了。）

当 Claude 调用 Workflow 工具时，它会传递一个完整的 JavaScript 编排脚本，该脚本是它为你任务编写的。运行时在后台执行该脚本：你的 query() 流保持活动状态，并且在代理运行时进度事件会回流。由于启动是一轮，完成是另一轮，因此流会为每一轮产生一个 ResultMessage。最后一个携带最终结果，下面的 run_agent() 辅助函数会保留它。在 SDK 中没有交互式权限提示。工具调用遵循你配置的权限规则，这就是我们设置显式 allowed_tools 列表的原因。（你仍然可能会看到允许列表之外的只读调用（如 Bash）成功：非交互式运行会自动允许只读命令并拒绝其余命令。）工作流生成的子代理以 acceptEdits 模式运行并继承该允许列表。

在首次运行之前，有三件事值得了解：

我们在这里手动编写一个小型记录器，而不是重用文件夹中的 agent_visualizer 辅助函数，因为工作流会发出这些辅助函数不覆盖的 TaskProgressMessage 和 TaskNotificationMessage 事件。下一个单元格定义了我们将用于运行的辅助函数：run_agent() 流式传输查询并打印紧凑的活动日志，workflow_options() 构建 ClaudeAgentOptions，show_script() 显示 Claude 生成的编排脚本。

# 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(r"Script file: (.+?\.js)")
TASK_ID_RE = re.compile(r"Task 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(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    # Show the first line of Claude's 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(f"```javascript\n{Path(path).read_text()}\n```"))


def workflow_options() -> ClaudeAgentOptions:
    """ClaudeAgentOptions for a workflow-enabled agent working in the OrbitCart workspace."""
    return ClaudeAgentOptions(
        cwd=str(WORKSPACE),
        model=MODEL,
        allowed_tools=["Read", "Write", "Edit", "Glob", "Grep", "Workflow"],
        permission_mode="acceptEdits",
        max_turns=40,
    )

示例：核实投资者更新

任务： 法律部门不会批准 investor_update.md，除非每项声明都能追溯到源文件。共有10项声明和4个源文件。

为什么用工作流： 单个代理检查10项声明时，会把所有源文件都加载到上下文中，到第7项声明时就开始走马观花。"再检查一遍你的发现"也只是一条指令，在上下文压力下会被跳过。下面的工作流让验证变得*结构化*：

  1. 提取： 一个代理阅读草稿，提取所有事实性声明
  2. 验证： 每个声明一个代理，各自在干净的上下文中重新阅读源文件
  3. 对抗性检查： 一个怀疑者代理在每项裁决被接受前挑战它
  4. 报告： 一个代理将裁决汇编成一份可行/不可行报告

这就是扇出 + 对抗性验证模式。注意下面的提示词中有多少是在描述工作的*形态*，而不是任务本身。使用工作流时，提示词描述的是你想要的框架。

FACT_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 draft's 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())

display(Markdown(factcheck_info["result"] or "_(no result returned)_"))

将判定结果与答案键进行核对，即我们在数据中埋下的那些曲折点：

| 声明 | 埋下的真相 |

|---|---|

| 1. "$4.8M，增长42%" | 矛盾。 CSV显示第二季度总计461万美元，较第一季度的360万美元增长28% |

| 4. "NPS为71" | 矛盾。 调查显示为62 |

| 5. "99.99%可用性" | 矛盾。 可靠性报告显示为99.71%，且5月份有一次6小时的事故 |

| 6. TechPedal引用 | 矛盾。 文章称其为"*增长最快的* *在线*自行车零售商之一"，而非"增长最快的" |

| 7. 闪购，8. 支持时间 | 无法验证。 没有来源涵盖这两项 |

| 2、3、9、10 | 有来源支持 |

声明6是值得深思的。一个仓促的智能体常常会忽略"增长最快的"和"增长最快的之一"之间的区别。一个专门的验证者，其上下文中没有其他内容，然后受到怀疑者的挑战，就很难被蒙混过关。这种精确性来自于工作流程的结构，而不是更聪明的模型。

如果工作流程在受支持的声明上比我们的答案键更严格，不要感到惊讶。声明3、9和10是怀疑者通行证往往会挑战的类型：调查确认了12,400这个数字，但没有来源为"增长"建立先前的基线（声明3）；新闻文章称公司"最近扩展"到三个类别，但没有确认这些发布发生在*第二季度内*（声明9）；虽然5月份没有数据丢失，但怀疑者可能会反对将六小时的中断称为"短暂"中断（声明10）。无论你的运行是确认它们还是标记它们，这都是对抗性检查在发挥作用。过度严格的标记会花费一分钟的人工审查；而宽松的通过则会损害投资者的信誉。

阅读 Claude 编写的脚本

最具启发性的产物是脚本本身，因为它展示了工作是如何组织的。Workflow 工具的结果包含了持久化脚本的路径，而 run_agent() 捕获了它。在阅读脚本时，请注意强制验证者返回结构化裁决的 schema 选项，以及阶段之间的纯 JavaScript 逻辑：仅将确认的裁决路由给怀疑者，成本为零 token。

show_script(factcheck_info)

如何阅读工作流脚本

每个工作流脚本都使用相同的构建块。完整的选项集记录在 Agent SDK 参考 的 Workflow 工具条目中，但你可以通过将常见原语映射到上面看到的内容来掌握大部分内容：

| 原语 | 作用 |

|---|---|

| export const meta = {name, description, phases} | 声明工作流的名称和阶段；进度 UI 将代理分组到这些阶段下 |

| agent(prompt, options) | 生成一个具有干净上下文的子代理：它只看到给定的提示字符串，其他什么都看不到。选项可以设置 label、phase、用于结构化输出的 JSON schema 以及 model |

| parallel([...]) | 并发运行一批代理，并在继续之前等待所有代理（屏障） |

| pipeline(items, stage1, stage2, ...) | 将每个项目独立地通过一系列阶段运行，因此项目 A 可以处于阶段 2，而项目 B 仍处于阶段 1 |

| phase("...") | 标记后续代理所属的阶段 |

| 调用之间的纯 JavaScript | 过滤、去重、合并和循环，全部精确、即时且无 token 成本 |

| return {...} | 脚本返回的内容将传回你的会话 |

这种设计产生了三个特性：

  1. 计划由代码强制执行。 怀疑者通行证在验证者之后运行，因为脚本的控制流如此排序。不依赖于 Claude 记住指令。
  2. Claude 的上下文保持干净。 每个裁决和引用的源行都存在于脚本变量中。只有最终编译的报告到达启动工作流的会话。
  3. 脚本是你的。 它是磁盘上的一个文件。你可以阅读它（我们刚刚读过）、编辑它、将其保存到项目的 .claude/workflows/ 目录中，并按名称重新运行。编排成为团队可复用的资产。

何时使用工作流

动态工作流位于难度曲线的顶端；单个代理和子代理在它之下仍然是合适的工具。一个合理的决策指南：

| 使用... | 当... |

|---|---|

| 单个代理 | 任务适合在一个上下文窗口中完成，且不需要独立验证。大多数任务属于此类。 |

| 子代理 | 你需要几个专门的工人（如研究员、审阅者），并且主导代理应根据结果调整计划。 |

| 动态工作流 | 任务超出单个上下文窗口（项目多、来源多），需要你无法跳过的验证，或者是一个你会重复的过程。 |

从这个笔记本中值得保留的三个成本控制习惯：

  1. 在提示中明确结构。 你描述的工作流就是你得到的工作流，包括运行多少验证以及哪些阶段可以使用更便宜的模型。
  2. 关注每次运行的成本。 上面的运行打印了一个。验证所有内容的工作流比不验证任何内容的工作流成本更高，而这个拨号是在提示中设置的。
  3. 保持数据规模诚实。 我们使用了10个声明和4个来源，以便这个笔记本在几分钟内运行。相同的结构可以扩展到数百个项目。脚本不介意大小；你的令牌预算介意。

清理

删除此笔记本创建的工作区。如果你想先浏览一下文件和事实核查证据，可以跳过此单元格，并在再次运行事实核查之前重新运行创建工作区的单元格。

shutil.rmtree(WORKSPACE, ignore_errors=True)
print("Workspace removed.")

你学到了什么

回顾引言中的目标：

后续步骤

常见问题

常见错误：1) SDK版本过低导致Workflow工具不可用，需升级claude-agent-sdk>=0.2.90并重启内核；2) 未在allowed_tools中允许Workflow工具，导致Claude无法启动工作流；3) 提示词未明确要求使用工作流，Claude可能选择逐轮执行；4) 子代理尝试写入报告文件而非返回结果，导致结果丢失，应指导代理将发现作为返回值；5) 运行失败时，注意日志中的[Failed]通知，Claude会自动重启工作流，已完成代理返回缓存结果，成本上升属正常。排查方法：检查SDK版本、确认工具列表、查看脚本路径和任务ID、监控进度事件中的token和工具调用数。生产优化：使用CLAUDE_CODE_SUBAGENT_MODEL环境变量统一覆盖子代理模型，对机械提取用便宜模型，对困难判断用强模型；在提示中明确阶段模型分配；控制并发和总代理数；将脚本保存到.claude/workflows/目录以便复用。

企业应用扩展

动态工作流可扩展至多种场景：文档审计（提取声明、并行验证、对抗检查）、内容迁移（分阶段处理大量文件）、研究综合（多来源信息提取与交叉验证）、分类队列（批量分类并验证）。可组合的模式包括流水线（pipeline）、锦标赛（多轮筛选）、分诊队列（按优先级处理）。进一步学习可参考Anthropic多代理系统工程博客，以及Agent SDK参考中的Workflow工具文档。

总结

本文介绍Claude Code的动态工作流功能，用于编排大规模子代理任务。通过JavaScript脚本定义工作流，由Workflow工具在后台执行，支持并行扇出、对抗性验证等模式。相比单代理和子代理，动态工作流将计划固化在代码中，确保验证强制运行，且脚本可复用。文章以事实核查投资者更新为例，展示从Python Agent SDK触发工作流、流式获取进度、阅读生成脚本的完整流程，并讨论适用场景、成本控制及常见问题。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程