🤖

ADKのチュートリアルを少し変更しつつやってみる 1. 最初のAgent

2025/04/14に公開

Agent Development Kit

tech

こんにちはサントリーこと大橋です。
前回はCloud Next 2025で発表されたAgent Development KitA(ADK)のクイックスタートをちょっと変えて触ってみました。
https://zenn.dev/soundtricker/articles/704ba75a69f061
これだとADKの全体像が把握できてないので、今度は↓の内容を翻訳しつつチュートリアルやってみたいと思います。

チュートリアルが結構長いので記事を分割して実行していきます。
https://google.github.io/adk-docs/get-started/tutorial
なおADKで準備されているチュートリアルはJupyter Notebook形式で作成されたものを、ドキュメントに表示しており、そのままだとNotebook外だとやりにくいので、自分なりにやりやすくしつつ触っていきたいと思います。
https://github.com/google/adk-docs/blob/main/examples/python/notebooks/adk_tutorial.ipynb
また、環境のセットアップには mise を使ってやります。

この記事を読みながら一緒に触っていく方で、mise をインストールしてない方は自分でインストールしてください。
https://mise.jdx.dev/

 Step 0: 環境準備まず環境準備します。

利用するのは以下の環境
Python: 3.12.5
プロジェクト管理: uv
プロジェクトディレクトリ作成とpython環境の準備
mkdir adk-tutorial
cd adk-tutorial
mise use python@3.12.5
mise use uv@latest
uv python pin 3.12.5
uv init --app .
source ./.venv/bin/activate
次にadkとLiteLLMをインストールします。
 adkとその他もろもろのインストール
uv add google-adk litellm python-dotenv
litellmは後で解説があります。

チュートリアルではJupyter Notebookなので、ファイルとかプロジェクト構成とかがありませんが、

今回はローカルフォルダで作っているので一旦以下のようなプロジェクト構成にします。
adk-tutorial # project root
├── .venv
├── .env #環境変数の設定
├── README.md
├── weather_agent # AI Agent module
│   ├── __init__.py
│   ├── agent.py # 実際のAIエージェント
│   └── tools.py
├── main.py # エージェントのランナー
├── mise.toml
├── pyproject.toml
└── uv.lock
適当にフォルダを作っていってください。
次に環境変数周りを設定していきます。

なおチュートリアルでは1箇所で各種ライブラリをimportしていますが、ファイルを分割しているのでそれぞれのファイルでimportをやっていきます。
まず環境変数は .env に書いていきます。

チュートリアルでは、 Gemini、OpenAI、Anthropic(Claude)のモデルを使ってマルチモデルエージェントを作成します。

それぞれのサイトでAPI Keyを取得してきてください。

OpenAIっていま無料枠有ったっけ...?

Anthropicは有った気がします。
https://aistudio.google.com/app/apikey

https://platform.openai.com/api-keys

https://console.anthropic.com/settings/keys
やだなーやだなー

変だなー変だなー

怖いなー怖いなー

と思う方は、GeminiのAPI Keyだけでも取得して、Geminiの複数モデル(2.0flashと2.5proとか)を利用してください。
取得したら .env に設定していきます。
.env
GOOGLE_GENAI_USE_VERTEXAI=False
GOOGLE_API_KEY=put your api key
OPENAI_API_KEY=put your api key
ANTHROPIC_API_KEY=put your api key
GOOGLE_GENAI_USE_VERTEXAIはGeminiをVertex AI経由で利用する場合に使いますが、今回は利用しないのでFalseを指定します。
次に利用するモデルを定義します。

これは特にやらずに、agentに直接文字列に指定してもいいです。
weather_agent/agent.py
MODEL_GEMINI_2_0_FLASH = "gemini-2.0-flash"
MODEL_CLAUDE_SONNET = "anthropic/claude-3-sonnet-20240229"
MODEL_GEMINI_2_5_PRO = "gemini-2.5-pro-exp-03-25"

 Step 1: 最初のAgent 基本的な天気予報チュートリアルではまず基本的なお天気Botを作成します。

質問に対して天気情報を取得し、答えるだけのシンプルなボットです。

このボットには2つのピースが存在します。
Tool: Python関数で天気情報をAgentへ提供します。
Agent: ユーザのリクエストを理解するための脳 天気情報ツールがあることを理解し、それをどの様に使い、いつ使うべきかを決定します。

 Toolの定義前の記事でも書きましたが、ToolはAgentに対して、テキスト作成以外の具体的な機能を提供する構成要素です。ToolはAPIの呼び出し、DBへのクエリ、計算の実行などなどの特定の機能を実行する標準的なPython関数です。
