Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

ADK 2 编排：图、协作和动态工作流

1. 概览

ADK 2 的标题是三种编排模式。在此 Codelab 中，您将通过构建一个应用（即 Marathon Race Day Coach）来学习这三种技术，每次学习一个可运行的环节。每个级别回答一个问题、添加一个创意，并自行运行。

学习内容

图工作流（支柱 1）- 当您可以在输入到达之前绘制流程时。
协作型代理（支柱 2）- 当您知道团队，但请求会选择子集时 - 以及所有三种协作模式（chat / task / single_turn），每种模式都会实时运行。
动态工作流（支柱 3）- 当工作本身的形态取决于输入时。
如何选择 - 包含一个问题的决策树，以及模式的组成方式。

贯穿始终的主线

已知结构 → 已知团队 / 变量子集 → 未知形状 → 选择正确的形状

您的学习路线图

构建内容

一款应用（即 Marathon Race Day Coach）一次组装一个可运行的关卡。每个级别都是一个纯 Python 模块，您可以在终端中运行；到 L5 时，您将拥有以下所有内容。

该图是根据正在运行的代码绘制的：每条实线都是从 Workflow.graph.edges 中读取的。这是第一个经验教训 - 可以提前绘制的部分正是支柱 1，而无法提前绘制的部分正是支柱 2 和 3 存在的原因。

整个应用 - 以及图表无法显示的内容

所需条件

Google 账号（用于 Colab）- 无需进行本地设置。
约 50 分钟（两个 L4 级别耗时较长，请预留充足的时间）。
访问 Gemini 模型的两种方式之一。选择您的跑道 - 您可以运行一个设置步骤，然后跳过另一个：

	🎓 研讨会	🏠 课后练习
参与者	您正在参加现场研讨会，并且讲师向您提供了学分兑换链接	其他所有人（包括研讨会参与者）
您需要	声明链接，以及可创建云项目的 Google 账号	免费的 AI Studio API 密钥
在以下设备上运行	Vertex AI，在通过研讨会积分付费的项目中	Google AI Studio
费用	折抵金额涵盖的费用	免费层级
设置步骤	研讨会设置（下一步）	回家后设置（后续步骤）

从序言开始，一切都相同，无论采用哪种方式 - 轨道仅决定笔记本与哪个模型端点通信。

两种跟进方式

以下每个步骤都对应于 Colab 笔记本中的一个单元格和 GitHub 代码库中的一个文件夹。您可以选择以下任一选项：

▶ Colab（推荐）： 打开笔记本 → 从上到下运行单元格。
💻 本地： git clone 代码库 ./setup_venv.sh，然后将每个级别作为模块 (python -m ...) 运行，或使用 ./run.sh (adk web) 浏览所有级别。

2. 研讨会设置 · 申领抵用金并切换到 Vertex AI

在研讨会上，您将获得 Google Cloud 赠金。您将声明该账号，创建一个向该账号付费的项目，并将笔记本指向 Vertex AI 而不是 AI Studio。一个单元格包含声明后的所有内容。

1 · 领取赠金（约 1 分钟）

打开教师分享给您的领取链接。该路径类似于 https://me.developers.google.com/benefits/claim/your-workshop-name。
登录并按照页面上的说明接受赠金。
记下您使用的 Google 账号。以下每个步骤都必须以该账号运行。

2 · 打开笔记本并安装 ADK 2 （约 1 分钟）

点击在 Colab 中打开 ▶，然后运行第一个代码单元。它会固定此 Codelab 经过验证的确切 ADK 2 版本，并打印 ✓ installed。

3 · 运行“Workshop setup”单元格（约 3 分钟）

这是标题为 🎓 路径 A · 研讨会的单元格。运行该代码，Colab 会要求您进行授权 - 选择您刚刚用于申领抵用金的同一 Google 账号，然后允许访问。

此代码块会执行以下四项操作：在您的信用额度下创建一个名为 adk-2-tutorial-XXXX 的项目，启用该项目中的 Vertex AI API，设置每个后续代码块读取的四个环境变量，然后向 Vertex 发出测试调用并等待其响应 - 因此，设置要么完成，要么会告知您原因，而不是在后续级别中失败。

预期输出 - 最后一行才是最重要的：

Signed in as: you@example.com
...
Successfully created GCP project 'adk-2-tutorial-4817'.
Successfully linked 'adk-2-tutorial-4817' to billing account '01ABCD-...'.
   waiting for Vertex AI to come up on the new project... (10s)
   waiting for Vertex AI to come up on the new project... (20s)

✅ Vertex AI on adk-2-tutorial-4817 · us-central1 · gemini-2.5-flash — answered a test call

403 Permission 'aiplatform.endpoints.predict' denied ... (or it may not exist)? 这是最有可能出错的地方，但不是结算、信用额度或权限问题 - 上面的行已确认您的项目存在并且信用额度已关联。刚刚创建的 Google Cloud 项目（几秒钟前）还不能提供服务：启用 API 并授予您所有者权限的操作在命令返回后仍会传播一两分钟。查看错误的 reason 以确认，它显示的是 IAM_PERMISSION_DENIED，而不是 SERVICE_DISABLED。操作步骤：无需执行任何操作。设置单元格已重试两分钟，这几乎涵盖了所有情况；如果仍出现 403 错误，请重新运行设置单元格，并再等待一分钟。（运行的是此笔记本的旧副本，即在重试功能推出之前创建的副本？请通过上述链接重新打开该页面。）

4 · 跳过“在家设置”步骤

