iT邦幫忙

2026 iThome 鐵人賽

DAY 14
0
AI Engineering

從 Prompt 到自主決策:用 Python × Agentic Workflow 實作生活助理系列 第 14 篇

第一個真實資料來源:串接中央氣象署天氣 API

  • 分享至 

  • xImage
  •  

昨天我們把 HTTP 的基礎和一個帶重試的請求函式準備好了。今天要拿它去做真的事:查台灣的天氣。

為什麼選天氣當第一個 API?

  1. 資料免費、申請簡單(中央氣象署開放資料平臺)
  2. 跟「生活助理」的情境直接相關——安排行程一定要看天氣
  3. 回傳的 JSON 結構夠複雜,剛好可以練習怎麼從深層巢狀資料裡挖東西

一、申請 API Key

到 中央氣象署開放資料平臺:https://opendata.cwa.gov.tw

  1. 註冊帳號(免費)
  2. 登入後點右上角「會員資訊」→「API 授權碼」
  3. 取得一串 CWA-XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX 格式的授權碼

拿到之後,照 Day 12 的做法放進 .env:

CWA_API_KEY=CWA-XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX

不要寫進程式碼裡。


二、先用瀏覽器看看資料長什麼樣

昨天說過,寫程式之前先看清楚資料結構。我們要用的是 F-C0032-001:一般天氣預報-今明 36 小時天氣預報。

在瀏覽器打開(把 YOUR_KEY 換成你的授權碼):

https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-C0032-001?Authorization=YOUR_KEY&locationName=臺北市

回傳的 JSON 結構大概長這樣(我把不重要的部分省略了):

{
  "success": "true",
  "records": {
    "location": [
      {
        "locationName": "臺北市",
        "weatherElement": [
          {
            "elementName": "Wx",
            "time": [
              {
                "startTime": "2026-09-28 12:00:00",
                "endTime": "2026-09-28 18:00:00",
                "parameter": {"parameterName": "多雲", "parameterValue": "4"}
              }
            ]
          },
          {"elementName": "PoP", "time": [...]},
          {"elementName": "MinT", "time": [...]},
          {"elementName": "MaxT", "time": [...]},
          {"elementName": "CI", "time": [...]}
        ]
      }
    ]
  }
}

五個天氣要素的意思:

elementName 中文 單位
Wx 天氣現象 文字(晴、多雲、陰時多雲短暫陣雨…)
PoP 降雨機率 %
MinT 最低溫度 °C
MaxT 最高溫度 °C
CI 舒適度 文字(舒適、悶熱…)

而 time 陣列有三個時段,各 12 小時,涵蓋未來 36 小時。

這個結構有點難用——資料是「以要素分組」,但我們想要的是「以時段分組」。這就是今天要處理的主要工作。

⚠️ 一個很容易卡住的地方:城市名稱必須用正體的「臺」,不是「台」。臺北市 可以,台北市 會查不到。而且要包含「市」或「縣」。


三、第一版:把資料抓下來

建立 tools/weather.py:

"""tools/weather.py — 中央氣象署天氣查詢。"""

import requests

CWA_BASE = "https://opendata.cwa.gov.tw/api/v1/rest/datastore"
FORECAST_36H = "F-C0032-001"


def fetch_raw_forecast(city, api_key):
    """向氣象署取得原始的 36 小時預報資料。"""
    response = requests.get(
        f"{CWA_BASE}/{FORECAST_36H}",
        params={
            "Authorization": api_key,
            "locationName": city,
            "format": "JSON",
        },
        timeout=10,
    )
    response.raise_for_status()
    return response.json()


# 試跑
import config
data = fetch_raw_forecast("臺北市", config.CWA_API_KEY)
print(data["records"]["location"][0]["locationName"])

能跑就代表串通了。接下來是把資料整理成好用的形狀。


四、第二版:把巢狀資料攤平

我們想要的結果是這樣:

{
    "city": "臺北市",
    "periods": [
        {
            "start": "2026-09-28 12:00:00",
            "end": "2026-09-28 18:00:00",
            "weather": "多雲",
            "rain_probability": 20,
            "min_temp": 26,
            "max_temp": 31,
            "comfort": "悶熱",
        },
        ...
    ],
}

轉換的程式碼:

# 氣象署的欄位名 → 我們自己的欄位名
ELEMENT_MAP = {
    "Wx": "weather",
    "PoP": "rain_probability",
    "MinT": "min_temp",
    "MaxT": "max_temp",
    "CI": "comfort",
}

NUMERIC_FIELDS = {"rain_probability", "min_temp", "max_temp"}


