🤖
ADKのチュートリアルを少し変更しつつやってみる 2. LiteLLMを利用してMulti-Model

2025/04/14に公開
Agent Development Kit
tech
こんにちはサントリーこと大橋です。
前回はCloud Next 2025で発表されたAgent Development KitA(ADK)のチュートリアルの1.をちょっと変えて触ってみました。

前回に引き続き、チュートリアルの2.をちょっと変えて触っていきたいと思います。

一緒に触る方は、前回記事を一通りやってください。 前回記事が実行済みであることを前提に書いていきます。
https://zenn.dev/soundtricker/articles/ae6a98eaefb6ec

 Step 2. LiteLLMを利用してMulti-Model AgentStep 1.では特定のGeminiモデルを利用したお天気Agentを作成しました。特定のモデルを利用することは効果的ですが、複数のモデルを柔軟に利用することは現実世界ではしばしばメリットがあります。
パフォーマンス: 特定のタスク(コーディング、推論、クリエイティブライティングなど)では一部のモデルのほうがパフォーマンスが優れています。
コスト: モデルによって価格帯が異なります。
機能：モデルは多様な機能、コンテキストウィンドウサイズ、微調整のためのオプションを提供します。
可用性／冗長性：代替モデルを用意することで、あるプロバイダーで問題が発生してもアプリケーションの機能を維持できます。

ADKはLiteLLMライブラリとの統合により、モデル間のシームレスな切り替えを実現しています。LiteLLMは、100を超える異なるLLMへの一貫したインターフェースとして機能します。
https://docs.litellm.ai/
このステップでは以下のことを行います。
LiteLLMラッパーを使用して、OpenAIで（GPT）やAnthropic（Claude）などのプロバイダーのモデルを使用するようにADK Agentを構成する方法を学習します。
それぞれ異なる LLM を基盤とする Weather Agent のインスタンスを定義、設定（独自のSessionとRunnerを使用）、すぐにテストできます。
これらの異なるAgentと対話することで、同じ基盤ツールを使用している場合でも、応答の潜在的な変化を観察できます。
なおOpenAIのAPI Keyを持ってないので、Gemini 2.0とGemini 2.5、あとClaude 3.7 Sonnectを利用してAgentを作成したいと思います。
ベストプラクティス

モデル名には定数（Step 0で定義したMODEL_GPT_4O、MODEL_CLAUDE_SONNETなど）を使用し、タイプミスを防ぎます。
エラー処理

エージェント定義をtry...exceptブロックで囲みます。これにより、特定のプロバイダーのAPIキーが欠落または無効であっても、コードセル全体が失敗することがなくなり、構成されたモデルでチュートリアルを続行できます。
まずはOpenAI's GPT-4o を利用したモデルを作るとチュートリアルでは書いてありますが、今回はGemini 2.0 flashを利用したAgentを作ります。また同時にclaudeを利用したAgentを作成します。
weather_agent/agent.py
import os
import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from . import tools


MODEL_GEMINI_2_0_FLASH = "gemini-2.0-flash"
MODEL_CLAUDE_SONNET = "anthropic/claude-3-5-sonnet-20240620"
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リスト
)

weather_agent_flash = Agent(
    name="weather_agent_flash",
    model=MODEL_GEMINI_2_0_FLASH,
    description="Provides weather information for specific cities. (Using Gemini-2.0 Flash)",
    instruction="You are a helpful weather assistant. Your primary goal is to provide current weather reports.  created by Gemini-2.0 Flash"
                "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リスト
)


weather_agent_claude = Agent(
    name="weather_agent_claude",
    model=LiteLlm(model=MODEL_CLAUDE_SONNET),
    description="Provides weather information for specific cities. (Using Claude 3 Sonnect)",
    instruction="You are a helpful weather assistant. Your primary goal is to provide current weather reports.  created by Claude 3 Sonnect"
                "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リスト
)
次にこれらを呼び出す処理をmain.pyに作成します。SessionServiceやRunnerはそれぞれに作成します。
main.py
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from weather_agent.agent import weather_agent, weather_agent_flash, weather_agent_claude
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}'.")

session_service_flash = InMemorySessionService()

APP_NAME_FLASH = "weather_tutorial_app_flash"
USER_ID_FLASH = "user_1_flash"
SESSION_ID_FLASH = "session_001_flash" # 今回は固定のセッションIDで実行

session = session_service_flash.create_session(
    app_name=APP_NAME_FLASH,
    user_id=USER_ID_FLASH,
    session_id=SESSION_ID_FLASH
)

print(f"Session created: App='{APP_NAME_FLASH}', User='{USER_ID_FLASH}', Session='{SESSION_ID_FLASH}'")

runner_flash = Runner(
    agent=weather_agent_flash, # 実行するAgent
    app_name=APP_NAME_FLASH,
    session_service=session_service_flash
)
print(f"Runner created for agent '{runner_flash.agent.name}'.")

session_service_claude = InMemorySessionService()

APP_NAME_CLAUDE = "weather_tutorial_app_claude"
USER_ID_CLAUDE = "user_1_claude"
SESSION_ID_CLAUDE = "session_001_claude" # 今回は固定のセッションIDで実行

session = session_service_claude.create_session(
    app_name=APP_NAME_CLAUDE,
    user_id=USER_ID_CLAUDE,
    session_id=SESSION_ID_CLAUDE
)

print(f"Session created: App='{APP_NAME_CLAUDE}', User='{USER_ID_CLAUDE}', Session='{SESSION_ID_CLAUDE}'")

runner_claude = Runner(
    agent=weather_agent_claude, # 実行するAgent
    app_name=APP_NAME_CLAUDE,
    session_service=session_service_claude
)
print(f"Runner created for agent '{runner_claude.agent.name}'.")


async def call_agent_async(query: str, runner: Runner, user_id: str, session_id: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():

    for r, user_id, session_id in [(runner, USER_ID, SESSION_ID), (runner_flash, USER_ID_FLASH, SESSION_ID_FLASH), (runner_claude, USER_ID_CLAUDE, SESSION_ID_CLAUDE)]:
        print(f">> Run runner: {r.app_name} {user_id} {session_id}")
        await call_agent_async("ロンドンの天気は?", runner=r, user_id=user_id, session_id=session_id)
        await call_agent_async("パリについては？", runner=r, user_id=user_id, session_id=session_id)
        await call_agent_async("ニューヨークの天気を教えて下さい。", runner=r, user_id=user_id, session_id=session_id)

async def main():

    await run_conversation()

if __name__ == "__main__":
    asyncio.run(main())
実行してみます。
uv run main.py
多分model毎に違う結果になったと思います。

ADK+LiteLLMを利用することで柔軟にモデルを切り替えることが可能になりました。
次のステップでは単一のAgentからあげんｔ同士がタスクを委任できる小規模なチームを構築します。
Step 2. LiteLLMを利用してMulti-Model Agent

Discussion