请勿运行 AI Studio 密钥单元格，否则笔记本会切换回 AI Studio，并撤消您刚刚所做的操作。（该单元格会防范这种情况并拒绝运行，但更简洁的做法是直接跳过它。）直接从此处前往共享构建块单元格。

5 · 运行“共享组成要素”单元格

运行一次。它定义了从 L2 开始的每个级别都会重复使用的 Pydantic 架构和预设的马拉松场景。您会看到 ✓ schemas + scenarios ready。

研讨会结束后

您的积分和使用积分创建的项目不会永久保留。如需在研讨会结束后继续免费重新运行这些关卡，请改为运行回家设置步骤 - 免费 AI Studio 密钥，无需云项目，无需结算信息。只有该单元格会发生变化。

如需更快地清理，请打开 Cloud 控制台，选择 adk-2-tutorial-XXXX，然后将其删除。本 Codelab 中的其他内容不会创建可计费资源。

3. 自学设置 · AI Studio API 密钥

此路径中的所有内容均使用免费的 Google AI Studio API 密钥运行，无需 Google Cloud 云项目、结算或本地安装。整个步骤大约需要 3 分钟。

1 · 打开笔记本

点击 在 Colab 中打开 ▶。您将进入笔记本，其中包含一个 Markdown 简介，然后每个级别对应一个可运行的单元。您从上到下运行单元格；每个单元格都会在其正下方打印自己的输出。

2 · 安装 ADK 2 （约 1 分钟）

运行第一个代码单元。它会固定此 Codelab 经过验证的确切版本：

%pip install -q "google-adk==2.3.0" python-dotenv pydantic nest_asyncio

等待此过程完成，您会看到 ✓ installed。（首次安装需要大约 30-60 秒；之后会缓存。）

3 · 从 AI Studio 获取 Gemini API 密钥（约 1 分钟）

在新浏览器标签页中打开 aistudio.google.com/app/apikey。
使用您的 Google 账号登录。
点击右上角的创建 API 密钥。
选择现有 Google 项目或让它创建一个项目。
复制密钥 - 密钥以 AIza... 开头，长度约为 40 个字符。

4 · 将密钥添加到 Colab （约 1 分钟）

选项 A - Colab Secret（推荐；密钥保持隐藏状态）：

点击 Colab 左侧边栏中的钥匙图标 🔑。
点击 + 添加新 Secret。
将名称设置为 GOOGLE_API_KEY。
将密钥粘贴到值中。
将笔记本访问权限切换为开启。

选项 B - 在系统提示时粘贴（快速）：跳过密钥；当您运行下一个单元格时，系统会显示一个隐藏的提示 🔑 Enter your Google AI Studio API key: - 粘贴并按 Enter 键。

5 · 运行密钥单元

它会读取密钥（或回退到粘贴提示），然后将 ADK 指向 AI Studio（而非 Vertex AI）：

import os

# 🏠 TAKE-HOME ONLY — if you ran the Workshop setup cell, skip this one.
if os.environ.get("GOOGLE_GENAI_USE_VERTEXAI") == "True":
    raise SystemExit("✋ You're set up on the workshop path (Vertex AI). Skip this cell.")

# Google AI Studio API key — add GOOGLE_API_KEY in the 🔑 Secrets panel (or paste when prompted).
try:
    from google.colab import userdata
    key = userdata.get("GOOGLE_API_KEY")
except Exception:
    import getpass
    key = getpass.getpass("Enter your Google AI Studio API key: ")

os.environ["GOOGLE_API_KEY"] = "".join(key.split())    # drop any stray whitespace/newlines
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "False"      # use AI Studio, not Vertex AI
print("✅ API key set — using Google AI Studio.")

预期输出： ✅ API key set — using Google AI Studio.

6 · 运行“共享构建块”单元格

运行一次 Shared building blocks 单元。它定义了从 L2 开始的每个级别都会重复使用的 Pydantic 架构和预设的马拉松场景。您会看到 ✓ schemas + scenarios ready。

设置完成！🎽 在介绍 L0（每个人首先构建的版本）之前，我们先简单介绍一下。

4. 序言 · 为什么不使用一个大型提示？

⚡ 在运行之前，请确定要关注的一件事：每个具体数字来自哪里？这就是整个练习，其他一切都是装饰。

在梯子之前，运行梯子所取代的内容：一个提示承诺一切的代理 - 获取天气、分析课程、读取训练日志、按条件规划路线、输出计划。

您会看到：一个自信、具体、格式良好的策略，但其中的数字是编造的。在一次实际运行中，它以 “I have pulled today's weather metrics”（我已提取今天的天气指标）开头，报告了 52°F 的气温、9 英里的风速，并分析了它从未见过的训练日志。这里没有天气 API，没有课程数据，也没有日志 - 一个不透明的模型调用要么捏造输入，要么将输入对冲为无用。

这种疾病有四种值得一提的症状：

您无法信任它 - 这些数据是流畅地编造出来的。
您无法对其进行测试 - 第 4 步的路由位于散文中；没有 if 可用于进行单元测试。
无法交换步骤 - 没有可插入真实天气 API 的接缝。
您每次都需为所有内容付费 - 五个步骤，一次巨大的调用，没有缓存确定性部分。

保持这种感觉。接下来的九个级别会从提示中逐步移除这些步骤：函数提取（L1-L2a）、if 语句路由（L2b）、专家分工（L3a-L3b）和代码限定形状（L4a-L4b）。

超级提示教练 - 自信，图表后没有任何内容

💻 本地： python -m shared.prologue

5. L0 · 您的首个 ADK 2 智能体

路线图 - 您目前处于：L0

⚡ 总结：智能体是模型 + 指令 + 可能调用的工具；Runner 执行智能体。此级别之后的所有内容都只是更多代理，以更出色的形状排列。