def parse_forecast(raw):
    """把氣象署的巢狀 JSON 攤平成好用的結構。"""
    locations = raw.get("records", {}).get("location", [])
    if not locations:
        raise ValueError("氣象署沒有回傳任何地點資料,請確認城市名稱(要用「臺」)")

    location = locations[0]

    # 用時段的 startTime 當 key,把各要素的值收集起來
    periods = {}

    for element in location.get("weatherElement", []):
        field = ELEMENT_MAP.get(element["elementName"])
        if field is None:
            continue    # 不認識的要素就跳過

        for slot in element.get("time", []):
            key = slot["startTime"]
            period = periods.setdefault(key, {
                "start": slot["startTime"],
                "end": slot["endTime"],
            })

            value = slot["parameter"]["parameterName"]
            if field in NUMERIC_FIELDS:
                try:
                    value = int(value)
                except ValueError:
                    value = None    # 氣象署偶爾會給空字串
            period[field] = value

    return {
        "city": location["locationName"],
        # 依開始時間排序(ISO 風格的字串排序剛好等於時間排序)
        "periods": [periods[k] for k in sorted(periods)],
    }

這段程式碼把 Day 5、6、8、10 學的東西全用上了:

  • 迴圈與巢狀迴圈:走訪要素、走訪時段
  • Dictionary 的 .get() 與 .setdefault():安全地取值、不存在就建立
  • 例外處理:int() 轉換失敗時不要讓整件事崩掉
  • sorted():把 dict 的 key 排序

setdefault() 這個方法很好用:「如果這個 key 存在就拿出來,不存在就用預設值建立再拿出來」。一行取代了三行 if-else。


五、第三版:加上快取

天氣預報不會每秒變。如果使用者在十分鐘內問三次天氣,我們打三次 API 是浪費——浪費時間(每次要等 1~2 秒)、也浪費額度。

用 Day 9、10 學的檔案與 JSON 來做快取:

import json
import time
from pathlib import Path

CACHE_DIR = Path("data/cache")
CACHE_TTL = 30 * 60      # 30 分鐘


def _cache_path(city):
    return CACHE_DIR / f"weather_{city}.json"


def read_cache(city):
    """讀取快取,過期或不存在回傳 None。"""
    path = _cache_path(city)
    if not path.exists():
        return None
    try:
        with open(path, "r", encoding="utf-8") as f:
            cached = json.load(f)
    except json.JSONDecodeError:
        return None

    if time.time() - cached.get("cached_at", 0) > CACHE_TTL:
        return None      # 過期了

    return cached["data"]


def write_cache(city, data):
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    with open(_cache_path(city), "w", encoding="utf-8") as f:
        json.dump(
            {"cached_at": time.time(), "data": data},
            f, ensure_ascii=False, indent=2,
        )

快取的核心概念就三件事:

  1. 存的時候記下時間(cached_at)
  2. 讀的時候檢查有沒有過期(現在時間 − 存的時間 > TTL)
  3. 過期就當作沒有,重新去抓

六、完整版

把三個部分組合起來,就是可以直接用的工具:

"""tools/weather.py — 中央氣象署天氣查詢。"""

import json
import time
from pathlib import Path

import requests

CWA_BASE = "https://opendata.cwa.gov.tw/api/v1/rest/datastore"
FORECAST_36H = "F-C0032-001"

CACHE_DIR = Path("data/cache")
CACHE_TTL = 30 * 60

ELEMENT_MAP = {
    "Wx": "weather",
    "PoP": "rain_probability",
    "MinT": "min_temp",
    "MaxT": "max_temp",
    "CI": "comfort",
}
NUMERIC_FIELDS = {"rain_probability", "min_temp", "max_temp"}


class WeatherError(Exception):
    """天氣查詢失敗。"""


# --- 快取 ---(略,同上)

# --- 主要入口 ---

def get_weather(city, api_key, use_cache=True):
    """查詢城市的 36 小時天氣預報。

    Args:
        city: 城市名稱,例如「臺北市」(注意是「臺」不是「台」)
        api_key: 中央氣象署授權碼
        use_cache: 是否使用快取

    Returns:
        dict: {"city": ..., "periods": [...]}

    Raises:
        WeatherError: 查詢失敗
    """
    city = normalize_city(city)

    if use_cache:
        cached = read_cache(city)
        if cached is not None:
            return cached

    try:
        response = requests.get(
            f"{CWA_BASE}/{FORECAST_36H}",
            params={"Authorization": api_key, "locationName": city, "format": "JSON"},
            timeout=10,
        )
        response.raise_for_status()
        raw = response.json()
    except requests.Timeout as e:
        raise WeatherError("查詢天氣逾時,請稍後再試") from e
    except requests.ConnectionError as e:
        raise WeatherError("無法連線到氣象署,請檢查網路") from e
    except requests.HTTPError as e:
        code = e.response.status_code
        if code == 401:
            raise WeatherError("氣象署授權碼無效,請檢查 CWA_API_KEY") from e
        raise WeatherError(f"氣象署回應錯誤(HTTP {code})") from e
    except ValueError as e:
        raise WeatherError("氣象署回應格式異常") from e

    if raw.get("success") not in ("true", True):
        raise WeatherError("氣象署回報查詢失敗")

    try:
        data = parse_forecast(raw)
    except (KeyError, ValueError) as e:
        raise WeatherError(f"無法解析天氣資料:{e}") from e

    write_cache(city, data)
    return data


