1. 總覽
ADK 2 的主打功能是三種自動化調度管理模式。本程式碼研究室會逐步建構「馬拉松賽事教練」應用程式，同時說明這三種概念。每個層級都會回答單一問題、新增一個想法，並自行運作。
課程內容
- 圖形工作流程 (第 1 支柱) - 您可以在輸入內容抵達前繪製流程。
- 協作代理程式 (第 2 支柱) - 當您知道團隊，但要求會挑選子集時，以及所有三種協作模式 (
chat/task/single_turn)，都會即時執行。 - 動態工作流程 (第 3 支柱) - 工作本身的形狀取決於輸入內容。
- 如何選擇：單一問題的決策樹，以及模式的組成方式。
貫穿的主題
已知結構 → 已知團隊 / 變數子集 → 不明形狀 → 選擇正確的形狀
建構項目
一個應用程式 (馬拉松賽事教練) 一次組裝一個可執行的層級。每個層級都是您從終端機執行的純 Python 模組；到了 L5，您就能擁有下列所有片段。
圖片是從執行的程式碼繪製：每條實線都是從 Workflow.graph.edges 讀取。這就是第一課的內容：可預先繪製的內容正是支柱 1，而無法預先繪製的內容則說明瞭支柱 2 和 3 的存在意義。
軟硬體需求
- Google 帳戶 (適用於 Colab) - 無須進行本機設定。
- 約 50 分鐘 (L4 級別的兩項作業耗時較長，請預留時間)。
- 存取 Gemini 模型的方式有兩種，選擇路徑：您會執行一個設定步驟，並略過其他步驟：
🎓 研討會 | 🏠 居家 | |
參與者 | 你正在參加現場研討會，講師提供抵免學分連結 | 其他人 (包括研討會參與者) |
需求條件 | 聲明連結，以及可建立雲端專案的 Google 帳戶 | 免費的 AI Studio API 金鑰 |
執行 | Vertex AI，專案費用會從研討會抵免額扣除 | Google AI Studio |
費用 | 抵免額適用範圍 | 免費方案 |
設定步驟 | 工作坊設定 (下一步) | 在家設定 (後續步驟) |
無論使用哪種方式，從序章開始的所有內容都相同，通道只會決定筆記本與哪個模型端點通訊。
兩種跟著做的方式
下列每個步驟都會對應至 Colab 筆記本中的一個儲存格，以及 GitHub 存放區中的一個資料夾。請選擇下列其中一項：
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 中開啟」，然後執行第一個程式碼儲存格。這會固定本程式碼研究室驗證的確切 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
4 · 略過「帶回家設定」步驟
請勿執行 AI Studio 金鑰儲存格，否則筆記本會切換回 AI Studio，並還原您剛才執行的操作。(儲存格會防範這種情況，並拒絕執行，但更整齊的做法是直接略過)。直接從這裡前往「共用建構區塊」儲存格。
5 · 執行「Shared building blocks」儲存格
執行一次。這會定義 Pydantic 結構定義 + 罐頭馬拉松情境，L2 以上的每個層級都會重複使用。你會看到 ✓ schemas + scenarios ready。
研討會結束後
抵免額和抵免額建立的專案都有期限。如要在研討會結束後繼續免費重新執行這些層級，請改為執行「Take-home setup」步驟，取得免費的 AI Studio 金鑰，不必使用雲端專案，也不會產生任何費用。只有該儲存格會變更。
如要提早清理，請開啟 Cloud 控制台，選取 adk-2-tutorial-XXXX，然後刪除。本程式碼研究室中的其他項目不會建立計費資源。
3. 居家設定 · AI Studio API 金鑰
這條路徑上的所有內容都會使用免費的 Google AI Studio API 金鑰執行，不需要 Google Cloud 雲端專案、帳單資訊或在本機安裝。這個步驟大約需要 3 分鐘。
1. 開啟筆記本
按一下「在 Colab 中開啟」圖示 ▶。系統會將您帶往筆記本，其中包含 Markdown 簡介，以及每個等級各一個可執行的儲存格。您會從上到下執行儲存格，每個儲存格都會在下方列印自己的輸出內容。
2. 安裝 ADK 2 (約 1 分鐘)
執行第一個程式碼儲存格。這會固定本程式碼研究室驗證的確切版本：
%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 Secrets (建議使用；金鑰會保持隱藏)：
- 按一下 Colab 左側邊欄中的鑰匙圖示 🔑。
- 按一下「+ 新增密碼」。
- 將「Name」設為
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」儲存格
執行「Shared building blocks」(共用建構區塊) 儲存格一次。這會定義 Pydantic 結構定義 + 罐頭馬拉松情境，L2 以上的每個層級都會重複使用。你會看到 ✓ schemas + scenarios ready。
設定完成！🎽 在 L0 (大家首先建構的版本) 之前，我們來快速瞭解一下。
4. 序言 · 為什麼不使用一個大型提示？
⚡ 執行前，請鎖定要觀察的「一件事」：每個特定數字的來源。這就是整個練習，其他都是裝飾。
在階梯之前，請先執行階梯要取代的項目：一個提示可完成所有工作的代理程式，包括擷取天氣資訊、分析課程、讀取訓練記錄、依條件設定路徑，以及輸出計畫。
你會看到：自信、具體、格式正確的策略，但其中的數字是捏造的。在一次即時運作執行中，它以「我已擷取今天的氣象指標」開頭，並回報 52°F、風速 9 英里/小時，以及從未見過的訓練記錄數據分析。這裡沒有天氣 API、沒有課程資料，也沒有記錄，只有一個不透明的模型呼叫，不是捏造輸入內容，就是將輸入內容變成無用資訊。
這就是疾病，有四種值得命名的症狀：
- 不可信：這些資料是 AI 流暢地編造出來的。
- 您無法測試 - 步驟 4 的路徑位於散文內，沒有
if可進行單元測試。 - 無法交換步驟：沒有可插入實際天氣 API 的接縫。
- 每次都必須支付所有費用：五個步驟、一次大型呼叫，且不會快取決定性部分。
請保持這種感覺。接下來的九個層級會逐步從提示中移除這些步驟：函式擷取 (L1 至 L2a)、if 陳述式路徑 (L2b)、專家分工 (L3a 至 L3b)，以及程式碼界定形狀 (L4a 至 L4b)。
💻 當地商家： python -m shared.prologue
5. L0 · 您的第一個 ADK 2 代理
⚡ 簡而言之：代理程式 = 模型 + 指令 + 可呼叫的工具；Runner 則會執行代理程式。這個層級之後的所有內容都只是更多代理程式，以更完善的形狀排列。
問題：模型能否回答問題，並在需要算術運算時實際編寫程式碼？
一個概念，三個部分：
Agent：會推理的項目 (Gemini 模型 + 指令)。Runner：在工作階段中執行代理程式並串流事件的項目。- 工具：模型決定呼叫的簡單 Python 函式 (
pace_splits)。ADK 會讀取簽章和說明字串，並將宣告傳遞給模型，無須編寫結構定義。
序言之後的第一項修復作業：LLM 會在腦中進行配速算術，但很可能會出錯。pace_splits 是確定性的 Python，因此答案中的數字是計算出來的，而非即興。
▶ Colab：執行 L0 儲存格 · 📁 GitHub：L0_first_agent/ · 💻 本機：python -m L0_first_agent.agent
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 來自您的程式碼，而非來自權杖統計資料。
❓ 你可能會想知道： 模型是否一律會呼叫工具？不會，系統會根據每個問題決定是否顯示。如果詢問的內容不含數字，🔧 符號就會消失 (Playground 會要求您嘗試這麼做)。
👀 閱讀： pace_splits (一般函式) 和 tools=[pace_splits] 行。· ▶ 執行。· ✏️ 變更：詢問一般問題 (沒有目標時間) - 請注意 🔧 行消失：模型會決定何時值得呼叫工具。然後重新編寫 instruction 並重新執行，指令就是程式的其餘部分。
6. 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
函式節點會列印產生的資料 (沒有模型呼叫)，然後代理程式會根據收到的實際溫度和風速提供建議：
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 的資料。重點在於貴族制度，而非順序。
👀 閱讀：fetch_conditions 會傳回沒有模型呼叫的資料；advise 則有 input_schema=Conditions。· ▶ 執行。· ✏️ 變更：在函式中設定 temp_f=30 並重新執行，建議就會翻轉，且函式仍會耗用 0 次 LLM 呼叫。
7. L2a · 平行扇出 + JoinNode (Pillar 1a)
⚡ 摘要：平行分散 (免費)、等待「所有」結果、打包，然後將完整資訊交給一個代理程式。
問題：你可以在輸入內容抵達前繪製流程。從架構開始：平行收集資料、將資料打包，然後交給一個代理程式。
形狀：
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
🔍 標記：三個邊緣都從 START 開始，這是扇出，而 JoinNode 是會合點。
- 這三項擷取作業是函式，會平行執行，且不會呼叫 LLM。
JoinNode會等待這三項作業完成，然後將其彙整為一個以函式名稱做為鍵的型別酬載 (BundledRunData)。- 一個
strategy代理會讀取套件並撰寫RaceStrategy。
顯示內容：每次擷取都會列印 started / finished 時間戳記。這三項作業都從 0.0 秒開始，並在 2.0 秒結束 (最慢的擷取作業，而非總和為 4.5 秒 的時間長度)。重疊部分就是平行處理量。(最後印出的 total 實際時間為約 8 秒，因為當中也包含策略代理程式的 LLM 呼叫，請讀取平行聲明的擷取時間戳記，而非總時間。)
💡 序言回呼：巨型提示發明了天氣。這裡的溫度來自擷取 函式，是實際的程式碼和接縫。將罐頭字典換成實際的天氣 API，其他都不會變更。
❓ 你可能會想知道：
JoinNode
我需要瞭解哪些概念？簡單來說，這個函式會等待每個平行分支完成，將輸出內容封裝到以串流函式名稱做為鍵的字典中，本身不會計算任何內容。L2b 的路由器可以寫入 node_input["fetch_weather"]["temp_f"]，就是因為這個字典。
👀 讀取：三條邊緣從 START 扇出；JoinNode 將這些邊緣綁定至一個代理。· ▶ 執行並讀取時間戳記，而非總和。· ✏️ 變更：進行一次擷取休眠 3.0 - 先預測新的擴散傳遞功能結束時間，然後驗證。
8. L2b · 新增確定性路由器 (第 1b 支柱)
⚡ 簡而言之：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
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 邊緣 {"HOT": ..., "NORMAL": ..., "COLD": ...}。
重點：三種工作，三種住家：
- 可預測的工作 → 函式 (3 個平行擷取作業)
- 明確規則 → 明確轉送 (
route_by_weather是if陳述式，而非模型決策) - 推論 → 模型 (執行一個策略代理)
顯示內容： 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
⚡ TL;DR：同一支隊伍，一面旗幟。chat 將整段對話交給一位專家，然後就此結束；single_turn 將每位專家變成工具，並行子集、自動回覆、一次合成。
問題：您知道團隊，但要求會決定哪些成員應回覆。如何讓 LLM 挑選子集，並同時執行？
形狀：一位協調員負責六位專家 (醫療、天氣、配速、裝備、營養、心理)。這個層級會讓同一組成員執行兩次，也就是使用相同的協調員提示和六位專家。唯一的差別在於子代理程式上的一個旗標。對比就是重點。
▶ Colab：執行 L3a 儲存格 · 📁 GitHub：L3a_collaborative/ · 💻 本機：python -m L3a_collaborative.concierge --mode chat "What about fueling?"
🔍 標記： 工廠中的 mode="single_turn"，以及輸出中的 TRANSFER → (節拍 1)，與共用一個時間戳記的 DISPATCH → 行爆發 (節拍 2)。
第 1 拍：先執行預設值，並觀察作業失敗
未撰寫 mode= → 子代理預設為 chat。你會看到：
TRANSFER → nutrition_specialist (transfer_to_agent — the only tool chat subagents provide)
Final speaker: nutrition_specialist
協調員沒有委派工具，即時通訊子代理只會將整個對話依序轉交給一位專員。transfer_to_agent該專員會直接回覆使用者，並結束執行程序。不可同時出貨。恕不退貨。沒有合成內容。如果問的問題範圍很廣，情況會更糟：六位專員，一次轉接。
這不是錯誤，而是對話模式的正常運作方式。會話群組屬於持有者，直到有人明確轉移為止。適合開放式助理，但不適合管道步驟。
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 會並行執行這些呼叫，每個呼叫都會自動傳回結果，協調器則會合成結果。
問題 | 會開除員工的專家 |
「那加油呢？」 | 僅限營養 |
「My knee hurts at mile 18」(第 18 英里時膝蓋疼痛) | 僅限醫療 |
「我今天應該參加比賽嗎？」 | 醫療 + 天氣 + 配速 |
「有什麼需要注意的地方嗎？」 | 全部 6 |
為何每位專家都會收到完整簡報：每個 single_turn 子代理都會在各自獨立的會話分支中執行，無法查看對話或同儕。沒有任何環境：協調器必須將整個 SpecialistInput (問題 + 策略 + 執行器資料) 分別轉送至每個平行呼叫。
💡 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 會在伺服器端強制執行這項操作。這是輸入：協調器必須為每個平行呼叫逐字重現整個巢狀 SpecialistInput，有時會出錯。ADK 會將錯誤當做該工具的結果傳回，協調器會復原，合成作業仍會完成。
❓ 您可能會好奇：
chat
一次只能委派給一個代理？基本上是：這是 1.x 的預設行為，現在有了名稱。與 single_turn 的差距是三維的：協調員持有的內容 (每個專家一個 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 · 任務模式：有終點的對話 - 支柱 2
⚡ 摘要：中間模式 - 與使用者對話直到收集欄位為止，然後自動傳回經過驗證的物件。
問題：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
🔍 標記： 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驗證。輸入完成線的對話，控制權會自動返回給協調員，並附上結果。
選擇模式的單一問題規則
💡 「使用者是否需要與其對話？需要對話到何時？」對話 = 無期限 · 工作 = 直到收集完欄位 · 單次對話 = 永不。
模式 | 人機迴圈 | 平行？ | 返回上層 |
| 完整對話 | 否 | 手動 (透過轉移) |
| 僅限釐清問題 | 否 | 自動 (透過 |
| 無 | yes | 自動 (附上結果) |
mode 只能在子代理程式上執行，絕不能在協調器上執行。工作流程節點預設為 single_turn (這就是 L1 至 L2b 從未撰寫的原因)，而子代理程式預設為 chat (這就是 L3a 必須撰寫的原因)。
⚠️ 在開始建構之前，請先注意以下兩點：(1) task
做為靜態圖形節點時，會因版本而異：在 2.0.0b1 至 2.3.0 (本程式碼研究室的釘選) 中，Workflow(...) 會在建構時引發；請使用這個層級的確切內容 (具有工作子代理程式的聊天協調器)，或透過 ctx.run_node 傳送。已在 2.5.0 版中移除。(2) 「工作代理必須是葉節點代理」 (沒有自己的子代理) 是 ADK 的記錄限制，但合約並非執行階段防護措施：2.3.0 和 2.5.0 都不會阻止您。請勿將沒有錯誤視為權限。
💡 深入瞭解：task 嵌入圖形工作流程的代理 (2.5.0 以上版本)，具有可將對話迴圈返回以重試的路由：隨附的存放區 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)
⚡ 簡而言之：骨架仍是三個靜態步驟，動態部分隱藏在中間步驟，寬度由執行階段的資料決定。
⚠️ 注意：這是階梯中最陡峭的一步。先前的層級為 44 行，這個層級則約為 120 行，包含三個代理程式和兩個工作流程節點，且沒有任何填補。請預留約 15 分鐘，並在最後一行讀取/執行/變更：您不需要在第一次傳遞時吸收每一行。
問題：作品的形狀取決於輸入內容。你無法事先繪製圖表。首先是執行階段寬度：讓大型語言模型決定要提出多少子問題。
形狀 (深一層)：
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
🔍 標記 - 沒有
dynamic=True
切換。動態是撰寫方式，不是設定。只有兩個標記：@node(parallel_worker=True) (採用執行階段大小的清單，每個項目執行一個 worker) 和 ctx.run_node(...) (直接排定程式碼節點)。看到任一畫面 → 你已進入動態模式。
你會看到：分解器會列印 5 個子問題，並同時進行研究，然後綜合整理成簡報。每次執行時，數字都會不同，這是固定圖表無法做到的。
工作站上有兩個值得瞭解的標記：
- 在呼叫
ctx.run_node的任何節點上，rerun_on_resume=True為必要項目，否則 ADK 會引發ValueError。恢復執行時，必須重新執行分派節點，重建所產生的子項，因為這些子項不在靜態圖表中。 retry_config=bounds how this FAILS. 平行工作站會取消所有同層級項目，並在其中一個子項失敗時重新引發例外狀況，因此如果沒有重試機制，即使只有一個暫時性的 429 錯誤，也會導致整個執行程序遭到捨棄，包括所有已付費的呼叫。重試會落在內部每個項目的節點上，因此每個分支都會獨立重試。
❓ 您可能會想： ADK 是如何「得知」這是動態？不需要，因為任何地方都沒有宣告任何內容。分解器會在執行階段產生清單；平行工作站會根據收到的內容調整大小。動態性是您編寫的資料流程屬性，而非您開啟的模式。
👀 閱讀：research_topic 上的兩個旗標 - parallel_worker 和 rerun_on_resume。· ▶ 執行。· ✏️ 變更：換成您自己的開放式問題 - N 會變更，因為輸入決定了寬度。
12. L4b · Add recursive spawning (Pillar 3b)
⚡ 簡而言之：遞迴是「編寫」而非「提供」，工作人員會透過 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
@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, ...) 本身 (即遞迴) 內的自我參照 is，以及上方一行中的防護措施 depth < MAX_DEPTH。research_topic
您會看到：研究節點列印 spawning N deeper - 遞迴即時發生 - 然後是執行階段樹狀結構形狀 (例如 5 top-level + 10 recursive children)。每次執行時，樹狀結構都會有所不同。
⚠️ 調高旋鈕前：上限會快速增加，最糟的情況是從約 30 次呼叫增加到約 93 次。MAX_DEPTH=3在執行作業的最後，您可能會看到 cancelling N leftover tasks 記錄行：這是 ADK 在結果完成後，拆除平行工作群組。無害，而且視記錄設定而定，您可能永遠不會看到這項訊息。
❓ 你可能會想： 動態遞迴不是預設為啟用嗎？否，L4a 完全是動態的，且沒有遞迴。動態只會提供一般的 Python 控制流程；L4b 選擇使用它編寫遞迴。由於遞迴是「您」撰寫的，因此「您」必須撰寫遞迴的邊界，這時「讓 LLM 塑造工作，將邊界保留在程式碼中」就不再只是口號。
👀 閱讀：警衛：if finding.needs_deeper and depth < MAX_DEPTH。· ▶ 執行。· ✏️ 變更：設定 MAX_DEPTH = 1 並重新執行，樹狀結構就會扁平化 (執行費用也會降低)。邊界是「您」的，以程式碼表示。
13. L5 · Which pattern should you use?
⚡ 簡而言之：一個軸向決定一切，也就是由誰選擇下一個步驟：您繪製的圖表、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)
1.x 與 2 的真實比較
這並非「2.0 可以做到 1.x 無法做到的事」— 1.x 可以建構所有內容。2.0 版會為每個形狀提供更直接的歸屬位置，因此已知的控制流程會離開提示，成為可查看及測試的結構。
模式 | 1.x 版的費用 | ADK 2 首頁 |
圖形 | 一般建構作業中會進行 4 次 LLM 呼叫；提示詞中隱藏了路徑 | function + agent 節點做為同層級節點 → 1 次呼叫， |
協作 | 可透過 | 已宣告的團隊： |
動態 | 遞迴會將您從架構中移除 | 架構內有 |
整個應用程式，以及圖表無法顯示的內容
您現在已建構下列所有項目。Workflow 會在 graph.edges 露出結構，因此這張圖片是從程式碼產生，而非手繪，而內省找到的內容就是本實驗室的摘要：
支柱 |
| 原因 |
1 · Graph (L2b) | 10 個邊緣、路線等 | 您在收到任何輸入內容前繪製了該筆劃 |
2 · 協作 (L3a) | 0 邊緣 - 僅限 | LLM 會根據要求選取子集 |
3 · Dynamic (L4a/L4b) | 3 個邊緣 - 兩者相同 | 遞迴是以 Python 編寫，而非在圖表中連線 |
最後一列是 L4b 回答問題的證明：L4a 和 L4b 具有相同的圖表，且只有其中一個會遞迴。
現在可以建構的內容
您剛執行的每個模式都是實際的產品形狀：
你練習了 | 在實際情況中，這表示 | 開始於 |
圖表 + 路由器 (L2a/L2b) | 文件管道、ETL (含 LLM 步驟)、審查/核准鏈、評估架構 | 這個存放區的 L2b |
統籌專員 + | 支援 Copilot，可與專家團隊、分流服務台、多鏡頭審查功能搭配使用 | 馬拉松示範模式 2 |
| 填寫表單、預訂流程、新手上路、瞭解你的客戶 (KYC) - 任何「收集資訊後採取行動」的 | |
動態寬度/深度 (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 建構的應用程式，並提供瀏覽器 UI，可即時顯示所有三種模式：github.com/cuppibla/adk-2-marathon-demo。
- 深入瞭解：adk-workflows-compared - 所有 23 個官方 ADK 2 工作流程範例，每個範例都有 1.x 移植版本和使用時機指南。請先從
docs/three-pillars.md開始，然後是本程式碼研究室略過的內容：07_loop、17_request_input、22_agent_in_workflow。 - 自行移植問題：哪些部分是已知結構 (L2)、已知團隊 (L3a/L3b)、未知形狀 (L4)？
- 瀏覽程式碼：github.com/cuppibla/adk2-tutorial。
- 參加過工作坊嗎？你的抵免額和抵免額建立的專案不會永久存在。如要繼續免費重新執行這些關卡，請改為執行「居家設定」步驟：免費 AI Studio 金鑰，無需雲端專案，也不會產生費用。唯一變更就是交換該儲存格。