问题：能否让模型回答问题，并在算术运算很重要时使用真实代码？

一个想法 - 三个部分：

Agent - 可进行推理的实体（Gemini 模型 + 指令）。
Runner - 在会话中执行代理并流式传输事件的实体。
工具 - 模型决定调用的普通 Python 函数 (pace_splits)。ADK 会读取签名和文档字符串，并向模型提供声明；无需编写架构。

在序言之后，这是第一个修复：在脑海中进行配速计算的 LLM 很容易出错 - pace_splits 是确定性的 Python，因此答案中的数字是计算出来的，而不是临时想出来的。

▶ Colab：运行 L0 代码单元 · 📁 GitHub：L0_first_agent/ · 💻 本地：python -m L0_first_agent.agent

L0 流程

def pace_splits(target_finish: str) -> dict:
    """Convert a goal time like '3:30:00' into exact per-mile / per-km paces."""
    ...                                  # deterministic Python — no LLM

pace_coach = Agent(
    name="pace_coach", model=MODEL,
    tools=[pace_splits],                 # the model may call it; ADK reads the signature
    instruction="You are a friendly, concise marathon coach. ... If the runner "
                "mentions a goal time, call pace_splits — never do arithmetic yourself.",
)
runner = Runner(node=pace_coach, session_service=InMemorySessionService(), auto_create_session=True)
async for event in runner.run_async(user_id="u1", session_id="s1", new_message=msg):
    ...  # events carry the model's text

🔍 标记： Agent(...) · tools=[pace_splits] · Runner(...)。在输出中，🔧 行表示模型在回答过程中决定调用您的代码。

您将看到的内容：

   🔧 model called tool → pace_splits({'target_finish': '3:30:00'})
   🔧 tool returned     → {'per_mile': '8:00', 'per_km': '4:58', ...}
🧠 Coach: To finish in 3:30:00, you need an average pace of 8:00 per mile...

带有 🔧 的行是本课的重点：在回答过程中，模型选择调用您的函数，并且回答中的确切 8:00/mile 来自您的代码，而不是来自令牌统计信息。

❓ 您可能想知道： 模型是否总是会调用工具？不会，系统会根据每个问题来决定。询问不含数字的问题，🔧 行就会消失（园地会要求您尝试这样做）。

👀 阅读： pace_splits（一个普通函数）和 tools=[pace_splits] 行。· ▶ 运行。· ✏️ 更改：提出一般问题（没有目标时间）- 注意到 🔧 行消失了：模型会自行决定何时值得调用工具。然后重写 instruction 并重新运行，该指令是程序的其余部分。

6. L1 · 您的第一个工作流

路线图 - 您目前处于：L1

⚡ 总结：普通函数和 LLM 代理是同类节点。可预测的工作 → 函数（0 LLM，确定性）；推理 → 代理。

问题：如何在同一流程中混合使用纯代码和 LLM，而无需为仅包含代码的部分支付模型调用费用？

一个想法：在 Workflow 中，纯 Python 函数和 LLM 代理都只是同一 edges 列表中的节点。

START ──► fetch_conditions (function, 0 LLM) ──► advise (agent, 1 LLM)

▶ Colab：运行 L1 代码单元 · 📁 GitHub：L1_graph_basics/ · 💻 本地：python -m L1_graph_basics.workflow

L1 流程

函数节点会输出其生成的数据（无模型调用），然后智能体给出建议，其中引用了其收到的实际温度和风力：

def fetch_conditions(node_input):                # function node — 0 LLM
    return Event(output=Conditions(temp_f=78, wind_mph=12, conditions="sunny").model_dump())

advise = Agent(name="advise", model=MODEL, mode="single_turn",
               input_schema=Conditions, instruction="...give pacing + gear advice...")

workflow = Workflow(edges=[(START, fetch_conditions, advise)])

🔍 标记：一个边缘元组 ((START, fetch_conditions, advise))，中间是一个裸 Python 函数，input_schema= 用于验证移交。

与 L0 相比，新版有何变化： Workflow(edges=[...])、START（输入进入的位置）、返回 Event(output=...) 的函数节点，以及 input_schema=Conditions，因此在代理看到函数输出之前，系统会根据该架构验证函数输出（以 JSON 文本的形式 - input_schema 验证边界，不会向代理提供 Python 对象）。

❓ 您可能想知道： 是否必须先调用函数，然后再调用代理？否 - 任何顺序、任何组合、任何数量。advise 仅在第二位运行，因为它需要 fetch_conditions 的数据。教训是贵族制度，而不是序列。

👀 阅读：fetch_conditions 返回的数据不包含模型调用；advise 包含 input_schema=Conditions。· ▶ 运行。· ✏️ 更改：在函数中设置 temp_f=30 并重新运行 - 建议会发生变化，但函数仍需 0 次 LLM 调用。

7. L2a · 并行扇出 + JoinNode（支柱 1a）

路线图 - 您目前处于：L2a

⚡ 总结：并行分发（免费），等待所有结果，打包，将完整信息交给一个代理。

问题：您能否在输入到达之前绘制流程？从框架开始：并行收集数据，将其打包，然后交给一个代理。

形状：

START ──► fetch_weather ──┐
START ──► analyze_course ─┼─► JoinNode ─► strategy (1 agent)
START ──► pull_fitness ───┘   (bundles)

▶ Colab：运行 L2a 代码单元 · 📁 GitHub：L2a_parallel_join/ · 💻 本地：python -m L2a_parallel_join.workflow

L2a 流程

🔍 标记：全部从 START 开始的三个边（即扇出）和交汇点 JoinNode。

