Sourav Kohli
PythonAI/LLMFastAPIProduction Engineering

Integrating LLMs into Production Python Backends Without the Chaos

The gap between a working prototype and a reliable LLM feature in production is wider than it looks. Here are the patterns that actually close it.

The first LLM integration is always surprisingly easy. You grab the OpenAI client, write 10 lines of code, and it works. Then you try to ship it, and the problems start: latency spikes, rate limit errors in production, costs that scale unexpectedly, and a model that confidently outputs malformed JSON that breaks your downstream parser.

LLM features aren’t hard to prototype. They’re hard to make reliable. Here are the patterns I use when integrating them into production Python backends.

The Production Checklist

Before anything else, here’s what separates a demo from a deployable feature:

  1. Timeouts — LLM API calls can hang for 30+ seconds. Your web framework’s defaults aren’t set for this.
  2. Retries with backoff — Transient errors and rate limits are facts of life; they need to be handled, not surfaced to users.
  3. Structured output validation — Never trust freeform LLM text in code that expects a specific format.
  4. Cost observability — Track token usage per feature, not just globally.
  5. Graceful degradation — What does your product do when the LLM call fails entirely?

Let me go through each.

Timeouts

The default httpx and requests timeout in Python is typically None — no timeout at all. For an LLM call, this can mean a user-facing request hangs for a minute or longer.

from openai import OpenAI

client = OpenAI(
    timeout=30.0,       # total request timeout in seconds
    max_retries=0,      # we'll handle retries ourselves
)

For streaming responses, set the timeout on the connection, not the read — a streaming call legitimately takes time to fully receive.

In FastAPI, set a background task for long calls rather than holding the HTTP request open, or use streaming responses with Server-Sent Events if you want the user to see tokens as they arrive.

Retries with Exponential Backoff

Rate limit errors (HTTP 429) and transient server errors (HTTP 500, 502, 503) from LLM providers are not exceptional — they’re operational reality, especially under load. The right response is to retry with backoff, not to surface the error.

import time
import random
from openai import OpenAI, RateLimitError, APIStatusError

def call_with_retry(client: OpenAI, **kwargs) -> str:
    max_attempts = 3
    base_delay = 1.0

    for attempt in range(max_attempts):
        try:
            response = client.chat.completions.create(**kwargs)
            return response.choices[0].message.content or ''
        except RateLimitError:
            if attempt == max_attempts - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            time.sleep(delay)
        except APIStatusError as e:
            if e.status_code in (500, 502, 503) and attempt < max_attempts - 1:
                time.sleep(base_delay * (2 ** attempt))
            else:
                raise

For production use, tenacity gives you more declarative retry logic:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError

@retry(
    retry=retry_if_exception_type(RateLimitError),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    stop=stop_after_attempt(4),
)
def call_model(client, messages):
    return client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
    )

Structured Output Validation

If your code needs a specific structure from the LLM — a JSON object with particular fields — don’t parse freeform text. Use either:

1. Instructor / response_model pattern (cleanest):

import instructor
from pydantic import BaseModel
from openai import OpenAI

class ExtractedData(BaseModel):
    category: str
    confidence: float
    reason: str

client = instructor.from_openai(OpenAI())

result = client.chat.completions.create(
    model='gpt-4o-mini',
    response_model=ExtractedData,
    messages=[{'role': 'user', 'content': 'Classify: "Error 500 on checkout page"'}],
)
# result is a validated ExtractedData instance — or a ValidationError

2. JSON mode with manual validation:

import json
from pydantic import BaseModel, ValidationError

response = client.chat.completions.create(
    model='gpt-4o-mini',
    response_format={'type': 'json_object'},
    messages=[...],
)

raw = response.choices[0].message.content or '{}'
try:
    data = ExtractedData.model_validate_json(raw)
except (json.JSONDecodeError, ValidationError) as e:
    # log and handle — never let this reach your business logic unhandled
    logger.warning('LLM returned invalid structure: %s | raw: %s', e, raw[:200])
    return fallback_value

Never use raw string parsing or eval() on LLM output in production. The model will occasionally produce output that doesn’t match your prompt, no matter how carefully written the prompt is.

Cost Observability

Token costs add up fast and non-uniformly across features. Track them at the feature level, not just globally.

import logging

logger = logging.getLogger(__name__)

def tracked_completion(client, feature_name: str, **kwargs):
    response = client.chat.completions.create(**kwargs)
    usage = response.usage

    logger.info(
        'llm_usage feature=%s model=%s prompt_tokens=%d completion_tokens=%d total_tokens=%d',
        feature_name,
        response.model,
        usage.prompt_tokens,
        usage.completion_tokens,
        usage.total_tokens,
    )
    return response

Feed this into your existing observability stack (Datadog, Grafana, CloudWatch) and set alerts on per-feature daily spend. Surprises in LLM costs always come from a single feature that scaled more than expected, not from the total.

Graceful Degradation

Design every LLM feature with the explicit question: “What does the product do when this fails entirely?”

For some features, the answer is “show a generic fallback and log it.” For others, it might be “queue the request for async processing.” For others, the feature shouldn’t exist if it can’t be made reliable enough.

async def classify_ticket(text: str) -> str:
    try:
        result = await call_model_async(text)
        return result.category
    except Exception:
        logger.exception('ticket classification failed, using fallback')
        return 'uncategorised'  # known, safe value

The worst outcome is an unhandled exception that returns a 500 to the user because the LLM was temporarily unavailable. That’s a choice you’re making — usually the wrong one.

Caching

LLM calls are expensive and slow. If you’re making the same call with the same inputs more than once, cache the output.

import hashlib
import json
from django.core.cache import cache

def cached_completion(messages: list[dict], model: str = 'gpt-4o-mini', ttl: int = 3600):
    cache_key = 'llm:' + hashlib.sha256(
        json.dumps({'messages': messages, 'model': model}, sort_keys=True).encode()
    ).hexdigest()[:16]

    cached = cache.get(cache_key)
    if cached is not None:
        return cached

    result = client.chat.completions.create(model=model, messages=messages)
    content = result.choices[0].message.content or ''
    cache.set(cache_key, content, timeout=ttl)
    return content

Cache at the semantic level when possible: the cache key should reflect the logical input, not the raw API request structure, so minor prompt changes don’t unnecessarily bust the cache.

Putting It Together

The prototype shows that the LLM can do the job. The engineering is everything that makes it reliable at scale: timeouts, retries, validated outputs, cost tracking, graceful degradation, and caching. None of these are complicated on their own — the work is in making them habitual, part of how you build LLM features from the start rather than bolted on when production incidents force your hand.

← All Posts