[note] Google Agent Development Kit (ADK)

基本功能

定義 Tool

基本的 Tool

# https://google.github.io/adk-docs/get-started/tutorial/


def get_weather(city: str) -> dict:
    # Best Practice：為工具撰寫清晰、描述性且準確的 docstring，這對於 LLM 正確使用工具至關重要。
    """
    Retrieves weather, converts temp unit based on session state.

    Args:
        city (str): The name of the city (e.g., "New York", "London", "Tokyo").

    Returns:
        dict: 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.
    """
    # Best Practice: 記錄工具執行日誌以便於除錯
    print(f"--- Tool: get_weather called for {city} ---")

    city_normalized = city.lower().replace(" ", "")  # Basic input normalization

    # Mock weather data (always stored in Celsius internally)
    mock_weather_db = {
        "newyork": {"temp_c": 25, "condition": "sunny"},
        "london": {"temp_c": 15, "condition": "cloudy"},
        "tokyo": {"temp_c": 18, "condition": "light rain"},
    }

    # Best Practice: Handle potential errors gracefully within the tool
    if city_normalized in mock_weather_db:
        data = mock_weather_db[city_normalized]
        temp_value = data["temp_c"]
        condition = data["condition"]

        report = f"The weather in {city.capitalize()} is {condition} with a temperature of {temp_value:.0f}°C."
        result = {"status": "success", "report": report}
        return result
    else:
        # Handle city not found
        error_msg = f"Sorry, I don't have weather information for '{city}'."
        print(f"--- Tool: City '{city}' not found. ---")

        result = {"status": "error", "error_message": error_msg}
        return result


print("✅'get_weather' tool defined.")

使用 state 的 Tool

• SessionService

# https://google.github.io/adk-docs/get-started/tutorial/

from google.adk.tools.tool_context import ToolContext


# Key Concept：tool function 的最後一個參數會被 ADK 注入 ToolContext，ToolContext 讓 tool 能夠和 session's context 溝通的橋樑，其中包含讀取和寫入資料狀態。
def get_weather(city: str, tool_context: ToolContext) -> dict:
    """
    Retrieves weather, converts temp unit based on session state.

    Args:
        city (str): The name of the city (e.g., "New York", "London", "Tokyo").

    Returns:
        dict: 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.
    """
    print(f"--- Tool: get_weather called for {city} ---")

    # Best Practice: 當從 state 中讀取資料時，使用 dictionary.get('key', default_value) 來處理 key 可能不存在的情況，確保 tool 不會爆掉。
    preferred_unit = tool_context.state.get(
        "user_preference_temperature_unit", "Celsius"
    )
    print(
        f"--- Tool: Reading state 'user_preference_temperature_unit': {preferred_unit} ---"
    )

    city_normalized = city.lower().replace(" ", "")  # Basic input normalization

    # Mock weather data (always stored in Celsius internally)
    mock_weather_db = {
        "newyork": {"temp_c": 25, "condition": "sunny"},
        "london": {"temp_c": 15, "condition": "cloudy"},
        "tokyo": {"temp_c": 18, "condition": "light rain"},
    }

    if city_normalized in mock_weather_db:
        data = mock_weather_db[city_normalized]
        temp_c = data["temp_c"]
        condition = data["condition"]

        # Format temperature based on state preference
        if preferred_unit == "Fahrenheit":
            temp_value = round((temp_c * 9 / 5) + 32)
            temp_unit = "°F"
        else:
            temp_value = temp_c
            temp_unit = "°C"

        report = f"The weather in {city.capitalize()} is {condition} with a temperature of {temp_value:.0f}{temp_unit}."
        result = {"status": "success", "report": report}
        print(f"--- Tool: Generated report in {preferred_unit}. Result: {result} ---")

        # Example of writing back to state (optional for this tool)
        tool_context.state["last_city_checked_stateful"] = city
        print(f"--- Tool: Updated state 'last_city_checked_stateful': {city} ---")

        return result
    else:
        # Handle city not found
        error_msg = f"Sorry, I don't have weather information for '{city}'."
        print(f"--- Tool: City '{city}' not found. ---")

        result = {"status": "error", "error_message": error_msg}
        return result