这三个提取操作是函数，它们并行运行，0 次 LLM 调用。
JoinNode 会等待所有这三个结果，并将它们捆绑到一个类型化的载荷 (BundledRunData) 中，该载荷以函数名称为键。
一个 strategy 代理读取该软件包并写入 RaceStrategy。

您会看到：每次提取都会输出一个 started / finished 时间戳。这三个请求都从 0.0 秒开始，扇出在 2.0 秒结束，这是最慢的提取时间，而不是它们时长之和 4.5 秒。这种重叠就是并行性。（最后打印的总实际时间约为 8 秒，因为其中还包含策略代理的 LLM 调用 - 请读取并行声明的提取时间戳，而不是总时间。）

💡 序幕回调：超级提示编造了天气。这里，温度来自提取函数 function - 真实代码，真实接缝。将预设字典替换为实际的天气 API，其他内容保持不变。

❓ 您可能想知道：

JoinNode

需要了解哪些内容？一句话总结：它会一直等到每个并行分支都完成，将输出打包到一个以源函数名称为键的字典中，并且本身不进行任何计算。正是因为这个字典，L2b 的路由器才能写入 node_input["fetch_weather"]["temp_f"]。

👀 读取：三条边从 START 扇出；JoinNode 将它们捆绑在一起，供一个代理使用。· ▶ 运行该脚本并读取时间戳，而不是总时间。· ✏️ 更改：使一次提取休眠 3.0 - 先预测新的扇出结束时间，然后进行验证。

8. L2b · 添加确定性路由器（支柱 1b）

路线图 - 您目前处于：L2b

⚡ 总结：L2a 未触及 + 一个简单的 if 决定运行哪个代理。分支，但不询问模型。

问题：应根据天气炎热或寒冷来调整方案。如何在不要求模型做出决定的情况下进行分支？

形状（L2a + 路由器）：

... JoinNode ─► route_by_weather ─► hot_strategy
               (if-statement)   ─► normal_strategy
                                ─► cold_strategy

▶ Colab：运行 L2b 代码单元 - 尝试 run("NORMAL") / run("COLD") · 📁 GitHub：L2b_router/ · 💻 本地：python -m L2b_router.workflow COLD

L2b 流程

def route_by_weather(node_input):                        # an if-statement, 0 LLM
    temp = node_input["fetch_weather"]["temp_f"]
    route = "HOT" if temp >= 70 else "COLD" if temp <= 40 else "NORMAL"
    return Event(output=node_input, route=route)

(route_by_weather, {"HOT": hot_strategy, "NORMAL": normal_strategy, "COLD": cold_strategy})

🔍 标记： Event(output=..., route=...) - 用于命名路径的函数节点，以及将名称映射到节点的 dict-edge {"HOT": ..., "NORMAL": ..., "COLD": ...}。

总结 - 三种工作，三种住宅：

可预测的工作 → 函数（3 个并行提取）
明确的规则 → 显式路由（route_by_weather 是 if 语句，而不是模型决策）
推理 → 模型（运行的策略代理数量为 1）

您会看到： temp=78F -> route=HOT，然后是结构化的 RaceStrategy。净成本：1 次 LLM 调用。

⚠️ 如果您添加了第四个分支，请为 route-dict 也添加一个 DEFAULT_ROUTE 条目。字典不匹配的路由不是错误，分支只是结束，程序以 0 退出，没有输出，这对于调试来说是一个令人困惑的死胡同。

❓ 您可能想知道： 那么 L2b 实际上就是 L2a 加上一个路由器？可以 - 提取和联接保持不变，仍然只有 1 次 LLM 调用。更改内容：“始终是同一位客服人员”变为“三位客服人员之一，由数据选择”。

👀 阅读： route_by_weather - 路由器是 if 语句，而不是代理。· ▶ 运行 run("COLD")。· ✏️ 更改：添加包含第四个代理的 WINDY 分支 - 之前请先阅读上方的 DEFAULT_ROUTE 警告。

9. L3a · 协作型智能体：一面旗帜，两个世界 - 支柱 2

路线图 - 您目前处于：L3a

⚡ 总结：同一团队，一面旗帜。chat 将整个对话交给一位专家，然后不再返回；single_turn 将每位专家变成一个工具 - 并行子集、自动返回、一次合成。

问题：您知道团队，但请求决定了哪些成员应该回答。如何让 LLM 选择子集并并发运行？

形状：一名协调员负责六位专家（医疗、天气、配速、装备、营养、心理）。此级别会让同一团队运行两次 - 相同的协调员提示，相同的六位专家。唯一的区别在于子代理上的一个标志。对比就是经验。

▶ Colab：运行 L3a 代码单元 · 📁 GitHub：L3a_collaborative/ · 💻 本地：python -m L3a_collaborative.concierge --mode chat "What about fueling?"

L3a 流

🔍 标记： 工厂中的 mode="single_turn" 和输出中的 TRANSFER →（第 1 拍）与共享一个时间戳的 DISPATCH → 行突发（第 2 拍）。

Beat 1 · 先运行默认值 - 并观察它是否会使作业失败

未写入 mode= → 分代理默认值为 chat。您将看到的内容：

TRANSFER → nutrition_specialist   (transfer_to_agent — the only tool chat subagents provide)
Final speaker: nutrition_specialist

协调器没有委托工具 - 聊天子代理只会向一位专家提供 transfer_to_agent，即整个对话的序列化移交。该专家直接回答用户的问题，然后运行结束。不并行调度。不可退货。无合成。如果提出宽泛的问题，情况会更糟：需要 6 位专家，转接 1 次。

