diff --git a/docs/design.md b/docs/design.md index 348b14ac..4f6f12a7 100644 --- a/docs/design.md +++ b/docs/design.md @@ -181,6 +181,8 @@ Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活 CLI 负责关闭自己创建的模型服务对象及其网络资源。测试或外部调用方注入的对象视为借用，不由应用关闭。 +配置边界将 `LLM_FALLBACK_PROVIDER` 和 `LLM_FALLBACK_MODEL` 作为一组解析。可选的 `LLM_FALLBACK_API_KEY` 和 `LLM_FALLBACK_BASE_URL` 只传给备用适配器；未配置时由 any-llm 按所选 provider 的规则解析默认连接配置。组合根用 `FallbackLLMProvider` 包装两个独立持有 SDK 资源的适配器；首次主适配器请求失败后，该包装器在生命周期内固定使用备用适配器。关闭包装器时会尝试释放两侧资源；两侧都发生普通清理异常时通过 `ExceptionGroup` 聚合，取消异常保持传播并优先于普通清理异常，竞争的普通异常只按异常类型记录。 + ## 调度与投递 `daemon` 创建 forecast 和 briefing 两类 APScheduler 任务，并保持常驻。它不接受立即运行参数。 diff --git a/docs/notes.md b/docs/notes.md index a68afa3e..a83b6606 100644 --- a/docs/notes.md +++ b/docs/notes.md @@ -48,6 +48,14 @@ 如果以后发布明确允许删除旧配置的重大版本，并且迁移说明已经给已有部署留出足够时间，就可以移除这两个后备变量。在此之前，修改模型配置边界时必须保留并测试这一优先级。 +## FallbackLLMProvider 在进程内保持粘性 + +这是一个有意保留的自定义外部服务集成。选择 `FallbackLLMProvider` 的原因是 any-llm 只统一调用单个 provider，不编排跨 provider 的故障切换。包装器捕获主适配器的 `LLMRequestError`，切换后在剩余生命周期内固定使用备用适配器，使同一进程里的契约修复不会回到刚刚失败的主服务。 + +它替代的是每个调用点手写的故障切换分支，不替代 any-llm 的厂商适配器，也不接管 SDK 凭据、请求重试或输出验证。 + +这个选择成立的条件是一次主服务请求失败足以让当前模型对象的后续调用继续使用备用服务，恢复主服务交给下次创建模型对象。如果 any-llm 提供可观察的跨 provider 路由，或者常驻进程需要在不重启的情况下探测并恢复主服务，就重新评估并用带健康状态和冷却时间的路由替换当前粘性开关。 + ## 模型对象的关闭使用兼容探测 any-llm 当前没有统一的关闭接口。应用先尝试 provider 的 `aclose()` 或 `close()`，再清理它直接持有的已知 SDK 资源。 diff --git a/docs/requirements.md b/docs/requirements.md index 1a8ae9c9..199de483 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -64,6 +64,7 @@ - 服务状态必须与天气简报独立调度，默认每五分钟采集一次，并允许两类任务同时启用。 - 状态页发生变化后仍须判断是否值得打扰用户；没有新增影响或行动价值的变化不能推送。 - 故障、恢复等官方消息使用英语或用户指定语言时必须原样转发；语言不匹配时可以用大语言模型忠实翻译，但不得添加事实。 +- 用户可以配置一个备用大语言模型；主模型请求失败时，系统应自动尝试备用模型，但不得用切换供应商掩盖模型已经返回的无效内容。 - 服务状态必须可以投递到一个或多个独立于天气简报的平台。 - 没有配置可选来源，或可选来源暂时失败时，天气简报仍应运行。 - 每轮只读取按地点时区判断为当天发布的文章。 diff --git a/env.example b/env.example index 1752a880..a1e17586 100644 --- a/env.example +++ b/env.example @@ -2,10 +2,17 @@ # python-dotenv and Docker Compose accept quoted values, but Docker CLI # can pass the wrapping quotes into the container as literal characters. -# Required: select an any-llm provider and model. Provider credentials and API -# bases use the environment names documented by any-llm for that provider. +# Required: select an any-llm provider and model. Primary-provider credentials +# and API bases use the environment names documented by any-llm. LLM_PROVIDER=deepseek LLM_MODEL=deepseek-v4-flash +# Optional request-failure fallback, disabled by default. Configure both values +# or neither. +# LLM_FALLBACK_PROVIDER=openai +# LLM_FALLBACK_MODEL=gpt-5-mini +# Optional fallback-only credentials and endpoint overrides. +# LLM_FALLBACK_API_KEY=replace-in-runtime-environment +# LLM_FALLBACK_BASE_URL= # DeepSeek example. DEEPSEEK_API_BASE is optional. DEEPSEEK_API_KEY=replace-in-runtime-environment # DEEPSEEK_API_BASE= diff --git a/tests/test_cli.py b/tests/test_cli.py index 3a58aed7..1a20df92 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -53,6 +53,7 @@ from weather_briefing.composition.providers import weather_provider_metadata as _weather_provider_metadata from weather_briefing.config import ConfigurationError, Settings from weather_briefing.delivery import BarkTextRenderer +from weather_briefing.llm import FallbackLLMProvider from weather_briefing.models import LocationSpec, ResolvedLocation from weather_briefing.persistence import StateDirectoryInUseError, daemon_state_owner from weather_briefing.registries import PublisherName, WeatherProviderName @@ -763,6 +764,10 @@ async def fail_run( llm_provider="deepseek", llm_model="m", llm_base_url=None, + llm_fallback_provider=None, + llm_fallback_model=None, + llm_fallback_api_key=None, + llm_fallback_base_url=None, llm_max_output_tokens=8192, llm_max_attempts=3, http_timeout_seconds=30.0, @@ -1204,6 +1209,62 @@ async def test_arbitrary_any_llm_provider_is_forwarded(self, monkeypatch) -> Non "diagnostics": None, } + async def test_configured_fallback_llm_provider_is_composed(self, monkeypatch) -> None: + providers: list[SimpleNamespace] = [] + calls: list[tuple[tuple[object, ...], dict[str, object]]] = [] + + def create_provider(*args, **kwargs): + provider = SimpleNamespace() + providers.append(provider) + calls.append((args, kwargs)) + return provider + + monkeypatch.setattr("weather_briefing.llm.any_llm.create_any_llm_provider", create_provider) + settings = replace( + _make_fake_settings(), + llm_fallback_provider="openai", + llm_fallback_model="gpt-fallback", + ) + + provider = _llm_provider(settings) + + assert isinstance(provider, FallbackLLMProvider) + assert len(providers) == 2 + assert calls[1] == ( + ("openai", "gpt-fallback", 8192), + { + "api_key": None, + "api_base": None, + "diagnostics": None, + }, + ) + + async def test_deepseek_fallback_forwards_normalized_connection_settings(self, monkeypatch) -> None: + calls: list[tuple[tuple[object, ...], dict[str, object]]] = [] + monkeypatch.setattr( + "weather_briefing.llm.any_llm.create_any_llm_provider", + lambda *args, **kwargs: calls.append((args, kwargs)) or SimpleNamespace(), + ) + settings = replace( + _make_fake_settings(), + llm_fallback_provider="deepseek", + llm_fallback_model="deepseek-fallback", + llm_fallback_api_key="fallback-key", + llm_fallback_base_url="https://deepseek.example/v1", + ) + + provider = _llm_provider(settings) + + assert isinstance(provider, FallbackLLMProvider) + assert calls[1] == ( + ("deepseek", "deepseek-fallback", 8192), + { + "api_key": "fallback-key", + "api_base": "https://deepseek.example/v1", + "diagnostics": None, + }, + ) + class TestDeliveryProvider: async def test_stdout(self, async_client: httpx.AsyncClient) -> None: diff --git a/tests/test_config.py b/tests/test_config.py index 61ea2451..ec011aae 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -61,6 +61,10 @@ def test_mainland_weather_providers_default_to_qweather_then_open_meteo(monkeypa assert [feed.id for feed in settings.feeds] == ["authority-weather"] assert settings.llm_provider == "deepseek" assert settings.llm_base_url is None + assert settings.llm_fallback_provider is None + assert settings.llm_fallback_model is None + assert settings.llm_fallback_api_key is None + assert settings.llm_fallback_base_url is None assert settings.llm_max_attempts == 3 assert settings.qweather_jwt_lifetime_seconds == 900 assert settings.llm_history_max_documents == 8 @@ -1052,6 +1056,41 @@ def test_any_llm_provider_uses_sdk_managed_configuration(monkeypatch) -> None: assert settings.llm_base_url is None +def test_llm_fallback_provider_and_model_are_loaded(monkeypatch) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_FALLBACK_PROVIDER", "openai") + monkeypatch.setenv("LLM_FALLBACK_MODEL", "gpt-fallback") + + settings = Settings.from_env() + + assert settings.llm_fallback_provider == "openai" + assert settings.llm_fallback_model == "gpt-fallback" + assert settings.llm_fallback_api_key is None + assert settings.llm_fallback_base_url is None + + +def test_llm_fallback_uses_dedicated_connection_settings(monkeypatch) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_FALLBACK_PROVIDER", "openai") + monkeypatch.setenv("LLM_FALLBACK_MODEL", "gpt-fallback") + monkeypatch.setenv("LLM_FALLBACK_API_KEY", "'fallback-key'") + monkeypatch.setenv("LLM_FALLBACK_BASE_URL", "https://gateway.example.invalid/v1/") + + settings = Settings.from_env() + + assert settings.llm_fallback_api_key == "fallback-key" + assert settings.llm_fallback_base_url == "https://gateway.example.invalid/v1" + + +@pytest.mark.parametrize("name", ("LLM_FALLBACK_API_KEY", "LLM_FALLBACK_BASE_URL")) +def test_llm_fallback_connection_settings_require_fallback(monkeypatch, name: str) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv(name, "configured") + + with pytest.raises(ConfigurationError, match=rf"{name} requires LLM_FALLBACK_PROVIDER and LLM_FALLBACK_MODEL"): + Settings.from_env() + + def test_deepseek_model_name_remains_compatible(monkeypatch) -> None: _required_environment(monkeypatch) monkeypatch.delenv("LLM_MODEL") @@ -1197,6 +1236,42 @@ def test_llm_provider_without_completion_raises_error(self, monkeypatch) -> None with pytest.raises(ConfigurationError, match="does not support completion"): Settings.from_env() + @pytest.mark.parametrize( + ("configured_name", "missing_name"), + ( + ("LLM_FALLBACK_PROVIDER", "LLM_FALLBACK_MODEL"), + ("LLM_FALLBACK_MODEL", "LLM_FALLBACK_PROVIDER"), + ), + ) + def test_incomplete_llm_fallback_configuration_raises_error( + self, + monkeypatch, + configured_name: str, + missing_name: str, + ) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv(configured_name, "openai") + monkeypatch.delenv(missing_name, raising=False) + + with pytest.raises(ConfigurationError, match="must be configured together"): + Settings.from_env() + + def test_unsupported_llm_fallback_provider_raises_error(self, monkeypatch) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_FALLBACK_PROVIDER", "unsupported") + monkeypatch.setenv("LLM_FALLBACK_MODEL", "fallback-model") + + with pytest.raises(ConfigurationError, match="Unsupported LLM_FALLBACK_PROVIDER"): + Settings.from_env() + + def test_llm_fallback_provider_without_completion_raises_error(self, monkeypatch) -> None: + _required_environment(monkeypatch) + monkeypatch.setenv("LLM_FALLBACK_PROVIDER", "voyage") + monkeypatch.setenv("LLM_FALLBACK_MODEL", "fallback-model") + + with pytest.raises(ConfigurationError, match="LLM_FALLBACK_PROVIDER does not support completion"): + Settings.from_env() + def test_invalid_float_env_value_raises_error(self, monkeypatch) -> None: _required_environment(monkeypatch) monkeypatch.setenv("HTTP_TIMEOUT_SECONDS", "not-a-number") diff --git a/tests/test_llm_fallback.py b/tests/test_llm_fallback.py new file mode 100644 index 00000000..1a99c149 --- /dev/null +++ b/tests/test_llm_fallback.py @@ -0,0 +1,335 @@ +import asyncio +import logging +from unittest.mock import AsyncMock, call + +import pytest + +from weather_briefing.llm import FallbackLLMProvider, LLMError, LLMRequestError +from weather_briefing.notifications import NotificationDecision + + +def _provider() -> AsyncMock: + provider = AsyncMock() + provider.summarize.return_value = {"provider": "result"} + provider.assess_notification.return_value = NotificationDecision(True) + provider.translate_service_status.return_value = ("title", "body") + return provider + + +@pytest.mark.parametrize( + ("operation", "args", "expected"), + ( + ("summarize", ("system", {"input": "value"}), {"provider": "fallback"}), + ("assess_notification", ({"input": "value"},), NotificationDecision(True)), + ("translate_service_status", ("title", "body", "en"), ("translated", "content")), + ), +) +async def test_request_failure_uses_fallback(operation: str, args: tuple[object, ...], expected: object) -> None: + primary = _provider() + fallback = _provider() + getattr(primary, operation).side_effect = LLMRequestError("primary unavailable") + getattr(fallback, operation).return_value = expected + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + result = await getattr(provider, operation)(*args) + + assert result == expected + getattr(primary, operation).assert_awaited_once_with(*args) + getattr(fallback, operation).assert_awaited_once_with(*args) + + +async def test_successful_primary_request_does_not_use_fallback() -> None: + primary = _provider() + fallback = _provider() + primary.summarize.return_value = {"provider": "primary"} + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + result = await provider.summarize("system", {"input": "value"}) + + assert result == {"provider": "primary"} + fallback.summarize.assert_not_awaited() + + +async def test_request_failure_pins_fallback_for_later_repairs() -> None: + primary = _provider() + fallback = _provider() + primary.summarize.side_effect = LLMRequestError("primary unavailable") + fallback.summarize.side_effect = ( + {"provider": "invalid"}, + {"provider": "repaired"}, + ) + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + first_result = await provider.summarize("system", {"attempt": 1}) + second_result = await provider.summarize("repair", {"attempt": 2}) + + assert first_result == {"provider": "invalid"} + assert second_result == {"provider": "repaired"} + primary.summarize.assert_awaited_once_with("system", {"attempt": 1}) + assert fallback.summarize.await_args_list == [ + call("system", {"attempt": 1}), + call("repair", {"attempt": 2}), + ] + + +async def test_request_failure_pins_fallback_across_operations() -> None: + primary = _provider() + fallback = _provider() + primary.summarize.side_effect = LLMRequestError("primary unavailable") + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + await provider.summarize("system", {"input": "value"}) + decision = await provider.assess_notification({"notification": "value"}) + + assert decision == NotificationDecision(True) + primary.assess_notification.assert_not_awaited() + fallback.assess_notification.assert_awaited_once_with({"notification": "value"}) + + +async def test_output_contract_failure_does_not_use_fallback() -> None: + primary = _provider() + fallback = _provider() + primary.summarize.side_effect = LLMError("invalid response") + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with pytest.raises(LLMError, match="invalid response"): + await provider.summarize("system", {"input": "value"}) + + fallback.summarize.assert_not_awaited() + + +async def test_fallback_failure_preserves_primary_as_context() -> None: + primary = _provider() + fallback = _provider() + primary.summarize.side_effect = LLMRequestError("primary unavailable") + fallback.summarize.side_effect = LLMRequestError("fallback unavailable") + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with pytest.raises(LLMRequestError, match="fallback unavailable") as exc_info: + await provider.summarize("system", {"input": "value"}) + + assert isinstance(exc_info.value.__context__, LLMRequestError) + assert str(exc_info.value.__context__) == "primary unavailable" + + +async def test_fallback_log_excludes_exception_details(caplog) -> None: + primary = _provider() + fallback = _provider() + try: + raise RuntimeError("private upstream detail") + except RuntimeError as exc: + request_error = LLMRequestError("private request detail") + request_error.__cause__ = exc + primary.summarize.side_effect = request_error + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with caplog.at_level(logging.WARNING, logger="weather_briefing.llm"): + await provider.summarize("system", {"input": "value"}) + + assert "operation=summarize primary=primary fallback=fallback" in caplog.text + assert "error_type=RuntimeError" in caplog.text + assert "private upstream detail" not in caplog.text + assert "private request detail" not in caplog.text + + +async def test_close_releases_both_providers() -> None: + primary = _provider() + fallback = _provider() + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + await provider.aclose() + + primary.aclose.assert_awaited_once_with() + fallback.aclose.assert_awaited_once_with() + + +async def test_primary_close_failure_still_closes_fallback() -> None: + primary = _provider() + fallback = _provider() + primary.aclose.side_effect = RuntimeError("cleanup failed") + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with pytest.raises(RuntimeError, match="cleanup failed"): + await provider.aclose() + + fallback.aclose.assert_awaited_once_with() + + +async def test_both_close_failures_are_preserved() -> None: + primary = _provider() + fallback = _provider() + primary.aclose.side_effect = RuntimeError("primary cleanup failed") + fallback.aclose.side_effect = ValueError("fallback cleanup failed") + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with pytest.raises(ExceptionGroup) as exc_info: + await provider.aclose() + + assert [str(error) for error in exc_info.value.exceptions] == [ + "primary cleanup failed", + "fallback cleanup failed", + ] + + +async def test_primary_close_cancellation_still_closes_fallback() -> None: + primary = _provider() + fallback = _provider() + cancellation = asyncio.CancelledError() + primary.aclose.side_effect = cancellation + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with pytest.raises(asyncio.CancelledError) as exc_info: + await provider.aclose() + + assert exc_info.value is cancellation + fallback.aclose.assert_awaited_once_with() + + +async def test_fallback_close_failure_does_not_mask_primary_cancellation(caplog) -> None: + primary = _provider() + fallback = _provider() + cancellation = asyncio.CancelledError() + primary.aclose.side_effect = cancellation + fallback.aclose.side_effect = RuntimeError("private cleanup detail") + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with ( + caplog.at_level(logging.WARNING, logger="weather_briefing.llm"), + pytest.raises(asyncio.CancelledError) as exc_info, + ): + await provider.aclose() + + assert exc_info.value is cancellation + assert "error_type=RuntimeError" in caplog.text + assert "private cleanup detail" not in caplog.text + + +async def test_fallback_cancellation_does_not_replace_primary_cancellation(caplog) -> None: + primary = _provider() + fallback = _provider() + primary_cancellation = asyncio.CancelledError("secret-primary-detail") + fallback_cancellation = asyncio.CancelledError("secret-fallback-detail") + primary.aclose.side_effect = primary_cancellation + fallback.aclose.side_effect = fallback_cancellation + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with ( + caplog.at_level(logging.WARNING, logger="weather_briefing.llm"), + pytest.raises(asyncio.CancelledError) as exc_info, + ): + await provider.aclose() + + assert exc_info.value is primary_cancellation + assert "cleanup was cancelled while preserving primary cancellation" in caplog.text + assert "secret-primary-detail" not in caplog.text + assert "secret-fallback-detail" not in caplog.text + + +async def test_fallback_close_cancellation_propagates() -> None: + primary = _provider() + fallback = _provider() + cancellation = asyncio.CancelledError() + fallback.aclose.side_effect = cancellation + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with pytest.raises(asyncio.CancelledError) as exc_info: + await provider.aclose() + + assert exc_info.value is cancellation + primary.aclose.assert_awaited_once_with() + + +async def test_fallback_cancellation_logs_primary_close_failure_type(caplog) -> None: + primary = _provider() + fallback = _provider() + cancellation = asyncio.CancelledError("private cancellation detail") + primary.aclose.side_effect = RuntimeError("private primary cleanup detail") + fallback.aclose.side_effect = cancellation + provider = FallbackLLMProvider( + primary, + fallback, + primary_name="primary", + fallback_name="fallback", + ) + + with ( + caplog.at_level(logging.WARNING, logger="weather_briefing.llm"), + pytest.raises(asyncio.CancelledError) as exc_info, + ): + await provider.aclose() + + assert exc_info.value is cancellation + assert "primary=primary fallback=fallback" in caplog.text + assert "error_type=RuntimeError" in caplog.text + assert "private primary cleanup detail" not in caplog.text + assert "private cancellation detail" not in caplog.text diff --git a/weather_briefing/composition/providers.py b/weather_briefing/composition/providers.py index c0ffd852..96d08bad 100644 --- a/weather_briefing/composition/providers.py +++ b/weather_briefing/composition/providers.py @@ -21,7 +21,7 @@ TelegramHTMLRenderer, TelegramPublisher, ) -from ..llm import AnyLLMStructuredProvider, SensitiveLLMDiagnostics, any_llm +from ..llm import CompleteLLMProvider, FallbackLLMProvider, SensitiveLLMDiagnostics, any_llm from ..models import ResolvedLocation from ..registries import LOCAL_WEATHER_CAPABILITY_PROVIDERS, PublisherName, WeatherProviderName from ..weather import ( @@ -45,9 +45,9 @@ def llm_provider( settings: Settings, diagnostics: SensitiveLLMDiagnostics | None = None, -) -> AnyLLMStructuredProvider: - """Build the configured any-llm adapter.""" - return any_llm.create_any_llm_provider( +) -> CompleteLLMProvider: + """Build the configured primary and optional fallback LLM adapters.""" + primary = any_llm.create_any_llm_provider( settings.llm_provider, settings.llm_model, settings.llm_max_output_tokens, @@ -55,6 +55,22 @@ def llm_provider( api_base=settings.llm_base_url, diagnostics=diagnostics, ) + if settings.llm_fallback_provider is None or settings.llm_fallback_model is None: + return primary + fallback = any_llm.create_any_llm_provider( + settings.llm_fallback_provider, + settings.llm_fallback_model, + settings.llm_max_output_tokens, + api_key=settings.llm_fallback_api_key, + api_base=settings.llm_fallback_base_url, + diagnostics=diagnostics, + ) + return FallbackLLMProvider( + primary, + fallback, + primary_name=settings.llm_provider, + fallback_name=settings.llm_fallback_provider, + ) def delivery_provider( diff --git a/weather_briefing/config/settings.py b/weather_briefing/config/settings.py index 3e3b1af3..cf6e3242 100644 --- a/weather_briefing/config/settings.py +++ b/weather_briefing/config/settings.py @@ -48,6 +48,10 @@ class Settings: llm_provider: str llm_model: str llm_base_url: str | None + llm_fallback_provider: str | None + llm_fallback_model: str | None + llm_fallback_api_key: str | None + llm_fallback_base_url: str | None llm_max_output_tokens: int llm_max_attempts: int http_timeout_seconds: float @@ -147,10 +151,7 @@ def from_env(cls) -> Settings: hourly_cron = cron_hour("BRIEFING_CRON", "9-23") service_status_cron = cron_expression("SERVICE_STATUS_CRON", "*/5 * * * *") llm_provider = clean_env(os.getenv("LLM_PROVIDER", "deepseek")) - if llm_provider not in AnyLLM.get_supported_providers(): - raise ConfigurationError(f"Unsupported LLM_PROVIDER: {llm_provider}") - if not AnyLLM.get_provider_class(llm_provider).SUPPORTS_COMPLETION: - raise ConfigurationError(f"LLM_PROVIDER does not support completion: {llm_provider}") + _validate_llm_provider("LLM_PROVIDER", llm_provider) llm_model = clean_env(os.getenv("LLM_MODEL")) if llm_provider == "deepseek": api_key = clean_env(os.getenv("DEEPSEEK_API_KEY")) or None @@ -161,6 +162,18 @@ def from_env(cls) -> Settings: llm_base_url = None if not llm_model: raise ConfigurationError("Missing required environment variable: LLM_MODEL") + llm_fallback_provider = clean_env(os.getenv("LLM_FALLBACK_PROVIDER")) or None + llm_fallback_model = clean_env(os.getenv("LLM_FALLBACK_MODEL")) or None + if (llm_fallback_provider is None) != (llm_fallback_model is None): + raise ConfigurationError("LLM_FALLBACK_PROVIDER and LLM_FALLBACK_MODEL must be configured together") + if llm_fallback_provider is not None: + _validate_llm_provider("LLM_FALLBACK_PROVIDER", llm_fallback_provider) + llm_fallback_api_key = clean_env(os.getenv("LLM_FALLBACK_API_KEY")) or None + llm_fallback_base_url = clean_env(os.getenv("LLM_FALLBACK_BASE_URL")) or None + if llm_fallback_api_key is not None and llm_fallback_provider is None: + raise ConfigurationError("LLM_FALLBACK_API_KEY requires LLM_FALLBACK_PROVIDER and LLM_FALLBACK_MODEL") + if llm_fallback_base_url is not None and llm_fallback_provider is None: + raise ConfigurationError("LLM_FALLBACK_BASE_URL requires LLM_FALLBACK_PROVIDER and LLM_FALLBACK_MODEL") locations = load_locations(locations_path) if weather_briefings_enabled else () location_ids = {location.id for location in locations} unknown_feed_locations = { @@ -224,6 +237,10 @@ def from_env(cls) -> Settings: llm_provider=llm_provider, llm_model=llm_model, llm_base_url=llm_base_url.rstrip("/") if llm_base_url else None, + llm_fallback_provider=llm_fallback_provider, + llm_fallback_model=llm_fallback_model, + llm_fallback_api_key=llm_fallback_api_key, + llm_fallback_base_url=llm_fallback_base_url.rstrip("/") if llm_fallback_base_url else None, llm_max_output_tokens=llm_max_output_tokens, llm_max_attempts=positive_integer("LLM_MAX_ATTEMPTS", 3), http_timeout_seconds=positive_float("HTTP_TIMEOUT_SECONDS", 30), @@ -272,3 +289,11 @@ def from_env(cls) -> Settings: service_status_cron=service_status_cron, debug=boolean("DEBUG", False), ) + + +def _validate_llm_provider(setting_name: str, provider: str) -> None: + """Require a known any-llm provider with completion support.""" + if provider not in AnyLLM.get_supported_providers(): + raise ConfigurationError(f"Unsupported {setting_name}: {provider}") + if not AnyLLM.get_provider_class(provider).SUPPORTS_COMPLETION: + raise ConfigurationError(f"{setting_name} does not support completion: {provider}") diff --git a/weather_briefing/llm/__init__.py b/weather_briefing/llm/__init__.py index e3a06492..a5e0c6b2 100644 --- a/weather_briefing/llm/__init__.py +++ b/weather_briefing/llm/__init__.py @@ -9,12 +9,15 @@ SensitiveLLMDiagnostics, serialize_llm_payload, ) +from .fallback import CompleteLLMProvider, FallbackLLMProvider from .lazy import LazyServiceStatusLLM from .result import parse_result from .schema import LLMStructuredOutput __all__ = [ "AnyLLMStructuredProvider", + "CompleteLLMProvider", + "FallbackLLMProvider", "LLMError", "LLMOutputLimitError", "LLMProvider", diff --git a/weather_briefing/llm/fallback.py b/weather_briefing/llm/fallback.py new file mode 100644 index 00000000..8d25f593 --- /dev/null +++ b/weather_briefing/llm/fallback.py @@ -0,0 +1,146 @@ +"""Request-failure fallback across two complete LLM providers.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable +from typing import Protocol, TypeVar + +from ..notifications import NotificationDecision +from .base import LLMRequestError + +_LOGGER = logging.getLogger("weather_briefing.llm") +_Result = TypeVar("_Result") + + +class CompleteLLMProvider(Protocol): + """Expose every LLM operation used by application composition.""" + + async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dict[str, object]: + """Return one structured briefing response.""" + ... + + async def assess_notification(self, payload: dict[str, object]) -> NotificationDecision: + """Return whether an official message change merits a notification.""" + ... + + async def translate_service_status( + self, + title: str, + body: str, + target_language: str, + ) -> tuple[str, str]: + """Translate one official service-status message.""" + ... + + async def aclose(self) -> None: + """Close owned resources.""" + ... + + +class FallbackLLMProvider: + """Switch permanently after a primary request failure; see docs/notes.md.""" + + def __init__( + self, + primary: CompleteLLMProvider, + fallback: CompleteLLMProvider, + *, + primary_name: str, + fallback_name: str, + ) -> None: + """Retain provider order and diagnostic names.""" + self._primary = primary + self._fallback = fallback + self._primary_name = primary_name + self._fallback_name = fallback_name + self._using_fallback = False + + async def _request( + self, + operation: str, + primary_call: Callable[[], Awaitable[_Result]], + fallback_call: Callable[[], Awaitable[_Result]], + ) -> _Result: + if self._using_fallback: + return await fallback_call() + try: + return await primary_call() + except LLMRequestError as exc: + self._using_fallback = True + error_type = type(exc.__cause__ or exc).__name__ + _LOGGER.warning( + "Primary LLM request failed; trying fallback operation=%s primary=%s fallback=%s error_type=%s", + operation, + self._primary_name, + self._fallback_name, + error_type, + ) + return await fallback_call() + + async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dict[str, object]: + """Generate a briefing, falling back only after a request failure.""" + return await self._request( + "summarize", + lambda: self._primary.summarize(system_prompt, payload), + lambda: self._fallback.summarize(system_prompt, payload), + ) + + async def assess_notification(self, payload: dict[str, object]) -> NotificationDecision: + """Assess notification value, falling back only after a request failure.""" + return await self._request( + "assess-notification", + lambda: self._primary.assess_notification(payload), + lambda: self._fallback.assess_notification(payload), + ) + + async def translate_service_status( + self, + title: str, + body: str, + target_language: str, + ) -> tuple[str, str]: + """Translate a status message, falling back only after a request failure.""" + return await self._request( + "translate-service-status", + lambda: self._primary.translate_service_status(title, body, target_language), + lambda: self._fallback.translate_service_status(title, body, target_language), + ) + + async def aclose(self) -> None: + """Close both providers without letting cleanup mask cancellation.""" + errors: list[Exception] = [] + try: + await self._primary.aclose() + except asyncio.CancelledError: + try: + await self._fallback.aclose() + except asyncio.CancelledError: + _LOGGER.warning("Fallback LLM provider cleanup was cancelled while preserving primary cancellation") + except Exception as exc: + _LOGGER.warning( + "Failed to close fallback LLM provider during cancellation error_type=%s", + type(exc).__name__, + ) + raise + except Exception as exc: + errors.append(exc) + try: + await self._fallback.aclose() + except asyncio.CancelledError: + if errors: + _LOGGER.warning( + "Fallback LLM provider cleanup was cancelled after primary close failure " + "primary=%s fallback=%s error_type=%s", + self._primary_name, + self._fallback_name, + type(errors[0]).__name__, + ) + raise + except Exception as exc: + errors.append(exc) + if len(errors) == 1: + raise errors[0] + if errors: + raise ExceptionGroup("Failed to close fallback LLM providers", errors)