🤖
ADKのチュートリアルを少し変更しつつやってみる 3. Agent Teamの作成 - 挨拶と送別の委譲
こんにちはサントリーこと大橋です。
前回、前々回とCloud Next 2025で発表されたAgent Development KitA(ADK)のチュートリアルの1.,2.をちょっと変えて触ってみました。
前回から少し時間が空きましたが
前回に引き続き、チュートリアルの3.をちょっと変えて触っていきたいと思います。
一緒に触る方は、前々回の記事を一通りやってください。 前々回の記事が実行済みであることを前提に書いていきます。
特に今回はSub Agentをやっていきます。
Sub Agent(とAuto Delegation)はADKのキーコンセプトとも言える機能なのでちゃんとやっていきたいです。
なお、最近ADKのリポジトリを見ているのですが、そろそろ0.2.0がリリースされそうですね。
0.2.0ではadk createで新しいAgentを作成する機能が追加されたり、
Dev UIの改善などが行われているようです。
Step 3. Agent Teamの作成 - 挨拶と送別の委譲
Step 1、2では天気予報の検索に特化したAgentを作成し、実験しました。
このAgentは特定のタスクには効果的ですが、実際のアプリケーションでは、より多様な処理が必要になります。この天気予報AgentにToolや複雑な指示を追加し続けることも可能ですが、すぐに管理が困難になり、効率が低下する可能性があります。
より堅牢なアプローチは、Agent Teamを構築することです。これは以下の手順で作成されます。
- それぞれ特定の機能向けに設計された複数の特化Agent（例：天気予報用、挨拶用、計算用など）を作成する。
- 最初のリクエストを受信するRoot Agent（またはOrchestrator）を指定する。
- Root Agentがユーザーの意図に基づいて、最も適切な特化SubAgentにリクエストを以上できるようにする。
Agent Teamを構築する理由は？
- Modularity(モジュール性): 個々のAgentの開発、テスト、保守が容易になる
- Specialization(特化): 各Agentは特定のタスクに合わせて微調整(指示やモデルの選択)ができる。
- Scalability(スケーラビリティー): 新しいAgentを追加することで機能追加を容易にできる。
- Efficiency(効率性): より単純なタスク(挨拶等)をより単純で安価なモデルに任せることができる
このStepでは以下のことを行います。
- 挨拶(
say_hello)と、さよなら(say_goodbye)を行うToolを定義します。 -
greeting_agentとfarewell_agentという2つの特化Sub Agentを定義します。 - 作成した天気予報Agent(
weather_agent_v2)をRoot Agentとして更新します。 - Root AgentとSub Agentを設定し、automatic delegation(自動委任)を有効にします。
- Root Agentに様々なリクエストを送信して委任フローを確認します。
Sub Agent用のTool定義
まず、特化AgentのToolとして機能するシンプルなPython関数を作成します。
明確なDocstringはそれらを使うAgentによってとても重要なことを忘れないようにしてください。
weather_agent.py
# 以下追加
def say_hello(name: str = "there") -> str:
"""Provides a simple greeting, optionally addressing the user by name.
Args:
name (str, optional): The name of the person to greet. Defaults to "there".
Returns:
str: A friendly greeting message.
"""
print(f"--- Tool: say_hello called with name: {name} ---")
return f"Hello, {name}!"
def say_goodbye() -> str:
"""Provides a simple farewell message to conclude the conversation."""
print(f"--- Tool: say_goodbye called ---")
return "Goodbye! Have a great day."
Sub Agent(Greeting & Farewell)の定義
次に特化Agentを作成します。非常に焦点を絞った指示と、特に明確な説明に注目してください。この説明は、Root AgentがSub Agentにいつ委任するかを判断する際に使用する主要な情報です。
これらのSub Agentには異なるLLMを使用することもできます。
が今回は面倒なので同じにします。
ベストプラクティス
Sub Agentのdescriptionは、Sub Agentの特定の機能を正確かつ簡潔に要約する必要があります。これは、効果的なautomatic delegation(自動委任)に不可欠です。
ベストプラクティス
Sub AgentのinstructionはSub Agentの限定されたスコープに合わせて調整し、何をすべきか、何をすべきでないかを正確に伝える必要があります（例：「あなたの唯一のタスクは…」）。
まずモデル名をconstants.pyに移動します。
weather_agent/constants.py
MODEL_GEMINI_2_0_FLASH = "gemini-2.0-flash-001"
MODEL_CLAUDE_SONNET = "anthropic/claude-3-sonnet-20240229"
MODEL_GEMINI_2_5_PRO = "gemini-2.5-pro-exp-03-25"
次にAgentを定義します。
weather_agent/sub_agents/greeting_agent.py
from google.adk.agents import Agent
from .. import tools
from ..constants import MODEL_GEMINI_2_0_FLASH
greeting_agent = Agent(
# Using a potentially different/cheaper model for a simple task
model=MODEL_GEMINI_2_0_FLASH,
name="greeting_agent",
instruction="You are the Greeting Agent. Your ONLY task is to provide a friendly greeting to the user. "
"Use the 'say_hello' tool to generate the greeting. "
"If the user provides their name, make sure to pass it to the tool. "
"Do not engage in any other conversation or tasks.",
description="Handles simple greetings and hellos using the 'say_hello' tool.", # Crucial for delegation
tools=[tools.say_hello],
)
weather_agent/sub_agents/farewell_agent.py
from google.adk.agents import Agent
from .. import tools
from ..constants import MODEL_GEMINI_2_0_FLASH
farewell_agent = Agent(
# Can use the same or a different model
model=MODEL_GEMINI_2_0_FLASH, # Sticking with GPT for this example
name="farewell_agent",
instruction="You are the Farewell Agent. Your ONLY task is to provide a polite goodbye message. "
"Use the 'say_goodbye' tool when the user indicates they are leaving or ending the conversation "
"(e.g., using words like 'bye', 'goodbye', 'thanks bye', 'see you'). "
"Do not perform any other actions.",
description="Handles simple farewells and goodbyes using the 'say_goodbye' tool.", # Crucial for delegation
tools=[tools.say_goodbye],
)
これらのAgentを__init__.pyで宣言しておきます。
weather_agent/sub_agents/__init__.py
from .farewell_agent import farewell_agent
from .greeting_agent import greeting_agent
__all__ = [
"greeting_agent",
"farewell_agent"
]
Root AgentとSub Agentの定義
次にweather_agentを更新します。
主要な変更点は、
-
sub_agentsパラメータの追加-
greeting_agentとfarewell_agentをリストで渡します。
-
-
instructionパラメータの更新- Root AgentにSub Agentについて、またSub Agentにタスクを委任するタイミングについて明示的に伝えます。
キーコンセプト
Automatic Delegation(Auto Flow) 自動委任
ADKはsub_agentsを渡すと、Automatic Delegation(自動委任)を可能にします。
Root Agentがユーザークエリを受け取ると、LLMは自身の指示とToolだけではなく、各Sub Agentの記述も考慮します。
LLMは、クエリがSub Agentの記述された機能（例：「簡単な挨拶を処理する」）とより合致していると判断した場合、そのターンでそのSub Agentに制御を移すための特別な内部アクションを自動的に生成します。Sub Agentは、自身のモデル、指示、およびToolを使用してクエリを処理します。
ベストプラクティス
Root Agentのinstructionsが委任の決定を明確に導くようにしてください。
Sub Agentの名前を上げて委任が行われる記述をします。
weather_agent/agent.py
from google.adk.agents import Agent
from . import tools
from .sub_agents import greeting_agent, farewell_agent
from .constants import MODEL_GEMINI_2_0_FLASH
weather_agent_team = Agent(
name="weather_agent_v2", # Give it a new version name
model=MODEL_GEMINI_2_0_FLASH,
description="The main coordinator agent. Handles weather requests and delegates greetings/farewells to specialists.",
instruction="You are the main Weather Agent coordinating a team. Your primary responsibility is to provide weather information. "
"Use the 'get_weather' tool ONLY for specific weather requests (e.g., 'weather in London'). "
"You have specialized sub-agents: "
"1. 'greeting_agent': Handles simple greetings like 'Hi', 'Hello'. Delegate to it for these. "
"2. 'farewell_agent': Handles simple farewells like 'Bye', 'See you'. Delegate to it for these. "
"Analyze the user's query. If it's a greeting, delegate to 'greeting_agent'. If it's a farewell, delegate to 'farewell_agent'. "
"If it's a weather request, handle it yourself using 'get_weather'. "
"For anything else, respond appropriately or state you cannot handle it.",
tools=[tools.get_weather], # Root agent still needs the weather tool for its core task
# Key change: Link the sub-agents here!
sub_agents=[greeting_agent, farewell_agent]
)
Agent Teamとのやり取り
Root Agentと専用のSub Agentを定義したので、委任をテストします。
- 非同期関数
run_team_conversationを定義します。 - この関数内に、このテスト実行専用の新しい
InMemorySessionServiceと特定のセッション（session_001_agent_team）を作成します。これにより、Agent Teamをテストするための会話履歴が分離されます。 -
weather_agent_team（RootAgent）と専用のSessionServiceを使用するように構成されたRunner（runner_agent_team）を作成します。 - 更新した
call_agent_async関数を使用して、runner_agent_teamに色々な種類のクエリ（挨拶、天気予報のリクエスト、別れ）を送信します。この特定のテストでは、Runner、UserID、SessionIDを明示的に渡します。 -
run_team_conversation関数を直ちに実行します。
想定されるフローは以下のとおりです。
- 「こんにちは!」クエリが
runner_agent_teamに送信されます。 - Root Agent（
weather_agent_team）はそれを受け取り、その指示とgreeting_agentの説明に基づいてタスクを委任します。 -
greeting_agentはクエリを処理し、say_helloToolを呼び出して応答を生成します。 - 「ニューヨークの天気は？」クエリは委任されず、Root Agentが
get_weatherToolを使用して直接処理します。 - 「ありがとう。さようなら!」クエリは
farewell_agentに委任され、say_goodbyeToolを使用します。
main.py
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from weather_agent.agent import weather_agent_team
import asyncio
from google.genai import types # メッセージを作成するため
from dotenv import load_dotenv
import logging
import warnings
warnings.filterwarnings("ignore")
logging.basicConfig(level=logging.ERROR)
load_dotenv()
async def call_agent_async(query: str, runner, user_id: str, session_id: str):
"""Sends a query to the agent and prints the final response."""
print(f"\n>>> User Query: {query}")
# Prepare the user's message in ADK format
content = types.Content(role='user', parts=[types.Part(text=query)])
final_response_text = "Agent did not produce a final response." # Default
# Key Concept: run_async executes the agent logic and yields Events.
# We iterate through events to find the final answer.
async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
# You can uncomment the line below to see *all* events during execution
# print(f" [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")
# Key Concept: is_final_response() marks the concluding message for the turn.
if event.is_final_response():
if event.content and event.content.parts:
# Assuming text response in the first part
final_response_text = event.content.parts[0].text
elif event.actions and event.actions.escalate: # Handle potential errors/escalations
final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
# Add more checks here if needed (e.g., specific error codes)
break # Stop processing events once the final response is found
print(f"<<< Agent Response: {final_response_text}")
async def run_conversation():
# conversation関数を定義する前にroot_agent変数が存在するかどうかを確認します
root_agent_var_name = 'root_agent' # Default name from Step 3 guide
if 'weather_agent_team' in globals(): # Check if user used this name instead
root_agent_var_name = 'weather_agent_team'
elif 'root_agent' not in globals():
print("⚠️ Root agent ('root_agent' or 'weather_agent_team') not found. Cannot define run_team_conversation.")
# コードブロックが実行される場合に NameError を防ぐためにダミー値を割り当てます。
root_agent = None
if root_agent_var_name in globals() and globals()[root_agent_var_name]:
async def run_team_conversation():
print("\n--- Testing Agent Team Delegation ---")
# InMemorySessionService は、このチュートリアル用のシンプルな非永続ストレージです。
session_service = InMemorySessionService()
# インタラクションコンテキストを識別するための定数を定義する
APP_NAME = "weather_tutorial_agent_team"
USER_ID = "user_1_agent_team"
SESSION_ID = "session_001_agent_team"
# 会話が行われる特定のセッションを作成します
session = session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=SESSION_ID
)
print(f"Session created: App='{APP_NAME}', User='{USER_ID}', Session='{SESSION_ID}'")
# --- 実際のRoot Agentオブジェクトを取得する ---
# 決定された変数名を使用する
actual_root_agent = globals()[root_agent_var_name]
# このAgent Teamテスト専用のRunnerを作成する
runner_agent_team = Runner(
agent=actual_root_agent,
app_name=APP_NAME,
session_service=session_service
)
print(f"Runner created for agent '{actual_root_agent.name}'.")
await call_agent_async(query = "こんにちは!",
runner=runner_agent_team,
user_id=USER_ID,
session_id=SESSION_ID)
await call_agent_async(query = "ニューヨークの天気は？",
runner=runner_agent_team,
user_id=USER_ID,
session_id=SESSION_ID)
await call_agent_async(query = "ありがとう。さようなら!",
runner=runner_agent_team,
user_id=USER_ID,
session_id=SESSION_ID)
# 会話の実行
# Note: これには、Root AgentとSub Agentによって使用されるモデルの API キーが必要になる場合があります。
await run_team_conversation()
else:
print("\n⚠️ Skipping agent team conversation as the root agent was not successfully defined in the previous step.")
async def main():
await run_conversation()
if __name__ == "__main__":
asyncio.run(main())
実行してみます。
uv run main.py
--- Testing Agent Team Delegation ---
Session created: App='weather_tutorial_agent_team', User='user_1_agent_team', Session='session_001_agent_team'
Runner created for agent 'weather_agent_v2'.
>>> User Query: こんにちは!
--- Tool: say_hello called with name: there ---
<<< Agent Response: Hello, there!
>>> User Query: ニューヨークの天気は？
--- Tool: get_weatherが呼び出されました city: New York ---
<<< Agent Response: The weather in New York is sunny with a temperature of 25°C.
>>> User Query: ありがとう。さようなら!
--- Tool: say_goodbye called ---
<<< Agent Response: Goodbye! Have a great day.
各挨拶や、サヨナラはSub Agentにて実行され、天気の場合のみRoot Agentで実行されました。
正常にAutomatic Delegationが動作しています。
これでやっとADKらしくなってきましたね
Discussion