今回は擬似的な天気情報を提供するToolを作成します。

今回はADKのチュートリアルなので、面倒なところは一旦簡易な実装で済まそうという感じです。

実際に天気情報を取得したければあとから修正してください。
さてここで大事になってくるのが Docstring です。AgentのLLMは以下の点を理解するためにPython関数のDocstringを利用します。
ツールの機能
いつ利用するか
必要な引数 (city: str等)
返却される情報
ベストプラクティス

ToolのDocstringは 明確で説明的かつ、正確なものを書くことが大事です。これによりLLMは正しくToolを利用することができます。

現時点だと、Geminiなど一部のLLMは日本語より英語が得意なため、英語で書いていたほうがより無難だと思います。チュートリアル中ではDocstringはGoogleスタイルを利用していますが、LLMが理解できる形式であれば何でも大丈夫だと思います。今回はあえて reStructuredTextスタイルに変更して書いてみたいと思います。
今回Toolは weather_agent/tools.py に書いていきます。
weather_agent/tools.py
def get_weather(city: str) -> dict:
    """
    Retrieves the current weather report for a specified city.

    :param str city:  The name of the city (e.g., "New York", "London", "Tokyo").
    :return: A dictionary containing the weather information.
              Includes a 'status' key ('success' or 'error').
              If 'success', includes a 'report' key with weather details.
              If 'error', includes an 'error_message' key.
    :rtype: dict
    """    
    # docstringは英語で
    # 戻り値はJSON 変換可能な値で

    # ベストプラクティス: Tool内でログ書いておいたほうがデバッグしやすいよ
    print(f"--- Tool: get_weatherが呼び出されました city: {city} ---")
    city_normalized = city.lower().replace(" ", "") # とりあえずDict Keyなので軽くNormalizeする

    # モック天気情報
    mock_weather_db = {
        "newyork": {"status": "success", "report": "The weather in New York is sunny with a temperature of 25°C."},
        "london": {"status": "success", "report": "It's cloudy in London with a temperature of 15°C."},
        "tokyo": {"status": "success", "report": "Tokyo is experiencing light rain and a temperature of 18°C."},
    }

    # ベストプラクティス:Tool内でわかってるエラーハンドリングはなるべくやっておく
    if city_normalized in mock_weather_db:
        return mock_weather_db[city_normalized]
    else:
        return {"status": "error", "error_message": f"Sorry, I don't have weather information for '{city}'."}

 Agentの定義Agentを次は作ります。

ADKにおけるAgentはユーザと、LLMそして、Toolを対話させるためのものです。
Agentには以下のようなパラメータがあります。

name: 一意となるAgent名 (e.g., "weather_agent_v1").

model: LLM モデル

description: Agentの目的を完結にまとめたもの 他のAgentがこのAgentにタスクを委任するかどうか決定する際に使います。ということは英語が良いかな

instruction: システムプロンプトとか呼ばれるもの LLMの行動、役割、人格、目標 Toolを具体的にいつ、どの様に使うか、LLMに伝えるセキュリティ情報(情報の秘匿等)などの詳細なガイダンス

tools: Agentが利用するPythonで書かれた、Tool関数リスト
ベストプラクティス

instructionはとても重要で、明確かつ具体的に指示を記載する必要があります。指示が詳細であればあるほどLLMは自分の役割とToolの利用方法を理解します。必要に応じてエラー処理についても書いておいたほうが良いです。

また、情報の秘匿などのセキュリティ情報も記載しておいたほうが良いです。
ベストプラクティス

わかりやすい名前(name)と説明(description)の値を設定すること。

これらはADK内で使用され、automatic delegationなどの機能に必要です。

なので英語で書いておいたほうが無難だと思います。
今回Agentは weather_agent/agent.py に記述します。
weather_agent/agent.py
import os
import asyncio
from google.adk.agents import Agent
from . import tools


MODEL_GEMINI_2_0_FLASH = "gemini-2.0-flash"
MODEL_CLAUDE_SONNET = "anthropic/claude-3-sonnet-20240229"
MODEL_GEMINI_2_5_PRO = "gemini-2.5-pro-exp-03-25"