print("✅ State-aware 'get_weather' tool defined.")

需要的話，可以透過 stored_session = session_service.sessions[APP_NAME][USER_ID][SESSION_ID] 來取得和修改 state，例如：

async def run_conversation_with_state():
    print("\n--- Testing State: Temp Unit Conversion & output_key ---")

    # 1. Check weather (Uses initial state: Celsius)
    print("--- Turn 1: Requesting weather in London (expect Celsius) ---")
    await call_agent_async(
        query="What is the weather like in London?",
        runner=runner,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )

    # 2. Manually update state to Fahrenheit
    print("\n--- Manually Updating State: Setting unit to Fahrenheit ---")
    try:
        # Access the internal storage directly - THIS IS SPECIFIC TO InMemorySessionService for testing
        stored_session = session_service.sessions[APP_NAME][USER_ID][SESSION_ID]
        stored_session.state["user_preference_temperature_unit"] = "Fahrenheit"
        # Optional: You might want to update the timestamp as well if any logic depends on it

        stored_session.last_update_time = time.time()
        print(
            f"--- Stored session state updated. Current 'user_preference_temperature_unit': {stored_session.state['user_preference_temperature_unit']} ---"
        )
    except KeyError:
        print(
            f"--- Error: Could not retrieve session '{SESSION_ID}' from internal storage for user '{USER_ID}' in app '{APP_NAME}' to update state. Check IDs and if session was created. ---"
        )
    except Exception as e:
        print(f"--- Error updating internal session state: {e} ---")

    # 3. Check weather again (Tool should now use Fahrenheit)
    # This will also update 'last_weather_report' via output_key
    print("\n--- Turn 2: Requesting weather in New York (expect Fahrenheit) ---")
    await call_agent_async(
        query="Tell me the weather in New York.",
        runner=runner,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )

    # 4. Test basic delegation (should still work)
    # This will update 'last_weather_report' again, overwriting the NY weather report
    print("\n--- Turn 3: Sending a greeting ---")
    await call_agent_async(
        query="Hi!",
        runner=runner,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )

async def main():
    await run_conversation_with_state()

asyncio.run(main())

定義 Agent

官方文件

LLM Agents

建立 Agent

# https://google.github.io/adk-docs/get-started/tutorial/#2-define-the-agent
from google.adk.agents import Agent

from weather_agent.constants import MODEL_GEMINI
from weather_agent.guardrail.llm import block_keyword_guardrail
from weather_agent.guardrail.tool import block_paris_tool_guardrail
from weather_agent.sub_agent.farewell_agent.agent import farewell_agent
from weather_agent.sub_agent.greeting_agent.agent import greeting_agent
from weather_agent.tools.get_current_time import get_current_time
from weather_agent.tools.get_weather import get_weather

model = MODEL_GEMINI

# https://google.github.io/adk-docs/agents/llm-agents/#defining-the-agents-identity-and-purpose
root_agent = Agent(
    model=model,
    # 最佳實踐: 選擇具有描述性的 "name" 和 "description"。
    # 它們會被 ADK 內部使用，並且對於 automatic delegation 這樣的功能非常重要。
    name="weather_agent_v6_tool_guardrail",
    description="The main coordinator agent. Handles weather requests and delegates greetings/farewells to specialists.",
    # Best Practice: 提供清晰且具體的 instruction prompts。指令越詳細，LLM 越能理解其角色以及如何有效使用工具。
    # 如果需要，請明確說明錯誤處理方式。
    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=[get_weather, get_current_time],
    sub_agents=[greeting_agent, farewell_agent],
    output_key="last_weather_report",  # <<< Auto-save agent's final weather response
    before_model_callback=block_keyword_guardrail,  # <<< Assign the guardrail callback
    before_tool_callback=block_paris_tool_guardrail,  # <<< Add tool guardrail
)
print(f"✅ Root Agent '{root_agent.name}' created using stateful tool and output_key.")

