Disaster Recovery for LLM Applications: A Developer's Guide
Large Language Model (LLM) applications have become mission-critical components of modern software systems. From customer support chatbots to code generation assistants, these applications now handle workloads that directly impact revenue, user experience, and business operations. However, LLM applications introduce unique failure modes that traditional disaster recovery (DR) strategies were never designed to handle. A sudden API outage from your model provider, a rate-limit storm, a corrupted vector index, or a prompt regression can take your entire application offline in seconds. This tutorial walks you through building a robust disaster recovery strategy tailored specifically for LLM-powered systems.
What Is Disaster Recovery for LLM Applications?
Disaster recovery for LLM applications is the set of policies, tools, and architectural patterns that ensure your AI-powered services remain available—or degrade gracefully—when failures occur. Unlike conventional DR, which focuses primarily on infrastructure redundancy, LLM DR must account for the unique dependencies of AI workloads: external model APIs, embedding services, vector databases, prompt templates, fine-tuned model weights, and cached responses.
At its core, LLM disaster recovery answers one question: when your primary model or provider fails, how quickly can your application continue serving users with acceptable quality?
Why It Matters
- Provider concentration risk: Most applications depend on a single LLM provider. When that provider experiences an outage, your application goes down with it.
- Rate limiting and quota exhaustion: Traffic spikes can exhaust your token quota, effectively creating a self-inflicted outage.
- Model deprecations: Providers regularly deprecate model versions, sometimes with little notice, breaking applications that hardcode model identifiers.
- Vector store corruption: Retrieval-augmented generation (RAG) systems depend on vector indexes that can become corrupted or inconsistent.
- Prompt regressions: A seemingly innocent prompt change can degrade output quality in ways that only surface in production.
- Cost spikes: Without safeguards, a runaway agent loop or a malicious input can generate catastrophic billing.
Core Architecture for LLM Resilience
A resilient LLM application is built on three layers: a provider abstraction layer, a fallback chain, and a caching layer. Let's examine each and implement them in code.
1. Provider Abstraction Layer
The first step in LLM disaster recovery is decoupling your application logic from any specific model provider. Never call provider SDKs directly from your business logic. Instead, build an abstraction that normalizes requests and responses across providers.
# llm_provider.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class LLMRequest:
prompt: str
system_prompt: Optional[str] = None
max_tokens: int = 1024
temperature: float = 0.7
@dataclass
class LLMResponse:
text: str
provider: str
model: str
tokens_used: int
latency_ms: int
class BaseProvider(ABC):
@abstractmethod
def generate(self, request: LLMRequest) -> LLMResponse:
pass
@abstractmethod
def health_check(self) -> bool:
pass
With this abstraction in place, you can implement concrete providers for OpenAI, Anthropic, Google, and even local models behind a unified interface.
# providers.py
import time
import openai
import anthropic
from llm_provider import BaseProvider, LLMRequest, LLMResponse
class OpenAIProvider(BaseProvider):
def __init__(self, model: str = "gpt-4o"):
self.client = openai.OpenAI()
self.model = model
def generate(self, request: LLMRequest) -> LLMResponse:
start = time.time()
messages = []
if request.system_prompt:
messages.append({"role": "system", "content": request.system_prompt})
messages.append({"role": "user", "content": request.prompt})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=request.max_tokens,
temperature=request.temperature,
)
latency = int((time.time() - start) * 1000)
return LLMResponse(
text=response.choices[0].message.content,
provider="openai",
model=self.model,
tokens_used=response.usage.total_tokens,
latency_ms=latency,
)
def health_check(self) -> bool:
try:
self.client.models.list()
return True
except Exception:
return False
class AnthropicProvider(BaseProvider):
def __init__(self, model: str = "claude-3-5-sonnet-20241022"):
self.client = anthropic.Anthropic()
self.model = model
def generate(self, request: LLMRequest) -> LLMResponse:
start = time.time()
kwargs = {
"model": self.model,
"max_tokens": request.max_tokens,
"messages": [{"role": "user", "content": request.prompt}],
}
if request.system_prompt:
kwargs["system"] = request.system_prompt
response = self.client.messages.create(**kwargs)
latency = int((time.time() - start) * 1000)
return LLMResponse(
text=response.content[0].text,
provider="anthropic",
model=self.model,
tokens_used=response.usage.input_tokens + response.usage.output_tokens,
latency_ms=latency,
)
def health_check(self) -> bool:
try:
self.client.messages.create(
model=self.model,
max_tokens=1,
messages=[{"role": "user", "content": "ping"}],
)
return True
except Exception:
return False
2. Fallback Chain with Circuit Breakers
The fallback chain is the heart of your DR strategy. When the primary provider fails, the chain automatically routes requests to the next provider. A circuit breaker prevents your application from repeatedly hammering a failing provider, which would waste time and potentially worsen the outage.
# fallback_chain.py
import time
from enum import Enum
from typing import List
from llm_provider import BaseProvider, LLMRequest, LLMResponse
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(self, failure_threshold: int = 3, recovery_timeout: int = 60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.state = CircuitState.CLOSED
self.last_failure_time = 0
def can_execute(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
return True
return False
return True # HALF_OPEN: allow one test request
def record_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
class FallbackChain:
def __init__(self, providers: List[BaseProvider]):
self.providers = providers
self.breakers = {p.__class__.__name__: CircuitBreaker() for p in providers}
def generate(self, request: LLMRequest) -> LLMResponse:
errors = []
for provider in self.providers:
name = provider.__class__.__name__
breaker = self.breakers[name]
if not breaker.can_execute():
errors.append(f"{name}: circuit open")
continue
try:
response = provider.generate(request)
breaker.record_success()
return response
except Exception as e:
breaker.record_failure()
errors.append(f"{name}: {str(e)}")
continue
raise RuntimeError(
f"All providers failed. Errors: {'; '.join(errors)}"
)
Using the fallback chain is straightforward. You define your preferred order of providers, and the chain handles the rest.
# usage.py
from providers import OpenAIProvider, AnthropicProvider
from fallback_chain import FallbackChain
from llm_provider import LLMRequest
# Define provider priority: OpenAI first, Anthropic as fallback
chain = FallbackChain([
OpenAIProvider(model="gpt-4o"),
AnthropicProvider(model="claude-3-5-sonnet-20241022"),
])
request = LLMRequest(
prompt="Explain quantum computing in one paragraph.",
system_prompt="You are a helpful science communicator.",
)
try:
response = chain.generate(request)
print(f"[{response.provider}/{response.model}] {response.text}")
except RuntimeError as e:
print(f"DR activated - all providers unavailable: {e}")
# Trigger deeper DR protocols here
3. Semantic Caching for Graceful Degradation
When all providers are down, a semantic cache can serve responses to queries that are similar to previously answered ones. This is far more valuable than exact-match caching because user queries rarely repeat word-for-word.
# semantic_cache.py
import hashlib
import json
import sqlite3
import numpy as np
from typing import Optional
class SemanticCache:
def __init__(self, db_path: str = "cache.db", similarity_threshold: float = 0.92):
self.conn = sqlite3.connect(db_path)
self.threshold = similarity_threshold
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS cache (
id TEXT PRIMARY KEY,
prompt TEXT,
prompt_hash TEXT,
embedding BLOB,
response TEXT,
provider TEXT,
created_at REAL
)
""")
self.conn.commit()
def _hash(self, text: str) -> str:
return hashlib.sha256(text.encode()).hexdigest()
def get(self, prompt: str, prompt_embedding: np.ndarray) -> Optional[str]:
# First try exact match
row = self.conn.execute(
"SELECT response FROM cache WHERE prompt_hash = ?",
(self._hash(prompt),)
).fetchone()
if row:
return row[0]
# Then try semantic match
rows = self.conn.execute(
"SELECT embedding, response FROM cache"
).fetchall()
for emb_bytes, response in rows:
cached_emb = np.frombuffer(emb_bytes, dtype=np.float32)
similarity = np.dot(prompt_embedding, cached_emb) / (
np.linalg.norm(prompt_embedding) * np.linalg.norm(cached_emb)
)
if similarity >= self.threshold:
return response
return None
def put(self, prompt: str, prompt_embedding: np.ndarray,
response: str, provider: str):
self.conn.execute(
"INSERT OR REPLACE INTO cache VALUES (?, ?, ?, ?, ?, ?, ?)",
(
self._hash(prompt),
prompt,
self._hash(prompt),
prompt_embedding.tobytes(),
response,
provider,
__import__("time").time(),
)
)
self.conn.commit()
Putting It All Together: A Resilient LLM Service
Now let's combine the provider abstraction, fallback chain, and semantic cache into a single resilient service that can survive provider outages.
# resilient_llm.py
import time
import logging
from typing import Optional
from providers import OpenAIProvider, AnthropicProvider
from fallback_chain import FallbackChain
from semantic_cache import SemanticCache
from llm_provider import LLMRequest, LLMResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ResilientLLMService:
def __init__(self, cache_ttl: int = 86400):
self.chain = FallbackChain([
OpenAIProvider(model="gpt-4o"),
AnthropicProvider(model="claude-3-5-sonnet-20241022"),
])
self.cache = SemanticCache()
self.cache_ttl = cache_ttl
def generate(self, prompt: str, system_prompt: Optional[str] = None) -> str:
request = LLMRequest(prompt=prompt, system_prompt=system_prompt)
# Step 1: Check semantic cache
# In production, compute embedding via a lightweight embedding model
# For simplicity, we use a zero vector placeholder here
embedding = self._get_embedding(prompt)
cached = self.cache.get(prompt, embedding)
if cached:
logger.info("Cache hit - serving cached response")
return cached
# Step 2: Try the fallback chain
try:
response = self.chain.generate(request)
# Cache the successful response
self.cache.put(prompt, embedding, response.text, response.provider)
logger.info(f"Served by {response.provider}/{response.model}")
return response.text
except RuntimeError as e:
logger.error(f"All providers failed: {e}")
# Step 3: Last resort - return a degraded but functional response
return self._degraded_response(prompt)
def _get_embedding(self, text: str):
# Placeholder: use OpenAI embeddings or a local model
import numpy as np
return np.zeros(1536, dtype=np.float32)
def _degraded_response(self, prompt: str) -> str:
return (
"I'm experiencing temporary connectivity issues with my "
"language models. Here's what I can tell you based on cached "
"knowledge: Your request has been logged and will be processed "
"once service is restored. Please try again in a few moments."
)
# Usage
if __name__ == "__main__":
service = ResilientLLMService()
result = service.generate(
prompt="What are the best practices for Python error handling?",
system_prompt="You are a senior software engineer."
)
print(result)
Disaster Recovery for RAG Systems
Retrieval-augmented generation applications have additional DR concerns because they depend on vector databases. If your vector store goes down, your LLM loses its context. You need a strategy for vector store redundancy and index backup.
Vector Store Backup and Recovery
# vector_dr.py
import json
import os
from datetime import datetime
from typing import List, Dict
import chromadb
class VectorStoreDR:
def __init__(self, collection_name: str, backup_dir: str = "vector_backups"):
self.client = chromadb.PersistentClient(path="chroma_db")
self.collection = self.client.get_or_create_collection(collection_name)
self.backup_dir = backup_dir
os.makedirs(backup_dir, exist_ok=True)
def backup(self) -> str:
"""Create a full backup of the vector collection."""
all_data = self.collection.get(include=["documents", "embeddings", "metadatas"])
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = os.path.join(self.backup_dir, f"backup_{timestamp}.json")
backup_data = {
"collection": self.collection.name,
"timestamp": timestamp,
"ids": all_data["ids"],
"documents": all_data["documents"],
"embeddings": all_data["embeddings"],
"metadatas": all_data["metadatas"],
}
with open(backup_path, "w") as f:
json.dump(backup_data, f)
print(f"Backup created: {backup_path} ({len(all_data['ids'])} records)")
return backup_path
def restore(self, backup_path: str):
"""Restore the vector collection from a backup file."""
with open(backup_path, "r") as f:
data = json.load(f)
# Clear existing data
self.client.delete_collection(data["collection"])
self.collection = self.client.get_or_create_collection(data["collection"])
# Reinsert all records
self.collection.add(
ids=data["ids"],
documents=data["documents"],
embeddings=data["embeddings"],
metadatas=data["metadatas"],
)
print(f"Restored {len(data['ids'])} records from {backup_path}")
def verify_integrity(self) -> bool:
"""Check that the collection is not corrupted."""
try:
count = self.collection.count()
sample = self.collection.peek(limit=5)
return count > 0 and len(sample["ids"]) > 0
except Exception as e:
print(f"Integrity check failed: {e}")
return False
Monitoring and Alerting
Disaster recovery is only effective if you detect failures quickly. Your monitoring stack should track LLM-specific metrics that go beyond traditional uptime checks.
# monitoring.py
import time
import logging
from collections import defaultdict, deque
from dataclasses import dataclass, field
logger = logging.getLogger("llm_monitor")
@dataclass
class LLMMetrics:
request_count: int = 0
failure_count: int = 0
fallback_count: int = 0
cache_hit_count: int = 0
total_tokens: int = 0
total_latency_ms: float = 0
provider_failures: dict = field(default_factory=lambda: defaultdict(int))
recent_latencies: deque = field(default_factory=lambda: deque(maxlen=100))
def record_request(self, provider: str, success: bool,
tokens: int, latency_ms: float, from_cache: bool):
self.request_count += 1
self.total_tokens += tokens
self.total_latency_ms += latency_ms
self.recent_latencies.append(latency_ms)
if from_cache:
self.cache_hit_count += 1
if not success:
self.failure_count += 1
self.provider_failures[provider] += 1
def error_rate(self) -> float:
if self.request_count == 0:
return 0.0
return self.failure_count / self.request_count
def p95_latency(self) -> float:
if not self.recent_latencies:
return 0.0
sorted_latencies = sorted(self.recent_latencies)
idx = int(len(sorted_latencies) * 0.95)
return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
class AlertManager:
def __init__(self, error_threshold: float = 0.1, latency_threshold: float = 5000):
self.error_threshold = error_threshold
self.latency_threshold = latency_threshold
def check_and_alert(self, metrics: LLMMetrics):
if metrics.error_rate() > self.error_threshold:
self._send_alert(
f"HIGH ERROR RATE: {metrics.error_rate():.2%} "
f"(threshold: {self.error_threshold:.2%})"
)
if metrics.p95_latency() > self.latency_threshold:
self._send_alert(
f"HIGH LATENCY: p95 = {metrics.p95_latency():.0f}ms "
f"(threshold: {self.latency_threshold}ms)"
)
for provider, failures in metrics.provider_failures.items():
if failures > 5:
self._send_alert(
f"PROVIDER DEGRADATION: {provider} has {failures} failures"
)
def _send_alert(self, message: str):
# Integrate with PagerDuty, Slack, or your alerting system
logger.critical(f"[ALERT] {message}")
print(f"ALERT: {message}")
Best Practices for LLM Disaster Recovery
- Never hardcode model identifiers. Use configuration management to map logical model names (e.g., "primary-chat-model") to specific provider models. This lets you swap models without deploying code changes.
- Implement request-level timeouts. LLM calls can hang indefinitely. Always set explicit timeouts and treat timeouts as failures that trigger fallback.
- Version and back up your prompts. Store prompt templates in version control. A bad prompt change is a disaster. Enable instant rollback to previous prompt versions.
- Back up fine-tuned model weights externally. If you fine-tune models, store weights in your own object storage. Provider-side fine-tuned models can disappear when accounts are suspended or models are deprecated.
- Use idempotency keys for all write operations. If your LLM application triggers side effects (e.g., sending emails, writing to databases), use idempotency keys to prevent duplicate actions during retries.
- Run regular DR drills. Simulate provider outages in staging by blocking API endpoints. Measure your recovery time objective (RTO) and refine your fallback configuration.
- Implement cost circuit breakers. Track spending in real time and automatically switch to cheaper models or reject requests when daily spend exceeds a threshold.
- Maintain a local model fallback. For critical applications, run a small open-source model (e.g., Llama, Mistral) locally as a last-resort provider. It will produce lower-quality output, but it keeps your service alive.
- Log all provider transitions. When a fallback occurs, log the full context so you can analyze patterns and proactively address recurring issues.
- Document your DR runbook. Write step-by-step procedures for common disaster scenarios: provider outage, vector store corruption, prompt regression, cost spike. Your on-call engineers need actionable instructions, not vague guidelines.
Conclusion
Disaster recovery for LLM applications is not optional—it is a fundamental engineering discipline that separates toy demos from production-grade systems. By building a provider abstraction layer, implementing a fallback chain with circuit breakers, maintaining a semantic cache, backing up your vector stores, and monitoring LLM-specific metrics, you create an architecture that can absorb failures without breaking the user experience. The key insight is that LLM DR is not about preventing failures entirely—providers will go down, models will be deprecated, and costs will spike. Instead, it is about ensuring that when these inevitable disruptions occur, your application degrades gracefully, recovers automatically, and keeps serving your users. Start by implementing the provider abstraction and fallback chain described in this tutorial, then iteratively add caching, monitoring, and runbook documentation as your application scales. Your future on-call engineers will thank you.