AGENT_MODEL = MODEL_GEMINI_2_5_PRO # とりあえずGemini 2.5 Proで

weather_agent = Agent(
    name="weather_agent_v1",
    model=AGENT_MODEL,
    description="Provides weather information for specific cities.",
    instruction="You are a helpful weather assistant. Your primary goal is to provide current weather reports. "
                "When the user asks for the weather in a specific city, "
                "you MUST use the 'get_weather' tool to find the information. "
                "Analyze the tool's response: if the status is 'error', inform the user politely about the error message. "
                "If the status is 'success', present the weather 'report' clearly and concisely to the user. "
                "Only use the tool when a city is mentioned for a weather request.",
    tools=[tools.get_weather], # Toolリスト
)


 RunnerとSession Serviceの準備実際に作成したAgentで、ユーザとの会話を管理して、実行にするには、2つの要素が必要です。

SessionService: ユーザ毎の会話履歴と状態を管理します。InMemorySessionService は、全てをメモリに保存するシンプルな実装で、テストやシンプルなアプリケーションに適しています。状態の永続化については後で

Runner: RunnerはAIとの会話を調整する為のエンジンで、ユーザの入力を受け取って、適切なAgentへルーティングしAgentのロジックに基づいてLLMとツールへの呼び出しを管理し、SessionServiceを使ってセッションの更新を処理し、会話の進行状況を示すイベントを生成します。
一旦 SessionService と Runner をmain.pyに書いてみます。
main.py
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from weather_agent.agent import weather_agent
from dotenv import load_dotenv
import logging
import warnings

warnings.filterwarnings("ignore")

logging.basicConfig(level=logging.ERROR)

load_dotenv()

session_service = InMemorySessionService()

APP_NAME = "weather_tutorial_app"
USER_ID = "user_1"
SESSION_ID = "session_001" # 今回は固定のセッションIDで実行

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}'")

runner = Runner(
    agent=weather_agent, # 実行するAgent
    app_name=APP_NAME,
    session_service=session_service
)
print(f"Runner created for agent '{runner.agent.name}'.")


 Agentとの会話会話として成立させるには、Agentにメッセージを送り、その結果を受取る必要があります。

LLMの呼び出しとToolの実行には時間がかかることがあるため、ADKのRunnerは非同期で動作します。
今回は以下の機能を持つ非同期ヘルパー関数(call_agent_async)を作成します。
ユーザのクエリ文字列を受け取る
ADKへContentの形式で渡す

runner.run_asyncを ユーザとセッション情報、2.とともに呼び出し

Runnerによって作成(返却)された Events を繰り返し処理します。Eventsはエージェントの実行におけるステップを表します。(例: Tool呼び出しの要求、Tool結果の受信、LLMの中間処理、最終応答 等) event.is_final_response() を使用して、そのEventが最終応答であるか判別して、出力します。
なぜ？

なぜ非同期なのか？ LLMやTool(主に外部API呼び出し等)とのやり取りはI/O依存の処理で、asyncioを使用することで、プログラムは実行をブロックすることなくこれらの操作を効率的に処理できるからです。
ということで main.py を更新して、 call_agent_asyncを作成します。
main.py
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from weather_agent.agent import weather_agent
from dotenv import load_dotenv

### ここから追加
import asyncio
from google.genai import types # メッセージを作成するため
### ここまで

import logging
import warnings

warnings.filterwarnings("ignore")

logging.basicConfig(level=logging.ERROR)

load_dotenv()

session_service = InMemorySessionService()

APP_NAME = "weather_tutorial_app"
USER_ID = "user_1"
SESSION_ID = "session_001" # 今回は固定のセッションIDで実行

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}'")

runner = Runner(
    agent=weather_agent, # 実行するAgent
    app_name=APP_NAME,
    session_service=session_service
)
print(f"Runner created for agent '{runner.agent.name}'.")

### ここから追加
import asyncio
from google.genai import types # メッセージを作成するため
async def call_agent_async(query: str):
  """クエリをAgentへ送信し、コンソールへ出力します。 Sends a query to the agent and prints the final response."""
  print(f"\n>>> User Query: {query}")

  # ADKのフォーマットでメッセージを作成
  content = types.Content(role='user', parts=[types.Part(text=query)])

  final_response_text = "Agent did not produce a final response." # Default

  # 重要: `run_async` は Agentロジックを実行して、Eventを作成します。
  # イベントを反復処理して最終応答を見つけます。
  async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
      # コメントインすると全イベントが出力されます。
      # print(f"  [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")

      # 重要: is_final_response() で反復の終了を判定
      if event.is_final_response():
          if event.content and event.content.parts:
             # 最初の部分ではテキスト応答を想定
             final_response_text = event.content.parts[0].text
          elif event.actions and event.actions.escalate: # 潜在的なエラー/エスカレーションを処理する
             final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
          # 他のエラー処理があれば記述する
          break # 最終応答が見つかったら反復処理を終了

  print(f"<<< Agent Response: {final_response_text}")
