Structuring OpenAI and Gemini Calls in FastAPI: Prompts, Fallbacks, and Retries

The fallback was supposed to be temporary. When the OpenAI API was unavailable, the service would return the last cached result for that user rather than failing the request. Clean degraded experience, no errors surfaced to the client.

What nobody noticed: the cache TTL was set to 7 days. When the API came back up, the service kept serving cached results for users whose inputs had changed — because the cache key was too coarse and the invalidation logic had a bug. No errors, no alerts. Clients were seeing stale AI-generated content for days.

The fallback worked exactly as coded. The design was wrong.

Building LLM integrations that hold up in production means getting three things right: how you call the model, how you retry when it fails, and what you actually do when it's unavailable. All three are worth more thought than they usually get.

---

Structuring the LLM Client

Don't scatter openai.chat.completions.create() calls across your service layer. Wrap the client once, in one place, with all the configuration and error handling it needs:

# services/llm_client.py
from openai import AsyncOpenAI
from config import settings
import logging

logger = logging.getLogger(__name__)

client = AsyncOpenAI(
    api_key=settings.openai_api_key,
    timeout=30.0,        # hard timeout — don't let slow responses hang indefinitely
    max_retries=0,       # handle retries yourself, not via the SDK's built-in retry
)

async def call_model(
    prompt: str,
    system: str,
    model: str = "gpt-4o-mini",
    max_tokens: int = 1000,
    temperature: float = 0.2,
) -> str:
    response = await client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
        max_tokens=max_tokens,
        temperature=temperature,
    )
    content = response.choices[0].message.content
    logger.info(
        "LLM call complete",
        extra={
            "model": model,
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
        }
    )
    return content

max_retries=0 is intentional — the SDK's built-in retry is a black box. You want explicit control over when, how many times, and with what backoff you retry. More on that below.

timeout=30.0 is a hard limit on how long a single API call can take. Without it, a slow model response holds your Cloud Run instance's concurrency slot indefinitely.

For Gemini, the structure is the same pattern with a different client:

import google.generativeai as genai
from config import settings

genai.configure(api_key=settings.gemini_api_key)
gemini_model = genai.GenerativeModel("gemini-1.5-flash")

async def call_gemini(prompt: str, system: str) -> str:
    response = await gemini_model.generate_content_async(
        f"{system}\n\n{prompt}"
    )
    return response.text

The key difference: Gemini doesn't have a separate system role in the same way — you prepend system instructions to the user prompt. If you're abstracting across both providers, account for this in your prompt builder.

---

Prompt Management

Prompts hardcoded as string constants in service files are a maintenance problem. They can't be iterated without a code deployment. They can't be versioned or rolled back independently. They can't be tested in isolation.

Store prompts as versioned config, loaded at startup:

# prompts/store.py
import json
from pathlib import Path
from config import settings

_prompts: dict = {}

def load_prompts():
    prompt_file = Path(f"prompts/{settings.environment}.json")
    with open(prompt_file) as f:
        global _prompts
        _prompts = json.load(f)

def get_prompt(key: str, version: str = "latest") -> dict:
    prompt_versions = _prompts.get(key, {})
    if version == "latest":
        version = max(prompt_versions.keys())
    return prompt_versions[version]

// prompts/prod.json
{
  "document_extraction": {
    "v1": {
      "system": "You are a document extraction assistant. Extract structured data from the provided document. Return only valid JSON matching the schema provided.",
      "user_template": "Extract the following fields from this document:\n{fields}\n\nDocument:\n{document}"
    },
    "v2": {
      "system": "You are a precise document extraction assistant. Extract only the explicitly stated fields. If a field is not present, return null. Return only valid JSON.",
      "user_template": "Extract these fields: {fields}\n\nDocument content:\n{document}\n\nReturn JSON only, no explanation."
    }
  }
}

# Usage in service
from prompts.store import get_prompt

prompt_config = get_prompt("document_extraction", version="v2")
system = prompt_config["system"]
user = prompt_config["user_template"].format(
    fields=", ".join(required_fields),
    document=document_text
)
result = await call_model(prompt=user, system=system)

Rolling back a bad prompt is now a config change, not a code deployment. A/B testing two prompt versions is straightforward. The service code doesn't change when prompts do.

---

Output Validation

Model output is untrusted input. Validate it before using it:

# services/extraction.py
from pydantic import BaseModel, ValidationError
from services.llm_client import call_model
from prompts.store import get_prompt
import json
import logging

logger = logging.getLogger(__name__)

class ExtractionResult(BaseModel):
    invoice_number: str
    total_amount: float
    vendor_name: str
    date: str
    line_items: list[dict]

async def extract_document(document_text: str) -> ExtractionResult:
    prompt_config = get_prompt("document_extraction")
    raw = await call_model(
        prompt=prompt_config["user_template"].format(document=document_text),
        system=prompt_config["system"],
    )

    # Strip markdown fences if model wraps output in ```json ... ```
    clean = raw.strip().removeprefix("```json").removesuffix("```").strip()

    try:
        data = json.loads(clean)
        return ExtractionResult(**data)
    except (json.JSONDecodeError, ValidationError) as e:
        logger.error("Output validation failed", extra={"raw_output": raw, "error": str(e)})
        raise ValueError(f"Model returned unparseable output: {e}")

The removeprefix/removesuffix for markdown fences is not optional — models frequently wrap JSON output in code fences even when instructed not to. Strip them before parsing.

---

Retry Logic

Retries on LLM APIs need explicit backoff. The failure modes worth retrying are rate limit errors (429) and server errors (500, 503). Don't retry on invalid request errors (400) or auth errors (401) — those won't resolve with a retry.

# services/llm_client.py (extended)
import asyncio
from openai import RateLimitError, APIStatusError

async def call_model_with_retry(
    prompt: str,
    system: str,
    max_attempts: int = 3,
    base_delay: float = 1.0,
    **kwargs,
) -> str:
    last_error = None

    for attempt in range(max_attempts):
        try:
            return await call_model(prompt, system, **kwargs)

        except RateLimitError as e:
            last_error = e
            delay = base_delay * (2 ** attempt)   # exponential backoff
            logger.warning(f"Rate limited, retrying in {delay}s (attempt {attempt + 1}/{max_attempts})")
            await asyncio.sleep(delay)

        except APIStatusError as e:
            if e.status_code in (500, 503):
                last_error = e
                delay = base_delay * (2 ** attempt)
                logger.warning(f"API error {e.status_code}, retrying in {delay}s")
                await asyncio.sleep(delay)
            else:
                raise   # don't retry 400, 401, 422

    raise last_error

Cap retries at 3. Add jitter if you're making many concurrent calls (multiply delay by random.uniform(0.8, 1.2)) to avoid thundering herd when multiple instances hit a rate limit simultaneously.

---

Fallbacks That Don't Silently Mislead

The opening story failed because the fallback served data that was no longer valid without telling anyone. A good fallback is explicit about what it's doing:

# services/extraction.py (with fallback)
from db.queries.extractions import get_cached_extraction
from datetime import datetime, timedelta

MAX_CACHE_AGE = timedelta(hours=1)   # not 7 days

async def extract_document_with_fallback(
    document_id: str,
    document_text: str,
) -> tuple[ExtractionResult, bool]:
    """Returns (result, is_fresh) — callers decide what to do with stale results."""

    try:
        result = await call_model_with_retry(
            prompt=..., system=...
        )
        fresh_result = ExtractionResult(**json.loads(result))
        await cache_extraction(document_id, fresh_result)
        return fresh_result, True

    except Exception as e:
        logger.error(f"LLM call failed, attempting cache fallback: {e}")
        cached = await get_cached_extraction(document_id)

        if cached and (datetime.utcnow() - cached.created_at) < MAX_CACHE_AGE:
            return cached.result, False   # is_fresh=False signals to the caller

        raise   # no usable cache — fail explicitly

The caller receives is_fresh=False and decides what to do — surface a "results may be outdated" indicator to the user, skip the result entirely, or queue a refresh job. The fallback doesn't hide that it's a fallback.

---

Wiring It Into FastAPI

# routers/documents.py
@router.post("/documents/{document_id}/extract")
async def extract(document_id: str, request: Request):
    document = await get_document(document_id)

    result, is_fresh = await extract_document_with_fallback(
        document_id=document_id,
        document_text=document.text,
    )

    return {
        "result": result.model_dump(),
        "fresh": is_fresh,
        "extracted_at": datetime.utcnow().isoformat(),
    }

fresh in the response gives the client full information. A dashboard can show "results from 47 minutes ago" instead of presenting stale data as current.