这不是错误，而是对话模式在正常运行。在有人明确转移会话之前，会话归持有者所有。适合开放式助理；不适合流水线步骤。

Beat 2 · 一面旗帜，两个世界

唯一区别：每位专家的 mode="single_turn"。相同的问题，再次运行：

[t= 7.8s] DISPATCH → medical_specialist      ← same timestamp =
[t= 7.8s] DISPATCH → weather_specialist         one turn, many calls
[t=14.5s]   ↩ medical_specialist replied     ← replies land inside
[t=14.5s]   ↩ weather_specialist replied        one short window
🧠 Concierge (synthesized): <one answer>

现在，ADK 会为每位专家注入一个委托工具，该工具以子代理命名，并由其 description= 进行描述（协调器在选择子集时会读取该文本；跳过该文本，您将仅根据名称进行路由）。协调器在一个回合中发出多个调用，ADK 并行运行这些调用，每个调用都会自动返回结果，然后协调器进行合成。

问题	开除员工的专家
“那加油呢？”	仅限营养
“我在跑完 18 英里后膝盖开始疼痛”	仅限医疗用途
“我今天应该参加比赛吗？”	医疗 + 天气 + 配速
“我需要担心什么吗？”	全部 6 个

为什么每个专家都会收到完整的简报：每个 single_turn 子代理都在其自己的隔离会话分支中运行，无法看到对话或其同级代理。没有什么是环境：协调器必须将整个 SpecialistInput（问题 + 策略 + runner 数据）单独转发到每个并行调用中。

💡 ADK 2 为此提供了一个直接的归宿：LLM 选择一个按请求划分的子集并并行运行该子集 - 通过 sub_agents + mode="single_turn" 进行声明。在 1.x 中，您可以通过将每个专家封装在 AgentTool 中来组装相同的形状；不同之处在于，现在它是一个声明，而不是管道。（ParallelAgent 始终为 all，transfer_to_agent 为 serial。）

⚠️ 有两点需要注意：(1) 模型会选择子集，因此它比 L2 的硬编码路由器更不确定，确切的子集可能会因运行而异。(2) 有时，您会看到一位专家的 Error validating input: ... 行。它几乎从不属于专业人士的输出 - output_schema 使 Gemini 在服务器端强制执行此操作。这是输入：协调器必须为每个并行调用原封不动地重现整个嵌套的 SpecialistInput，但有时会出错。ADK 将该错误作为工具的结果返回，协调器会恢复，并且合成仍会完成。

❓ 您可能想知道：

chat

仅限 1.x 样式的委托 - 一次只能有一个代理？基本上是的：这是 1.x 的默认行为，现在有了名称。single_turn 与 transfer_to_agent 之间的差距是三维的：协调员持有的内容（一个 transfer_to_agent 与每个专家一个工具 ）· 可以工作的数量（一个，负责对话与 N 个并行）· 控制权是否返回（从不返回与自动返回，并附带结果）。关于代码：工厂的 if mode == 分支仅用于对比，以便一个团队可以采用两种方式进行构建 - 实际应用会硬编码一种模式，而 if 会消失。

👀 读取：_specialist 工厂 - mode 参数是整个级别。· ▶ 运行两个节拍。· ✏️ 更改：询问“我在跑完 18 英里后膝盖疼痛”- 先预测子集，然后检查 DISPATCH 行。

# The factory's mode parameter is THE variable this level teaches:
def _specialist(name, domain, focus, mode):
    kwargs = {}
    if mode == "single_turn":   # the structured contract only makes sense for a TOOL
        kwargs = dict(mode="single_turn",
                      input_schema=SpecialistInput, output_schema=SpecialistResponse)
    return Agent(name=name, model=MODEL,
                 description=f"Marathon {domain} specialist. Consult for: {focus}.",
                 instruction=..., **kwargs)

race_concierge = Agent(name="race_concierge", model=MODEL,
                       sub_agents=[...six specialists...],   # NOTE: no `mode` on the coordinator
                       instruction="...DECIDE which specialists are relevant... call them IN PARALLEL... SYNTHESIZE...")

10. L3b · 任务模式：有终点的对话 - Pillar 2

路线图 - 您目前处于：L3b

⚡ 简而言之：中间模式 - 与用户对话直到收集到字段，然后自动返回经过验证的对象。

问题：L3a 留下了空白。chat 拥有整个对话；single_turn 从不与用户对话。但实际的接收工作介于两者之间：“与用户对话，直到收集到 X，然后返回经过验证的对象。”这是什么模式？

形状：

race_desk (coordinator)
  └─ gear_fitter (mode="task", output_schema=GearOrder)

▶ Colab：运行 L3b 代码单元 · 📁 GitHub：L3b_task_desk/ · 💻 本地：python -m L3b_task_desk.desk

L3b 流程

gear_fitter 保持任务打开状态 - 暂停的任务，而不是挂起

🔍 标记： 同一代理上的 mode="task" + output_schema=，以及输出中的 ⏸ 暂停和 finish_task 调用。

您将看到的内容：

━━ TURN 1 ━━  user: 'I need shoes for the marathon.'
  race_desk → delegate: gear_fitter
  gear_fitter: What is your shoe size?
  ⏸  The run ENDED — but nothing failed. This is a PAUSED task.

━━ TURN 2 ━━  user: 'Size 9, wide.'   (same session → resumes the task)
  gear_fitter → finish_task   (payload validates as GearOrder)
  race_desk: Your order ... in size 9 Wide has been confirmed.

发生了以下三件事，而 L3a 模式无法做到：

