feat: агент «Погода» для AMS (Open-Meteo + met.no)
This commit is contained in:
Maksim Totmin
@@ -0,0 +1,4 @@
|
||||
{
|
||||
"name": "Погода",
|
||||
"description": "Присылает сводку о текущей погоде по названию города (кириллица или латиница)."
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
"""AMS-агент «Погода».
|
||||
|
||||
Назначение: по названию города вернуть короткую сводку о текущей погоде.
|
||||
|
||||
Архитектура (разделение ответственности):
|
||||
* weather.py — получение данных (геокодинг + прогноз); ничего не знает про AMS;
|
||||
* handler.py — тонкий адаптер AMS: вызывает сервис и отдаёт ответ стримом.
|
||||
|
||||
Модуль не выпускает исключений наружу: любая ошибка превращается в текстовый
|
||||
чанк. Это принципиально, потому что воркер AMS при исключении из step()
|
||||
прерывает диалог и не отправляет клиенту ничего.
|
||||
|
||||
Контракт AMS: модуль экспортирует генератор step(...), который AMS вызывает
|
||||
по именам аргументов: step(params=..., state=..., agent=...).
|
||||
Отдавать нужно словари вида {"type": "chunk", "data": "..."}.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from typing import Any, Iterator
|
||||
|
||||
from pydantic_ai import ModelSettings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
|
||||
def _load_sibling(module_name: str, filename: str) -> Any:
|
||||
"""Импортировать модуль из каталога агента.
|
||||
|
||||
AMS загружает handler.py как отдельный модуль (не пакет), а каталог агента
|
||||
не входит в sys.path. Поэтому соседний модуль подключаем явно и под
|
||||
уникальным именем — это исключает конфликты имён с другими агентами.
|
||||
|
||||
Модуль обязательно регистрируем в sys.modules: этого требуют dataclasses
|
||||
(при отложенных аннотациях) и pickle.
|
||||
"""
|
||||
spec = importlib.util.spec_from_file_location(module_name, os.path.join(CURRENT_DIR, filename))
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[module_name] = module
|
||||
try:
|
||||
spec.loader.exec_module(module)
|
||||
except Exception:
|
||||
sys.modules.pop(module_name, None) # не оставляем «полумодуль» в реестре
|
||||
raise
|
||||
return module
|
||||
|
||||
|
||||
weather = _load_sibling("ams_weather_service", "weather.py")
|
||||
|
||||
|
||||
def _read_text(filename: str) -> str:
|
||||
"""Прочитать текстовый ресурс, лежащий рядом с обработчиком."""
|
||||
with open(os.path.join(CURRENT_DIR, filename), encoding="utf-8") as file:
|
||||
return file.read()
|
||||
|
||||
|
||||
SYSTEM_PROMPT = _read_text("system_prompt.txt")
|
||||
|
||||
# temperature — модель только оформляет факты, без «творчества».
|
||||
# enable_thinking отключён: иначе Qwen тратит время на reasoning-токены,
|
||||
# а max_tokens ограничивает длину ответа (ответ по заданию короткий).
|
||||
MODEL_SETTINGS = ModelSettings(
|
||||
temperature=0.2,
|
||||
max_tokens=256,
|
||||
extra_body={"chat_template_kwargs": {"enable_thinking": False}},
|
||||
)
|
||||
|
||||
|
||||
def _facts(location: weather.Location, current: weather.Weather) -> str:
|
||||
"""Факты для языковой модели.
|
||||
|
||||
Все значения передаём явно, чтобы модель не додумывала числа.
|
||||
"""
|
||||
return (
|
||||
f"Город: {location.title}\n"
|
||||
f"Состояние: {current.description}\n"
|
||||
f"Температура: {current.temperature_c} °C\n"
|
||||
f"Влажность: {current.humidity_pct} %\n"
|
||||
f"Ветер: {current.wind_ms} м/с, направление {current.wind_dir_deg}°\n"
|
||||
f"Осадки: {current.precipitation_mm} мм/час"
|
||||
)
|
||||
|
||||
|
||||
def _run(agent: Any, params: dict[str, Any]) -> Iterator[dict[str, Any]]:
|
||||
"""Основной сценарий: город -> сводка о погоде."""
|
||||
# Запрос относится к конкретному городу, поэтому не тащим контекст прошлых.
|
||||
agent.history = []
|
||||
agent.set_system_prompt(SYSTEM_PROMPT)
|
||||
|
||||
try:
|
||||
location = weather.geocode(params["city"])
|
||||
current = weather.fetch_weather(location)
|
||||
except weather.WeatherError as exc:
|
||||
logger.warning("Погода недоступна: %s", exc.detail)
|
||||
yield {"type": "chunk", "data": f"Не удалось получить погоду: {exc.message}."}
|
||||
return
|
||||
|
||||
summary = weather.format_summary(location, current)
|
||||
|
||||
# Языковая модель только переформулирует факты. Если она недоступна —
|
||||
# отдаём детерминированную сводку, чтобы ответ был в любом случае.
|
||||
try:
|
||||
with agent.run_stream_sync(_facts(location, current), model_settings=MODEL_SETTINGS) as stream:
|
||||
for chunk in stream.stream_text(delta=True):
|
||||
yield {"type": "chunk", "data": chunk}
|
||||
except Exception: # noqa: BLE001 — намеренно широкий: нужен гарантированный ответ
|
||||
logger.exception("LLM недоступен, отдаю детерминированную сводку")
|
||||
yield {"type": "chunk", "data": summary}
|
||||
|
||||
|
||||
def step(agent: Any, state: dict[str, Any], params: dict[str, Any]) -> Iterator[dict[str, Any]]:
|
||||
"""Точка входа AMS. Гарантированно не поднимает исключений."""
|
||||
try:
|
||||
yield from _run(agent, params)
|
||||
except Exception: # noqa: BLE001 — последний рубеж: диалог не должен падать
|
||||
logger.exception("Непредвиденная ошибка обработки запроса")
|
||||
yield {"type": "chunk", "data": "Не удалось обработать запрос, попробуйте позже."}
|
||||
@@ -0,0 +1,21 @@
|
||||
"""Входные параметры агента «Погода».
|
||||
|
||||
AMS валидирует входящие params этой моделью до вызова handler.step.
|
||||
|
||||
Важно: loader AMS создаёт `Params()` без аргументов (для интроспекции UI),
|
||||
поэтому у всех полей обязаны быть значения по умолчанию. Пустой город
|
||||
отсекается уже в handler'е понятным сообщением.
|
||||
"""
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
class Params(BaseModel):
|
||||
# Обрезаем случайные пробелы по краям — « Казань » и «Казань» эквивалентны.
|
||||
model_config = ConfigDict(str_strip_whitespace=True)
|
||||
|
||||
city: str = Field(
|
||||
default="",
|
||||
max_length=100,
|
||||
description="Название города (кириллица или латиница), например «Казань»",
|
||||
json_schema_extra={"ui_type": "input", "ui_main": True},
|
||||
)
|
||||
@@ -0,0 +1,9 @@
|
||||
Ты — погодный ассистент. На вход приходят уже готовые данные о погоде.
|
||||
|
||||
Правила:
|
||||
- Составь короткую сводку на русском языке (2–3 предложения), тон — спокойный и деловой.
|
||||
- Используй ТОЛЬКО переданные числа и факты. Ничего не додумывай и не добавляй.
|
||||
- Обязательно назови город, температуру и общее состояние погоды.
|
||||
- Уместно добавь один короткий практический совет (зонт, ветровка, шапка) — без лишних слов.
|
||||
- Не задавай вопросов и не предлагай выбирать другие города.
|
||||
- Если данных не хватает — просто перечисли то, что есть, без домыслов.
|
||||
@@ -0,0 +1,212 @@
|
||||
"""Погодный сервис: геокодинг Open-Meteo + прогноз met.no.
|
||||
|
||||
Модуль намеренно ничего не знает про AMS и pydantic-ai — это чистый слой
|
||||
получения данных. Его можно запустить и проверить отдельно:
|
||||
|
||||
python weather.py Казань Москва
|
||||
|
||||
Поддерживаются русские (кириллица) и латинские названия городов.
|
||||
API-ключи не нужны: оба сервиса бесплатные.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# --- Внешние сервисы ------------------------------------------------------
|
||||
GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search"
|
||||
FORECAST_URL = "https://api.met.no/weatherapi/locationforecast/2.0/compact"
|
||||
|
||||
# met.no просит осмысленный User-Agent с контактом, иначе может блокировать.
|
||||
USER_AGENT = "ams-weather-agent/1.0 (+https://git.totmin.ru/en2zmax/ams-weather)"
|
||||
|
||||
# (connect, read) в секундах. Держим короткими, чтобы не подвешивать диалог.
|
||||
TIMEOUT = (4, 7)
|
||||
RETRIES = 2 # общее число попыток на один HTTP-запрос
|
||||
|
||||
# Переиспользуем TCP-соединения между вызовами — заметно быстрее серии запросов.
|
||||
_session = requests.Session()
|
||||
_session.headers["User-Agent"] = USER_AGENT
|
||||
|
||||
|
||||
class WeatherError(Exception):
|
||||
"""Ожидаемая ошибка получения погоды.
|
||||
|
||||
Разделяем два уровня информации:
|
||||
* message — короткий безопасный текст для пользователя;
|
||||
* detail — подробности для лога (пользователю не показываем).
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, detail: str | None = None) -> None:
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
self.detail = detail or message
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Location:
|
||||
"""Найденный населённый пункт."""
|
||||
|
||||
name: str
|
||||
region: str | None
|
||||
country: str | None
|
||||
latitude: float
|
||||
longitude: float
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
"""Человекочитаемое имя места: «Казань, Татарстан, Россия»."""
|
||||
return ", ".join(part for part in (self.name, self.region, self.country) if part)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Weather:
|
||||
"""Текущие погодные условия."""
|
||||
|
||||
temperature_c: float
|
||||
humidity_pct: int
|
||||
wind_ms: float
|
||||
wind_dir_deg: int
|
||||
precipitation_mm: float
|
||||
description: str
|
||||
|
||||
|
||||
# met.no отдаёт symbol_code вида «partlycloudy_day»; суффикс дня/ночи отбрасываем.
|
||||
_SYMBOLS: dict[str, str] = {
|
||||
"clearsky": "ясно",
|
||||
"fair": "малооблачно",
|
||||
"partlycloudy": "переменная облачность",
|
||||
"cloudy": "облачно",
|
||||
"fog": "туман",
|
||||
"lightrain": "небольшой дождь",
|
||||
"rain": "дождь",
|
||||
"heavyrain": "сильный дождь",
|
||||
"lightrainshowers": "небольшой ливень",
|
||||
"rainshowers": "ливень",
|
||||
"heavyrainshowers": "сильный ливень",
|
||||
"lightsleet": "небольшой дождь со снегом",
|
||||
"sleet": "дождь со снегом",
|
||||
"heavysleet": "сильный дождь со снегом",
|
||||
"lightsnow": "небольшой снег",
|
||||
"snow": "снег",
|
||||
"heavysnow": "сильный снег",
|
||||
"lightsnowshowers": "небольшой снегопад",
|
||||
"snowshowers": "снегопад",
|
||||
"heavysnowshowers": "сильный снегопад",
|
||||
"rainandthunder": "дождь с грозой",
|
||||
"sleetandthunder": "дождь со снегом и грозой",
|
||||
"snowandthunder": "снег с грозой",
|
||||
"thunderstorm": "гроза",
|
||||
}
|
||||
_UNKNOWN_DESCRIPTION = "нет данных"
|
||||
|
||||
|
||||
def _get(url: str, params: dict[str, Any]) -> dict[str, Any]:
|
||||
"""GET с короткими ретраями и единообразной обработкой ошибок.
|
||||
|
||||
Повторяем только временные сбои: сетевые ошибки, 5xx и 429.
|
||||
Постоянные 4xx (например, 400) сразу превращаем в WeatherError —
|
||||
повторять их бессмысленно.
|
||||
"""
|
||||
last_error: Exception | None = None
|
||||
for attempt in range(1, RETRIES + 1):
|
||||
try:
|
||||
response = _session.get(url, params=params, timeout=TIMEOUT)
|
||||
except requests.RequestException as exc:
|
||||
last_error = exc
|
||||
else:
|
||||
status = response.status_code
|
||||
if 400 <= status < 500 and status != 429:
|
||||
raise WeatherError("внешний сервис погоды вернул ошибку", f"{url}: HTTP {status}")
|
||||
if status < 500 and status != 429:
|
||||
try:
|
||||
return response.json()
|
||||
except ValueError as exc: # ответ не является корректным JSON
|
||||
last_error = exc
|
||||
else:
|
||||
last_error = requests.HTTPError(f"HTTP {status}")
|
||||
|
||||
logger.warning("Запрос %s не удался (попытка %d/%d): %s", url, attempt, RETRIES, last_error)
|
||||
if attempt < RETRIES:
|
||||
time.sleep(0.4 * attempt) # короткий нарастающий backoff
|
||||
|
||||
raise WeatherError("внешний сервис погоды недоступен", f"{url}: {last_error}")
|
||||
|
||||
|
||||
@lru_cache(maxsize=256)
|
||||
def geocode(city: str) -> Location:
|
||||
"""Найти город по названию (кириллица или латиница).
|
||||
|
||||
Результат кэшируется: повторный запрос того же города отвечает мгновенно.
|
||||
"""
|
||||
city = city.strip()
|
||||
if not city:
|
||||
raise WeatherError("не указан город")
|
||||
|
||||
payload = _get(GEOCODE_URL, {"name": city, "count": 1, "language": "ru", "format": "json"})
|
||||
results = payload.get("results") or []
|
||||
if not results:
|
||||
raise WeatherError(f"город «{city}» не найден")
|
||||
|
||||
top = results[0]
|
||||
return Location(
|
||||
name=top.get("name", city),
|
||||
region=top.get("admin1"),
|
||||
country=top.get("country"),
|
||||
# met.no принимает координаты не более чем с 4 знаками после запятой.
|
||||
latitude=round(float(top["latitude"]), 4),
|
||||
longitude=round(float(top["longitude"]), 4),
|
||||
)
|
||||
|
||||
|
||||
def fetch_weather(location: Location) -> Weather:
|
||||
"""Текущая погода по координатам (met.no)."""
|
||||
payload = _get(FORECAST_URL, {"lat": location.latitude, "lon": location.longitude})
|
||||
try:
|
||||
series = payload["properties"]["timeseries"][0]
|
||||
details = series["data"]["instant"]["details"]
|
||||
next_hour = series["data"].get("next_1_hours") or {}
|
||||
symbol = (next_hour.get("summary") or {}).get("symbol_code", "")
|
||||
precipitation = (next_hour.get("details") or {}).get("precipitation_amount", 0.0)
|
||||
|
||||
return Weather(
|
||||
temperature_c=round(float(details["air_temperature"]), 1),
|
||||
humidity_pct=round(float(details["relative_humidity"])),
|
||||
wind_ms=round(float(details["wind_speed"]), 1),
|
||||
wind_dir_deg=round(float(details["wind_from_direction"])),
|
||||
precipitation_mm=round(float(precipitation), 1),
|
||||
description=_SYMBOLS.get(symbol.rsplit("_", 1)[0], _UNKNOWN_DESCRIPTION),
|
||||
)
|
||||
except (KeyError, IndexError, TypeError, ValueError) as exc:
|
||||
raise WeatherError("сервис прогноза вернул неожиданный формат", repr(exc)) from exc
|
||||
|
||||
|
||||
def format_summary(location: Location, weather: Weather) -> str:
|
||||
"""Детерминированная сводка.
|
||||
|
||||
Используется как фолбэк, если языковая модель недоступна, — пользователь
|
||||
в любом случае получает ответ по существу.
|
||||
"""
|
||||
return (
|
||||
f"{location.title}: {weather.description}, {weather.temperature_c:g} °C, "
|
||||
f"влажность {weather.humidity_pct} %, ветер {weather.wind_ms:g} м/с."
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__": # ручной smoke-тест: python weather.py Казань
|
||||
import sys
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
for query in sys.argv[1:] or ["Казань"]:
|
||||
try:
|
||||
found = geocode(query)
|
||||
print(format_summary(found, fetch_weather(found)))
|
||||
except WeatherError as exc:
|
||||
print(f"{query}: ошибка — {exc.message}")
|
||||
Reference in New Issue
Block a user