c522e1b285
getWeather now returns only the Korean sentence (지금 <곳> 날씨는 <상태>, 기온 N도 (체감 M도)입니다) with no English/°C source. A deterministic weather path in the engine returns it verbatim, bypassing the 7B which was rephrasing into multiple sentences and leaking 'Celsius'.
479 lines
20 KiB
Python
479 lines
20 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) + "입니다."
|
|
|
|
# The reply is the clean Korean sentence ONLY — no English/°C source
|
|
# for the model to echo ("25도 Celsius"), no forecast firehose to
|
|
# ramble over. The deterministic weather path in the engine returns
|
|
# this verbatim; on the LLM path the model just echoes one sentence.
|
|
lines = [ko_sentence]
|
|
|
|
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}"
|
|
)
|