跑步在任务中途真正停止 - 任务处于暂停状态，而不是挂起或失败。智能体提出了澄清问题，并保持任务处于打开状态。（在 adk web 中，您只需输入答案；测试框架会将其作为同一会话中的第二条消息编写脚本。）
后续消息会继续与同一任务代理互动，无需重新路由，也无需重新委派。会话知道谁在等待。
finish_task 结束了它 - ADK 因 mode="task" 而注入的工具。代理必须调用它才能完成，并且其载荷必须根据 output_schema 进行验证。与具有类型化完成行的对话 - 然后控制自动返回到协调器，并附上结果。

选择模式的单问题规则

💡 “用户是否需要与该实体对话，以及需要对话到什么时候？”聊天 = 无限期 · 任务 = 直到收集完字段 · 单轮 = 从不。

模式	人机协同 (human-in-the-loop)	平行？	返回到父级
`chat` （子代理默认）- 支持助理，开放式 Copilot	完整对话	否	手动（通过转移）
`task` - 咨询、预订、问题排查	仅限澄清式问题	否	自动（通过 `finish_task`，使用已验证的对象）
`single_turn` - 分类、提取、判断、生成	无	是	自动（附带结果）

mode 仅在子代理上运行，绝不在协调器上运行。工作流节点默认为 single_turn（这就是 L1-L2b 从未写入它的原因），而子代理默认为 chat（这就是 L3a 必须写入它的原因）。

⚠️ 在此基础上构建之前，请注意以下两个版本说明：(1) task

作为静态图节点是版本相关的 - 在 2.0.0b1-2.3.0（此 Codelab 的 pin）上，Workflow(...) 在构建时引发；使用此级别完全相同的操作（具有任务子代理的聊天协调器）或通过 ctx.run_node 进行调度。已在 2.5.0 中提升。(2) “任务代理必须是叶代理”（没有自己的子代理）是已记录在案的 ADK 限制，但属于合同，而不是运行时防护措施：2.3.0 和 2.5.0 都不会阻止您。请勿将未出现错误视为获得许可。

💡深入了解：嵌入在图工作流（2.5.0 及更高版本的形状）中的 task 代理，其路由可以使对话循环返回以进行重试：配套代码库 22_agent_in_workflow · 完整模式指南：docs/agent-modes.md。

❓ 您可能想知道：

task

买到其他两款产品买不到的商品？三件事：自动返回（聊天会继续对话）· 输入完成线（finish_task 的载荷必须根据架构进行验证 - 您会收到数据，而不是转写内容）· 暂停/恢复（⏸ 是等待人工干预的暂停任务，而不是挂起）。

👀 读作： gear_fitter — mode="task" + output_schema 是整个合同。· ▶ 运行。· ✏️ 更改： run_desk("I need a hydration vest", "2 liters, medium") - 澄清式问题会根据情况调整，终点线保持输入状态。

11. L4a · 运行时大小的并行扇出（支柱 3a）

路线图 - 您目前处于：L4a

⚡ 总结：骨架仍然是三个静态步骤 - 动态隐藏在中间步骤中，宽度由运行时的数据决定。

⚠️ 提醒：这是阶梯中最陡峭的一步。之前的级别有 44 行；这个级别有大约 120 行 - 三个代理和两个工作流节点，没有填充。预留约 15 分钟时间，并重点关注末尾的“读/运行/更改”行：您无需在第一次浏览时就理解每一行。

问题：作业的形状取决于输入。您无法提前绘制图表。从运行时宽度开始：让 LLM 决定子问题的数量。

形状（一层）：

START ─► decompose ─► research_topic (parallel_worker) ─► synthesize
                             │  │  │
                             └──┴──┴─ (flat: no children yet)

开放式问题会被分解为 N 个子问题，N 由 LLM 在运行时选择 (3-7)，每个子问题都会并行研究，然后合成为一份简报。

▶ Colab：运行 L4a 代码单元 · 📁 GitHub：L4a_flat_research/ · 💻 本地：python -m L4a_flat_research.deep_research

L4a 流程

🔍 标记 - 没有

dynamic=True

开关。动态是一种写入方式，而不是配置。只有两个标记：@node(parallel_worker=True)（接受运行时大小的列表，每个项运行一个工作器）和 ctx.run_node(...)（直接调度代码节点）。看到其中任一选项 → 您使用的是动态布局。

您会看到：分解器打印出 5 个子问题（例如），它们并行进行研究，然后生成一份合成简报。每次运行时的数字都不同，而固定图无法做到这一点。

工作器上有两个值得了解的标志：

在调用 ctx.run_node 的任何节点上，rerun_on_resume=True 是必需的，否则 ADK 会引发 ValueError。在恢复时，它必须重新执行调度节点以重建其生成的子节点，因为这些子节点不在静态图中。
retry_config= 限制了此 FAILS 的范围。并行工作器会取消每个同级，并在一个子级失败时立即重新引发异常 - 因此，如果不进行重试，单个暂时性 429 错误会舍弃整个运行，包括已付费的每个调用。重试会落在内部的每个商品节点上，因此每个分支都会独立重试。

❓ 您可能想知道： ADK 是如何“知道”这是动态的？不需要，因为任何地方都没有声明任何内容。分解器会在运行时生成一个列表；并行工作器会根据收到的内容调整自身的大小。动态性是您编写的数据传输的属性，而不是您开启的模式。

👀 阅读：research_topic 上的两个标志 - parallel_worker 和 rerun_on_resume。· ▶ 运行。· ✏️ 更改：换成您自己的开放式问题 - N 发生更改，因为输入决定了宽度。

12. L4b · 添加了递归生成功能（支柱 3b）

路线图 - 您目前处于：L4b

