前言

還記的我們第一支AI Agent嗎? 我們今天將進階一點，製作出 天氣 agent~

天氣 agent

如果忘記環境架設如何使用，可以前往我之前 [Day 5] 第一個 Python Agent，今天就不多做撰寫了~

• .env

# OpenWeatherMap API Key
OPENWEATHER_API_KEY="your_actual_api_key_here"

# Google API Key
GOOGLE_API_KEY="Your-Google-API-Key-Here"
GOOGLE_GENAI_USE_VERTEXAI="FALSE"

先去 OpenWeatherMap拿取API KEY，並貼上
也可在官方網址內查看 JSON 範例

• 新增 weather_agent.py 與 weather_prompt.py
  

• weather_agent

from google.adk.tools import ToolContext
import datetime, os
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
from urllib.parse import quote
import aiohttp
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from . import weather_prompt as prompt

load_dotenv()

# 台灣城市地名映射
TAIWAN_CITY_MAP = {
    "taipei": "台北",
    "taichung": "台中", 
    "kaohsiung": "高雄",
    "tainan": "台南",
    "taoyuan": "桃園",
    "hsinchu": "新竹",
    "keelung": "基隆",
    "hualien": "花蓮",
    "taitung": "台東",
    "chiayi": "嘉義",
    "changhua": "彰化",
    "yunlin": "雲林",
    "nantou": "南投",
    "pingtung": "屏東",
    "yilan": "宜蘭",
    "miaoli": "苗栗",
    "xianeibu": "台北",
    "jinping": "台中",
    "new taipei": "新北",
    "xinbei": "新北"
}

# 城市時區映射
TIMEZONE_MAP = {
    "new york": "America/New_York",
    "nyc": "America/New_York", 
    "london": "Europe/London",
    "tokyo": "Asia/Tokyo",
    "taipei": "Asia/Taipei",
    "taiwan": "Asia/Taipei",
    "taichung": "Asia/Taipei",
    "beijing": "Asia/Shanghai",
    "shanghai": "Asia/Shanghai",
    "china": "Asia/Shanghai",
    "hong kong": "Asia/Hong_Kong"
}

def get_localized_city_name(english_name: str, user_input: str = "") -> str:
    """獲取本地化的城市名稱"""
    # 先檢查是否為台灣城市
    english_lower = english_name.lower().strip()
    
    # 直接映射
    if english_lower in TAIWAN_CITY_MAP:
        return TAIWAN_CITY_MAP[english_lower]
    
    # 如果用戶輸入包含中文，優先使用用戶輸入
    if user_input and any('\u4e00' <= char <= '\u9fff' for char in user_input):
        # 提取中文部分
        chinese_part = ''.join([char for char in user_input if '\u4e00' <= char <= '\u9fff'])
        if chinese_part:
            return f"{chinese_part} ({english_name})"
    
    # 檢查是否包含台灣關鍵字，推測為台灣城市
    taiwan_keywords = ["taiwan", "tw", "roc"]
    if any(keyword in english_lower for keyword in taiwan_keywords):
        return f"{english_name} (台灣)"
    
    return english_name

