diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e0bfbb..5716bae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,6 +86,9 @@ This file records what changes **in the product** – process and session state existing `auth` schema only when the app owns it. ### Fixed +- AI service: model calls answered with 429 or 503 (provider overload) are retried up to three attempts + in total with backoff instead of failing the extraction at once; all attempts together stay within + the model timeout, and timeouts are not retried (#69). - AI verifier: a unit quoted together with the neighbouring table cell (e.g. `60 | Stk.`) is now confirmed as `found` when one cell of the quote is exactly the unit; quotes that differ from the source in real characters are still rejected (#50). diff --git a/docs/technical/deployment-vercel.md b/docs/technical/deployment-vercel.md index a855dfb..fd9414f 100644 --- a/docs/technical/deployment-vercel.md +++ b/docs/technical/deployment-vercel.md @@ -61,6 +61,7 @@ creation). The script is all-or-nothing; running it twice fails on "role already | `AI_ALLOW_GEMINI_API_DEV` | `true` | | `GEMINI_API_KEY` | the orchestrator's key – the orchestrator pipes it in from a file they created themselves; never in a chat, the repo or an issue | | `AI_PDF_PIPELINE` / `AI_PDF_OCR` | `textlines` / `off` – no model downloads on the showcase | + | `AI_MODEL_TIMEOUT_SECONDS` | `45` – budget of one model call including retries; it must stay below the web app's `AI_SERVICE_TIMEOUT_MS` (60 s) minus parsing, or the worker gives up while the model still runs | `VERTEX_PROJECT` stays **unset**: together with `AI_ALLOW_GEMINI_API_DEV=true` the service refuses to start. 4. **Exception (ADR-0001 D11 amendment 2026-09-26, exceptions register):** the Gemini API free tier is only diff --git a/services/ai/README.md b/services/ai/README.md index cea9c4a..e08e535 100644 --- a/services/ai/README.md +++ b/services/ai/README.md @@ -45,7 +45,9 @@ Proof (the same commands as CI): | `VERTEX_PROJECT` | – (required) | GCP project for Vertex AI. | | `VERTEX_LOCATION` | `eu` | Vertex location (`eu` multi-region; a region such as `europe-west3` also works). | | `VERTEX_MODEL` | `gemini-3.5-flash` | Model ID. | -| `AI_MODEL_TIMEOUT_SECONDS` | `60` | Timeout of one model call. The SDK does not retry; retries belong to the worker (pg-boss). | +| `AI_MODEL_TIMEOUT_SECONDS` | `60` | Time budget of one model call, **retries included**: every attempt only gets what is left, so the call never runs longer. Keep it below the caller's `AI_SERVICE_TIMEOUT_MS` minus parsing time. | +| `AI_MODEL_RETRY_ATTEMPTS` | `3` | Attempts in total (1–5) when the model answers 429 or 503 (overload, rate limit). Timeouts, network and all other errors are not retried here – they fail the call, and the worker (pg-boss) retries the job later. A retry only starts with at least 10 s of budget left. | +| `AI_MODEL_RETRY_INITIAL_DELAY_SECONDS` | `1` | First wait before a retry (≤ 5); it doubles per attempt, capped at 8 s, plus up to 25 % jitter. | | `AI_ALLOW_GEMINI_API_DEV` | `false` | **Local development with synthetic data only.** Uses the Gemini API (free tier) instead of Vertex. Needs this flag **and** `GEMINI_API_KEY`, and `VERTEX_PROJECT` must be unset (both set → the service refuses to start). It is never used as a fallback, and a key alone changes nothing. | | `GEMINI_API_KEY` | – | Only read when the dev flag is `true`. | | `AI_PDF_PIPELINE` | `textlines` | `textlines` (model-free) or `layout` (docling layout model, see below). | diff --git a/services/ai/src/requestflow_ai/config.py b/services/ai/src/requestflow_ai/config.py index c9c40c2..899b035 100644 --- a/services/ai/src/requestflow_ai/config.py +++ b/services/ai/src/requestflow_ai/config.py @@ -24,7 +24,12 @@ class Settings(BaseSettings): vertex_project: str | None = None vertex_location: str = "eu" vertex_model: str = "gemini-3.5-flash" + # Time budget of one model call, retries included (#72): no attempt runs past it. ai_model_timeout_seconds: float = Field(default=60, gt=0) + # Transient provider errors (429/503) are retried inside that budget (#69): total attempts incl. + # the first one, exponential backoff from the initial delay (capped at 8 s per wait). + ai_model_retry_attempts: int = Field(default=3, ge=1, le=5) + ai_model_retry_initial_delay_seconds: float = Field(default=1.0, gt=0, le=5) # Gemini API (free tier) - local development with synthetic data only. Needs BOTH the flag # and the key; never used as a fallback for Vertex. diff --git a/services/ai/src/requestflow_ai/extraction/model_client.py b/services/ai/src/requestflow_ai/extraction/model_client.py index 7f21079..f0f2fb7 100644 --- a/services/ai/src/requestflow_ai/extraction/model_client.py +++ b/services/ai/src/requestflow_ai/extraction/model_client.py @@ -13,6 +13,8 @@ from __future__ import annotations import logging +import random +import time from collections.abc import Callable from dataclasses import dataclass from typing import Any, Protocol @@ -63,30 +65,74 @@ def model_id(self) -> str: ... def extract(self, system_instruction: str, user_content: str) -> ModelResponse: ... +@dataclass(frozen=True) +class RetryPolicy: + """Retries inside one model call (#69). Only overload and rate-limit answers are transient; a + timeout or network error is not retried, because it already used the budget. All attempts + together stay within ``budget_seconds`` (the model timeout), each one only gets what is left.""" + + budget_seconds: float + attempts: int = 1 + initial_delay_seconds: float = 1.0 + + +_RETRYABLE_STATUS = frozenset({429, 503}) +_MAX_DELAY_SECONDS = 8.0 +# A retry needs room for a whole answer (one extraction takes about 8 s, measured in #67). +_MIN_ATTEMPT_SECONDS = 10.0 + + class GeminiModelClient: - def __init__(self, client: Client, model: str) -> None: + def __init__( + self, + client: Client, + model: str, + retry: RetryPolicy, + *, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + ) -> None: self._client = client self._model = model + self._retry = retry + self._clock = clock + self._sleep = sleep @property def model_id(self) -> str: return self._model def extract(self, system_instruction: str, user_content: str) -> ModelResponse: - config = types.GenerateContentConfig( - system_instruction=system_instruction, - response_mime_type="application/json", - response_schema=ModelExtraction, - temperature=0, - candidate_count=1, - automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), - ) - try: - response = self._client.models.generate_content( - model=self._model, contents=user_content, config=config - ) - except (errors.APIError, httpx.HTTPError, GoogleAuthError) as exc: - raise ModelClientError(type(exc).__name__) from exc + deadline = self._clock() + self._retry.budget_seconds + delay = self._retry.initial_delay_seconds + attempt = 1 + while True: + remaining = deadline - self._clock() + try: + response = self._client.models.generate_content( + model=self._model, + contents=user_content, + config=self._config(system_instruction, remaining), + ) + break + except errors.APIError as exc: + wait = min(delay, _MAX_DELAY_SECONDS) * (1 + random.random() / 4) # noqa: S311 - jitter, not security + left = deadline - self._clock() - wait + if ( + exc.code not in _RETRYABLE_STATUS + or attempt >= self._retry.attempts + or left < _MIN_ATTEMPT_SECONDS + ): + raise ModelClientError(type(exc).__name__) from exc + _log.warning( + "model_call_retry", + extra={"attempt": attempt, "status": exc.code, "waitMs": round(wait * 1000)}, + ) + self._sleep(wait) + delay *= 2 + attempt += 1 + except (httpx.HTTPError, GoogleAuthError) as exc: + raise ModelClientError(type(exc).__name__) from exc text = response.text if not text: @@ -108,6 +154,20 @@ def extract(self, system_instruction: str, user_content: str) -> ModelResponse: model_version=response.model_version, ) + @staticmethod + def _config(system_instruction: str, remaining_seconds: float) -> types.GenerateContentConfig: + return types.GenerateContentConfig( + system_instruction=system_instruction, + response_mime_type="application/json", + response_schema=ModelExtraction, + temperature=0, + candidate_count=1, + automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), + # This attempt gets only what is left of the budget (at least 1 s; the SDK needs a + # timeout). + http_options=types.HttpOptions(timeout=max(1000, int(remaining_seconds * 1000))), + ) + CredentialsLoader = Callable[..., tuple[Any, str | None]] @@ -120,9 +180,14 @@ def build_model_client( credentials_loader: CredentialsLoader = google.auth.default, ) -> ModelClient: """Create the configured model client or raise ``ModelClientInitError`` (fail-closed).""" + # No SDK retry: it would also repeat timeouts outside the budget (#72 review); see RetryPolicy. http_options = types.HttpOptions( - timeout=int(settings.ai_model_timeout_seconds * 1000), - httpx_client=httpx_client, + timeout=int(settings.ai_model_timeout_seconds * 1000), httpx_client=httpx_client + ) + retry = RetryPolicy( + budget_seconds=settings.ai_model_timeout_seconds, + attempts=settings.ai_model_retry_attempts, + initial_delay_seconds=settings.ai_model_retry_initial_delay_seconds, ) if settings.ai_allow_gemini_api_dev and settings.vertex_project: @@ -145,7 +210,7 @@ def build_model_client( client = Client(vertexai=False, api_key=key, http_options=http_options) except Exception as exc: raise ModelClientInitError("Gemini API client init failed") from exc - return GeminiModelClient(client, settings.vertex_model) + return GeminiModelClient(client, settings.vertex_model, retry) if not settings.vertex_project: raise ModelClientInitError( @@ -172,4 +237,4 @@ def build_model_client( # Defence in depth: an API key from the environment must never switch Vertex to key mode. if not client.vertexai or getattr(client._api_client, "api_key", None): raise ModelClientInitError("Vertex AI client resolved to API-key mode; refusing to start") - return GeminiModelClient(client, settings.vertex_model) + return GeminiModelClient(client, settings.vertex_model, retry) diff --git a/services/ai/tests/test_model_retry.py b/services/ai/tests/test_model_retry.py new file mode 100644 index 0000000..ce22bf2 --- /dev/null +++ b/services/ai/tests/test_model_retry.py @@ -0,0 +1,142 @@ +"""Transient model errors are retried inside one extraction call (#69), replayed over HTTP.""" + +from __future__ import annotations + +from typing import Any + +import httpx +import pytest +from conftest import fake_credentials, make_settings, no_adc, recorded + +from requestflow_ai.extraction.model_client import ( + ModelClient, + ModelClientError, + build_model_client, +) + +FAST = {"ai_model_retry_initial_delay_seconds": 0.01} + + +def _error(status: int) -> dict[str, Any]: + return {"error": {"code": status, "message": "synthetic upstream error"}} + + +class Sequence: + """Answers with the given (status, body) pairs in order; the last pair repeats.""" + + def __init__(self, *responses: tuple[int, dict[str, Any]]) -> None: + self._responses = list(responses) + self.requests: list[httpx.Request] = [] + + def handler(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + status, body = self._responses.pop(0) if len(self._responses) > 1 else self._responses[0] + return httpx.Response(status, json=body) + + def client(self) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(self.handler)) + + +def _vertex(sequence: Sequence, **overrides: Any) -> ModelClient: + return build_model_client( + make_settings(**FAST, **overrides), + credentials=fake_credentials(), + httpx_client=sequence.client(), + credentials_loader=no_adc, + ) + + +def test_a_503_is_retried_and_the_call_succeeds() -> None: + sequence = Sequence((503, _error(503)), (200, recorded("musterbau_pdf.json"))) + + _vertex(sequence).extract("S", "U") + + assert len(sequence.requests) == 2 + + +def test_a_429_is_retried() -> None: + sequence = Sequence((429, _error(429)), (200, recorded("musterbau_pdf.json"))) + + _vertex(sequence).extract("S", "U") + + assert len(sequence.requests) == 2 + + +def test_a_persistent_503_gives_up_after_three_attempts() -> None: + sequence = Sequence((503, _error(503))) + + with pytest.raises(ModelClientError): + _vertex(sequence).extract("S", "U") + + assert len(sequence.requests) == 3 + + +@pytest.mark.parametrize("status", [400, 401, 403, 404, 500]) +def test_other_errors_are_not_retried(status: int) -> None: + sequence = Sequence((status, _error(status))) + + with pytest.raises(ModelClientError): + _vertex(sequence).extract("S", "U") + + assert len(sequence.requests) == 1 + + +def test_a_timeout_is_not_retried() -> None: + # A hanging model already used the whole budget; another attempt would outlive the worker + # (#72 review). + requests: list[httpx.Request] = [] + + def hang(request: httpx.Request) -> httpx.Response: + requests.append(request) + raise httpx.ReadTimeout("synthetic timeout", request=request) + + client = build_model_client( + make_settings(**FAST), + credentials=fake_credentials(), + httpx_client=httpx.Client(transport=httpx.MockTransport(hang)), + credentials_loader=no_adc, + ) + + with pytest.raises(ModelClientError): + client.extract("S", "U") + + assert len(requests) == 1 + + +def test_no_retry_without_enough_budget_left() -> None: + # Budget below the minimum a retry needs: a 503 fails at once instead of eating the worker's + # time. + sequence = Sequence((503, _error(503)), (200, recorded("musterbau_pdf.json"))) + + with pytest.raises(ModelClientError): + _vertex(sequence, ai_model_timeout_seconds=5).extract("S", "U") + + assert len(sequence.requests) == 1 + + +def test_each_attempt_gets_only_the_remaining_budget() -> None: + sequence = Sequence((503, _error(503)), (200, recorded("musterbau_pdf.json"))) + + _vertex(sequence, ai_model_timeout_seconds=30).extract("S", "U") + + first, second = (request.extensions["timeout"]["read"] for request in sequence.requests) + assert first <= 30 + assert second < first + + +def test_the_gemini_api_dev_path_retries_too() -> None: + sequence = Sequence((503, _error(503)), (200, recorded("musterbau_pdf.json"))) + client = build_model_client( + make_settings( + **FAST, + ai_allow_gemini_api_dev=True, + gemini_api_key="synthetic-dev-key", + vertex_project=None, + ), + httpx_client=sequence.client(), + credentials_loader=no_adc, + ) + + client.extract("S", "U") + + assert len(sequence.requests) == 2