⚡ 总结：递归是编写出来的，而不是给定的 - 工作器通过 ctx.run_node（普通的 Python）调用自身，因此必须编写制动器。即，MAX_DEPTH。

问题：一项研究发现有时会揭示一个值得单独研究的狭窄子主题。如何让分支生成更多并行工作，并保持在一定范围内？

形状（现在是递归的）：

START ─► decompose ─► research_topic (parallel_worker, recursive) ─► synthesize
                             │  │  │
                             │  │  └─ research(q3) ─► maybe spawn children
                             │  └─── research(q2) ─► maybe spawn children
                             └────── research(q1) ─► maybe spawn children

▶ Colab：运行 L4b 代码单元 · 📁 GitHub：L4b_recursion/ · 💻 本地：python -m L4b_recursion.deep_research

L4b 流程

@node(parallel_worker=True, rerun_on_resume=True)
async def research_topic(ctx, node_input):
    finding = coerce(await ctx.run_node(research_agent, node_input=...), ResearchFinding)
    if finding.needs_deeper and finding.deeper_questions and depth < MAX_DEPTH:   # boundary in CODE
        children = await ctx.run_node(research_topic, node_input=deeper)          # recursive fan-out
    yield Event(output={..., "children": children})

🔍 标记：ctx.run_node(research_topic, ...) 位于research_topic 本身内部 - 自引用是递归 - 以及上一行的防护 depth < MAX_DEPTH。

您会看到：研究节点打印 spawning N deeper - 实时发生递归 - 然后是运行时树形结构（例如 5 top-level + 10 recursive children）。每次运行树形结构都不同。

⚠️ 在调高旋钮之前：上限会快速增长，MAX_DEPTH=3 最坏情况从大约 30 次调用增加到大约 93 次。在运行结束时，您可能会看到 cancelling N leftover tasks 日志行：这是 ADK 在结果已完成之后拆解其并行任务组。无害 - 根据您的日志记录配置，您可能永远不会看到此消息。

❓ 您可能想知道： 动态递归不是默认开启的吗？否 - L4a 是完全动态的，具有零递归。动态只提供普通的 Python 控制流；L4b 选择使用它来编写递归。由于递归是您编写的，因此您必须编写其边界，此时“让 LLM 塑造工作，将边界保留在代码中”不再是一句口号。

👀 读取：防护措施：if finding.needs_deeper and depth < MAX_DEPTH。· ▶ 运行。· ✏️ 更改：设置 MAX_DEPTH = 1 并重新运行 - 树变平（运行成本降低）。边界由您在代码中定义。

13. L5 · 您应该使用哪种模式？

路线图 - 您目前处于：L5

⚡ 总结：一个轴决定一切 - 谁来选择下一步：您绘制的图表、LLM 还是您的代码。

您已构建了所有三个。正是这种模型让它们变得有用：将模式与问题的形状相匹配。

轴：谁决定接下来运行什么？

柱子	谁决定接下来运行什么	内置
1 · 图表	您绘制的图表	L2a / L2b
2. 协作	LLM	L3a / L3b
3 · 动态	您的 Python 代码（在运行时）	L4a / L4b

第 0 步：您是否真的需要图表？

ADK 随附预构建的工作流智能体 - SequentialAgent、ParallelAgent、LoopAgent。对于简单的代理链，这是最便宜的正确答案，无需组装图表。当您需要显式路由（L2b 的路由器）、联接（L2a 的 JoinNode）或非代理节点（一个简单的函数，零 LLM 调用）时，请超越它们。最后一种情况通常是原因所在。

Would a prebuilt SequentialAgent / ParallelAgent / LoopAgent do?
│
├─ YES ──────────────────────────────► use it; stop here
│
└─ NO — I need routing, a join, or non-agent nodes
   │
   Can you draw the workflow before the input arrives?
   │
   ├─ YES ───────────────────────────► Pillar 1 · Graph workflow    (L2a/L2b)
   │
   └─ NO
      ├─ Known team, request picks the subset? ─► Pillar 2 · Collaborative  (L3a/L3b)
      └─ Does the shape depend on the input?  ──► Pillar 3 · Dynamic        (L4a/L4b)

L5 · which pattern

1.x 与 2 的真实对比

这不是“2.0 可以做 1.x 做不到的事情”——1.x 可以构建所有这些内容。不同之处在于，2.0 为每个形状提供了更直接的归宿，因此已知的控制流会离开提示，成为您可以查看和测试的结构。

模式	1.x 版费用	ADK 2 首页
图表	在通用 build 中进行 4 次 LLM 调用；路由隐藏在提示中	函数节点 + 代理节点作为对等节点 → 1 次调用，`if` 语句路由器
协作	可通过 `AgentTool` 管道构建；`ParallelAgent` 始终为 all，`transfer_to_agent` 序列	已声明的团队：`sub_agents` + `mode="single_turn"`
动态	递归会使您脱离框架	框架内的 `parallel_worker` + 递归 `ctx.run_node`

整个应用以及图表无法显示的内容

您现在已构建以下所有内容。Workflow 在 graph.edges 中公开了其结构，因此此图片是根据代码生成的，而不是手工绘制的；自省功能找到的是本实验的摘要：

柱子	`graph.edges` 包含的内容	原因
1 · 图表 (L2b)	10 条边、路线和所有内容	您在收到任何输入之前就绘制了它
2 · 协作 (L3a)	0 条边 - 仅限 `sub_agents` + `mode`	LLM 会根据每个请求选择子集
3 · 动态 (L4a/L4b)	3 条边 - 在两个图中完全相同	递归是用 Python 编写的，而不是在图中连接的