### ここまで


 会話の実行最後に、作成したAgentにいくつかのクエリを送って、テストしてみます。

非同期呼び出しをmainの非同期関数でラップしてawaitを使って実行します。
そして出力を確認します。
ユーザクエリを確認
AgentがToolを使用すると「--- Tool: get_weatherが呼び出されました ---」というログが出力されます
Agentの最終的な応答 特に天気データが利用できない場合(パリの場合)の処理方法を確認します。
main.py
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from weather_agent.agent import weather_agent

### ここから追加
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()

session_service = InMemorySessionService()

APP_NAME = "weather_tutorial_app"
USER_ID = "user_1"
SESSION_ID = "session_001" # 今回は固定のセッションIDで実行

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}'")

runner = Runner(
    agent=weather_agent, # 実行するAgent
    app_name=APP_NAME,
    session_service=session_service
)
print(f"Runner created for agent '{runner.agent.name}'.")

async def call_agent_async(query: str):
  """クエリをAgentへ送信し、コンソールへ出力します。 Sends a query to the agent and prints the final response."""
  print(f"\n>>> User Query: {query}")

  # ADKのフォーマットでメッセージを作成
  content = types.Content(role='user', parts=[types.Part(text=query)])

  final_response_text = "Agent did not produce a final response." # Default

  # 重要: `run_async` は Agentロジックを実行して、Eventを作成します。
  # イベントを反復処理して最終応答を見つけます。
  async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
      # コメントインすると全イベントが出力されます。
      # print(f"  [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}")

      # 重要: is_final_response() で反復の終了を判定
      if event.is_final_response():
          if event.content and event.content.parts:
             # 最初の部分ではテキスト応答を想定
             final_response_text = event.content.parts[0].text
          elif event.actions and event.actions.escalate: # 潜在的なエラー/エスカレーションを処理する
             final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
          # 他のエラー処理があれば記述する
          break # 最終応答が見つかったら反復処理を終了

  print(f"<<< Agent Response: {final_response_text}")

### ここから追加
async def run_conversation():
    await call_agent_async("ロンドンの天気は?")
    await call_agent_async("パリについては？") # エラーの確認
    await call_agent_async("ニューヨークの天気を教えて下さい。")

async def main():

    # 会話を非同期実行
    await run_conversation()

if __name__ == "__main__":
    asyncio.run(main())

### ここまで
これを実行します。
uv run main.py
以下が出力されるはずです。
期待する結果

Session created: App='weather_tutorial_app', User='user_1', Session='session_001'
Runner created for agent 'weather_agent_v1'.

>>> User Query: ロンドンの天気は?
--- Tool: get_weatherが呼び出されました city: London ---
<<< Agent Response: はい、ロンドンの現在の天気をお伝えします。

ロンドンの天気は曇りで、気温は15℃です。

>>> User Query: パリについては？
--- Tool: get_weatherが呼び出されました city: Paris ---
<<< Agent Response: 申し訳ありません、パリの天気情報は見つかりませんでした。

>>> User Query: ニューヨークの天気を教えて下さい。
--- Tool: get_weatherが呼び出されました city: New York ---
<<< Agent Response: ニューヨークの天気は晴れで、気温は25℃です。
実施できましたか？

Toolに渡すパラメータがクエリから取り出したものではなく、Parisなどの引数の形式になっているあたりがユーザの問を理解し、Toolの引数をわかったうえで渡してると言えます。

 まとめとりあえず最初のAgentを作ってみました。
今回のチュートリアルで大事な要素は
Tool
Agent
SessionService
Runner
となります。

またasync/awaitを利用した非同期呼び出しである点も大事なところですね

Step 0: 環境準備

Step 1: 最初のAgent 基本的な天気予報

Toolの定義

Agentの定義

RunnerとSession Serviceの準備

Agentとの会話

会話の実行

まとめ

Discussion