async def get_weather(city: str, user_query: str = "", tool_context: ToolContext = None) -> str:
    key = os.getenv("WEATHER_API_KEY") or os.getenv("OPENWEATHER_API_KEY")
    if not key:
        return "Missing API Key"

    try:
        timeout = aiohttp.ClientTimeout(total=10)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            # 嘗試直接查詢
            url = f"https://api.openweathermap.org/data/2.5/weather?q={quote(city)}&appid={key}&units=metric&lang=zh_tw"
            async with session.get(url) as resp:
                if resp.status == 404:  # 城市找不到，用地理編碼
                    geo_url = f"https://api.openweathermap.org/geo/1.0/direct?q={quote(city)}&limit=1&appid={key}"
                    async with session.get(geo_url) as geo_resp:
                        geo_data = await geo_resp.json()
                        if not geo_data:
                            return f"找不到城市: {city}"
                        lat, lon = geo_data[0]["lat"], geo_data[0]["lon"]
                        url = f"https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={key}&units=metric&lang=zh_tw"
                        async with session.get(url) as final_resp:
                            data = await final_resp.json()
                elif resp.status == 200:
                    data = await resp.json()
                else:
                    return f"API 錯誤: {resp.status}"
    except Exception as e:
        return f"網路錯誤: {str(e)}"

    # 提取關鍵資料
    w = data["weather"][0]
    m = data["main"]
    wind = data["wind"]
    
    # 獲取本地化地名
    api_city_name = data.get('name', city)
    localized_city = get_localized_city_name(api_city_name, city)
    
    # 智能天氣分析
    conditions = []
    rain = data.get("rain", {}).get("1h", 0)
    if rain > 0 or w["main"] in {"Rain", "Drizzle", "Thunderstorm"}:
        conditions.append("大雨" if rain > 5 else "中雨" if rain > 1 else "小雨")
    elif w["main"] == "Clear":
        conditions.append("晴朗")
    elif w["main"] == "Clouds":
        clouds = data["clouds"]["all"]
        conditions.append("陰天" if clouds > 80 else "多雲" if clouds > 50 else "少雲")
    
    # 溫度和濕度狀況
    temp = m["temp"]
    if temp > 35: conditions.append("炎熱")
    elif temp > 25: conditions.append("溫暖")
    elif temp < 0: conditions.append("嚴寒")
    elif temp < 10: conditions.append("寒冷")
    
    if wind["speed"] >= 15: conditions.append("強風")
    elif wind["speed"] >= 10: conditions.append("多風")
    
    if m["humidity"] > 80: conditions.append("潮濕")
    elif m["humidity"] < 30: conditions.append("乾燥")

    # 檢查特定查詢
    queries = {
        "溫度": f"{temp:.1f}°C", "體感": f"{m['feels_like']:.1f}°C",
        "濕度": f"{m['humidity']}%", "氣壓": f"{m['pressure']} hPa",
        "風速": f"{wind['speed']:.1f} m/s", "雲量": f"{data['clouds']['all']}%"
    }
    
    for keyword, value in queries.items():
        if keyword in user_query:
            result = f"{localized_city} 的 {keyword} 是 {value}"
            print("#" * 50)
            print("Weather query result:")
            print(f"city: {city}")
            print(f"localized_city: {localized_city}")
            print(f"query: {user_query}")
            print(f"result: {result}")
            print("#" * 50)
            return result

    # 格式化摘要
    condition_text = "、".join(conditions) if conditions else w["description"]
    wind_dir = ""
    if wind.get("deg"):
        dirs = ["北", "東北", "東", "東南", "南", "西南", "西", "西北"]
        wind_dir = f" ({dirs[round(wind['deg'] / 45) % 8]}風)"
    
    summary = (
        f"地點: {localized_city}\n"
        f"天氣: {condition_text}\n"
        f"溫度: {temp:.1f}°C\n"
        f"體感: {m['feels_like']:.1f}°C\n"
        f"濕度: {m['humidity']}%\n"
        f"風速: {wind['speed']:.1f}m/s{wind_dir}\n"
        f"氣壓: {m['pressure']} hPa\n"
        f"雲量: {data['clouds']['all']}%\n"
        f"天氣狀況: {w['main']} ({w['description']})\n"
        f"查詢時間: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    )
    
    if rain > 0:
        summary += f"\n降雨: {rain:.1f} mm/h"
    
    print("#" * 50)
    print("Weather result:")
    print(f"original_city: {api_city_name}")
    print(f"localized_city: {localized_city}")
    print(f"user_input: {city}")
    print(f"summary: {summary}")
    print("#" * 50)
    
    return summary

# --------------------------------
# 天氣 Agent 輔助函數
# --------------------------------

async def weather_sync(city: str, user_query: str = "", tool_context: ToolContext = None) -> str:
    """同步版本的天氣查詢

    Args:
        city(str): 要查詢天氣的城市名稱.
        user_query(str): 特定的天氣查詢 (溫度、濕度等).
        tool_context(ToolContext): The function context.

    Returns:
        str: 天氣資訊.
    """
    return await get_weather(city, user_query, tool_context)

async def get_current_time(city: str, tool_context: ToolContext = None) -> str:
    """返回指定城市的當前時間

    Args:
        city(str): 要查詢時間的城市名稱.
        tool_context(ToolContext): The function context.

    Returns:
        str: 時間資訊.
    """
    tz_identifier = TIMEZONE_MAP.get(city.lower().strip())
    if not tz_identifier:
        cities = list(TIMEZONE_MAP.keys())
        suggestions = [c for c in cities if city.lower() in c][:3]
        hint = f" 可用城市：{', '.join(suggestions)}" if suggestions else f" 支援 {len(cities)} 個城市"
        return f"不支援 {city} 的時區{hint}"

    try:
        now = datetime.datetime.now(ZoneInfo(tz_identifier))
        weekdays = ["週一", "週二", "週三", "週四", "週五", "週六", "週日"]
        formatted = now.strftime("%Y年%m月%d日 %H:%M:%S")
        result = f"{city} 目前時間：{formatted} {weekdays[now.weekday()]} ({now.strftime('%Z')})"
        
        print("#" * 50)
        print("Time query result:")
        print(f"city: {city}")
        print(f"result: {result}")
        print("#" * 50)
        
        return result
    except Exception as e:
        return f"時間錯誤: {str(e)}"

weather_agent = Agent(
    model="gemini-2.0-flash",
    name="weather_agent",
    description=prompt.WEATHER_AGENT_DESCRIPTION,
    instruction=prompt.WEATHER_AGENT_INSTRUCTION,
    tools=[
        FunctionTool(func=weather_sync),
        FunctionTool(func=get_current_time)
    ],
)

• weather_prompt.py

WEATHER_AGENT_DESCRIPTION = """
這是一個天氣代理，專門處理天氣和時間相關的查詢請求，包括：
1. 全球城市天氣查詢
2. 特定天氣指標查詢（溫度、濕度、風速等）
3. 主要城市時間查詢
4. 智能天氣狀況分析
"""

WEATHER_AGENT_INSTRUCTION = """
## 角色定義
你是一個專業的天氣和時間查詢專家，提供準確的氣象和時間資訊。

## 核心職責
1. 識別用戶的天氣或時間查詢需求
2. 正確調用相應的查詢工具
3. 提供結構化的天氣和時間資訊

## 工具使用條件
- 當用戶詢問天氣時，使用 weather_sync 函數
- 當用戶詢問特定指標（溫度、濕度等）時，使用對應的查詢參數
- 當用戶詢問時間時，使用 get_current_time 函數

## 回應格式
天氣查詢回應格式：
地點: [城市名稱]
天氣: [天氣狀況描述]
溫度: [溫度]°C
體感: [體感溫度]°C
濕度: [濕度]%
風速: [風速]m/s ([風向])
氣壓: [氣壓] hPa
雲量: [雲量]%
天氣狀況: [主要天氣] ([詳細描述])
查詢時間: [查詢時間戳]


時間查詢回應格式：
[城市] 目前時間：[完整時間] [星期] ([時區])


## 限制條件
- 依賴 OpenWeatherMap API 提供準確數據
- 時間查詢限制於支援的時區城市
- 如果 API 調用失敗，提供清楚的錯誤說明
- 對於不支援的城市，提供建議替代方案
"""

root agent

• root_prompt.py
  大家可以參考之前prompt寫法去撰寫自己的prompt，以下是部分參考

## 回應格式要求
天氣查詢標準格式：
地點: [使用者所說地點(中文，非亂碼)]
天氣: [查詢到的天氣資訊]
溫度: [查詢到的溫度資訊]
體感: [查詢到的體感溫度資訊]
濕度: [查詢到的濕度資訊]
風速: [查詢到的風速資訊]
氣壓: [查詢到的氣壓資訊]
雲量: [查詢到的雲量資訊]
天氣狀況: [查詢到的天氣狀況資訊]
降雨: [查詢到的降雨資訊]
查詢時間: [使用當前時間]

## 核心原則
1. 自動識別查詢類型並選擇合適的子代理
2. 確保回應格式的一致性和準確性
3. 對於錯誤情況提供清楚的說明
4. 保持友善和專業的服務態度

## 限制條件
- 不直接處理具體功能，透過子代理完成
- 確保所有回應都經過適當的格式化
- 遇到不支援的請求時，禮貌地說明限制

• root_agent.py

from google.adk.agents import Agent
from . import prompt
from dotenv import load_dotenv
from .tools.weather_agent import weather_agent

load_dotenv()

root_agent = Agent(
    name="root_agent",
    model="gemini-2.0-flash-exp",
    description=prompt.ROOT_AGENT_DESCRIPTION,
    instruction=prompt.ROOT_AGENT_INSTRUCTION,
    sub_agents=[
        weather_agent
    ]
)

執行

一樣
回去上一層資料夾 cd ..
執行 adk web

前去網址，選擇自己撰寫的agent 資料夾，就可以成功使用它了!

你就可以成功收到天氣資訊了~

結尾

今天天氣agent是不是很有趣，以之前 hello agent更難一點，成功做到這裡的你也是很厲害~ 但也不能就此鬆懈!
我們下一篇見（＾∀＾●）ﾉｼ