1. 概要
ADK 2 の見出しは「3 つのオーケストレーション パターン」です。この Codelab では、マラソン レース当日のコーチという 1 つのアプリを 1 つずつ実行可能なステップで作成しながら、3 つすべてを学びます。各レベルは 1 つの質問に答え、1 つのアイデアを追加し、独自に実行されます。
学習内容
- グラフ ワークフロー（ピラー 1） - 入力が到着する前にフローを描画できる場合。
- コラボレーション エージェント（ピラー 2） - チームはわかっているが、リクエストがサブセットを選択する場合。3 つのコラボレーション モード（
chat/task/single_turn）がすべてライブで実行されます。 - 動的ワークフロー（第 3 の柱） - 作業自体の形状が入力に依存する場合。
- 選び方 - 1 つの質問で構成されるディシジョン ツリーと、パターンの構成方法。
共通のテーマ
既知の構造 → 既知のチーム / 変数のサブセット → 不明な形状 → 正しいものを選択
作成するアプリの概要
1 つのアプリ（マラソン レース当日コーチ）が、実行可能なレベルを 1 つずつ組み立てました。各レベルは、ターミナルから実行するプレーンな Python モジュールです。L5 までに、以下のすべての部分を自分で作成します。
この図は実行中のコードから描画されたもので、すべての実線は Workflow.graph.edges から読み取られています。これが最初の教訓です。事前に描画できる部分はまさに Pillar 1 であり、描画できない部分が Pillar 2 と Pillar 3 の存在理由です。
必要なもの
- Google アカウント（Colab 用） - ローカル セットアップは不要です。
- 約 50 分（L4 レベルの 2 つは時間がかかるため、予算を立ててください）。
- Gemini モデルにアクセスする 2 つの方法の 1 つ。レーンを選択する - 1 つのセットアップ手順を実行し、他の手順をスキップします。
🎓 ワークショップ | 🏠 まとめ | |
話し手 | ライブ ワークショップに参加していて、インストラクターからクレジット請求リンクが提供された | それ以外のユーザー（ワークショップの参加者を含む） |
必要なもの | 請求リンクと、Cloud プロジェクトを作成できる Google アカウント | 無料の AI Studio API キー |
実行環境 | Vertex AI（ワークショップ クレジットで課金されるプロジェクト内） | Google AI Studio |
費用 | クレジットの対象 | 無料枠 |
設定手順 | ワークショップの設定（次のステップ） | 持ち帰りセットアップ（次のステップ） |
プロローグ以降はどちらの場合も同じです。レーンは、ノートブックが通信するモデル エンドポイントを決定するだけです。
2 つのフォロー方法
以下の各ステップは、Colab ノートブックの 1 つのセルと GitHub リポジトリの 1 つのフォルダに対応しています。次のいずれかを選択します。
2. ワークショップの設定 · クレジットを申請して Vertex AI に切り替える
ワークショップでは、Google Cloud クレジットが付与されます。この割り当てをリクエストし、この割り当てに課金されるプロジェクトを作成して、AI Studio ではなく Vertex AI を指すようにノートブックを設定します。1 つのセルが申し立て後のすべての処理を行います。
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 を有効にし、後続のすべてのセルが読み取る 4 つの環境変数を設定し、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 · 「共有構成要素」セルを実行する
1 回実行します。このファイルは、L2 以降のすべてのレベルで再利用される Pydantic スキーマと事前定義されたマラソン シナリオを定義します。✓ schemas + scenarios ready が表示されます。
ワークショップの終了後
クレジットと、それによって作成されたプロジェクトは永続的ではありません。ワークショップ終了後もこれらのレベルを無料で再実行するには、代わりに持ち帰り用設定の手順を実行します。無料の AI Studio キー、クラウド プロジェクト、請求は不要です。変更されるのはその 1 つのセルのみです。
すぐにクリーンアップするには、Cloud コンソールを開き、adk-2-tutorial-XXXX を選択して削除します。この Codelab では、これ以外に課金対象リソースは作成されません。
3. 持ち帰り用セットアップ · AI Studio API キー
このパスのすべての処理は、無料の Google AI Studio API キーで実行されます。Google Cloud プロジェクト、課金、ローカル インストールは必要ありません。このステップ全体の所要時間は約 3 分です。
1 · ノートブックを開く
[Colab で開く ▶] をクリックします。ノートブックが表示されます。マークダウンの概要と、レベルごとに実行可能なセルが 1 つずつあります。セルは上から下に実行され、各セルはすぐ下に出力を出力します。
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 シークレット（推奨、キーは非表示のまま）:
- Colab の左側のサイドバーにある鍵アイコン 🔑 をクリックします。
- [+ 新しいシークレットを追加] をクリックします。
- [名前] を
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 · 「共有ビルディング ブロック」セルを実行する
[共有ビルディング ブロック] セルを 1 回実行します。このファイルは、L2 以降のすべてのレベルで再利用される Pydantic スキーマと事前定義されたマラソン シナリオを定義します。✓ schemas + scenarios ready が表示されます。
設定が完了しました。🎽 L0 の前に少し寄り道しましょう。これは、誰もが最初に構築するバージョンです。
4. プロローグ · 1 つの大きなプロンプトを使用しない理由
⚡ 実行する前に、注目すべき 1 つのポイントを絞り込みます。それは、各数値がどこから来ているのかということです。これが演習のすべてです。それ以外は装飾です。
ラダーの前に、ラダーが置き換えるものを実行します。プロンプトですべてを約束する 1 つのエージェントです。天気を取得し、コースを分析し、トレーニング ログを読み取り、条件に基づいてルートを設定し、プランを出力します。
何がわかるか: 自信に満ちた、具体的で、形式が整った戦略... ただし、数値はでっち上げです。1 回のライブ実行では、「今日の天気指標を取得しました」というメッセージで始まり、52°F、風速 9 mph、見たことのないトレーニング ログの分析が報告されました。天気 API も、コースデータも、ログもありません。不透明なモデル呼び出しが入力を作成するか、無意味なものにヘッジします。
これが病気であり、4 つの症状があります。
- 信頼できない - データは流暢に作成されます。
- テストできない - ステップ 4 のルーティングは prose 内に存在するため、単体テストを行う
ifがありません。 - ステップを入れ替えることはできません。実際の天気 API を接続できる継ぎ目がありません。
- すべてを毎回支払う - 5 つのステップ、1 つの巨大な呼び出し、決定論的な部分のキャッシュ保存なし。
その感覚を保ちます。次の 9 つのレベルでは、これらの手順をプロンプトから 1 つずつ削除します。関数が取得（L1 ～ L2a）、if ステートメントがルーティング（L2b）、スペシャリストが作業を分割（L3a ～ L3b）、コードが形状をバインド（L4a ～ L4b）します。
💻 ローカル: python -m shared.prologue
5. L0 · 初めての ADK 2 エージェント
⚡ 要約: エージェントは、モデル + 指示 + 呼び出す可能性のあるツールです。Runner がエージェントを実行します。このレベル以降は、より優れた形状で配置されたエージェントが増えるだけです。
質問: モデルに回答させ、算術演算が重要な場合は実際のコードにアクセスさせることはできますか？
1 つのアイデア - 3 つの部分:
Agent- 推論を行うもの（Gemini モデル + 指示）。Runner- セッション内でエージェントを実行し、イベントをストリーミングするものです。- ツール - モデルが呼び出すことを決定するプレーンな Python 関数（
pace_splits）。ADK はシグネチャと docstring を読み取り、モデルに宣言を渡します。スキーマの記述は不要です。
プロローグの後の最初の修正です。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 エージェントは同じ種類のノードです。予測可能な作業 → 関数（LLM なし、確定的）。推論 → エージェント。
質問: コードのみの部分でモデル呼び出しの料金を支払うことなく、1 つのフローでプレーン コードと LLM を混在させるにはどうすればよいですか？
1 つのアイデア: 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)])
🔍 マーカー: 1 つのエッジ タプル（(START, fetch_conditions, advise)）と、その中央にある Python 関数、ハンドオフを検証する input_schema=。
L0 との違い: Workflow(edges=[...])、START（入力がここに入る）、Event(output=...) を返す関数ノード、input_schema=Conditions。これにより、エージェントが関数出力を認識する前に、そのスキーマに対して検証されます（JSON テキストとして。input_schema は境界を検証しますが、エージェントに Python オブジェクトを渡しません）。
❓ 関数、エージェントの順序は必須ですか？ いいえ。任意の順序、任意の組み合わせ、任意の数で構いません。advise は fetch_conditions のデータを必要とするため、2 番目に実行されます。教訓は序列ではなく、順序です。
👀 読む: fetch_conditions はモデル呼び出しなしでデータを返します。advise には input_schema=Conditions があります。· ▶ [実行] をクリックします。· ✏️ 変更: 関数で temp_f=30 を設定して再実行すると、アドバイスが反転し、関数は引き続き 0 LLM 呼び出しになります。
7. L2a · 並列ファンアウト + JoinNode（ピラー 1a）
⚡ 要約: 並列でファンアウト（無料）、すべてを待機、バンドル、1 つのエージェントに全体像を渡します。
質問: 入力が届く前にフローを描画できます。まず、スケルトンから始めます。データを並列で収集し、バンドルして、1 つのエージェントに渡します。
形状:
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 で始まる 3 つのエッジ（ファンアウト）と、合流点 JoinNode。
- 3 つのフェッチは関数です。これらは並行して実行され、LLM 呼び出しは 0 回です。
JoinNodeは 3 つすべてを待機し、関数名でキー設定された 1 つの型付きペイロード（BundledRunData）にバンドルします。- 1 つの
strategyエージェントがバンドルを読み取り、RaceStrategyを書き込みます。
表示される内容: 各フェッチで started / finished タイムスタンプが出力されます。3 つすべてが 0.0 秒で始まり、ファンアウトは 2.0 秒で終わります。これは、合計すると 4.5 秒になる最も遅いフェッチです。この重複が並列処理です。（最後に表示される合計実時間は、戦略エージェントの LLM 呼び出しも含まれているため、約 8 秒です。合計ではなく、並列クレームのフェッチ タイムスタンプを読み取ってください）。
💡 プロローグのコールバック: メガプロンプトが天気を考案しました。ここでは、温度はフェッチ 関数から取得されます。実際のコード、実際のシームです。缶詰の辞書を実際の天気 API に置き換えるだけで、他は何も変更しません。
❓ よくある質問: どのくらいの
JoinNode
知っておくべきことはありますか？一文で説明すると、すべての並列ブランチが着地するまで待機し、出力を上流関数の名前をキーとする 1 つの辞書にパックし、自身では何も計算しません。この dict が、L2b のルーターが node_input["fetch_weather"]["temp_f"] を書き込める理由です。
👀 読み取り: 3 つのエッジが START から分岐し、JoinNode がそれらを 1 つのエージェントにバンドルします。· ▶ 実行して、合計ではなくタイムスタンプを読み取ります。· ✏️ 変更: 1 回のフェッチでスリープ 3.0 を行うようにします。まず新しいファンアウトの終了時刻を予測してから、検証します。
8. L2b · 決定論的ルーターを追加する（ピラー 1b）
⚡ 要約: L2a は変更されず、プレーンな if によって、実行するエージェントが1 つ決定されます。モデルに質問せずに分岐します。
質問: 暑い日と寒い日でプランを変えるべきですか？モデルに決定を依頼せずに分岐するにはどうすればよいですか？
形状（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-edge {"HOT": ..., "NORMAL": ..., "COLD": ...}。
まとめ - 3 種類の仕事、3 種類の家:
- 予測可能な作業 → 関数（3 つの並列フェッチ）
- 明確なルール → 明示的なルーティング（
route_by_weatherはモデルの決定ではなくifステートメント） - Reasoning（推論）→ モデル（1 つの戦略エージェントが実行される）
表示される内容: temp=78F -> route=HOT、次に構造化された RaceStrategy。純費用: 1 回の LLM 呼び出し。
⚠️ 4 つ目のブランチを追加する場合は、route-dict に DEFAULT_ROUTE エントリも指定します。辞書と一致しないルートはエラーではありません。ブランチが終了し、プログラムが 出力なしで 0 を返して終了します。これは、デバッグが難しいデッドエンドです。
❓ 疑問に思うかもしれません: つまり、L2b は文字どおり L2a にルーターを加えたものですか？はい。フェッチと結合は変更されず、1 回の LLM 呼び出しのままです。変更点: 「常に同じエージェント」が「データによって選択される 3 人のうちの 1 人」になりました。
👀 読み取り: route_by_weather - ルーターはエージェントではなく if ステートメントです。· ▶ run("COLD") も実行します。· ✏️ 変更: 4 番目のエージェントを含む WINDY ブランチを追加します。その前に、上記の DEFAULT_ROUTE 警告をお読みください。
9. L3a · コラボレーション エージェント: 1 つのフラグ、2 つの世界 - ピラー 2
⚡ 要約: 同じチームに 1 つのフラグ。chat は 会話全体を 1 人のスペシャリストに渡し、戻ってきません。single_turn は各スペシャリストをツール（並列サブセット、自動復帰、1 つの合成）に変えます。
質問: チームはわかっているが、どのメンバーが回答すべきかはリクエストによって決まる。LLM にサブセットを選択させ、同時に実行するにはどうすればよいですか？
形状: 6 人のスペシャリスト（医療、天気、ペース、ギア、栄養、メンタル）を統括するコーディネーター。このレベルでは、同じチームが 2 回実行されます。コーディネーターのプロンプトも、6 人のスペシャリストも同じです。唯一の違いは、サブエージェントの 1 つのフラグです。コントラストが教訓です。
▶ Colab: L3a セルを実行 · 📁 GitHub: L3a_collaborative/ · 💻 ローカル: python -m L3a_collaborative.concierge --mode chat "What about fueling?"
🔍 マーカー: ファクトリーの mode="single_turn" と、出力の TRANSFER →（ビート 1）と 1 つのタイムスタンプを共有する DISPATCH → 行のバースト（ビート 2）。
ビート 1: まずデフォルトを実行し、ジョブが失敗するのを確認する
mode= が記述されていない → サブエージェントはデフォルトで chat になります。表示される内容
TRANSFER → nutrition_specialist (transfer_to_agent — the only tool chat subagents provide)
Final speaker: nutrition_specialist
コーディネーターには委任ツールがありません。チャット サブエージェントは、1 人のスペシャリストに会話全体を順番に引き渡す transfer_to_agent のみをコーディネーターに提供します。スペシャリストがユーザーに直接回答し、実行はそこで終了します。並列ディスパッチはありません。返品不可。合成なし。質問を広げると、さらに悪化します。6 人のスペシャリストと 1 回の転送。
これはバグではなく、チャットモードが正常に機能している状態です。会話は、誰かが明示的に転送するまで、保持しているユーザーに属します。オープンエンドのアシスタントには適していますが、パイプライン ステップには適していません。
演出 2 · 1 つのフラグ、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= で記述されたスペシャリストごとに 1 つの委任ツールを挿入します（このテキストは、コーディネーターがサブセットを選択するときに読み取るものです。スキップすると、名前だけでルーティングされます）。コーディネーターは 1 ターンで複数の呼び出しを発行し、ADK はそれらを並行して実行します。各呼び出しは結果を自動的に返し、コーディネーターが結果を合成します。
問題 | 発射するスペシャリスト |
「燃料補給はどうすればいいですか？」 | 栄養のみ |
「18 マイル地点で膝が痛くなった」 | 医療のみ |
「今日はレースに出るべき？」 | 医療 + 天候 + ペース |
「何か心配すべきことはありますか？」 | すべて 6 |
各スペシャリストにブリーフィング全体が渡される理由: 各 single_turn サブエージェントは独自の分離されたセッション ブランチで実行されるため、会話やピアを確認できません。アンビエントなものはありません。コーディネーターは、 SpecialistInput 全体（質問 + 戦略 + ランナーデータ）を個別にすべての並列呼び出しに転送する必要があります。
💡 ADK 2 でこれが直接的に実現される場所: LLM がリクエストごとのサブセットを選択し、並行して実行します。これは sub_agents + mode="single_turn" を介して宣言されます。1.x では、各スペシャリストを AgentTool でラップすることで同じ形状を組み立てることができましたが、変更点は、それが配管ではなく宣言になったことです。（ParallelAgent は常に all で、transfer_to_agent は serial です）。
⚠️ 2 つの注意点があります。（1）モデルがサブセットを選択するため、L2 のハードコードされたルーターよりも決定論的ではありません。正確なサブセットは実行ごとに異なる可能性があります。（2）スペシャリスト 1 人に対して Error validating input: ... 行が表示されることがあります。スペシャリストの出力になることはほとんどありません。output_schema を使用すると、Gemini はサーバーサイドでそれを強制します。これは入力です。コーディネーターは、並列呼び出しごとにネストされた SpecialistInput 全体をそのまま再現する必要がありますが、失敗することもあります。ADK はそのツールの結果としてエラーを返し、コーディネーターは復元し、合成は引き続き行われます。
❓ 次のような疑問をお持ちかもしれません。
chat
1.x スタイルの委任のみ - 一度に 1 つのエージェント？基本的には同じです。1.x のデフォルトの動作に名前が付けられました。single_turn とのギャップは 3 次元です。コーディネーターが保持するもの（1 つの transfer_to_agent 対 1 つのツール（専門家ごと））、作業できる数（1 つ（会話を所有）対 N（並列））、制御が戻るかどうか（戻らない対自動（結果あり））。コードについて: ファクトリの if mode == ブランチは、このコントラストのために 1 つのチームが両方の方法でビルドできるようにするためだけに存在します。実際のアプリでは、1 つのモードがハードコードされ、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
⚡ 要約: 中間モードでは、フィールドが収集されるまでユーザーと会話してから、検証済みのオブジェクトを使用して自動的に戻ります。
質問: 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 モードではできない 3 つのことが行われました。
- タスクの途中で実行が実際に停止した - ハングアップや失敗ではなく、一時停止したタスク。エージェントは確認のための質問をし、タスクを開いたままにしています。（
adk webでは回答を入力するだけです。ハーネスは同じセッションの 2 番目のメッセージとしてスクリプト化します）。 - 次のメッセージで同じタスク エージェントが再開された - 再ルーティングも再委任も行われません。セッションは待機していたユーザーを認識します。
finish_taskが終了しました -mode="task"のため、ツール ADK が を挿入しました。エージェントはこれを呼び出して終了する必要があり、そのペイロードはoutput_schemaに対して検証される必要があります。入力された終了行を含む会話 - その後、制御が自動的にコーディネーターに戻り、結果が添付されます。
モードを選択するための 1 つの質問ルール
💡 「ユーザーはいつまで会話する必要があるか？」チャット = 無期限 · タスク = フィールドが収集されるまで · 単一ターン = なし。
モード | 人間参加型 | 並列？ | 親に戻る |
| 会話の全文 | × | 手動（転送経由） |
| 明確化のための質問のみ | × | 自動（ |
| なし | 必須 | 自動（結果を含む） |
mode はサブエージェントでのみ実行され、コーディネーターでは実行されません。また、ワークフローのノードはデフォルトで single_turn に設定されています（そのため、L1 ～ L2b は書き込みを行いませんでした）。一方、サブエージェントはデフォルトで chat に設定されています（そのため、L3a は書き込みを行う必要がありました）。
⚠️ この上に構築する前に、2 つのバージョンに関する注意事項があります。（1）task
静的グラフノードとしての Workflow(...) はバージョンに依存します - 2.0.0b1 ～ 2.3.0（この Codelab のピン）では、構築時に 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
他の 2 つにはない機能はありますか？3 つの要素があります。自動復帰（チャットが会話を継続する）· 入力された終了行（finish_task のペイロードはスキーマに対して検証される必要があります。トランスクリプトではなくデータが返されます）· 一時停止/再開（⏸ はハングではなく、人間を待機している保留中のタスクです）。
👀 読み取り: gear_fitter - mode="task" + output_schema は契約全体です。· ▶ [実行] をクリックします。· ✏️ 変更: run_desk("I need a hydration vest", "2 liters, medium") — 確認のための質問は適応するが、ゴールは入力されたままになる。
11. L4a · ランタイム サイズの並列ファンアウト（ピラー 3a）
⚡ 要約: スケルトンは 3 つの静的ステップのままです。動的ステップは中央のステップの内部に隠れており、幅は実行時のデータによって決まります。
⚠️ 注意: これはラダーの最も急なステップです。前のレベルは 44 行でしたが、今回は約 120 行です。3 つのエージェントと 2 つのワークフロー ノードがあり、パディングは一切ありません。15 分程度で、最後に Read/Run/Change の行に頼ります。最初のパスですべての行を吸収する必要はありません。
質問: 作業の形状は入力によって異なります。グラフを事前に描画することはできません。ランタイムの幅から始めます。LLM にサブ質問の数を決定させます。
シェイプ（1 レベルの深さ）:
START ─► decompose ─► research_topic (parallel_worker) ─► synthesize
│ │ │
└──┴──┴─ (flat: no children yet)
自由形式の質問は N 個のサブ質問に分解されます。N は実行時に LLM によって選択されます（3 ～ 7）。各サブ質問は並行して調査され、1 つのブリーフィングに合成されます。
▶ Colab: L4a セルを実行 · 📁 GitHub: L4a_flat_research/ · 💻 ローカル: python -m L4a_flat_research.deep_research
🔍 マーカー —
dynamic=True
スイッチ。Dynamic は構成ではなく、書き込みの方法です。2 つのマーカー（@node(parallel_worker=True)（実行時サイズのリストを取得し、アイテムごとに 1 つのワーカーを実行する）と ctx.run_node(...)（コード スケジューリング ノードを直接実行する）のみ。どちらか一方が表示された場合 → ダイナミック広告です。
表示される内容: 分解ツールが 5 つのサブ質問を出力し、それらが並行して調査され、要約されたブリーフィングが表示されます。数値は実行ごとに異なります。固定グラフではこのようなことはできません。
ワーカーの 2 つのフラグについて理解しておく必要があります。
rerun_on_resume=Trueは、ctx.run_nodeを呼び出すノードで必須です。これがないと、ADK はValueErrorを発生させます。再開時に、ディスパッチ ノードを再実行して、生成された子を再構築する必要があります。これは、それらが静的グラフにないためです。retry_config=は、この失敗を制限します。並列ワーカーはすべての兄弟をキャンセルし、1 つの子が失敗するとすぐに再発生させます。そのため、再試行がないと、1 つの 429 エラーで、すでに支払われたすべての呼び出しを含む実行全体が破棄されます。再試行は内部の項目ごとのノードに到達するため、各ブランチは個別に再試行されます。
❓ 次のような疑問が生じるかもしれません。 ADK は、これが動的であることをどのように「認識」するのでしょうか？どこにも宣言されていないため、必要ありません。分解ツールは実行時にリストを生成します。並列ワーカーは、到着した内容に合わせてサイズを調整します。動的性は、オンにしたモードではなく、作成したデータフローのプロパティです。
👀 読み取り: research_topic の 2 つのフラグ（parallel_worker と rerun_on_resume）。· ▶ [実行] をクリックします。· ✏️ 変更: 独自のオープン クエスチョンに置き換える - 入力によって幅が決まるため、N 回変更されます。
12. L4b · 再帰的スポーンを追加（ピラー 3b）
⚡ 要約: 再帰は与えられるものではなく、記述されるものです。ワーカーは ctx.run_node（通常の Python）を介して自身を呼び出すため、ブレーキも記述する必要があります。つまり、MAX_DEPTH です。
質問: 1 つの研究結果から、独自の調査に値する狭いサブトピックが浮上することがあります。ブランチで並列処理を増やし、その範囲を制限するにはどうすればよいですか？
形状（再帰的）:
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, ...) research_topic 自体の中 — 自己参照が再帰 — と、その 1 行上のガード depth < MAX_DEPTH。
表示される内容: spawning N deeper を出力する研究ノード（再帰がライブで発生）、ランタイム ツリーの形状（5 top-level + 10 recursive children など）。ツリーは実行ごとに異なります。
⚠️ ノブを上げる前: 上限が急速に増加します。MAX_DEPTH=3 は最悪のケースを約 30 回の呼び出しから約 93 回に増やします。実行の最後に cancelling N leftover tasks ログ行が表示されることがあります。これは、結果がすでに完了した後に ADK が並列タスク グループを破棄していることを示します。無害です。ロギング構成によっては、表示されないこともあります。
❓ 疑問に思うかもしれません: 動的再帰はデフォルトで有効ではないのですか？いいえ。L4a は完全に動的で、再帰はゼロです。Dynamic は通常の Python 制御フローのみを提供します。L4b はそれを使用して再帰を記述することを選択します。再帰はあなたが記述したため、境界もあなたが記述する必要があります。ここで、「LLM に作業を任せ、境界をコード内に保持する」というスローガンは意味をなさなくなります。
👀 読み取り: ガード: if finding.needs_deeper and depth < MAX_DEPTH。· ▶ [実行] をクリックします。· ✏️ 変更: MAX_DEPTH = 1 を設定して再実行すると、ツリーがフラット化され、実行コストが削減されます。境界はコードで定義します。
13. L5 · どのパターンを使用すべきか
⚡ 要約: 1 つの軸がすべてを決定します。つまり、次のステップを選択するのは、描画したグラフ、LLM、コードのどれかです。
3 つすべてを構築しました。このモデルが有用なのは、パターンを問題の形状に一致させるためです。
軸: 次に実行するものを決定するのは誰ですか？
柱 | 次に実行するものを決定するユーザー | 搭載 |
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 呼び出し。プロンプトに隠されたルーティング | 関数ノードとエージェント ノードをピアとして使用 → 1 回の呼び出し、 |
コラボレーション |
| 宣言されたチーム: |
ダイナミック | 再帰によりフレームワークから外れる | フレームワーク内の |
アプリ全体とグラフで表示できないもの
これで、以下のすべての部分が構築されました。Workflow は graph.edges で構造を公開しているため、この図は手描きではなくコードから生成されたものです。イントロスペクションで見つかったものは、このラボの概要です。
柱 |
| 理由 |
1 · グラフ（L2b） | 10 個のエッジ、ルートなど | 入力が届く前に描画した |
2 · Collaborative（L3a） | 0 エッジ - | LLM はリクエストごとにサブセットを選択します。 |
3 · Dynamic（L4a/L4b） | 3 つのエッジ - どちらも同じ | 再帰はグラフに組み込まれておらず、Python で記述されている |
最後の行は、L4b が回答する質問の証明です。L4a と L4b は同じグラフを持ち、そのうちの 1 つだけが再帰します。
現在構築できるもの
実行した各パターンは、実際の商品の形状です。
このアクティビティでは以下の方法を学びました。 | 実際には、 | 出発 |
グラフ + ルーター（L2a/L2b） | ドキュメント パイプライン、ETL-with-LLM-steps、レビュー/承認チェーン、評価ハーネス | このリポジトリの L2b |
コーディネーター + | 専門チーム、トリアージ デスク、マルチレンズ レビューを備えたサポート コパイロット | マラソン デモモード 2 |
| インテーク フォーム、予約フロー、オンボーディング、KYC など、「収集してから行動する」あらゆるもの | |
動的な幅/奥行き（L4a/L4b） | 研究エージェント、レポート ジェネレータ、不明なサイズの入力に対する監査スイープ | マラソン デモモード 3 |
作曲
3 つのパターンは相互に排他的ではありません。グラフノードはコラボレーション コーディネーターを呼び出すことができ、スペシャリストは動的ワークフローを開始できます。問題の部分ごとに適切なパターンを選択します。これにより、すべてのエージェント システムが 1 つの巨大なプロンプトになるのを防ぐことができます。
💡 独自のワークフローで試してみる: この図を描画したスクリプトは scripts/graph_dump.py です。任意の Workflow を指定すると、実際のエッジが出力されます。これは、構築したものの構造図を無料で作成できる機能です。
14. 完了
マラソン レース当日のコーチを構築し、その過程で ADK 2 の 3 つのオーケストレーション パターンをすべて構築しました。
学習した内容
- プロローグ - 独自の天気を作り出したメガプロンプト: 構造が存在する理由。
- L0 ～ L1 -
Agent、Runner、モデルが呼び出す実際のツール、最初のWorkflow（関数ノード + エージェント ノードをピアとして）。 - L2a / L2b - グラフ ワークフロー: 並列ファンアウト +
JoinNode、決定論的ルーティング - 1 回の LLM 呼び出し。 - L3a - コラボレーション エージェント: 同じチームが
chat（ストランド）で実行され、次にsingle_turn（並列サブセット + 合成）で実行されます。1 つのフラグ、2 つの世界。 - L3b -
taskモード: 確認のための質問の一時停止、スクリプト化された再開、検証済みのオブジェクトを返すfinish_task。 - L4a / L4b - 動的ワークフロー: ランタイム幅（ファンアウト）、ランタイム深度（再帰）、コード内の境界。
- L5 - ディシジョン ツリーとパターンの構成方法。
残しておくべき行
関数はコンテキストを準備します。エッジはワークフローを定義します。ルーターがパスを選択します。モデルが回答を書き込みます。
LLM に作業を任せますが、境界はコードで維持します。
パターンを問題の形状に合わせます。
次のステップ
- これらのレベルの元になったアプリ（マラソン レース当日コーチ）を実行します。これは、3 つのモードをすべてライブで表示するブラウザ UI を備えた 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。
- ワークショップから来ましたか？クレジットと、それによって作成されたプロジェクトは永続的なものではありません。これらのレベルを無料で繰り返し実行するには、代わりに持ち帰り用の設定の手順を行います。無料の AI Studio キー、クラウド プロジェクト、お支払い情報はありません。その 1 つのセルを入れ替えるだけです。