はじめに
ADK（Agent Development Kit）は、Google が開発している、特定のタスクを自律的に実行する「AIエージェント」を効率的に開発するためのツール、ライブラリ、およびフレームワークの総称です。
類似機能を持ったフレームワークとして、LangChain 、AutoGen が挙げられます。
Google Skills の、「ADK を使用してマルチエージェント システムを構築する」というラボを通して、
ADKでマルチエージェントを構築する場合どうなるのか試してみました。
ADKはチャット画面等のGUIもフレームワークとして含まれているため、GUIでの動作確認、及びトレースも可能です。
エージェントは下図の様なツリーの構成と記載の役割で構築して行きます。
前提
- Google Cloud プロジェクトは準備されている
- Google Cloud の APIは有効化完了している(本記事に有効化必要なAPIは記載しません)
- Google Cloud の Cloud Shell エディタ 上でソース配置したり、コマンド実行して動作確認する
Cloud Shell エディタ起動
- Google Cloud プロジェクト で、Cloud Shellのターミナルを表示します。
- Cloud Shellのターミナルで、下記コマンド実行します。
cloudshell workspace ~
ディレクトリ、ファイル構成
Cloud Shell エディタ 上で各ファイルを準備します。
adk_multiagent_systemsディレクトリ は、ユーザーホームディレクトリ配下に配置すると、後述のコマンドをそのまま実行できます。
adk_multiagent_systems
│ requirements.txt ･･･ 必要なライブラリ
│ callback_logging.py ･･･ ADK用のログインユーティリティ
└─parent_and_subagents
__init__.py ･･･ このディレクトリをエージェント パッケージとして識別させるための定義
agent.py ･･･ 各エージェント定義
- requirements.txt
langchain-community==0.3.20
- callback_logging.py
import logging
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmResponse, LlmRequest
def log_query_to_model(callback_context: CallbackContext, llm_request: LlmRequest):
if llm_request.contents and llm_request.contents[-1].role == 'user':
for part in llm_request.contents[-1].parts:
if part.text:
logging.info("[query to %s]: %s", callback_context.agent_name, part.text)
def log_model_response(callback_context: CallbackContext, llm_response: LlmResponse):
if llm_response.content and llm_response.content.parts:
for part in llm_response.content.parts:
if part.text:
logging.info("[response from %s]: %s", callback_context.agent_name, part.text)
elif part.function_call:
logging.info("[function call from %s]: %s", callback_context.agent_name, part.function_call.name)
- init.py
agent.py ファイルをインポートし、ADK は、このディレクトリをエージェント パッケージとして識別する。
from . import agent
- agent.py
- エージェント定義について
- root_agent: ADK の開発 UI とコマンドライン インターフェースでこのエージェントを識別するために steering という名前が付いています。ユーザーに質問をして（旅行先はもう決まっているか、それとも旅先選びのアドバイスが必要か）、ユーザーの回答に基づいて 2 つのサブエージェントのうちどちらに会話を誘導するかを判断します。
- travel_brainstormer: 旅行先を決めていないユーザーがブレインストーミングするのを手伝います。
- attractions_planner: 旅行先を決めたユーザーが、旅先でやりたいことのリストを作成するのを手伝います。
- エージェントのツリー指定
- root_agent の エージェント定義の中で、
sub_agents=[travel_brainstormer, attractions_planner]で子のエージェントを指定している。
- root_agent の エージェント定義の中で、
- エージェント定義について
import os
import sys
import logging
sys.path.append("..")
from callback_logging import log_query_to_model, log_model_response
from dotenv import load_dotenv
import google.cloud.logging
from google.adk import Agent
from google.genai import types
from typing import Optional, List, Dict
from google.adk.tools.tool_context import ToolContext
load_dotenv()
cloud_logging_client = google.cloud.logging.Client()
cloud_logging_client.setup_logging()
# ツール定義
def save_attractions_to_state(
tool_context: ToolContext,
attractions: List[str]
) -> dict[str, str]:
"""観光地 のリストを state["attractions"] に保存します。
Args:
attractions [str]: 観光地リストに追加する文字列のリスト
Returns:
None
"""
# stateから既存の観光地を読み込みます。存在しない場合は空のリストを作成します。
existing_attractions = tool_context.state.get("attractions", [])
# 'attractions' キーを、新旧のリストを組み合わせたもので更新します。
# ツールが実行されると、ADKはイベントを作成し、
# セッション状態に対応する更新を行います。
tool_context.state["attractions"] = existing_attractions + attractions
# ツールのベストプラクティスとして、戻り値の辞書でステータスメッセージを返します。
return {"status": "success"}
# エージェント定義
attractions_planner = Agent(
name="attractions_planner",
model=os.getenv("MODEL"),
description="ある国で訪れるべき観光地のリストを作成します。",
instruction="""
目的地を決めるのに助けが必要な場合は、
'travel_brainstormer' に案内してください。
目的地が決まっている場合は、
'attractions_planner' に案内してください。
- ユーザーが返信したら、ツールを使って選択された観光地を保存し、さらに可能な観光地を提案してください。
- リストを見たいと言われたら、 { attractions? } の箇条書きリストを提供し、さらにいくつか提案してください。
""",
before_model_callback=log_query_to_model,
after_model_callback=log_model_response,
# 指示があった場合、この行の下に tools パラメータを貼り付けてください
tools=[save_attractions_to_state]
)
travel_brainstormer = Agent(
name="travel_brainstormer",
model=os.getenv("MODEL"),
description="どの目的地を訪れるかユーザーが決めるのを手助けします。",
instruction="""
旅行者に人気のある国をいくつか提案してください。
ユーザーが旅行の主な目的（冒険、レジャー、学習、ショッピング、
あるいは芸術鑑賞など）を特定できるように手助けしてください。
彼らの優先順位に基づいて、素晴らしい旅行先となる国を特定してください。
""",
before_model_callback=log_query_to_model,
after_model_callback=log_model_response,
)
root_agent = Agent(
name="steering",
model=os.getenv("MODEL"),
description="ユーザーの旅行の冒険を開始します。",
instruction="""
どこに旅行したいか決まっているか、それとも決めるのに
助けが必要かユーザーに尋ねてください。
""",
generate_content_config=types.GenerateContentConfig(
temperature=0,
),
# 指示があった場合、この行の下に sub_agents パラメータを追加してください
sub_agents=[travel_brainstormer, attractions_planner]
)
ライブラリインストール、環境設定
- Cloud Shell エディタ のターミナルを起動して、ライブラリをインストールします。
cd ~/
export PATH=$PATH:"/home/${USER}/.local/bin"
python3 -m pip install google-adk -r adk_multiagent_systems/requirements.txt
- .envファイル作成
YOUR_GCP_PROJECT_IDには、実行するGoogleCloudのプロジェクトIDを設定してください。
GOOGLE_GENAI_USE_VERTEXAI=TRUE は、Gemini API キー認証ではなく、Vertex AI を認証に使用することを指定しています。
cd ~/adk_multiagent_systems cat << EOF > parent_and_subagents/.env GOOGLE_GENAI_USE_VERTEXAI=TRUE GOOGLE_CLOUD_PROJECT=YOUR_GCP_PROJECT_ID GOOGLE_CLOUD_LOCATION=global MODEL=gemini-2.5-flash EOF
ADK 開発UI を 起動して、動作確認
- Cloud Shell エディタ のターミナル下記コマンド実行して、ターミナルの出力にある http://127.0.0.1:8000 リンクをクリックして、開発UIを表示します。
cd ~/adk_multiagent_systems adk web
- チャット入力欄に、下記入力して挨拶します。
こんにちは
- 下記入力して、エージェント「travel_brainstormer」に振られて、どんなこと聞かれるか試します。
具体的にどこに旅行したいか決まっていません。
- 下記入力して、エージェント「travel_brainstormer」とやり取りします。
温泉が良いですね。
- 下記入力して、エージェント「attractions_planner」に振られるか試します。
台湾が良いですね。
入力と返答の全文は下図の通りです。
エージェント アイコン（ロボの様なアイコン）をクリックして、実際に、どのエージェントに振られたか確認します。
「travel_brainstormer」に振られています。
「attractions_planner」に振られています。
最後に
ADK（Agent Development Kit）を使って、マルチエージェントを試してみました。
構築が大変そうなエージェントシステムを容易に構築することができました。
プロンプトの指定で、どのエージェントに割り振るか、AIが判断して振り分けているのは興味深いと思いました。
また、ADK に、開発UI があることで、動作確認や、トレースしての検証がし易いのは便利なフレームワークだと思いました。
簡単30秒