はじめに
Agent Development Kit (ADK) の新しいバージョン ADK 2.0 Beta が発表されました。
このアップデートにより、「グラフベースのエージェントワークフロー」がサポートされました。
「AIの柔軟な推論」と「プログラムの厳密なロジック」をいいとこ取りできる機能ということで、今回は基本的なノードとエッジと条件分岐に絞って早速試してみました。
本記事は基本的に ADK 1.x を触ったことがある方 を対象にしています。
ADKの基本概念は既知として進めます。
ADK 2.0 Beta を使用する際の注意点
- ADK 2.0はベータ版リリースであり、以前のバージョンのADKと併用すると互換性の問題が発生する可能性があります。
グラフベースのワークフローとは？
これまでエージェントに複雑な作業をさせようとすると、どうしてもプロンプトが長くなりがちでした。
そして、指示が長くなればなるほど、AI が途中のステップを飛ばしたり、指示を無視したりして思った通りに動いてくれず、信頼性が低下する問題もあります。
そこで、全体のワークフローをコードで具体的に定義できるグラフワークフローが使えるようになりました。
全体の流れを一つの巨大なプロンプトで指示するのではなく、処理のステップを「ノード（Node）」として細かく分割し、それらを「エッジ（Edge）」で繋ぐようにコードで定義します。
これにより、AIに任せる部分（非決定論的な処理）と、自作コードやツールで確実に動かす部分（決定論的な処理）を組み合わせて、タスクの順序やデータの受け渡しを開発者側で完全にコントロールできるようになりました。
従来のプロンプトベースからの4つの利点
| 利点 | 詳細 |
|---|---|
| 正確なロジックの定義 | ノード間の遷移を管理するルーティングロジックを、コード上で明示的にマッピングできます。 |
| 複雑な構造の実装 | 条件分岐やステート（状態）管理をサポートする高度なエージェントワークフローの構築が容易になります。 |
| AIを介さない関数チェーンの実行 | LLM（生成AIモデル）を都度呼び出すことなく、エージェントツールや独自のコードのみを連携して実行できます。これにより、意図しないハルシネーションを防ぎつつ処理を完了させることが可能です。 |
| 信頼性の強化 | プロンプトの解釈（揺らぎ）だけに依存するのではなく、構造化されたノード定義に基づくため、エージェントの挙動の予測可能性と安定性が大幅に向上します。 |
「AIに判断させる範囲」と「コードで制御する範囲」のバランスを意識して設計することが重要です。
実装方法
Python 3.11以降で以下を実行。
インストール方法
pip install google-adk==2.0.0b1
or
pip install google-adk --pre
ADK の設定方法は以下を参照。
1. ノード（Nodes）とエッジ（Edges）の基本構文
グラフの構成要素となる「ノード」には、AIエージェント、ADKツール、人間の入力タスク、あるいは開発者が記述したPython関数（FunctionNode）などを割り当てることができます。
各ノードは Event オブジェクトを通じてデータを出力し、次のノードへ情報を伝播させます。
フローの実行経路は edges 配列で定義します。
最もシンプルな「順次実行（Sequence）」は以下のように書きます。
from google.adk import Workflow
root_agent = Workflow(
name="sequential_workflow",
edges=[("START", task_A_node, task_B_node, task_C_node)],
)
配列の最初に START キーワードを置くことでグラフの起点を宣言し、以降は記述された順序でノードが実行されます。
2. 条件分岐の実装
実際の業務プロセスでは、状況に応じた条件分岐が不可欠です。
ルーティング判定用の関数ノードを用意し、その出力する route の値に応じて次に実行するノードを振り分けることができます。
from google.adk import Event, Workflow
def router(node_input: str):
# 何らかの判定ロジック
return Event(route="RUN_TASK_C")
root_agent = Workflow(
name="routing_workflow",
edges=[
("START", task_A_node, router),
(router, {
"RUN_TASK_B": task_B_node,
"RUN_TASK_C": task_C_node, # 今回はこちらにルーティングされる
}),
],
)
このように辞書型を用いてマッピングを行うため、分岐ロジックがコード上で極めて見やすくなります。
試してみた
ユーザーの入力内容に応じて AI が判断し、適切なプログラム処理へ受け渡すワークフローを定義してみました。
例として、簡易的なカスタマーサポート自動ルーティングシステムとしました。
1. AIエージェントを「分類器（Classifier）」として定義
まず、ユーザーの意図を汲み取るための AI エージェントを定義しています。
# 判定を行うLLMエージェント
classifier_agent = Agent(
name="classifier_agent",
model="gemini-2.5-flash",
instruction="""ユーザーのメッセージを分析し、以下のいずれかのカテゴリに分類してください。
指定されたカテゴリ名（英字）のみを出力してください。
- COMPLAINT: クレームや不満、対応の遅れに関する内容
- INQUIRY: 商品の仕様や使い方などの一般的な問い合わせ
- OTHER: 上記に該当しないもの""",
output_schema=str,
)
ここでは LLM に対して「特定のキーワード（COMPLAINT, INQUIRY, OTHER）のみを出力せよ」と指示しています。
これにより、プログラムが扱いやすい構造化データ（特定の文字列）に変換しています。
2. ルーター（Router）の定義
AI の出力結果を受け取って「次にどのノードを実行するか」を決定する関数を定義しています。
def router(node_input: str):
"""エージェントの分類結果を受け取り、経路を決定するノード"""
category = node_input.strip()
# AIの出力結果に基づいてルーティングイベントを発行
if category == "COMPLAINT":
return Event(route="ROUTE_COMPLAINT")
elif category == "INQUIRY":
return Event(route="ROUTE_INQUIRY")
else:
return Event(route="ROUTE_OTHER")
Event(route="...") を返すことで、グラフ上の分岐を制御します。
3. ロジック（ハンドラー）の切り出し
各ルートの先にある処理（ノード）を独立した関数として定義しています。
def complaint_handler():
"""クレーム対応ノード"""
return Event(output="【優先対応部署】ご不便をおかけし申し訳ございません。至急確認いたします。")
def inquiry_handler():
"""一般問い合わせ対応ノード"""
return Event(output="【一般サポート】お問い合わせありがとうございます。お調べいたします。")
def other_handler():
"""その他対応ノード"""
return Event(output="【総合受付】メッセージを承りました。担当部署へ確認いたします。")
今回はテキストを返しているだけですが、実際にはここで「データベースへの登録」「Slack への通知」「API の呼び出し」など、AI を介さない処理を記述します。
4. ワークフローの定義
最後に、これら全ての部品を Workflow クラスで繋ぎます。
root_agent = Workflow(
name="ai_routing_workflow",
edges=[
# STARTから始まり、AI分類エージェント -> ルーターの順に実行
("START", classifier_agent, router),
# ルーターが出力した 'route' の値に基づいて処理を分岐
(router, {
"ROUTE_COMPLAINT": complaint_handler,
"ROUTE_INQUIRY": inquiry_handler,
"ROUTE_OTHER": other_handler,
}),
],
)
edges の定義を見るとどのようなワークフローか理解できます。
全コード
from google.adk import Agent, Workflow, Event
# ==========================================
# 1. エージェントとノード（タスク）の定義
# ==========================================
# 判定を行うLLMエージェント
classifier_agent = Agent(
name="classifier_agent",
model="gemini-2.5-flash",
instruction="""ユーザーのメッセージを分析し、以下のいずれかのカテゴリに分類してください。
指定されたカテゴリ名（英字）のみを出力してください。
- COMPLAINT: クレームや不満、対応の遅れに関する内容
- INQUIRY: 商品の仕様や使い方などの一般的な問い合わせ
- OTHER: 上記に該当しないもの""",
output_schema=str,
)
def router(node_input: str):
"""エージェントの分類結果を受け取り、経路を決定するノード"""
category = node_input.strip()
print(f"[*] AIによる分類結果: {category}")
# AIの出力結果に基づいてルーティングイベントを発行
if category == "COMPLAINT":
return Event(route="ROUTE_COMPLAINT")
elif category == "INQUIRY":
return Event(route="ROUTE_INQUIRY")
else:
return Event(route="ROUTE_OTHER")
def complaint_handler():
"""クレーム対応ノード"""
return Event(output="【優先対応部署】ご不便をおかけし申し訳ございません。至急確認いたします。")
def inquiry_handler():
"""一般問い合わせ対応ノード"""
return Event(output="【一般サポート】お問い合わせありがとうございます。お調べいたします。")
def other_handler():
"""その他対応ノード"""
return Event(output="【総合受付】メッセージを承りました。担当部署へ確認いたします。")
# ==========================================
# 2. ワークフロー（グラフ）の構築
# ==========================================
root_agent = Workflow(
name="ai_routing_workflow",
edges=[
# STARTから始まり、AI分類エージェント -> ルーターの順に実行
("START", classifier_agent, router),
# ルーターが出力した 'route' の値に基づいて処理を分岐
(router, {
"ROUTE_COMPLAINT": complaint_handler,
"ROUTE_INQUIRY": inquiry_handler,
"ROUTE_OTHER": other_handler,
}),
],
)
動作
「エアコンの使い方」と入力しました。
すると、最初にエージェントはこのテキストがどのカテゴリに属するかを分析し、"INQUIRY" と出力しています。
その後、エージェント が出力した "INQUIRY" という文字列が、router 関数（ルーターノード）に渡され、Event(route="ROUTE_INQUIRY") が選択されます。
そして、該当のハンドラー(inquiry_handler)が実行され、最終的に【一般サポート】お問い合わせありがとうございます。お調べいたします。と出力されます。
グラフも表示され、コードを読まなくてもグラフを見るだけでフロー全体が一瞬で把握でき、説明コストが下がりそうだと感じました。
他のルーティングの動作
"ROUTE_COMPLAINT"
"ROUTE_OTHER"
まとめ
今回は、新しく発表された ADK 2.0 Beta の「グラフベースのエージェントワークフロー」を実際に試し、基本構文から条件分岐を用いた簡易的なカスタマーサポートのルーティングまでを実装してみました。
これまでのすべてをプロンプトで制御するアプローチとは異なり、ワークフロー全体の流れや確実に行いたい処理を コード（決定論的） で制御し、高度な文脈理解が必要な部分だけを AI（非決定論的） に任せて信頼性を高めることができます。