Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down
1 change: 1 addition & 0 deletions docs/technical/deployment-vercel.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion services/ai/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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). |
Expand Down
5 changes: 5 additions & 0 deletions services/ai/src/requestflow_ai/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
103 changes: 84 additions & 19 deletions services/ai/src/requestflow_ai/extraction/model_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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]]

Expand All @@ -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:
Expand All @@ -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(
Expand All @@ -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)
142 changes: 142 additions & 0 deletions services/ai/tests/test_model_retry.py
Original file line number Diff line number Diff line change
@@ -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
Loading