建立 Session Service 和 Runner

SessionService 透過以下元素建立：

• app_name
• user_id
• session_id

Runner 透過以下元素建立：

• agent
• app_name
• session_service

# https://google.github.io/adk-docs/get-started/tutorial/#3-setup-runner-and-session-service
import logging
import warnings

from dotenv import load_dotenv
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

from weather_agent.agent import root_agent

load_dotenv()

warnings.filterwarnings("ignore")
logging.basicConfig(level=logging.ERROR)

# --- Session Management ---
# Key Concept: SessionService stores conversation history & state.
# InMemorySessionService is simple, non-persistent storage for this tutorial.
session_service = InMemorySessionService()
print("✅ New InMemorySessionService created for state demonstration.")

# Define constants for identifying the interaction context
APP_NAME = "weather_tutorial_app"
SESSION_ID = "session_state_demo_001"
USER_ID = "user_state_demo"


# Define initial state data - user prefers Celsius initially
initial_state = {"user_preference_temperature_unit": "Celsius"}

session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID,
    state=initial_state,  # <<< Initialize state during creation
)
print(f"Session created: App='{APP_NAME}', User='{USER_ID}', Session='{SESSION_ID}'")

runner = Runner(
    agent=root_agent,
    app_name=APP_NAME,
    session_service=session_service,
)
print(f"Runner created for agent '{runner.agent.name}'.")

整合 Agent 和 Runner 來進行對話

import asyncio
from google.adk.runners import Runner
from google.genai import types

# ...

# 取得 LLM 的 response
async def call_agent_async(query: str, runner: 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"\n<<< Agent Response: {final_response_text}")

# 和 Agent 進行對話
async def run_conversation():
    await call_agent_async(
        "What is the weather like in London?",
        runner=runner,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )
    await call_agent_async(
        "How about Paris?", runner=runner, user_id=USER_ID, session_id=SESSION_ID
    )  # Expecting the tool's error message
    await call_agent_async(
        "Tell me the weather in New York",
        runner=runner,
        user_id=USER_ID,
        session_id=SESSION_ID,
    )


async def main():
    await run_conversation()
    final_session = session_service.get_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
    if final_session:
        print(
            f"Final Preference: {final_session.state.get('user_preference_temperature_unit')}"
        )
        print(
            f"Final Last Weather Report (from output_key): {final_session.state.get('last_weather_report')}"
        )
        print(
            f"Final Last City Checked (by tool): {final_session.state.get('last_city_checked_stateful')}"
        )
        # Print full state for detailed view
        print(f"Full State: {final_session.state}")
    else:
        print("\n❌ Error: Could not retrieve final session state.")


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

驗證使用者輸入的內容（Input Guardrail）

keywords: before_model_callback

官方資料

Step 5: Adding Safety - Input Guardrail with before_model_callback

before_model_callback 主要用啦驗證使用者提供給 LLM 的 prompt 是否有不合法的內容。

定義會驗證使用者 Input Prompts 的 callback function。這個函式可以接受 callback_context: CallbackContext 和 llm_request: LlmRequest：

• 在 callback_context 中可以得到 Agent 相關的資訊（例如，callback_context.agent_name）或 session state 的資料（如，callback_context.state）
• 在 llm_request 中可以得到使用者輸入的訊息，例如，llm_request.contents、llm_request.config
• 根據判斷結果，如果需要 block 則回傳 LlmResponse ，否著回傳 None

# https://google.github.io/adk-docs/get-started/tutorial
# @title 1. Define the before_model_callback Guardrail

# Ensure necessary imports are available
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.genai import types  # For creating response content
from typing import Optional


