Introduction to Highly Available LLM Serving
Large Language Models (LLMs) have become the backbone of modern AI-powered applications, from customer support chatbots to code generation assistants. However, serving these models in production introduces unique challenges that traditional web services don't face. A single LLM inference request can consume significant GPU memory, take several seconds to complete, and cost orders of magnitude more than a typical API call. When an LLM serving endpoint goes down, the impact is immediate and costly.
Highly Available LLM Serving Architecture refers to a system design pattern that ensures LLM inference services remain operational, responsive, and reliable even in the face of hardware failures, traffic spikes, or maintenance events. This involves combining load balancing, redundancy, health monitoring, graceful degradation, and intelligent request routing to create a serving infrastructure that can withstand real-world operational challenges.
Why High Availability Matters for LLM Serving
LLM serving differs from traditional microservices in several critical ways. First, the hardware requirements are substantial — a single GPU can cost tens of thousands of dollars, making redundancy expensive. Second, model loading times can range from minutes to tens of minutes, meaning cold starts are painfully slow. Third, inference latency is inherently variable, with long output sequences creating head-of-line blocking issues.
Key Challenges
- GPU scarcity and cost: Redundancy requires duplicating expensive hardware resources
- Long cold starts: Loading multi-billion parameter models into GPU memory takes time
- Variable latency: Token generation rates vary based on input/output length
- Memory pressure: KV cache management can cause OOM errors under load
- Batching complexity: Continuous batching requires careful queue management
- Model updates: Deploying new model versions without downtime is non-trivial
When these challenges aren't addressed, the consequences include service outages, degraded user experience, wasted compute resources, and lost revenue. For businesses building products on top of LLMs, availability directly impacts customer trust and retention.
Core Architecture Components
A highly available LLM serving architecture consists of several interconnected layers, each addressing specific failure modes and performance concerns.
1. Load Balancer Layer
The entry point for all inference requests. This layer distributes traffic across multiple inference workers, handles TLS termination, and performs health checks. For LLM workloads, the load balancer must be session-aware and capable of handling long-lived streaming connections.
2. Inference Worker Pool
Multiple instances of inference engines (such as vLLM, TGI, or TensorRT-LLM) running on GPU-equipped nodes. Each worker manages its own model instance, KV cache, and request queue. The pool should span multiple availability zones to survive zone-level failures.
3. Request Router / API Gateway
An intelligent routing layer that can make decisions based on model type, request priority, token count, or tenant. This layer can implement features like rate limiting, request caching, and fallback routing to alternative models.
4. Health Monitoring System
Continuous monitoring of GPU utilization, memory usage, inference latency, and error rates. This system feeds data to the load balancer and orchestrator to make real-time routing and scaling decisions.
5. Model Registry and Storage
A centralized store for model weights, configurations, and version metadata. This enables fast model loading and consistent deployments across workers.
Building the Architecture: A Practical Implementation
Let's walk through a concrete implementation using Python, FastAPI, vLLM as the inference engine, and Redis for health tracking. This example demonstrates the key patterns for high availability.
Setting Up the Inference Worker
Each inference worker runs a vLLM server wrapped with health reporting. The worker registers itself with a service registry and reports metrics periodically.
# inference_worker.py
import asyncio
import json
import os
import time
import signal
import sys
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
import redis.asyncio as redis
from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams
# Configuration from environment
WORKER_ID = os.getenv("WORKER_ID", "worker-1")
MODEL_NAME = os.getenv("MODEL_NAME", "meta-llama/Llama-2-7b-chat-hf")
REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379")
PORT = int(os.getenv("PORT", "8000"))
AVAILABILITY_ZONE = os.getenv("AVAILABILITY_ZONE", "us-east-1a")
# Global state
engine: AsyncLLMEngine = None
redis_client: redis.Redis = None
health_task: asyncio.Task = None
is_healthy = False
is_shutting_down = False
async def report_health():
"""Periodically report health metrics to Redis."""
while not is_shutting_down:
try:
metrics = {}
if engine is not None:
# Collect engine metrics
engine_metrics = await engine.get_metrics()
metrics = {
"worker_id": WORKER_ID,
"model": MODEL_NAME,
"zone": AVAILABILITY_ZONE,
"status": "healthy" if is_healthy else "unhealthy",
"timestamp": time.time(),
"gpu_memory_used_mb": engine_metrics.get("gpu_memory_used_mb", 0),
"gpu_memory_total_mb": engine_metrics.get("gpu_memory_total_mb", 0),
"num_running_requests": engine_metrics.get("num_running_requests", 0),
"num_waiting_requests": engine_metrics.get("num_waiting_requests", 0),
"avg_inference_latency_ms": engine_metrics.get("avg_latency_ms", 0),
}
# Store health data with TTL of 15 seconds
await redis_client.setex(
f"llm:worker:{WORKER_ID}:health",
15,
json.dumps(metrics)
)
# Add to active workers set
await redis_client.sadd("llm:active_workers", WORKER_ID)
except Exception as e:
print(f"Health reporting error: {e}")
await asyncio.sleep(5)
async def check_internal_health() -> bool:
"""Perform internal health checks."""
try:
if engine is None:
return False
# Check if engine can process a minimal request
test_params = SamplingParams(max_tokens=1, temperature=0.0)
result = await engine.generate("test", test_params, f"health-check-{time.time()}")
# Consume the generator
async for _ in result:
break
return True
except Exception as e:
print(f"Health check failed: {e}")
return False
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Manage application lifecycle."""
global engine, redis_client, health_task, is_healthy
# Initialize Redis
redis_client = redis.from_url(REDIS_URL, decode_responses=True)
# Initialize vLLM engine
print(f"Loading model: {MODEL_NAME}")
engine_args = AsyncEngineArgs(
model=MODEL_NAME,
tensor_parallel_size=int(os.getenv("TENSOR_PARALLEL_SIZE", "1")),
gpu_memory_utilization=float(os.getenv("GPU_MEMORY_UTILIZATION", "0.90")),
max_num_seqs=int(os.getenv("MAX_NUM_SEQS", "256")),
max_num_batched_tokens=int(os.getenv("MAX_NUM_BATCHED_TOKENS", "8192")),
enable_prefix_caching=True,
)
engine = AsyncLLMEngine.from_engine_args(engine_args)
is_healthy = True
print("Model loaded successfully")
# Start health reporting
health_task = asyncio.create_task(report_health())
yield
# Shutdown
is_shutting_down = True
is_healthy = False
if health_task:
health_task.cancel()
try:
await health_task
except asyncio.CancelledError:
pass
# Remove from active workers
try:
await redis_client.srem("llm:active_workers", WORKER_ID)
await redis_client.delete(f"llm:worker:{WORKER_ID}:health")
except Exception:
pass
await redis_client.close()
app = FastAPI(lifespan=lifespan)
@app.get("/health")
async def health():
"""Health endpoint for load balancer probes."""
if is_healthy and not is_shutting_down:
return JSONResponse(
status_code=200,
content={"status": "healthy", "worker_id": WORKER_ID}
)
return JSONResponse(
status_code=503,
content={"status": "unhealthy", "worker_id": WORKER_ID}
)
@app.get("/ready")
async def ready():
"""Readiness endpoint — returns 503 during shutdown."""
if is_shutting_down:
return JSONResponse(
status_code=503,
content={"status": "shutting_down", "worker_id": WORKER_ID}
)
if not is_healthy:
return JSONResponse(
status_code=503,
content={"status": "not_ready", "worker_id": WORKER_ID}
)
return JSONResponse(
status_code=200,
content={"status": "ready", "worker_id": WORKER_ID, "model": MODEL_NAME}
)
@app.post("/v1/completions")
async def completions(request: Request):
"""Generate completions using vLLM."""
if is_shutting_down:
return JSONResponse(
status_code=503,
content={"error": "Worker is shutting down"}
)
body = await request.json()
prompt = body.get("prompt", "")
stream = body.get("stream", False)
sampling_params = SamplingParams(
max_tokens=body.get("max_tokens", 256),
temperature=body.get("temperature", 0.7),
top_p=body.get("top_p", 0.95),
)
request_id = f"req-{WORKER_ID}-{time.time()}"
if stream:
async def generate_stream():
try:
results_generator = await engine.generate(prompt, sampling_params, request_id)
async for output in results_generator:
text = output.outputs[0].text
yield f"data: {json.dumps({'text': text, 'request_id': request_id})}\n\n"
yield f"data: {json.dumps({'done': True, 'request_id': request_id})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
return StreamingResponse(generate_stream(), media_type="text/event-stream")
else:
try:
results_generator = await engine.generate(prompt, sampling_params, request_id)
final_output = None
async for output in results_generator:
final_output = output
return JSONResponse(content={
"text": final_output.outputs[0].text,
"request_id": request_id,
"worker_id": WORKER_ID,
})
except Exception as e:
return JSONResponse(
status_code=500,
content={"error": str(e), "worker_id": WORKER_ID}
)
@app.post("/drain")
async def drain():
"""Gracefully drain the worker — stop accepting new requests."""
global is_healthy
is_healthy = False
return JSONResponse(
content={"status": "draining", "worker_id": WORKER_ID}
)
# Handle shutdown signals
def handle_shutdown(signum, frame):
global is_shutting_down, is_healthy
print(f"Received signal {signum}, initiating graceful shutdown...")
is_shutting_down = True
is_healthy = False
signal.signal(signal.SIGTERM, handle_shutdown)
signal.signal(signal.SIGINT, handle_shutdown)
Implementing the Intelligent Load Balancer
The load balancer sits in front of all inference workers and routes requests based on real-time health data, current load, and availability zone awareness. Unlike simple round-robin load balancers, this one makes informed decisions to maximize availability and minimize latency.
# load_balancer.py
import asyncio
import json
import time
import random
from collections import defaultdict
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
import httpx
import redis.asyncio as redis
app = FastAPI()
REDIS_URL = "redis://redis:6379"
WORKER_BASE_PORT = 8000
redis_client: redis.Redis = None
http_client: httpx.AsyncClient = None
@app.on_event("startup")
async def startup():
global redis_client, http_client
redis_client = redis.from_url(REDIS_URL, decode_responses=True)
http_client = httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=5.0))
@app.on_event("shutdown")
async def shutdown():
if http_client:
await http_client.aclose()
if redis_client:
await redis_client.close()
async def get_healthy_workers() -> list:
"""Retrieve all healthy workers from Redis with their metrics."""
workers = []
try:
worker_ids = await redis_client.smembers("llm:active_workers")
for worker_id in worker_ids:
health_data = await redis_client.get(f"llm:worker:{worker_id}:health")
if health_data:
data = json.loads(health_data)
if data.get("status") == "healthy":
workers.append(data)
except Exception as e:
print(f"Error fetching workers: {e}")
return workers
def select_worker(workers: list, strategy: str = "least_loaded") -> dict:
"""Select a worker based on the chosen strategy."""
if not workers:
return None
if strategy == "least_loaded":
# Select worker with fewest running + waiting requests
return min(
workers,
key=lambda w: w.get("num_running_requests", 0) + w.get("num_waiting_requests", 0)
)
elif strategy == "zone_aware":
# Prefer workers in the same zone as the request origin
# Fall back to least loaded if no zone preference
return min(
workers,
key=lambda w: w.get("num_running_requests", 0) + w.get("num_waiting_requests", 0)
)
elif strategy == "round_robin":
return random.choice(workers)
else:
return workers[0]
async def forward_request(worker: dict, path: str, body: dict, stream: bool = False):
"""Forward request to selected worker."""
worker_host = worker.get("host", "localhost")
worker_port = worker.get("port", WORKER_BASE_PORT)
url = f"http://{worker_host}:{worker_port}{path}"
try:
if stream:
async def stream_response():
async with http_client.stream("POST", url, json=body, timeout=120.0) as response:
async for chunk in response.aiter_bytes():
yield chunk
return StreamingResponse(
stream_response(),
media_type="text/event-stream"
)
else:
response = await http_client.post(url, json=body, timeout=120.0)
return JSONResponse(
content=response.json(),
status_code=response.status_code
)
except httpx.ConnectError:
# Worker is unreachable, remove from active set
await redis_client.srem("llm:active_workers", worker.get("worker_id"))
raise HTTPException(status_code=503, detail="Selected worker unavailable")
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="Worker timeout")
async def forward_with_failover(path: str, body: dict, max_retries: int = 3, stream: bool = False):
"""Forward request with automatic failover to alternative workers."""
last_error = None
for attempt in range(max_retries):
workers = await get_healthy_workers()
if not workers:
raise HTTPException(status_code=503, detail="No healthy workers available")
# Exclude workers that failed in previous attempts
if attempt > 0:
workers = [w for w in workers if w.get("worker_id") not in failed_workers]
if not workers:
raise HTTPException(status_code=503, detail="All workers failed")
worker = select_worker(workers, strategy="least_loaded")
try:
result = await forward_request(worker, path, body, stream=stream)
return result
except (httpx.ConnectError, httpx.TimeoutException) as e:
last_error = e
failed_workers.add(worker.get("worker_id"))
print(f"Attempt {attempt + 1} failed for worker {worker.get('worker_id')}: {e}")
await asyncio.sleep(0.5 * (attempt + 1)) # Brief backoff
continue
raise HTTPException(status_code=503, detail=f"All retries exhausted: {last_error}")
@app.post("/v1/completions")
async def completions(request: Request):
"""Load-balanced completions endpoint with failover."""
body = await request.json()
stream = body.get("stream", False)
global failed_workers
failed_workers = set()
return await forward_with_failover("/v1/completions", body, max_retries=3, stream=stream)
@app.get("/health")
async def health():
"""Load balancer health endpoint."""
workers = await get_healthy_workers()
return JSONResponse(content={
"status": "healthy",
"active_workers": len(workers),
"workers": [
{
"worker_id": w.get("worker_id"),
"zone": w.get("zone"),
"load": w.get("num_running_requests", 0) + w.get("num_waiting_requests", 0),
}
for w in workers
]
})
Implementing a Circuit Breaker
A circuit breaker prevents cascading failures by temporarily stopping traffic to unhealthy workers. When a worker starts failing repeatedly, the circuit opens and routes traffic elsewhere, giving the failing worker time to recover.
# circuit_breaker.py
import time
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict
import asyncio
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing if recovered
@dataclass
class CircuitBreaker:
failure_threshold: int = 5
recovery_timeout: float = 30.0
half_open_max_calls: int = 3
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
success_count: int = 0
last_failure_time: float = 0.0
half_open_calls: int = 0
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
async def can_execute(self) -> bool:
"""Check if a request can be sent through this circuit."""
async with self.lock:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Check if recovery timeout has elapsed
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
return True
return False
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls < self.half_open_max_calls:
self.half_open_calls += 1
return True
return False
return False
async def record_success(self):
"""Record a successful request."""
async with self.lock:
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.half_open_max_calls:
# Worker has recovered
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
elif self.state == CircuitState.CLOSED:
self.failure_count = 0
async def record_failure(self):
"""Record a failed request."""
async with self.lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
# Worker is still failing
self.state = CircuitState.OPEN
self.success_count = 0
elif self.state == CircuitState.CLOSED:
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
class CircuitBreakerRegistry:
"""Registry of circuit breakers per worker."""
def __init__(self):
self._breakers: dict[str, CircuitBreaker] = defaultdict(CircuitBreaker)
def get(self, worker_id: str) -> CircuitBreaker:
return self._breakers[worker_id]
def get_healthy_workers(self, all_workers: list) -> list:
"""Filter out workers with open circuits."""
return [
w for w in all_workers
if self._breakers[w.get("worker_id")].state != CircuitState.OPEN
]
# Usage in the load balancer
breaker_registry = CircuitBreakerRegistry()
async def forward_with_circuit_breaker(worker: dict, path: str, body: dict):
"""Forward request with circuit breaker protection."""
worker_id = worker.get("worker_id")
breaker = breaker_registry.get(worker_id)
if not await breaker.can_execute():
raise HTTPException(
status_code=503,
detail=f"Circuit open for worker {worker_id}"
)
try:
response = await http_client.post(
f"http://{worker['host']}:{worker['port']}{path}",
json=body,
timeout=120.0
)
if response.status_code >= 500:
await breaker.record_failure()
else:
await breaker.record_success()
return response
except Exception as e:
await breaker.record_failure()
raise
Model Fallback Strategy
When primary models are unavailable, falling back to smaller, faster models ensures the service remains responsive. This is particularly important for cost optimization — you don't want to block all traffic because your largest model is overloaded.
# fallback_router.py
import asyncio
from dataclasses import dataclass
from typing import Optional
import redis.asyncio as redis
@dataclass
class ModelTier:
name: str
model_id: str
priority: int # Lower = higher priority
max_tokens: int
fallback_for: Optional[str] = None # Which model this is a fallback for
# Define model tiers with fallback chains
MODEL_TIERS = [
ModelTier(
name="primary",
model_id="meta-llama/Llama-2-70b-chat-hf",
priority=1,
max_tokens=4096,
),
ModelTier(
name="secondary",
model_id="meta-llama/Llama-2-13b-chat-hf",
priority=2,
max_tokens=4096,
fallback_for="primary",
),
ModelTier(
name="emergency",
model_id="meta-llama/Llama-2-7b-chat-hf",
priority=3,
max_tokens=2048,
fallback_for="secondary",
),
]
class FallbackRouter:
"""Routes requests with automatic model fallback."""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
self._tier_map = {t.name: t for t in MODEL_TIERS}
self._fallback_map = {t.fallback_for: t for t in MODEL_TIERS if t.fallback_for}
async def get_available_tier(self, requested_tier: str = "primary") -> Optional[ModelTier]:
"""Find the best available model tier, falling back if needed."""
tier = self._tier_map.get(requested_tier)
if tier is None:
return None
# Check if workers for this tier are available
while tier is not None:
workers = await self._get_workers_for_model(tier.model_id)
if workers:
return tier
# Fall back to next tier
tier = self._fallback_map.get(tier.name)
return None
async def _get_workers_for_model(self, model_id: str) -> list:
"""Get healthy workers serving a specific model."""
worker_ids = await self.redis.smembers("llm:active_workers")
workers = []
for worker_id in worker_ids:
health_data = await self.redis.get(f"llm:worker:{worker_id}:health")
if health_data:
import json
data = json.loads(health_data)
if data.get("model") == model_id and data.get("status") == "healthy":
workers.append(data)
return workers
async def route_request(self, prompt: str, max_tokens: int = 256,
requested_tier: str = "primary") -> dict:
"""Route a request with fallback logic."""
tier = await self.get_available_tier(requested_tier)
if tier is None:
return {"error": "No model tiers available", "status": 503}
workers = await self._get_workers_for_model(tier.model_id)
if not workers:
return {"error": f"No workers for {tier.name}", "status": 503}
# Select least loaded worker
worker = min(
workers,
key=lambda w: w.get("num_running_requests", 0) + w.get("num_waiting_requests", 0)
)
# Adjust max_tokens if using a fallback with lower limits
effective_max_tokens = min(max_tokens, tier.max_tokens)
return {
"worker": worker,
"model_tier": tier.name,
"model_id": tier.model_id,
"max_tokens": effective_max_tokens,
"is_fallback": tier.name != requested_tier,
}
Graceful Shutdown and Zero-Downtime Deployments
Zero-downtime deployments require careful coordination between the load balancer, the orchestrator, and individual workers. The key is to stop sending new requests to a worker before it shuts down, while allowing in-flight requests to complete.
# deployment_manager.py
import asyncio
import json
import time
import httpx
import redis.asyncio as redis
class DeploymentManager:
"""Manages zero-downtime deployments of LLM workers."""
def __init__(self, redis_url: str):
self.redis = redis.from_url(redis_url, decode_responses=True)
self.http = httpx.AsyncClient(timeout=httpx.Timeout(30.0, connect=5.0))
async def drain_worker(self, worker_id: str, drain_timeout: float = 300.0) -> bool:
"""
Gracefully drain a worker:
1. Mark worker as draining (stop new traffic)
2. Wait for in-flight requests to complete
3. Send shutdown signal
"""
print(f"Starting drain for worker {worker_id}")
# Step 1: Mark as draining
await self.redis.srem("llm:active_workers", worker_id)
await self.redis.set(
f"llm:worker:{worker_id}:status",
"draining",
ex=600 # 10 minute TTL
)
# Step 2: Wait for in-flight requests to complete
start_time = time.time()
while time.time() - start_time < drain_timeout:
health_data = await self.redis.get(f"llm:worker:{worker_id}:health")
if health_data:
data = json.loads(health_data)
running = data.get("num_running_requests", 0)
waiting = data.get("num_waiting_requests", 0)
if running == 0 and waiting == 0:
print(f"Worker {worker_id} drained successfully")
break
print(f"Worker {worker_id}: {running} running, {waiting} waiting")
else:
# Worker health data expired — likely already down
break
await asyncio.sleep(2)
else:
print(f"Drain timeout for worker {worker_id}, forcing shutdown")
# Step 3: Send drain endpoint call to stop accepting requests
worker_host = await self._get_worker_host(worker_id)
if worker_host:
try:
await self.http.post(f"http://{worker_host}/drain")
except Exception as e:
print(f"Drain endpoint call failed: {e}")
return True
async def rolling_deploy(self, worker_ids: list, new_model: str = None):
"""
Perform a rolling deployment across workers.
Only drains one worker at a time to maintain capacity.
"""
total = len(worker_ids)
for i, worker_id in enumerate(worker_ids):
print(f"\n=== Deploying to worker {i+1}/{total}: {worker_id} ===")
# Drain the worker
await self.drain_worker(worker_id)
# In a real system, this would trigger the orchestrator
# to restart the worker with the new image/model
print(f"Worker {worker_id} drained, waiting for restart...")
await asyncio.sleep(10)
# Wait for worker to become healthy again
await self._wait_for_healthy(worker_id)
print(f"Worker {worker_id} is healthy, moving to next")
print("\nRolling deployment complete!")
async def _wait_for_healthy(self, worker_id: str, timeout: float = 600.0):
"""Wait for a worker to report healthy after restart."""
start = time.time()
while time.time() - start < timeout:
health_data = await self.redis.get(f"llm:worker:{worker_id}:health")
if health_data:
data = json.loads(health_data)
if data.get("status") == "healthy":
return True
await asyncio.sleep(5)
raise TimeoutError(f"Worker {worker_id} did not become healthy in time")
async def _get_worker_host(self, worker_id: str) -> str:
"""Get the host address for a worker."""
# In production, this would query service discovery
return f"worker-{worker_id}:8000"
async def close(self):
await self.http.aclose()
await self.redis.close()
# Example: Trigger a rolling deployment
async def main():
manager = DeploymentManager("redis://redis:6379")
# List of workers to update
workers = ["worker-1", "worker-2", "worker-3", "worker-4"]
try:
await manager.rolling_deploy(workers, new_model="meta-llama/Llama-2-13b-chat-hf")
finally:
await manager.close()
if __name__ == "__main__":
asyncio.run(main())
Request Caching for Availability
Caching repeated requests reduces load on inference workers and provides instant responses when workers are degraded. Semantic caching can catch paraphrased queries, but even exact-match caching provides significant availability benefits.
# request_cache.py
import hashlib
import json
import time
import redis.asyncio as redis
class RequestCache:
"""Redis-based cache for LLM completions."""
def __init__(self, redis_url: str, default_ttl: int = 3600):
self.redis = redis.from_url(redis_url, decode_responses=True)
self.default_ttl = default_ttl
def _cache_key(self, prompt: str, model: str, params: dict) -> str:
"""Generate a deterministic cache key."""
# Normalize parameters that affect output
relevant_params = {
"max_tokens": params.get("max_tokens", 256),
"temperature": params.get("temperature", 0.0), # Only cache temp=0
"top_p": params.get("top_p", 1.0),
}
cache_input = json.dumps({
"prompt": prompt.strip(),
"model": model,
"params": relevant_params,
}, sort_keys=True)
hash_val = hashlib.sha256(cache_input.encode()).hexdigest()
return f"llm:cache:{hash_val}"
async def get(self, prompt: str, model: str, params: dict) -> dict | None:
"""Retrieve a cached response if available."""
# Only cache deterministic requests (temperature = 0)
if params.get("temperature", 0.7) > 0:
return None
key = self._cache_key(prompt, model, params)
cached = await self.redis.get(key)
if cached:
data = json.loads(cached)
data["cache_hit"] = True
data["cache_age_seconds"] = time.time() - data.get("cached_at", 0)
return data
return None
async def set(self, prompt: str, model: str, params: dict,
response: dict, ttl: int = None):
"""Store a response in cache."""
# Only cache deterministic requests
if params.get("temperature", 0.7) > 0:
return
key = self._cache_key(prompt, model, params)
cache_entry = {
"response": response,
"model": model,
"cached_at": time.time(),
}
await self.redis.setex(
key,
ttl or self.default_ttl,
json.dumps(cache_entry)
)
async def invalidate_model(self, model: str):
"""Invalidate all cache entries for a specific model."""
# Use SCAN to find and delete keys
async for key in self.redis.scan_iter(match="llm:cache:*", count=100):
data = await self.redis.get(key)
if data:
entry = json.loads(data)
if entry.get("model") == model:
await self.redis.delete(key)
async def close(self):
await self.redis.close()
# Integration with the load balancer
class CachedLoadBalancer:
"""Load balancer with caching layer."""
def __init__(self, cache: RequestCache):
self.cache = cache
async def handle_request(self, prompt: str, model: str, params: dict):
"""Handle a request with caching."""
# Check cache first
cached = await self.cache.get(prompt, model, params)
if cached:
print(f"Cache hit! Age: {cached['cache_age_seconds']:.1f}s")
return cached
# Forward to inference worker
response = await self._forward_to_worker(prompt, model, params)
# Cache the response
await self.cache.set(prompt, model, params, response)
return response
async def _forward_to_worker(self, prompt: str, model: str, params: dict):
"""Forward request to an inference worker."""
# Implementation would use the load balancer logic above
pass
Best Practices
Deployment and Infrastructure
- Distribute across availability zones: Never place all inference workers in a single AZ. A minimum of two zones with equal capacity ensures zone-level failures don't take down the service.
- Use GPU node pools with autoscaling: Configure cluster autoscaler with appropriate scale-up thresholds. GPU nodes take longer to provision, so preemptive scaling based on queue depth is better than reactive scaling.
- Keep model weights on local NVMe: Store model weights on fast local storage or pre-load them into a shared cache. Network-attached storage for model loading adds minutes to cold starts.
- Implement pod disruption budgets: In Kubernetes, set PDBs to ensure a minimum number of inference workers are always running during voluntary disruptions.
Request Handling
- Set aggressive timeouts: Configure request timeouts at every layer. A stuck request consuming GPU memory is worse than a failed request. Use 30-60 second connect timeouts and 120-300 second read timeouts for streaming.
- Implement request prioritization: Not all requests are equal. Premium users, real-time chat, and critical system prompts should take priority over batch processing jobs.
- Use continuous batching: Engines like vLLM support continuous batching, which dramatically improves throughput. Ensure your max_num_seqs and max_num_batched_tokens are tuned for your hardware.
- Enable prefix caching: Many LLM applications send repeated system prompts. Prefix caching avoids recomputing these tokens, reducing latency and GPU load.
Monitoring and Observability
- Track GPU-specific metrics: Monitor GPU memory utilization, GPU compute utilization, KV cache hit rate, and tokens per second. These metrics reveal bottlenecks before they become failures.
- Implement distributed tracing: Use OpenTelemetry to trace requests through the load balancer, router, and inference worker. This makes it easy to identify which component