diff --git a/docs/notes.md b/docs/notes.md index 3b7f4265..02aa284f 100644 --- a/docs/notes.md +++ b/docs/notes.md @@ -58,6 +58,8 @@ 这是一个有意保留的自定义外部服务集成。选择 `FallbackLLMProvider` 的原因是 any-llm 只统一调用单个 provider，不编排跨 provider 的故障切换。包装器捕获主适配器的 `LLMRequestError`，切换后在剩余生命周期内固定使用备用适配器，使同一进程里的契约修复不会回到刚刚失败的主服务。 +适配器把应用工厂创建的 AnyLLM provider client 在 completion 调用中抛出的异常归一化为 `LLMRequestError`，使配置正确的 fallback 能够启动。请求消息在进入该边界前完成序列化，因此应用自身的序列化错误不会触发 fallback；结构化输出的 Pydantic 校验错误和直接注入协议实现所抛出的异常继续原样传播。 + 它替代的是每个调用点手写的故障切换分支，不替代 any-llm 的厂商适配器，也不接管 SDK 凭据、请求重试或输出验证。 这个选择成立的条件是一次主服务请求失败足以让当前模型对象的后续调用继续使用备用服务，恢复主服务交给下次创建模型对象。如果 any-llm 提供可观察的跨 provider 路由，或者常驻进程需要在不重启的情况下探测并恢复主服务，就重新评估并用带健康状态和冷却时间的路由替换当前粘性开关。 diff --git a/tests/test_any_llm_provider.py b/tests/test_any_llm_provider.py index 91808ccd..859b67b6 100644 --- a/tests/test_any_llm_provider.py +++ b/tests/test_any_llm_provider.py @@ -1,20 +1,24 @@ import json import logging -from collections.abc import Mapping +import os +from collections.abc import Callable, Mapping from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import httpx import pytest +from anthropic import BadRequestError as AnthropicBadRequestError from any_llm import AnyLLM from any_llm.providers.openai.base import BaseOpenAIProvider -from openai import AsyncOpenAI -from pydantic import BaseModel +from openai import AsyncOpenAI, BadRequestError +from pydantic import BaseModel, ValidationError from weather_briefing.api_client import LoggedAsyncClient from weather_briefing.llm import ( AnyLLMStructuredProvider, + FallbackLLMProvider, LazyServiceStatusLLM, + LLMRequestError, LLMStructuredOutput, create_any_llm_provider, ) @@ -48,6 +52,28 @@ async def acompletion( return self._response +def _openai_bad_request(response: httpx.Response) -> Exception: + return BadRequestError("Upstream request failed", response=response, body={"error": "upstream"}) + + +def _anthropic_bad_request(response: httpx.Response) -> Exception: + return AnthropicBadRequestError("Upstream request failed", response=response, body={"error": "upstream"}) + + +def _httpx_connection_error(response: httpx.Response) -> Exception: + return httpx.ConnectError("Upstream connection failed", request=response.request) + + +class _ProviderStatusError(Exception): + def __init__(self, response: httpx.Response) -> None: + super().__init__("Upstream request failed") + self.response = response + + +def _provider_status_error(response: httpx.Response) -> Exception: + return _ProviderStatusError(response) + + async def test_service_status_llm_is_created_only_on_first_operation() -> None: provider = AsyncMock() provider.assess_notification.return_value = NotificationDecision(True) @@ -176,6 +202,167 @@ async def test_any_llm_provider_assesses_notification_value_with_a_narrow_schema assert client.calls[0]["max_tokens"] == 256 +@pytest.mark.parametrize( + ("provider_name", "error_factory", "operation", "args", "message"), + ( + ( + "openai", + _openai_bad_request, + "summarize", + ("Return JSON", {"input": "data"}), + "LLM request failed", + ), + ( + "anthropic", + _anthropic_bad_request, + "summarize", + ("Return JSON", {"input": "data"}), + "LLM request failed", + ), + ( + "openrouter", + _httpx_connection_error, + "summarize", + ("Return JSON", {"input": "data"}), + "LLM request failed", + ), + ( + "gemini", + _provider_status_error, + "summarize", + ("Return JSON", {"input": "data"}), + "LLM request failed", + ), + ( + "openai", + _openai_bad_request, + "assess_notification", + ({"current": {"status": "operational"}},), + "LLM notification decision request failed", + ), + ( + "openai", + _openai_bad_request, + "translate_service_status", + ("Incident", "Elevated errors", "en"), + "LLM translation request failed", + ), + ), +) +async def test_factory_normalizes_provider_native_request_errors( + monkeypatch, + provider_name: str, + error_factory: Callable[[httpx.Response], Exception], + operation: str, + args: tuple[object, ...], + message: str, +) -> None: + request = httpx.Request("POST", "https://api.example.invalid/chat/completions") + response = httpx.Response(400, request=request) + error = error_factory(response) + client = AsyncMock(spec=AnyLLM) + client.acompletion.side_effect = error + monkeypatch.delenv("ANY_LLM_UNIFIED_EXCEPTIONS", raising=False) + monkeypatch.setattr(AnyLLM, "create", lambda *args, **kwargs: client) + provider = create_any_llm_provider(provider_name, "requested-model", 4096) + + with pytest.raises(LLMRequestError, match=f"^{message}$") as exc_info: + await getattr(provider, operation)(*args) + + assert exc_info.value.__cause__ is error + assert "ANY_LLM_UNIFIED_EXCEPTIONS" not in os.environ + + +async def test_factory_preserves_sdk_output_validation_errors(monkeypatch) -> None: + with pytest.raises(ValidationError) as validation: + LLMStructuredOutput.model_validate({}) + client = AsyncMock(spec=AnyLLM) + client.acompletion.side_effect = validation.value + monkeypatch.setattr(AnyLLM, "create", lambda *args, **kwargs: client) + provider = create_any_llm_provider("openai", "requested-model", 4096) + + with pytest.raises(ValidationError) as propagated: + await provider.summarize("Return JSON", {"input": "data"}) + + assert propagated.value is validation.value + + +async def test_any_llm_client_does_not_mask_payload_serialization_errors() -> None: + client = AsyncMock(spec=AnyLLM) + provider = AnyLLMStructuredProvider( + client, + provider="openai", + model="requested-model", + max_output_tokens=4096, + ) + + with pytest.raises(TypeError, match="not JSON serializable"): + await provider.summarize("Return JSON", {"input": object()}) + + client.acompletion.assert_not_awaited() + + +async def test_protocol_client_preserves_completion_errors() -> None: + error = RuntimeError("Injected client failed") + client = AsyncMock() + client.acompletion.side_effect = error + provider = AnyLLMStructuredProvider( + client, + provider="wrapped-provider", + model="requested-model", + max_output_tokens=4096, + ) + + with pytest.raises(RuntimeError, match="Injected client failed") as exc_info: + await provider.summarize("Return JSON", {"input": "data"}) + + assert exc_info.value is error + + +async def test_provider_native_request_error_switches_to_fallback(monkeypatch) -> None: + request = httpx.Request("POST", "https://api.example.invalid/chat/completions") + response = httpx.Response(400, request=request) + error = BadRequestError( + "Upstream request failed", + response=response, + body={"error": "upstream"}, + ) + primary_client = AsyncMock(spec=AnyLLM) + primary_client.acompletion.side_effect = error + monkeypatch.delenv("ANY_LLM_UNIFIED_EXCEPTIONS", raising=False) + monkeypatch.setattr(AnyLLM, "create", lambda *args, **kwargs: primary_client) + fallback_result = { + "headline": "Fallback briefing", + "headline_source_ids": ["source"], + "conclusions": [], + "active_warnings": [], + "resolved_warning_ids": [], + "advice": [], + "disaster_tracking": [], + "should_publish": True, + } + fallback_client = _CompletionClientStub( + SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content=json.dumps(fallback_result)))]) + ) + provider = FallbackLLMProvider( + create_any_llm_provider("openai", "primary-model", 4096), + AnyLLMStructuredProvider( + fallback_client, + provider="openai", + model="fallback-model", + max_output_tokens=4096, + ), + primary_name="openai/primary-model", + fallback_name="openai/fallback-model", + ) + + result = await provider.summarize("Return JSON", {"input": "data"}) + + assert result == fallback_result + primary_client.acompletion.assert_awaited_once() + assert len(fallback_client.calls) == 1 + + async def test_factory_accepts_every_any_llm_completion_provider(monkeypatch) -> None: created: list[tuple[str, dict[str, object]]] = [] diff --git a/tests/test_llm.py b/tests/test_llm.py index e940b577..900bb827 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -2,7 +2,6 @@ import logging from copy import deepcopy from types import SimpleNamespace -from typing import Any import pendulum import pytest @@ -80,7 +79,7 @@ async def test_completion_client_stub_requires_a_configured_response() -> None: ) -def _valid_payload() -> dict[str, Any]: +def _valid_payload() -> dict[str, object]: return { "headline": "Briefing", "headline_source_ids": ["source"], diff --git a/tests/test_llm_fallback.py b/tests/test_llm_fallback.py index 04ddb052..096b04d7 100644 --- a/tests/test_llm_fallback.py +++ b/tests/test_llm_fallback.py @@ -163,8 +163,14 @@ async def test_output_contract_failure_does_not_use_fallback() -> None: 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") + primary_cause = TimeoutError("primary unavailable") + primary_error = LLMRequestError("primary request failed") + primary_error.__cause__ = primary_cause + fallback_cause = ConnectionError("fallback unavailable") + fallback_error = LLMRequestError("fallback request failed") + fallback_error.__cause__ = fallback_cause + primary.summarize.side_effect = primary_error + fallback.summarize.side_effect = fallback_error provider = FallbackLLMProvider( primary, fallback, @@ -172,11 +178,13 @@ async def test_fallback_failure_preserves_primary_as_context() -> None: fallback_name="fallback", ) - with pytest.raises(LLMRequestError, match="fallback unavailable") as exc_info: + with pytest.raises(LLMRequestError, match="fallback request failed") as exc_info: await provider.summarize("system", {"input": "value"}) + assert exc_info.value.__cause__ is fallback_cause assert isinstance(exc_info.value.__context__, LLMRequestError) - assert str(exc_info.value.__context__) == "primary unavailable" + assert str(exc_info.value.__context__) == "primary request failed" + assert exc_info.value.__context__.__cause__ is primary_cause async def test_fallback_log_excludes_exception_details(caplog) -> None: diff --git a/weather_briefing/llm/any_llm.py b/weather_briefing/llm/any_llm.py index e00cd40a..0afd445f 100644 --- a/weather_briefing/llm/any_llm.py +++ b/weather_briefing/llm/any_llm.py @@ -3,13 +3,14 @@ from __future__ import annotations import logging -from collections.abc import Mapping +from collections.abc import Iterator, Mapping +from contextlib import contextmanager from inspect import isawaitable from typing import Protocol from any_llm import AnyLLM from any_llm.exceptions import AnyLLMError, LengthFinishReasonError -from pydantic import BaseModel +from pydantic import BaseModel, ValidationError from ..api_client import api_call_context from ..data.any_llm_compatibility import UNSUPPORTED_DEFAULT_HEADER_PROVIDERS @@ -44,6 +45,25 @@ async def acompletion( ... +@contextmanager +def _normalize_request_errors( + message: str, + *, + normalize_completion_errors: bool, +) -> Iterator[None]: + """Normalize failures escaping an application-owned completion client.""" + try: + yield + except (LengthFinishReasonError, ValidationError): + raise + except AnyLLMError as exc: + raise LLMRequestError(message) from exc + except Exception as exc: + if normalize_completion_errors: + raise LLMRequestError(message) from exc + raise + + class AnyLLMStructuredProvider: """Adapt an any-llm provider to the application's structured LLM boundary.""" @@ -56,6 +76,7 @@ def __init__( max_output_tokens: int, diagnostics: SensitiveLLMDiagnostics | None = None, owns_client: bool = False, + normalize_completion_errors: bool = False, ) -> None: """Configure a reusable any-llm client and output limit.""" self._client = client @@ -64,12 +85,37 @@ def __init__( self._max_output_tokens = max_output_tokens self._diagnostics = diagnostics self._owns_client = owns_client + self._normalize_completion_errors = normalize_completion_errors @property def provider(self) -> str: """Return the application-facing provider name used for diagnostics.""" return self._provider + async def _complete( + self, + messages: list[dict[str, str]], + *, + response_format: type[BaseModel], + temperature: float, + max_tokens: int, + request_error_message: str, + ) -> object: + with ( + api_call_context(self._provider, "chat-completions"), + _normalize_request_errors( + request_error_message, + normalize_completion_errors=self._normalize_completion_errors, + ), + ): + return await self._client.acompletion( + model=self._model, + messages=[*messages], + response_format=response_format, + temperature=temperature, + max_tokens=max_tokens, + ) + async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dict[str, object]: """Request and decode one structured JSON response.""" log_sensitive = _sensitive_llm_diagnostics_enabled(self._diagnostics) @@ -87,18 +133,18 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic system_prompt, payload, ) + messages: list[dict[str, str]] = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": serialize_llm_payload(payload)}, + ] try: - with api_call_context(self._provider, "chat-completions"): - response = await self._client.acompletion( - model=self._model, - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": serialize_llm_payload(payload)}, - ], - response_format=LLMStructuredOutput, - temperature=0.2, - max_tokens=self._max_output_tokens, - ) + response = await self._complete( + messages, + response_format=LLMStructuredOutput, + temperature=0.2, + max_tokens=self._max_output_tokens, + request_error_message="LLM request failed", + ) except LengthFinishReasonError as exc: _LOGGER.warning( "LLM response reached output token limit: provider=%s model=%r max_output_tokens=%d error_type=%s", @@ -108,8 +154,6 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic type(exc).__name__, ) raise LLMOutputLimitError("LLM response reached output token limit") from exc - except AnyLLMError as exc: - raise LLMRequestError("LLM request failed") from exc result_payload = decode_structured_response(response).model_dump(mode="json") if log_sensitive: _LOGGER.debug( @@ -122,21 +166,21 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic async def assess_notification(self, payload: dict[str, object]) -> NotificationDecision: """Evaluate notification value independently from content generation.""" + messages: list[dict[str, str]] = [ + { + "role": "system", + "content": (f"{NOTIFICATION_POLICY}\n根据输入返回 should_notify。只返回请求的 JSON 对象。"), + }, + {"role": "user", "content": serialize_llm_payload(payload)}, + ] try: - with api_call_context(self._provider, "chat-completions"): - response = await self._client.acompletion( - model=self._model, - messages=[ - { - "role": "system", - "content": (f"{NOTIFICATION_POLICY}\n根据输入返回 should_notify。只返回请求的 JSON 对象。"), - }, - {"role": "user", "content": serialize_llm_payload(payload)}, - ], - response_format=NotificationDecisionOutput, - temperature=0.0, - max_tokens=min(self._max_output_tokens, 256), - ) + response = await self._complete( + messages, + response_format=NotificationDecisionOutput, + temperature=0.0, + max_tokens=min(self._max_output_tokens, 256), + request_error_message="LLM notification decision request failed", + ) except LengthFinishReasonError as exc: _LOGGER.warning( "LLM notification decision reached output token limit: " @@ -147,8 +191,6 @@ async def assess_notification(self, payload: dict[str, object]) -> NotificationD type(exc).__name__, ) raise LLMOutputLimitError("LLM notification decision reached output token limit") from exc - except AnyLLMError as exc: - raise LLMRequestError("LLM notification decision request failed") from exc return NotificationDecision(should_notify=decode_notification_decision(response)) async def translate_service_status( @@ -165,28 +207,28 @@ async def translate_service_status( }.get(target_language) if language_name is None: raise ValueError(f"Unsupported service-status translation language: {target_language}") + messages: list[dict[str, str]] = [ + { + "role": "system", + "content": ( + f"Translate the official service-incident explanation into concise {language_name}. " + "Preserve product names, incident facts, status, times, and technical terms. " + "Do not add analysis, advice, or facts. Return only the requested JSON object." + ), + }, + { + "role": "user", + "content": serialize_llm_payload({"title": title, "body": body}), + }, + ] try: - with api_call_context(self._provider, "chat-completions"): - response = await self._client.acompletion( - model=self._model, - messages=[ - { - "role": "system", - "content": ( - f"Translate the official service-incident explanation into concise {language_name}. " - "Preserve product names, incident facts, status, times, and technical terms. " - "Do not add analysis, advice, or facts. Return only the requested JSON object." - ), - }, - { - "role": "user", - "content": serialize_llm_payload({"title": title, "body": body}), - }, - ], - response_format=ServiceStatusTranslationOutput, - temperature=0.0, - max_tokens=min(self._max_output_tokens, 2048), - ) + response = await self._complete( + messages, + response_format=ServiceStatusTranslationOutput, + temperature=0.0, + max_tokens=min(self._max_output_tokens, 2048), + request_error_message="LLM translation request failed", + ) except LengthFinishReasonError as exc: _LOGGER.warning( "LLM translation reached output token limit: provider=%s model=%r max_output_tokens=%d error_type=%s", @@ -196,8 +238,6 @@ async def translate_service_status( type(exc).__name__, ) raise LLMOutputLimitError("LLM translation reached output token limit") from exc - except AnyLLMError as exc: - raise LLMRequestError("LLM translation request failed") from exc translated = decode_service_status_translation(response) return translated.title, translated.body @@ -282,4 +322,5 @@ def create_any_llm_provider( max_output_tokens=max_output_tokens, diagnostics=diagnostics, owns_client=True, + normalize_completion_errors=True, )