3d620dc4c7
getWeather returned a verbose multi-section English forecast that the 3B re-synthesised into long, CJK/°F-leaking answers. Hand it a ready-to-speak Korean one-liner (지금 <곳> 날씨는 <상태>, 기온 N도(체감 M도)입니다) and drop the hourly/7-day firehose from the default voice reply.
485 lines
21 KiB
Python
485 lines
21 KiB
Python
"""Weather tool implementation using Open-Meteo API (free, no API key required)."""
|
|
|
|
import requests
|
|
from typing import Dict, Any, Optional
|
|
from ...debug import debug_log
|
|
from ...utils.location import get_location_info
|
|
from ..base import Tool, ToolContext
|
|
from ..types import ToolExecutionResult
|
|
|
|
|
|
# Sentinel strings an LLM extractor may emit to mean "no place mentioned".
|
|
# Matched case-insensitively as whole-value comparisons, not substrings.
|
|
_NO_PLACE_SENTINELS = frozenset({
|
|
"none", "null", "no", "no place", "no location",
|
|
"n/a", "na", "unknown", "unspecified",
|
|
})
|
|
|
|
|
|
def _extract_place_from_user_text(text: str, cfg) -> Optional[str]:
|
|
"""Ask a small LLM to pull a place name out of the user's utterance.
|
|
|
|
Used as a last-ditch fallback when the tool-calling LLM didn't fill the
|
|
``location`` argument AND GeoIP auto-detect is unavailable. Small chat
|
|
models (e.g. gemma4:e2b) regularly fail to propagate a city into tool
|
|
args even when the user literally just said one — pulling the place
|
|
straight from the user's text sidesteps that weakness so the user
|
|
doesn't have to keep repeating themselves.
|
|
|
|
Returns ``None`` when no place is named, the call fails, or the
|
|
extractor gives back something that doesn't look like a place.
|
|
"""
|
|
if not isinstance(text, str) or not text.strip():
|
|
return None
|
|
if cfg is None:
|
|
return None
|
|
|
|
model = (
|
|
getattr(cfg, "tool_router_model", "")
|
|
or getattr(cfg, "intent_judge_model", "")
|
|
or getattr(cfg, "ollama_chat_model", "")
|
|
)
|
|
base_url = getattr(cfg, "ollama_base_url", "")
|
|
if not model or not base_url:
|
|
return None
|
|
|
|
try:
|
|
from ...llm import call_llm_direct
|
|
except Exception:
|
|
return None
|
|
|
|
sys_prompt = (
|
|
"You extract a single place name from a user's utterance so a weather "
|
|
"tool can look it up. Reply with ONLY the place name (city, town, or "
|
|
"country), with no punctuation, quotes, or explanation. If the user "
|
|
"did not name any place, reply with exactly: none"
|
|
)
|
|
user_prompt = f"User utterance: {text}\n\nPlace:"
|
|
|
|
try:
|
|
resp = call_llm_direct(
|
|
base_url, model, sys_prompt, user_prompt,
|
|
timeout_sec=float(getattr(cfg, "llm_tools_timeout_sec", 8.0)),
|
|
)
|
|
except Exception as e:
|
|
debug_log(f" ⚠️ place extraction failed: {e}", "tools")
|
|
return None
|
|
|
|
if not resp or not isinstance(resp, str):
|
|
return None
|
|
|
|
# Strip punctuation and quotes the extractor might wrap around the name.
|
|
place = resp.strip().strip("'\"`*.,:;!?()[]{}<>").split("\n", 1)[0].strip()
|
|
if not place:
|
|
return None
|
|
if place.lower() in _NO_PLACE_SENTINELS:
|
|
return None
|
|
# Reject multi-sentence or overly long replies — those are almost always
|
|
# the model explaining ("the user did not name a place") instead of
|
|
# answering. Place names are at most a handful of words (e.g. "New York",
|
|
# "Stratford-upon-Avon", "São Paulo"), so 5 words is a generous cap.
|
|
if len(place) > 60 or "." in place or len(place.split()) > 5:
|
|
return None
|
|
return place
|
|
|
|
|
|
def _romanise_place(name: str, cfg) -> Optional[str]:
|
|
"""Romanise a non-Latin place name for the geocoder.
|
|
|
|
Open-Meteo's geocoding API only matches Latin/English spellings, so a
|
|
Korean (or other non-Latin) city name like ``서울`` returns zero results
|
|
even though the place plainly exists. When OUTPUT_LANGUAGE locks replies to
|
|
Korean the tool-calling model naturally fills ``location`` with the Korean
|
|
name, which would otherwise dead-end. Ask the (already warm) small model for
|
|
the common English exonym so geocoding can succeed on a retry.
|
|
|
|
Returns the romanised name, or ``None`` if it is unavailable, unchanged, or
|
|
doesn't look like a place.
|
|
"""
|
|
if not isinstance(name, str) or not name.strip() or name.isascii():
|
|
return None
|
|
if cfg is None:
|
|
return None
|
|
|
|
model = (
|
|
getattr(cfg, "tool_router_model", "")
|
|
or getattr(cfg, "intent_judge_model", "")
|
|
or getattr(cfg, "ollama_chat_model", "")
|
|
)
|
|
base_url = getattr(cfg, "ollama_base_url", "")
|
|
if not model or not base_url:
|
|
return None
|
|
|
|
try:
|
|
from ...llm import call_llm_direct
|
|
except Exception:
|
|
return None
|
|
|
|
sys_prompt = (
|
|
"You romanise place names for a geocoding API that only understands "
|
|
"English/Latin spellings. Reply with ONLY the common English name of "
|
|
"the place, no punctuation, quotes, or explanation. "
|
|
"Examples: 서울 -> Seoul, 도쿄 -> Tokyo, 뮌헨 -> Munich."
|
|
)
|
|
user_prompt = f"Place: {name}\n\nEnglish name:"
|
|
|
|
try:
|
|
resp = call_llm_direct(
|
|
base_url, model, sys_prompt, user_prompt,
|
|
timeout_sec=float(getattr(cfg, "llm_tools_timeout_sec", 8.0)),
|
|
)
|
|
except Exception as e:
|
|
debug_log(f" ⚠️ place romanisation failed: {e}", "tools")
|
|
return None
|
|
|
|
if not resp or not isinstance(resp, str):
|
|
return None
|
|
out = resp.strip().strip("'\"`*.,:;!?()[]{}<>").split("\n", 1)[0].strip()
|
|
if not out or out.lower() in _NO_PLACE_SENTINELS:
|
|
return None
|
|
if len(out) > 60 or len(out.split()) > 5 or out == name:
|
|
return None
|
|
return out
|
|
|
|
|
|
# WMO Weather interpretation codes
|
|
# https://open-meteo.com/en/docs
|
|
WMO_CODES = {
|
|
0: "Clear sky",
|
|
1: "Mainly clear",
|
|
2: "Partly cloudy",
|
|
3: "Overcast",
|
|
45: "Foggy",
|
|
48: "Depositing rime fog",
|
|
51: "Light drizzle",
|
|
53: "Moderate drizzle",
|
|
55: "Dense drizzle",
|
|
56: "Light freezing drizzle",
|
|
57: "Dense freezing drizzle",
|
|
61: "Slight rain",
|
|
63: "Moderate rain",
|
|
65: "Heavy rain",
|
|
66: "Light freezing rain",
|
|
67: "Heavy freezing rain",
|
|
71: "Slight snow",
|
|
73: "Moderate snow",
|
|
75: "Heavy snow",
|
|
77: "Snow grains",
|
|
80: "Slight rain showers",
|
|
81: "Moderate rain showers",
|
|
82: "Violent rain showers",
|
|
85: "Slight snow showers",
|
|
86: "Heavy snow showers",
|
|
95: "Thunderstorm",
|
|
96: "Thunderstorm with slight hail",
|
|
99: "Thunderstorm with heavy hail",
|
|
}
|
|
|
|
# Korean conditions for the concise spoken reply.
|
|
WMO_CODES_KO = {
|
|
0: "맑음", 1: "대체로 맑음", 2: "구름 조금", 3: "흐림",
|
|
45: "안개", 48: "서리 안개",
|
|
51: "약한 이슬비", 53: "이슬비", 55: "강한 이슬비",
|
|
56: "약한 어는 이슬비", 57: "강한 어는 이슬비",
|
|
61: "약한 비", 63: "비", 65: "강한 비",
|
|
66: "약한 어는 비", 67: "강한 어는 비",
|
|
71: "약한 눈", 73: "눈", 75: "강한 눈", 77: "싸락눈",
|
|
80: "약한 소나기", 81: "소나기", 82: "강한 소나기",
|
|
85: "약한 눈소나기", 86: "강한 눈소나기",
|
|
95: "천둥번개", 96: "우박 동반 천둥번개", 99: "강한 우박 천둥번개",
|
|
}
|
|
|
|
|
|
class WeatherTool(Tool):
|
|
"""Tool for getting current weather using Open-Meteo API."""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "getWeather"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return (
|
|
"Weather only (current + forecast). NOT for time-of-day, date, or "
|
|
"location questions — those are already in the assistant's context. "
|
|
"Use for ANY weather question: now, later today, tomorrow, this week. "
|
|
"Call with {} — user location is auto-detected. Do NOT ask the user "
|
|
"where they are or request a city; just call this tool with empty args."
|
|
)
|
|
|
|
@property
|
|
def inputSchema(self) -> Dict[str, Any]:
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"location": {
|
|
"type": "string",
|
|
"description": "OPTIONAL. City name or location (e.g., 'London', 'New York', 'Tokyo'). Only set this if the user explicitly named a place different from their own location. If omitted, the tool auto-uses the user's current detected location — never ask the user for this argument."
|
|
}
|
|
},
|
|
"required": []
|
|
}
|
|
|
|
def _get_user_location(self, context: ToolContext) -> Optional[Dict[str, Any]]:
|
|
"""Get user's current location from config/auto-detection.
|
|
|
|
Returns dict with 'lat', 'lon', and 'display_name' keys, or None if unavailable.
|
|
"""
|
|
try:
|
|
location_info = get_location_info(
|
|
config_ip=getattr(context.cfg, 'location_ip_address', None),
|
|
auto_detect=getattr(context.cfg, 'location_auto_detect', True),
|
|
resolve_cgnat_public_ip=getattr(context.cfg, 'location_cgnat_resolve_public_ip', True),
|
|
location_cache_minutes=getattr(context.cfg, 'location_cache_minutes', 60),
|
|
)
|
|
|
|
if "error" in location_info:
|
|
debug_log(f" ⚠️ location detection failed: {location_info.get('error')}", "tools")
|
|
return None
|
|
|
|
# Use coordinates directly (avoids geocoding issues with district names)
|
|
lat = location_info.get("latitude")
|
|
lon = location_info.get("longitude")
|
|
if lat is None or lon is None:
|
|
return None
|
|
|
|
# Build display name from available fields (handle None values)
|
|
city = location_info.get("city") or ""
|
|
region = location_info.get("region") or ""
|
|
country = location_info.get("country") or ""
|
|
|
|
# Prefer city, but fall back to region if city is a district
|
|
display_parts = []
|
|
if city:
|
|
display_parts.append(city)
|
|
if region and region != city:
|
|
display_parts.append(region)
|
|
if country:
|
|
display_parts.append(country)
|
|
|
|
display_name = ", ".join(display_parts) if display_parts else "your location"
|
|
|
|
return {"lat": lat, "lon": lon, "display_name": display_name}
|
|
except Exception as e:
|
|
debug_log(f" ⚠️ location detection error: {e}", "tools")
|
|
return None
|
|
|
|
def run(self, args: Optional[Dict[str, Any]], context: ToolContext) -> ToolExecutionResult:
|
|
"""Get current weather for a location."""
|
|
context.user_print("🌤️ Checking weather...")
|
|
|
|
try:
|
|
# Get location from args, or fall back to user's detected location
|
|
location_str = ""
|
|
if args and isinstance(args, dict):
|
|
raw_location = args.get("location")
|
|
# Handle None values (LLM may pass location: null/None)
|
|
location_str = str(raw_location).strip() if raw_location else ""
|
|
|
|
# Determine coordinates and display name
|
|
lat: Optional[float] = None
|
|
lon: Optional[float] = None
|
|
location_display: str = ""
|
|
|
|
# Track whether we inferred the place name from the user's text
|
|
# rather than receiving it from the caller — used only for the
|
|
# debug log, doesn't change behaviour downstream.
|
|
place_from_fallback = False
|
|
|
|
if not location_str:
|
|
# No location provided - try auto-detected coordinates first.
|
|
user_loc = self._get_user_location(context)
|
|
if user_loc:
|
|
lat = user_loc["lat"]
|
|
lon = user_loc["lon"]
|
|
location_display = user_loc["display_name"]
|
|
debug_log(
|
|
f" 📍 using detected location: {location_display} ({lat}, {lon})",
|
|
"tools",
|
|
)
|
|
else:
|
|
# Auto-detect failed. Last resort: scrape a place name from
|
|
# the user's current utterance. Small tool-calling models
|
|
# often drop the city from tool args even when the user
|
|
# just said one, so doing this on the tool side stops the
|
|
# "I need it for London" → "please tell me which city"
|
|
# ping-pong loop.
|
|
user_text = getattr(context, "redacted_text", "") or ""
|
|
cfg = getattr(context, "cfg", None)
|
|
extracted = _extract_place_from_user_text(user_text, cfg)
|
|
if extracted:
|
|
debug_log(
|
|
f" 📍 auto-detect unavailable; extracted place from user text: '{extracted}'",
|
|
"tools",
|
|
)
|
|
location_str = extracted
|
|
place_from_fallback = True
|
|
else:
|
|
# Auto-detect genuinely failed and the user didn't name
|
|
# a place in this utterance. Asking is the right move.
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text=(
|
|
"I couldn't auto-detect your location. "
|
|
"Please tell me which city to check the weather for."
|
|
),
|
|
)
|
|
|
|
if location_str:
|
|
# User specified a location (or we pulled one from their text) — geocode it.
|
|
debug_log(
|
|
f" 🌤️ geocoding location: '{location_str}'"
|
|
+ (" (from user text fallback)" if place_from_fallback else ""),
|
|
"tools",
|
|
)
|
|
|
|
geocode_url = "https://geocoding-api.open-meteo.com/v1/search"
|
|
# Intentionally English — tool results are processed by the LLM,
|
|
# not shown to the user. All models handle English data well.
|
|
geocode_params = {
|
|
"name": location_str,
|
|
"count": 1,
|
|
"language": "en",
|
|
"format": "json"
|
|
}
|
|
|
|
geo_response = requests.get(geocode_url, params=geocode_params, timeout=10)
|
|
geo_response.raise_for_status()
|
|
geo_data = geo_response.json()
|
|
|
|
# Open-Meteo only matches Latin spellings, so a non-Latin name
|
|
# (e.g. Korean "서울") returns nothing. Retry once with an
|
|
# LLM-romanised name before giving up.
|
|
if not geo_data.get("results"):
|
|
romanised = _romanise_place(location_str, getattr(context, "cfg", None))
|
|
if romanised:
|
|
debug_log(
|
|
f" 🌤️ geocode empty for '{location_str}'; retrying romanised '{romanised}'",
|
|
"tools",
|
|
)
|
|
geocode_params["name"] = romanised
|
|
geo_response = requests.get(geocode_url, params=geocode_params, timeout=10)
|
|
geo_response.raise_for_status()
|
|
geo_data = geo_response.json()
|
|
if geo_data.get("results"):
|
|
location_str = romanised
|
|
|
|
if not geo_data.get("results"):
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text=f"Could not find location '{location_str}'. Try a different city name or spelling."
|
|
)
|
|
|
|
place = geo_data["results"][0]
|
|
lat = place["latitude"]
|
|
lon = place["longitude"]
|
|
place_name = place.get("name", location_str)
|
|
country = place.get("country", "")
|
|
admin1 = place.get("admin1", "") # State/region
|
|
|
|
# Build display name
|
|
location_display = place_name
|
|
if admin1 and admin1 != place_name:
|
|
location_display += f", {admin1}"
|
|
if country:
|
|
location_display += f", {country}"
|
|
|
|
debug_log(f" 📍 resolved to {location_display} ({lat}, {lon})", "tools")
|
|
|
|
# Step 2: Get current weather + forecast
|
|
weather_url = "https://api.open-meteo.com/v1/forecast"
|
|
weather_params = {
|
|
"latitude": lat,
|
|
"longitude": lon,
|
|
"current": "temperature_2m,relative_humidity_2m,apparent_temperature,weather_code,wind_speed_10m,wind_gusts_10m",
|
|
"hourly": "temperature_2m,weather_code",
|
|
"daily": "weather_code,temperature_2m_max,temperature_2m_min",
|
|
"forecast_days": 7,
|
|
"temperature_unit": "celsius",
|
|
"wind_speed_unit": "kmh",
|
|
"timezone": "auto"
|
|
}
|
|
|
|
weather_response = requests.get(weather_url, params=weather_params, timeout=10)
|
|
weather_response.raise_for_status()
|
|
weather_data = weather_response.json()
|
|
|
|
current = weather_data.get("current", {})
|
|
if not current:
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text=f"Weather data temporarily unavailable for {location_display}."
|
|
)
|
|
|
|
# Extract current weather values
|
|
temp_c = current.get("temperature_2m")
|
|
feels_like_c = current.get("apparent_temperature")
|
|
humidity = current.get("relative_humidity_2m")
|
|
weather_code = current.get("weather_code", 0)
|
|
wind_speed = current.get("wind_speed_10m")
|
|
wind_gusts = current.get("wind_gusts_10m")
|
|
|
|
# Convert to Fahrenheit as well
|
|
temp_f = round(temp_c * 9/5 + 32, 1) if temp_c is not None else None
|
|
feels_like_f = round(feels_like_c * 9/5 + 32, 1) if feels_like_c is not None else None
|
|
|
|
# Get weather description
|
|
weather_desc = WMO_CODES.get(weather_code, "Unknown conditions")
|
|
|
|
# Concise, ready-to-speak Korean one-liner for the voice path. The
|
|
# tool result is normally re-synthesised by the LLM, but a small
|
|
# model rambles and leaks °F / CJK fragments, so we hand it a clean
|
|
# Korean sentence it can echo verbatim (one-sentence system rule).
|
|
_ko = WMO_CODES_KO.get(weather_code, weather_desc)
|
|
_short_loc = location_display.split(",")[0].strip() or location_display
|
|
_ko_parts = [f"지금 {_short_loc} 날씨는 {_ko}"]
|
|
if temp_c is not None:
|
|
_t = f"기온 {round(temp_c)}도"
|
|
if feels_like_c is not None and round(feels_like_c) != round(temp_c):
|
|
_t += f"(체감 {round(feels_like_c)}도)"
|
|
_ko_parts.append(_t)
|
|
ko_sentence = ", ".join(_ko_parts) + "입니다."
|
|
|
|
# Build response text — concise current conditions (Korean sentence
|
|
# first so the model echoes it; English detail kept for any follow-up
|
|
# reasoning but the forecast firehose is dropped to curb rambling).
|
|
lines = [
|
|
f"한국어로 정확히 이 한 문장만 답하세요: {ko_sentence}",
|
|
f"(참고 데이터 — 답변에 추가하지 말 것: {location_display}, {weather_desc}, "
|
|
f"{temp_c}°C feels {feels_like_c}°C, humidity {humidity}%, wind {wind_speed}km/h)",
|
|
]
|
|
# Forecast (hourly / 7-day) is intentionally omitted from the default
|
|
# voice reply to keep it to one spoken sentence; current conditions
|
|
# are what "날씨 알려줘" asks for.
|
|
|
|
reply_text = "\n".join(lines)
|
|
|
|
debug_log(f" ✅ weather retrieved: {weather_desc}, {temp_c}°C", "tools")
|
|
# Use first part of location_display for concise output
|
|
short_name = location_display.split(",")[0].strip()
|
|
context.user_print(f"✅ Weather for {short_name}: {weather_desc}, {temp_c}°C")
|
|
|
|
return ToolExecutionResult(success=True, reply_text=reply_text)
|
|
|
|
except requests.exceptions.Timeout:
|
|
debug_log("weather request timed out", "tools")
|
|
context.user_print("⚠️ Weather service timeout.")
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text="Weather service is taking too long to respond. Please try again."
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
debug_log(f"weather request failed: {e}", "tools")
|
|
context.user_print("⚠️ Weather service unavailable.")
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text="Weather service is temporarily unavailable. Please try again later."
|
|
)
|
|
except Exception as e:
|
|
debug_log(f"weather error: {e}", "tools")
|
|
context.user_print("⚠️ Error getting weather.")
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text=f"Error getting weather: {e}"
|
|
)
|