[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
官方文件
建立 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}'.")