def block_keyword_guardrail(
    callback_context: CallbackContext, llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """
    Inspects the latest user message for 'BLOCK'. If found, blocks the LLM call
    and returns a predefined LlmResponse. Otherwise, returns None to proceed.
    """
    agent_name = (
        callback_context.agent_name
    )  # Get the name of the agent whose model call is being intercepted
    print(f"--- Callback: block_keyword_guardrail running for agent: {agent_name} ---")

    # 取得使用者輸入的內容
    last_user_message_text = ""
    if llm_request.contents:
        # Find the most recent message with role 'user'
        for content in reversed(llm_request.contents):
            if content.role == "user" and content.parts:
                # Assuming text is in the first part for simplicity
                if content.parts[0].text:
                    last_user_message_text = content.parts[0].text
                    break  # Found the last user message text

    print(
        f"--- Callback: Inspecting last user message: '{last_user_message_text[:100]}...' ---"
    )  # Log first 100 chars

    # --- Guardrail Logic ---
    keyword_to_block = "BLOCK"
    if keyword_to_block in last_user_message_text.upper():  # Case-insensitive check
        print(f"--- Callback: Found '{keyword_to_block}'. Blocking LLM call! ---")
        # Optionally, set a flag in state to record the block event
        callback_context.state["guardrail_block_keyword_triggered"] = True
        print(f"--- Callback: Set state 'guardrail_block_keyword_triggered': True ---")

        # 如果碰到有被 Block 就回傳這個
        return LlmResponse(
            content=types.Content(
                role="model",  # Mimic a response from the agent's perspective
                parts=[
                    types.Part(
                        text=f"I cannot process this request because it contains the blocked keyword '{keyword_to_block}'."
                    )
                ],
            )
            # Note: You could also set an error_message field here if needed
        )
    else:
        # 沒有被 Block 就回傳 None
        return None  # Returning None signals ADK to continue normally


print("✅ block_keyword_guardrail function defined.")

將上面定義好的 callback function 放到 Agent 中：

root_agent = Agent(
    model=model,
    name="weather_agent_v4",
    description="Main agent: Provides weather (state-aware unit), delegates greetings/farewells, saves report to state.",
    instruction="...",
    tools=[get_weather_stateful, get_current_time],
    sub_agents=[greeting_agent, farewell_agent],
    output_key="last_weather_report",  # <<< Auto-save agent's final weather response
    before_model_callback=block_keyword_guardrail,  # <<< 把上面定義好的 callback 放進來
)

驗證 Tool 使用的參數（Tool Argument Guardrail）

keywords: before_tool_callback

官方文件

Step 6: Adding Safety - Tool Argument Guardrail (before_tool_callback)

before_tool_callback 主要用來驗證 LLM 提供給 tool 的參數，常用來：

• Argument Validation：驗證 LLM 提供的給 tool 的參數是否合法
• Resource Protection：避免使用者帶入的參數會導致 tool 執行後耗費太多資源
• Dynamic Argument Modification：根據 session state 或 context 內的其他資訊來調整原本代入給 tool 的參數

# https://google.github.io/adk-docs/get-started/tutorial/#step-6-adding-safety-tool-argument-guardrail-before_tool_callback
from typing import Any, Dict, Optional

from google.adk.tools.base_tool import BaseTool
from google.adk.tools.tool_context import ToolContext


def block_paris_tool_guardrail(
    tool: BaseTool,  # 拿到 tool 相關的資訊，例如 tool.name
    args: Dict[str, Any],  # 拿到 LLM 提供給 tool 的參數
    tool_context: ToolContext,  # 拿到 agent 相關的資訊，例如 tool_context.agent_name
) -> Optional[Dict]:
    """
    Checks if 'get_weather' is called for 'Paris'.
    If so, blocks the tool execution and returns a specific error dictionary.
    Otherwise, allows the tool call to proceed by returning None.
    """
    tool_name = tool.name
    agent_name = tool_context.agent_name  # 拿到 agent 相關的資訊
    print(
        f"--- Callback(tool): block_paris_tool_guardrail running for tool '{tool_name}' in agent '{agent_name}' ---"
    )
    print(f"--- Callback(tool): Inspecting args: {args} ---")

    # --- Guardrail Logic ---
    target_tool_name = "get_weather"
    blocked_city = "paris"

    if tool_name == target_tool_name:
        city_argument = args.get("city", "")
        if city_argument and city_argument.lower() == blocked_city:
            print(
                f"--- Callback(tool): Detected blocked city '{city_argument}'. Blocking tool execution! ---"
            )
            # Optionally update state
            tool_context.state["guardrail_tool_block_triggered"] = True
            print(
                "--- Callback(tool): Set state 'guardrail_tool_block_triggered': True ---"
            )

            # 返回符合 tool 預期錯誤輸出格式的 dict
            # 這個 dict 會成為 tool 的結果，跳過實際的工具執行。
            return {
                "status": "error",
                "error_message": f"Policy restriction: Weather checks for '{city_argument.capitalize()}' are currently disabled by a tool guardrail.",
            }
        else:
            print(
                f"--- Callback(tool): City '{city_argument}' is allowed for tool '{tool_name}'. ---"
            )
    else:
        print(
            f"--- Callback(tool): Tool '{tool_name}' is not the target tool. Allowing. ---"
        )

    # If the checks above didn't return a dictionary, allow the tool to execute
    print(f"--- Callback(tool): Allowing tool '{tool_name}' to proceed. ---")
    return None  # 返回 None 允許實際的工具函數執行


print("✅ block_paris_tool_guardrail function defined.")

Agents

官方文件

• Agents @ ADK
• Multi-Agent Systems in ADK @ ADK

Multi-agent System

• ⁠一個 Agent Instance 只能被添加為一次 sub-agent，如果一個 sub-agent 有多個 parent 將導致 ValueError
• 在 ADK 中從 BaseAgent 提供多個特化的 agents，它們本身不執行任務，而是會協調 sub-agents 的執行流程。

Tools

官方文件

Tools

Function Tools

keywords: FunctionTool

官方文件

Function Tools

Basic Tools

• Docstring

  • 函式的命名最好是有意義、人類能理解的

• Parameters

• Function Parameter 盡量不要帶 default value，因為 LLM 目前還不支援解析預設值。

• 參數的命名最好是有意義、人類能理解的

• 參數越少越好

• 參數的型別越簡單愈好，最好是 str、int 這種型別，盡可能不要是 custom classes

• Return Type

  • Function 最好都回傳 dict，否則也會被轉成 Key 為 result 的 dict
  • 回傳的內容最好是人類（LLM）看得懂的內容，而不是 error code
  • 回傳的 dict 中，最好包含 status key，值則是 success、error 或 pending，如此讓 LLM 可以很清楚知道目前的狀態

Long Running Function Tools

• LongRunningFunctionTool 是 FunctionTool 的 subclass
• 透過 generator function 來達到
• 回傳的結構最好包含：
  • status：pending, running, waiting_for_input
  • progress：已完成的百分比或步驟數
  • message：給 LLM 看的訊息
  • estimated_completion_time：例如，「還剩下 5 分鐘」

Agent-as-a-Tool

Agent-as-a-Tool 的概念就類似建立一個 Python 函式來呼叫另一個 Agent，並把該 Agent 的 response 當成該函式的回傳值。

Agent-as-a-Tool 和 Sub-agent 的差異在於：

• Agent-as-a-Tool：當 Agent A 把 Agent B 當成工具呼叫時，Agent B 的答案會回傳給 Agent A，Agent A 會把 Agent B 的答案整理後作為使用者的回覆。
• Sub-agent：當 Agent A 把 Agent B 作為 sub-agent 呼叫時，回覆 user 的責任已經交給 Agent B 了，Agent A 不會知道 Agent B 回覆及做了些什麼。