diff --git a/backend/services/embedding.py b/backend/services/embedding.py index f3774868..f0326f7a 100644 --- a/backend/services/embedding.py +++ b/backend/services/embedding.py @@ -4,6 +4,7 @@ from core.config import settings from services.llm_provider_urls import build_llm_provider_http_client from services.exceptions import EmbeddingGenerationError +from services.retry import retry_transient STORAGE_EMBEDDING_DIMENSION = 1536 @@ -55,8 +56,11 @@ async def generate_embeddings( ) try: - response = await client.embeddings.create( - model=model or settings.OPENAI_EMBEDDING_MODEL, input=texts + response = await retry_transient( + lambda: client.embeddings.create( + model=model or settings.OPENAI_EMBEDDING_MODEL, input=texts + ), + operation_name="embedding generation", ) return [data.embedding for data in response.data] except openai.OpenAIError as e: diff --git a/backend/services/llm_service.py b/backend/services/llm_service.py index c4bbd93d..f86c463d 100644 --- a/backend/services/llm_service.py +++ b/backend/services/llm_service.py @@ -7,6 +7,7 @@ from openai import AsyncOpenAI from core.config import settings from core.exceptions import LLMServiceError +from services.retry import retry_transient from pydantic import BaseModel, Field from services.llm_provider_urls import build_llm_provider_http_client @@ -57,7 +58,8 @@ async def extract_todos_and_summary( ) selected_model = model or settings.OPENAI_MODEL try: - response = await client.beta.chat.completions.parse( + response = await retry_transient( + lambda: client.beta.chat.completions.parse( model=selected_model, messages=[ { @@ -71,6 +73,8 @@ async def extract_todos_and_summary( {"role": "user", "content": email_body}, ], response_format=ExtractionResult, + ), + operation_name="summary extraction", ) except Exception as e: logger.error(f"Error calling LLM API for extraction: {e}") @@ -127,10 +131,13 @@ async def translate_email_body( http_client=http_client, ) try: - response = await client.chat.completions.create( - model=selected_model, - messages=messages, - temperature=0.3, + response = await retry_transient( + lambda: client.chat.completions.create( + model=selected_model, + messages=messages, + temperature=0.3, + ), + operation_name="translation", ) except Exception as e: logger.error(f"Error calling LLM API for translation: {e}") @@ -186,9 +193,12 @@ async def draft_reply( http_client=http_client, ) try: - response = await client.chat.completions.create( - model=selected_model, - messages=messages, + response = await retry_transient( + lambda: client.chat.completions.create( + model=selected_model, + messages=messages, + ), + operation_name="reply drafting", ) except Exception as e: logger.error(f"Error calling LLM API for drafting: {e}") diff --git a/backend/services/retry.py b/backend/services/retry.py new file mode 100644 index 00000000..3c6870c7 --- /dev/null +++ b/backend/services/retry.py @@ -0,0 +1,67 @@ +"""Bounded retry for transient LLM/embedding provider failures. + +Stdlib-only (no new dependency): retries ONLY transient provider errors — +connection drops, timeouts, rate limits, and 5xx — with exponential backoff and +jitter. Auth, bad-request, and other 4xx errors fail immediately so callers keep +their existing fail-closed behavior. +""" + +import asyncio +import logging +import random +from collections.abc import Awaitable, Callable +from typing import TypeVar + +import openai + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + +# 429 and 5xx are worth one more attempt; 4xx auth/validation errors are not. +TRANSIENT_OPENAI_ERRORS: tuple[type[Exception], ...] = ( + openai.APIConnectionError, + openai.APITimeoutError, + openai.RateLimitError, + openai.InternalServerError, +) + +DEFAULT_PROVIDER_RETRIES = 2 +DEFAULT_BASE_DELAY_SECONDS = 0.5 +MAX_DELAY_SECONDS = 8.0 + + +async def retry_transient( + operation: Callable[[], Awaitable[T]], + *, + retries: int = DEFAULT_PROVIDER_RETRIES, + base_delay_seconds: float = DEFAULT_BASE_DELAY_SECONDS, + retryable: tuple[type[Exception], ...] = TRANSIENT_OPENAI_ERRORS, + operation_name: str = "provider call", +) -> T: + """Run ``operation``, retrying transient failures with backoff + jitter. + + ``retries`` is the number of retries after the first attempt (so the + operation runs at most ``retries + 1`` times). Non-retryable exceptions + propagate immediately; the last transient failure propagates unchanged + after the budget is exhausted. + """ + attempt = 0 + while True: + try: + return await operation() + except retryable as exc: + if attempt >= retries: + raise + delay = min(base_delay_seconds * (2**attempt), MAX_DELAY_SECONDS) + delay += random.uniform(0, delay / 2) + attempt += 1 + logger.warning( + "Transient %s failure (attempt %d/%d), retrying in %.2fs: %s", + operation_name, + attempt, + retries, + delay, + exc, + ) + await asyncio.sleep(delay) diff --git a/backend/tests/test_retry.py b/backend/tests/test_retry.py new file mode 100644 index 00000000..cc1c7025 --- /dev/null +++ b/backend/tests/test_retry.py @@ -0,0 +1,99 @@ +"""Tests for the transient provider retry helper.""" + +import asyncio + +import httpx +import openai +import pytest + +from services.retry import retry_transient + + +def _connection_error() -> openai.APIConnectionError: + return openai.APIConnectionError(request=httpx.Request("POST", "https://api.test")) + + +@pytest.mark.asyncio +async def test_returns_result_on_first_success(monkeypatch): + async def operation(): + return "ok" + + assert await retry_transient(operation) == "ok" + + +@pytest.mark.asyncio +async def test_retries_transient_error_then_succeeds(monkeypatch): + sleeps: list[float] = [] + + async def fake_sleep(delay: float) -> None: + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + attempts = 0 + + async def operation(): + nonlocal attempts + attempts += 1 + if attempts < 3: + raise _connection_error() + return "recovered" + + assert await retry_transient(operation, retries=2) == "recovered" + assert attempts == 3 + assert len(sleeps) == 2 + # Exponential: second delay window starts at 2x the first base. + assert sleeps[1] > sleeps[0] * 0.9 + + +@pytest.mark.asyncio +async def test_raises_after_retry_budget_exhausted(monkeypatch): + async def fake_sleep(delay: float) -> None: + return None + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + attempts = 0 + + async def operation(): + nonlocal attempts + attempts += 1 + raise _connection_error() + + with pytest.raises(openai.APIConnectionError): + await retry_transient(operation, retries=2) + assert attempts == 3 # initial + 2 retries + + +@pytest.mark.asyncio +async def test_non_transient_error_fails_immediately(): + attempts = 0 + + async def operation(): + nonlocal attempts + attempts += 1 + raise ValueError("bad request shape") + + with pytest.raises(ValueError): + await retry_transient(operation, retries=5) + assert attempts == 1 + + +@pytest.mark.asyncio +async def test_custom_retryable_tuple(monkeypatch): + async def fake_sleep(delay: float) -> None: + return None + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + attempts = 0 + + class Flaky(Exception): + pass + + async def operation(): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise Flaky() + return "done" + + assert await retry_transient(operation, retryable=(Flaky,)) == "done" + assert attempts == 2