def normalize_city(city):
    """把常見的城市寫法修正成氣象署認得的格式。"""
    city = city.strip()
    city = city.replace("台", "臺")
    if not city.endswith(("市", "縣")):
        # 簡單的補字規則
        counties = {"新竹", "嘉義", "苗栗", "彰化", "南投", "雲林",
                    "屏東", "宜蘭", "花蓮", "臺東", "澎湖", "金門", "連江"}
        city += "縣" if city in counties else "市"
    return city


def summarize(data):
    """把天氣資料轉成一段給人(或給 AI)讀的文字。"""
    lines = [f"【{data['city']}】未來 36 小時天氣"]
    for p in data["periods"]:
        start = p["start"][5:16]      # 只保留「09-28 12:00」
        end = p["end"][11:16]
        lines.append(
            f"{start}~{end} {p['weather']},"
            f"{p['min_temp']}–{p['max_temp']}°C,"
            f"降雨機率 {p['rain_probability']}%,{p['comfort']}"
        )
    return "\n".join(lines)

試跑:

import config
from tools.weather import get_weather, summarize, WeatherError

try:
    data = get_weather("台北", config.CWA_API_KEY)     # 故意用「台」
    print(summarize(data))
except WeatherError as e:
    print(f"查詢失敗:{e}")

輸出:

【臺北市】未來 36 小時天氣
09-28 12:00~18:00 多雲,26–31°C,降雨機率 20%,悶熱
09-28 18:00~06:00 多雲,25–28°C,降雨機率 10%,舒適
09-29 06:00~18:00 短暫陣雨,25–30°C,降雨機率 60%,悶熱

七、這個設計的三個重點

今天這支程式看起來只是「查天氣」,但它的結構是刻意設計的,值得回頭看一下:

1. 自訂例外 WeatherError

class WeatherError(Exception):
    """天氣查詢失敗。"""

我們把所有可能的失敗(逾時、斷線、授權錯誤、格式異常)統一包裝成一種例外,而且訊息是人看得懂的中文。

這樣呼叫端只要寫:

except WeatherError as e:
    print(f"查詢失敗:{e}")

就能處理所有情況,不需要知道底層用了 requests、也不需要知道氣象署的 API 長什麼樣。這就是 Day 7 講的「封裝」。

而且這些錯誤訊息,等到 Day 21 之後可以直接丟回給 AI:「工具執行失敗,原因是『氣象署授權碼無效』」——AI 就知道不該再重試這個工具。

2. normalize_city() 幫使用者擦屁股

使用者不會知道要打「臺」不是「台」。AI 也常常會給出「台北」。與其讓查詢失敗,不如在工具裡面幫忙修正。

好的工具應該對輸入寬容、對輸出嚴格。

3. summarize() 分離資料與呈現

get_weather() 回傳結構化資料(給程式用),summarize() 把它轉成文字(給人或 AI 用)。兩件事分開,以後要改顯示格式不用動到查詢邏輯。

這在 Agent 裡特別重要——同一份資料,可能要:

  • 塞進 prompt 給 AI 分析(要精簡)
  • 顯示給使用者看(要好讀)
  • 存進 log(要完整)

三種格式,一份資料。


八、這一步在整個專案裡的位置

我們的生活助理,現在有了第一個真實的資訊來源。

回想一下 Day 1 的情境:

使用者:「幫我安排明天的行程」
AI 取得天氣、行程等資訊 → 分析 → 給出建議

今天完成的是第一個箭頭的一部分。接下來:

  • Day 15:把這個函式包成「工具」的標準形式,讓它可以被 Agent 呼叫
  • Day 18:讓 AI 能讀懂這份天氣資料
  • Day 21–22:讓 AI 自己決定什麼時候該去查天氣

注意最後這一點——現在是我們決定什麼時候呼叫 get_weather()。到了 Day 21,會變成AI 自己決定。那就是「Workflow」跟「Agent」的差別。


小結

  • 中央氣象署開放資料平臺免費,F-C0032-001 提供 36 小時預報
  • 城市名稱要用「臺」不是「台」,而且要帶「市」或「縣」
  • 氣象署的資料是「以要素分組」,要自己攤平成「以時段分組」
  • 用檔案 + 時間戳記做快取,省時間也省額度
  • 自訂例外(WeatherError)把底層細節包起來,錯誤訊息要人看得懂
  • 資料(get_weather)與呈現(summarize)分開

明天我們要把今天寫的東西,正式升格成 Agent 可以使用的工具(Tool)。


上一篇
讓程式連上外面的世界:HTTP 與 API 基礎
下一篇
什麼才算一個「工具」?Tool 的介面設計
系列文
從 Prompt 到自主決策:用 Python × Agentic Workflow 實作生活助理 共 18 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言