Introduction
Building an LLM application that works in a demo is relatively straightforward. Building one that is reliable, cost-effective, and observable in production is an entirely different challenge. Production LLM applications require architectural patterns that address latency, cost, reliability, and observability concerns that don’t arise in prototyping.
This article covers the essential architecture patterns for production LLM systems, with practical implementation guidance for each.
Caching
Caching is the single most effective way to reduce latency and cost in LLM applications. Many real-world workloads have significant repetition—users ask similar questions, and the same context is processed repeatedly.
Semantic Caching
Traditional exact-match caching misses semantically equivalent queries. Semantic caching uses embeddings to identify similar past queries:
import hashlib
from datetime import timedelta
from redis import Redis
import numpy as np
redis = Redis()
def get_cached_response(query: str, query_embedding: list, threshold: float = 0.95):
# Check for exact match first
cache_key = f"llm:exact:{hashlib.md5(query.encode()).hexdigest()}"
cached = redis.get(cache_key)
if cached:
return cached.decode()
# Semantic cache lookup
# Store embeddings in a vector index (e.g., Redis with RediSearch)
similar = search_similar_queries(query_embedding, threshold)
if similar:
return similar[0]["response"]
return None
def cache_response(query: str, query_embedding: list, response: str, ttl: int = 3600):
cache_key = f"llm:exact:{hashlib.md5(query.encode()).hexdigest()}"
redis.setex(cache_key, ttl, response)
store_embedding(query, query_embedding, response)Cache Invalidation
Cache invalidation is critical for accuracy:
- Time-based: Expire entries after a set period
- Content-based: Invalidate when source documents change
- Version-based: Tag cache entries with model versions
Rate Limiting
LLM APIs are expensive and have rate limits. Implementing rate limiting protects your budget and prevents abuse.
Token-Based Rate Limiting
from datetime import datetime, timedelta
from collections import defaultdict
class TokenRateLimiter:
def __init__(self, max_tokens_per_minute: int = 100000):
self.limits = defaultdict(list)
self.max_tokens = max_tokens_per_minute
def check_and_consume(self, user_id: str, estimated_tokens: int) -> bool:
now = datetime.now()
window_start = now - timedelta(minutes=1)
# Clean old entries
self.limits[user_id] = [
(ts, tokens) for ts, tokens in self.limits[user_id]
if ts > window_start
]
current_usage = sum(tokens for _, tokens in self.limits[user_id])
if current_usage + estimated_tokens > self.max_tokens:
return False
self.limits[user_id].append((now, estimated_tokens))
return TrueTiered Rate Limiting
Implement different limits for different user tiers:
- Free tier: 10 requests/day, 1,000 tokens/day
- Pro tier: 100 requests/hour, 50,000 tokens/hour
- Enterprise: Custom limits with burst capacity
Fallbacks and Model Routing
No single LLM is perfect for every task. Production systems should route requests to the most appropriate model and fall back gracefully when failures occur.
Multi-Model Routing
from enum import Enum
from dataclasses import dataclass
class ModelTier(Enum):
FAST = "gpt-4o-mini"
BALANCED = "gpt-4o"
POWERFUL = "gpt-4o"
FALLBACK = "gpt-4o-mini"
class ModelRouter:
def __init__(self):
self.fallback_chain = [
ModelTier.BALANCED,
ModelTier.FAST,
ModelTier.FALLBACK
]
def route(self, query: str, context_length: int, complexity: str) -> ModelTier:
# Route based on task complexity
if complexity == "simple" or context_length < 500:
return ModelTier.FAST
elif complexity == "complex":
return ModelTier.POWERFUL
return ModelTier.BALANCED
def execute_with_fallback(self, prompt: str, model: ModelTier) -> str:
start_index = self.fallback_chain.index(model)
for tier in self.fallback_chain[start_index:]:
try:
return self.call_model(tier, prompt)
except Exception as e:
log_error(tier, e)
continue
raise RuntimeError("All models failed")Circuit Breakers
Prevent cascading failures when an LLM provider is down:
import time
from circuitbreaker import CircuitBreaker
class LLMCircuitBreaker(CircuitBreaker):
FAILURE_THRESHOLD = 5
RECOVERY_TIMEOUT = 60
@LLMCircuitBreaker
def call_llm(self, prompt: str) -> str:
return self.llm.invoke(prompt)Streaming Responses
Streaming improves perceived performance by showing tokens as they’re generated. This is essential for chat interfaces.
Server-Sent Events Streaming
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
app = FastAPI()
@app.post("/chat")
async def chat(request: dict):
async def generate():
async for chunk in llm.astream(request["message"]):
data = json.dumps({"token": chunk.content})
yield f"data: {data}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")Client-Side Consumption
const eventSource = new EventSource('/chat');
let fullResponse = '';
eventSource.onmessage = (event) => {
if (event.data === '[DONE]') {
eventSource.close();
return;
}
const data = JSON.parse(event.data);
fullResponse += data.token;
updateUI(fullResponse);
};Observability
Observability in LLM applications requires tracking inputs, outputs, latency, costs, and quality metrics.
Logging Structure
import logging
import json
from datetime import datetime
class LLMLogger:
def __init__(self):
self.logger = logging.getLogger("llm_app")
def log_request(self, request_id: str, model: str, prompt: str,
response: str, latency_ms: int, tokens_in: int,
tokens_out: int, cost: float):
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"request_id": request_id,
"model": model,
"prompt_length": len(prompt),
"response_length": len(response),
"latency_ms": latency_ms,
"tokens_in": tokens_in,
"tokens_out": tokens_out,
"cost_usd": cost
}
self.logger.info(json.dumps(log_entry))Key Metrics to Track
- Latency: Time to first token and total generation time
- Token usage: Input and output tokens per request
- Cost: Dollar cost per request and aggregate
- Error rate: Percentage of failed requests
- Cache hit rate: Percentage of requests served from cache
- Quality scores: User feedback or automated quality metrics
Using LangSmith or Langfuse
from langfuse import Langfuse
from langfuse.decorators import observe
langfuse = Langfuse()
@observe()
def generate_response(query: str) -> str:
# This function is automatically traced
docs = retrieve(query)
context = "\n".join([d.page_content for d in docs])
response = llm.invoke(f"Context: {context}\n\nQuestion: {query}")
return responseCost Optimization
Prompt Compression
Reduce input tokens by compressing prompts:
def compress_context(documents: list, max_tokens: int = 4000) -> str:
"""Compress documents to fit within token budget."""
total_text = ""
for doc in documents:
# Summarize each document to key sentences
summary = summarize_document(doc)
if count_tokens(total_text + summary) > max_tokens:
break
total_text += summary + "\n"
return total_textBatch Processing
For non-real-time workloads, batch requests to reduce per-request overhead:
async def batch_process(queries: list[str]) -> list[str]:
"""Process multiple queries in a single batch."""
messages = [[{"role": "user", "content": q}] for q in queries]
responses = await llm.abatch(messages)
return [r.content for r in responses]Conclusion
Production LLM applications require architectural patterns that address the unique challenges of working with large language models. Caching reduces cost and latency. Rate limiting protects budgets. Fallbacks and model routing improve reliability. Streaming enhances user experience. Observability provides visibility into system behavior.
Start by implementing caching and observability—they provide the most immediate value. Then add rate limiting, fallbacks, and model routing as your application scales. The key insight is that production LLM architecture is about building systems that degrade gracefully, cost predictably, and can be debugged when things go wrong.