最后一行是问题 L4b 的答案的证明：L4a 和 L4b 具有相同的图，但只有其中一个会递归。

您现在可以构建的内容

您刚刚运行的每个图案都是真实的产品形状：

您练习了	在实际应用中，	起点
图表 + 路由器（L2a/L2b）	文档流水线、包含 LLM 步骤的 ETL、审核/审批链、评估框架	相应代码库的 L2b
协调员 + `single_turn` 团队 (L3a)	支持专家团队、分诊台、多视角审核的支持 Copilot	马拉松演示模式 2
`task` 个代理（L3b）	客户信息收集表单、预订流程、新客户引导、KYC（了解您的客户）— 任何“先收集后行动”	`22_agent_in_workflow`
动态宽度/深度 (L4a/L4b)	研究代理、报告生成器、对未知大小的输入进行审核扫描	马拉松演示模式 3

他们撰写

这三种模式并不相互排斥。图节点可以调用协作协调器；专家可以启动动态工作流。根据问题的每个部分选择合适的模式，这样可以避免将每个智能体系统都变成一个巨大的提示。

整个应用 - 以及图表无法显示的内容

💡在您自己的工作流程中尝试一下：绘制此图表的脚本为 scripts/graph_dump.py。将它指向任何 Workflow，它都会打印出真实边缘，从而生成您构建的任何事物的免费结构图。

14. 恭喜

九名选手、一根接力棒、井然有序的终点

您构建了一个马拉松比赛日教练，并在此过程中使用了 ADK 2 的全部三种编排模式。

要点回顾

序言 - 创造了自己的天气的超级提示：为什么结构会存在。
L0-L1 - Agent、Runner、模型选择调用的真实工具，以及您的第一个 Workflow（作为对等方的函数节点 + 代理节点）。
L2a / L2b - 图形工作流：并行扇出 + JoinNode，然后确定性路由 - 一次 LLM 调用。
L3a - 协作型代理：同一团队在 chat（搁浅）和 single_turn（并行子集 + 综合）中运行 - 一个标志，两个世界。
L3b - task 模式：暂停的澄清式问题、脚本化的继续、finish_task 返回已验证的对象。
L4a / L4b - 动态工作流：运行时宽度（扇出），然后是运行时深度（递归），代码中包含边界。
L5 - 决策树，以及模式的组成方式。

值得保留的台词

函数准备上下文。边定义了工作流。路由器选择路径。模型撰写回答。

让 LLM 塑造工作，但要将边界保留在代码中。

将模式与问题的形状相匹配。

后续步骤

运行这些关卡所基于的完整应用，即 Marathon Race Day Coach，这是一个 FastAPI + SSE 构建，具有浏览器界面，可实时显示所有三种模式：github.com/cuppibla/adk-2-marathon-demo。
更广泛的示例：adk-workflows-compared - 所有 23 个官方 ADK 2 工作流示例，每个示例都包含 1.x 端口和使用时机指南。先从 docs/three-pillars.md 开始，然后学习本 Codelab 跳过的部分：07_loop、17_request_input、22_agent_in_workflow。
移植您自己的问题：哪些部分是已知结构 (L2)、已知团队 (L3a/L3b)、未知形状 (L4)？
探索代码：github.com/cuppibla/adk2-tutorial。
参加过本研讨会？您的赠金以及使用赠金创建的项目不会永久有效。如需继续免费重新运行这些关卡，请改为执行Take-home setup步骤：免费 AI Studio 密钥，无需云项目，无需结算信息。交换该单元格是唯一的变化。

ADK 2 编排：图、协作和动态工作流

1. 概览

学习内容

贯穿始终的主线

构建内容

所需条件

两种跟进方式

2. 研讨会设置 · 申领抵用金并切换到 Vertex AI

1 · 领取赠金 （约 1 分钟）

2 · 打开笔记本并安装 ADK 2 （约 1 分钟）

3 · 运行“Workshop setup”单元格 （约 3 分钟）

4 · 跳过“在家设置”步骤

5 · 运行“共享组成要素”单元格

研讨会结束后

3. 自学设置 · AI Studio API 密钥

1 · 打开笔记本

2 · 安装 ADK 2 （约 1 分钟）

3 · 从 AI Studio 获取 Gemini API 密钥 （约 1 分钟）

4 · 将密钥添加到 Colab （约 1 分钟）

5 · 运行密钥单元

6 · 运行“共享构建块”单元格

4. 序言 · 为什么不使用一个大型提示？

5. L0 · 您的首个 ADK 2 智能体

6. L1 · 您的第一个工作流

7. L2a · 并行扇出 + JoinNode（支柱 1a）

8. L2b · 添加确定性路由器（支柱 1b）

9. L3a · 协作型智能体：一面旗帜，两个世界 - 支柱 2

Beat 1 · 先运行默认值 - 并观察它是否会使作业失败

Beat 2 · 一面旗帜，两个世界

10. L3b · 任务模式：有终点的对话 - Pillar 2

选择模式的单问题规则

11. L4a · 运行时大小的并行扇出（支柱 3a）

12. L4b · 添加了递归生成功能（支柱 3b）

13. L5 · 您应该使用哪种模式？

轴：谁决定接下来运行什么？

第 0 步：您是否真的需要图表？

1.x 与 2 的真实对比

整个应用以及图表无法显示的内容

您现在可以构建的内容

他们撰写

14. 恭喜

要点回顾

值得保留的台词

后续步骤

1 · 领取赠金（约 1 分钟）

3 · 运行“Workshop setup”单元格（约 3 分钟）

3 · 从 AI Studio 获取 Gemini API 密钥（约 1 分钟）