Agent Development Kitを触ってみる #1

uv init adk_test -p 3.11
cd adk_test
uv add google-adk

mkdir multi_tool_agent/
echo "from . import agent" > multi_tool_agent/__init__.py

import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city.

    Args:
        city (str): The name of the city for which to retrieve the weather report.

    Returns:
        dict: status and result or error msg.
    """
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": (
                "The weather in New York is sunny with a temperature of 25 degrees"
                " Celsius (41 degrees Fahrenheit)."
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather information for '{city}' is not available.",
        }


def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city.

    Args:
        city (str): The name of the city for which to retrieve the current time.

    Returns:
        dict: status and result or error msg.
    """

    if city.lower() == "new york":
        tz_identifier = "America/New_York"
    else:
        return {
            "status": "error",
            "error_message": (
                f"Sorry, I don't have timezone information for {city}."
            ),
        }

    tz = ZoneInfo(tz_identifier)
    now = datetime.datetime.now(tz)
    report = (
        f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
    )
    return {"status": "success", "report": report}


root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description=(
        "Agent to answer questions about the time and weather in a city."
    ),
    instruction=(
        "You are a helpful agent who can answer user questions about the time and weather in a city."
    ),
    tools=[get_weather, get_current_time],
)

GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY={取得したAPIキー}

uv run adk web

def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city.

    Args:
        city (str): The name of the city for which to retrieve the weather report.

    Returns:
        dict: status and result or error msg.
    """
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": (
                "The weather in New York is sunny with a temperature of 25 degrees"
                " Celsius (41 degrees Fahrenheit)."
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather information for '{city}' is not available.",
        }

root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description=(
        "Agent to answer questions about the time and weather in a city."
    ),
    instruction=(
        "You are a helpful agent who can answer user questions about the time and weather in a city."
    ),
    tools=[get_weather, get_current_time],
)

uv add litellm

from google.adk.models.lite_llm import LiteLLM

try:
    root_agent = Agent(
        name="weather_time_agent",
        model=LiteLlm("provider/model_name"),
        description=(
            "Agent to answer questions about the time and weather in a city."
        ),
        instruction=(
            "You are a helpful agent who can answer user questions about the time and weather in a city."
        ),
        tools=[get_weather, get_current_time],
    )
except Exception as e:
    print(f"Counld not create Agent. Error: {e}")

from google.adk.sessions import InMemorySessionService

session_service = InMemorySessionService()

session = session_service.create_session(
    app_name="weather_time_agent",
    user_id="user_1",
    session_id="session_1"
)

from google.adk.runners import Runner

runner = Runner(
    agent=root_agent,
    app_name="weather_time_agent",
    session_service=session_service,
)

import asyncio
from google.genai import types

async def call_agent_async(query: str):
    """Sends a query to the agent and prints the final response."""
    print(f"\n>>> User Query: {query}")

    content = types.Content(role='user', parts=[types.Part(text=query)])

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

    # 最終的な回答が得られるまで反復
    async for event in runner.run_async(user_id="user_1", session_id="session_1", 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:
                # 最初のpartに応答があると仮定
                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("What is the weather like in London?")
    await call_agent_async("How about Paris?")
    await call_agent_async("Tell me the weather in New York")

await run_conversation()

>>> User Query: What is the weather like in London?
<<< Agent Response: I am sorry, I cannot get the weather information for London. The weather information for 'London' is not available.

>>> User Query: How about Paris?
<<< Agent Response: I am sorry, I cannot get the weather information for Paris. The weather information for 'Paris' is not available.

>>> User Query: Tell me the weather in New York
<<< Agent Response: OK. The weather in New York is sunny with a temperature of 25 degrees Celsius (41 degrees Fahrenheit).