系列:30 天用 Google AI 打造臺灣防災速報 App(Day 10/30)
前兩天的速報與分級,都是我們把資料準備好再交給模型。問答功能不一樣:使用者的問法有很多種(「屏東現在安全嗎」「最近有比較大的地震嗎」),答案卻在我們的資料裡,模型本身並不知道現在有哪些示警。
今天用 **Function Calling(函式呼叫)**來處理這件事。做法是先寫好幾個查詢用的函式,把它們交給模型當工具。模型讀了使用者的問題之後,自己決定要呼叫哪一個工具、帶什麼參數;程式執行完,把結果交回給模型,模型再寫成回答。
工具就是一般的 Python 函式。第一個查示警,第二個查地震報告,資料來源是前幾天寫好的下載與整理程式。以下節錄核心的部分,省略了匯入套件與建立 client 的幾行:
MAX_RETURN = 8 # 一次最多回傳幾則,避免把幾百則示警全部交給模型
# 執行時下載一次,之後的查詢都用這一份
ACTIVE = [a for a in pipeline(fetch_alerts()) if a.status == "active"]
def get_active_alerts(county: str | None = None, group: str | None = None,
category: str | None = None) -> dict:
"""查詢目前生效中的防災示警。
Args:
county: 縣市全名,例如「屏東縣」「臺北市」;不給則查全臺。
group: 六大類之一:氣象、地震海嘯、水利、火災事故、交通、民生;不給則查全部。
category: 發布單位的原始分類,比六大類更細,例如「停水」「水庫放流」「道路封閉」
「高溫」「土石流及大規模崩塌」「停班停課」;使用者問的是特定種類時用這個。
Returns:
total 是符合條件的總筆數,alerts 是其中最新的幾則。
"""
hits = ACTIVE
if group:
hits = [a for a in hits if a.group == group]
if category:
hits = [a for a in hits if a.raw_category == category]
if county:
names = {county, county.replace("臺", "台"), county.replace("台", "臺")}
hits = [a for a in hits if any(n in a.title + a.summary for n in names)]
return {
"total": len(hits),
"alerts": [{"group": a.group, "category": a.raw_category, "sender": a.sender,
"updated": a.updated.isoformat(), "summary": a.summary[:120]}
for a in hits[:MAX_RETURN]],
}
def get_recent_quakes(days: int = 7, min_magnitude: float = 0) -> dict:
"""查詢中央氣象署近期的顯著有感地震報告。
Args:
days: 往回查幾天,預設 7 天。
min_magnitude: 最小芮氏規模,預設 0(全部)。
Returns:
total 是符合條件的筆數,quakes 是每一筆的時間、位置、規模、深度與最大震度。
"""
since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S")
data = cwa_get("E-A0015-001", timeFrom=since)
quakes = [flatten(eq) for eq in data["records"]["Earthquake"]]
quakes = [q for q in quakes if q["magnitude"] >= min_magnitude]
return {"total": len(quakes), "quakes": quakes}
函式開頭用三個引號包起來的那段說明文字叫 docstring。google-genai SDK 可以直接把 Python 函式當工具傳入,它會把函式名稱、參數型別和 docstring 一起交給模型。模型只看得到這些,看不到函式裡面的程式,所以 docstring 要把每個參數能填什麼寫清楚。
SYSTEM = """你是臺灣防災速報 App 的問答助理。
規則:
- 除了地震預測的問題之外,回答示警或地震的問題前,一定要先呼叫工具查詢,不得憑記憶回答。
- 只能根據工具回傳的內容回答,不得補充工具沒有回傳的資訊或建議。
- get_active_alerts 回傳 total 為 0 時,回覆「目前查不到相關的生效示警」,不得說成當地沒有示警(查詢方式可能有遺漏),也不得自行編寫示警內容。
- get_recent_quakes 回傳 total 為 0 時,說明查詢的天數與規模門檻,回覆查不到符合條件的地震。
- total 大於實際列出的則數時,要先說明總共有幾則,再說明以下只列出最新的幾則。
- 使用者問到地震預測(例如「明天會不會地震」)時,不呼叫工具,回覆:目前的科學無法預測地震發生的時間與地點,建議平時做好防災準備。
- 回答簡短,用一般民眾看得懂的話。"""
resp = client.models.generate_content(
model="gemini-3.1-flash-lite",
contents="屏東現在有什麼警報?",
config={"tools": [get_active_alerts, get_recent_quakes],
"system_instruction": SYSTEM},
)
print(resp.text)
system_instruction 是這次請求交給模型的固定規則;這支程式每次呼叫模型時,都會在 config 裡帶上它。tools 傳入函式之後,SDK 預設會自動處理整個來回:模型決定要查詢,產生 get_active_alerts(county="屏東縣") 這個呼叫;SDK 替我們執行函式,把結果交回給模型;模型再寫成回答。想知道模型實際呼叫了什麼,可以讀 resp.automatic_function_calling_history。
上面的工具參數與規則是實測之後的定稿,其中 category 參數和一半的規則,是實測之後才加的或改的。下面照實記錄過程。
9/20 用五個問題測試。模型每一題都選對了工具,參數也合理,但回答有三項缺失。
第一,沒有說總數。 問「屏東現在有什麼警報?」,模型呼叫 get_active_alerts(county='屏東縣'),工具回傳 total=32,但只附上最新的 8 則。模型的回答只列了兩則,沒有提到總共有 32 則,使用者會以為只有兩則。修正方式是加一條規則:total 大於列出的則數時,要先說明總數。
第二,工具的參數太粗,模型還自己加了建議。 問「全臺的停水資訊」,當時工具只有六大類可以選,模型只能用 group='民生' 去查,查到的 635 則裡混了停班停課和其他公告。回答的最後,模型還加了一句「建議您可以前往自來水公司官網的停水公告頁面查詢」。這句話不在工具回傳的內容裡,是模型自己寫的。修正方式有兩個:工具加上 category 參數,可以直接查「停水」;規則加上「不得補充工具沒有回傳的資訊或建議」。
第三,回答裡出現已經過期的示警。 問「連江縣現在有什麼警報?」,模型回答有強風示警:「16 日晨至 16 日晚上,連江縣局部地區有平均風 6 級以上」。測試當天是 9 月 20 日,這則示警四天前就結束了。模型只是照工具回傳的內容回答,問題出在我們的資料。
回頭檢查 NCDR 的原始資料,才發現每一則示警其實都有 expires(到期時間)欄位。Day 2 寫下載程式時,我只挑了標題、摘要、分類這幾個欄位,沒有把它留下來。Day 5 判斷「生效中」的方式,也只是排除摘要裡有「已結案」的示警。
補上的程式分兩處。下載程式多保留一個欄位:
"expires": e.get("expires", ""), # 到期時間,格式如「2026/9/20 下午 10:19:00」
整理程式把這個中文格式的時間轉成可以比較的時間,再多一種「已過期」的狀態。程式裡的 TW 代表臺灣所在的時區(UTC+8),時間帶上時區才能和現在的時間正確比較:
TW = timezone(timedelta(hours=8))
def parse_expires(text: str) -> datetime | None:
try:
date, ampm, clock = text.split()
y, m, d = (int(x) for x in date.split("/"))
hh, mm, ss = (int(x) for x in clock.split(":"))
if ampm == "下午" and hh < 12:
hh += 12
if ampm == "上午" and hh == 12:
hh = 0
return datetime(y, m, d, hh, mm, ss, tzinfo=TW)
except (ValueError, AttributeError):
return None
# normalize() 裡決定狀態的部分
expires = parse_expires(raw.get("expires", ""))
if "已結案" in summary:
status = "closed"
elif expires and expires < datetime.now(TW):
status = "expired"
else: # 沒有到期時間或解析失敗時,仍視為生效中,避免漏掉示警
status = "active"
重新執行的結果(9/20 19:33):
原始 924 筆 → 去除完全重複後 922 筆
狀態:已過期 876 筆、已結案 16 筆、生效中 30 筆
生效中的六大類分布:民生 20、水利 4、氣象 3、火災事故 3
922 則裡有 876 則已經過期,真正還在生效的只有 30 則。過期的示警以停水(607 則)和水庫放流(168 則)最多,NCDR 會把這些示警保留好幾天。也就是說,Day 5 的「生效中」判定沒有檢查到期時間,筆數被高估了。當時存下來的資料沒有保留 expires 欄位,無法回頭確認高估了多少,這裡照實更正。
如果沒有做問答功能,這個問題可能要到推播上線、使用者收到過期通知時才會發現。
補上過期過濾、category 參數和兩條規則之後,同樣五個問題再執行一次。這一輪又找到一處要改:連江縣查到 0 則時,模型回答「目前連江縣沒有發布任何防災示警」。工具只能確定「標題和摘要裡找不到連江縣」,不能確定當地真的沒有示警,這樣回答說得太滿。所以查無資料的規則改成現在的寫法,之後又把「查無資料」的規則依工具分開寫,最後一次執行的結果如下:
| 使用者的問題 | 模型的呼叫 | 工具回傳 | 回答摘要 |
|---|---|---|---|
| 屏東現在有什麼警報? | get_active_alerts(county='屏東縣') |
1 則 | 鹽埔漁港有不明油污的海洋污染通報 |
| 最近有比較大的地震嗎 | get_recent_quakes(days=7, min_magnitude=5) |
0 則 | 過去 7 天內,沒有規模 5 以上的顯著有感地震 |
| 全臺的停水資訊 | get_active_alerts(category='停水') |
19 則 | 先說明共 19 則,再列出最新 8 則 |
| 連江縣現在有什麼警報? | get_active_alerts(county='連江縣') |
0 則 | 目前查不到相關的生效示警 |
| 明天會不會地震? | 沒有呼叫工具 | 照規則回覆「目前的科學無法預測地震」 |
第一輪的三項缺失都解決了,查無資料時的說法也不再過度肯定。
min_magnitude=5,查到 0 筆,回答查不到規模 5 以上的地震;有一次沒有帶門檻,列出 9/14 的兩起地震(規模 4.9 與 4.7,最大震度 4 級,和 Day 6 查到的報告一致)。兩種回答都有依據,但同一個問題得到不同的答案。比較好的做法是把「比較大」的定義寫進 docstring,之後會處理。(本篇實測於 2026-09-20,模型 gemini-3.1-flash-lite,溫度維持預設值。示警資料隨時在變,重新執行時筆數會不同。)
expires 欄位,先前沒有保留。補上之後,922 則示警裡真正生效中的只有 30 則。system_instruction:查不到就說查不到(不說成當地沒有示警)、不補充工具沒回傳的內容、地震預測的問題用固定說法回覆。明天 Day 11 開始建立防災知識庫:RAG 的第一步,整理語料。