Chaos Engineering for AI: What Happens When You Pull the Plug on GPT-4?

Your chatbot runs on GPT-4. It's great. Users love it. You're scaling fast. Then OpenAI has an outage. Five minutes. Not long. Your app returns 500 errors for five minutes. Users see a broken bot. Some leave negative reviews. Your customer support queue explodes. This is the moment you learn what you should have learned in a test environment: your system is not designed for the failure modes that actually happen in production. Chaos engineering for AI is about systematically injecting those failures before they happen in production. It's not about finding bugs. It's about building systems that degrade gracefully when something goes wrong. And something will go wrong.

Why AI Systems Need Chaos Engineering

Traditional software is deterministic. You can test edge cases. You can trace logic paths. When it fails, it usually fails loud. AI systems are different: - **Non-deterministic**: The same input produces different outputs - **Black box**: You can't always predict when or why it'll fail - **Cascading failures**: A small error in one step (bad embedding) cascades into completely wrong answers downstream - **Fail silently**: Your LLM might hallucinate with full confidence. You won't know it's wrong until production This is why chaos engineering matters for AI. You can't test your way to confidence. You need to inject failures and watch how your system responds.

The Failure Modes You Should Test

Model Timeout

Your LLM takes too long to respond. The request times out after 30 seconds. What happens?

class AIWithTimeout:
 def __init__(self, model, timeout_seconds=30):
 self.model = model
 self.timeout = timeout_seconds

 def get_response(self, prompt):
 try:
 response = self.model.complete(prompt, timeout=self.timeout)
 return response
 except TimeoutError:
 # What's your fallback?
 # Option 1: Return cached response
 # Option 2: Return a default message
 # Option 3: Escalate to human
 return self._fallback_response(prompt)

 def _fallback_response(self, prompt):
 # Test this path!
 cached = self.cache.get(prompt)
 if cached:
 return cached
 return "I'm having trouble answering that right now. A human will help you shortly."

**Chaos test**: Inject timeouts randomly. Verify your fallback works. Measure how long users wait.

Provider Outage

OpenAI, Anthropic, any provider can have outages. Your system should survive them.

async def call_llm_with_failover(prompt, providers):
 """
 Try primary provider, fall back to secondary, then to cached/local model
 Chaos test: fail providers in sequence and verify graceful degradation
 """
 for provider in providers:
 try:
 response = await provider.complete(prompt, timeout=5)
 return {"source": "primary", "response": response}
 except Exception as e:
 logger.warning(f"Provider {provider.name} failed: {e}")
 continue

 # All providers failed
 try:
 response = self.local_fallback_model.complete(prompt)
 return {"source": "fallback", "response": response}
 except:
 return {"source": "error", "response": "Service temporarily unavailable"}

**Chaos test**: Kill your primary provider. Kill your secondary. Verify users still get responses, even if degraded.

Degraded Embedding Quality

Your RAG system embeds documents. An embedding model goes down or gets corrupted. What happens?

class RAGWithEmbeddingFailure:
 def __init__(self, embedding_model, semantic_threshold=0.8):
 self.embeddings = embedding_model
 self.threshold = semantic_threshold

 def retrieve(self, query, top_k=5):
 try:
 query_embedding = self.embeddings.embed(query)
 except Exception:
 # Embedding model failed
 # Fallback: use BM25 keyword search instead
 return self.keyword_search(query, top_k)

 results = self.search(query_embedding, top_k=top_k)

 # Verify quality of results
 for result in results:
 if result["score"] < self.threshold:
 logger.warning(f"Low-confidence retrieval: {result['score']:.2f}")

 return results

 def keyword_search(self, query, top_k):
 # Keyword-based search when semantic search fails
 # It's not perfect, but it's better than nothing
 pass

**Chaos test**: Return garbage embeddings. Verify your retrieval degrades gracefully, not crashes.

Corrupted Context

Long-context models are powerful but fragile. If the context window gets corrupted, the model will hallucinate with confidence:

def validate_context(context, expected_size):
 """
 Verify context wasn't corrupted before passing to LLM
 Chaos test: inject corrupted context and verify this catches it
 """
 if not context or len(context) < expected_size * 0.8:
 # Context suspiciously small
 logger.error(f"Context size mismatch: expected {expected_size}, got {len(context)}")
 return False

 # Check for common corruption patterns
 if "�" in context or context.count("\x00") > 0:
 logger.error("Context contains null bytes or encoding errors")
 return False

 if context.count("\n") < expected_size // 1000:
 # Context suspiciously uniform
 logger.warning("Context lacks newlines; may be truncated")
 return False

 return True

def get_response(prompt, context):
 if not validate_context(context, expected_size=5000):
 return "I couldn't process that document. Please try again."

 return self.model.complete(f"{context}\n\nQ: {prompt}")

**Chaos test**: Truncate context randomly. Verify validation catches it and you don't return hallucinations as fact.

Rate Limit Hits

Your API provider rate-limits you. You're calling faster than you should. What's your backoff strategy?

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
 stop=stop_after_attempt(5),
 wait=wait_exponential(multiplier=1, min=2, max=60),
)
def call_model_with_backoff(prompt):
 """
 Exponential backoff: wait 2s, 4s, 8s, 16s, 32s on failures
 Chaos test: trigger rate limits and verify backoff prevents cascade failures
 """
 try:
 return self.model.complete(prompt)
 except RateLimitError as e:
 logger.warning(f"Rate limited. Backing off: {e}")
 raise # Trigger retry with exponential backoff

# Test: What happens when you call this 1000 times in parallel?
# Do you retry forever? Do you give up? Do you queue?

**Chaos test**: Simulate rate limit responses. Verify your backoff doesn't hammer the provider. Verify you don't lose requests.

Building Game Day Playbooks

Chaos engineering discovers failure modes. Game day playbooks turn discovery into muscle memory. A game day is a scheduled exercise where your team simulates a failure and practices the response: **Day: Thursday 2pm** **Scenario**: OpenAI API is completely down for 2 hours. **Steps**: 1. Engineer 1 blocks OpenAI API responses at the network layer (use tc, iptables, or a proxy) 2. Engineer 2 monitors dashboards and records the timeline of what breaks 3. Engineer 3 runs the incident playbook: check failover models, verify cached responses work, monitor error rates 4. Entire team joins war room call at 2:15pm 5. Playbook execution: - T+5min: Detect OpenAI down, switch to local fallback model - T+15min: Degrade service quality but maintain availability - T+60min: Check OpenAI status page, update users - T+120min: OpenAI restored, verify cache consistency, switch back 6. Post-game review: What worked? What surprised us? What do we change? **Expected outcome**: If this happens in production, your team has muscle memory. You don't panic. You execute. Here's what a playbook looks like in code:

class AISystemPlaybook:
 def __init__(self):
 self.fallback_models = {
 "primary": GPT4Client(),
 "secondary": Claude3Client(),
 "local": LocalLlamaModel(),
 }

 def handle_provider_failure(self, provider_name):
 """
 Executed on game day when provider goes down
 Chaos test: actually fails the provider and runs this
 """
 logger.critical(f"Provider {provider_name} failed. Executing failover playbook.")

 # Step 1: Disable primary provider
 self.fallback_models["primary"].disable()

 # Step 2: Switch traffic to secondary
 self.current_provider = self.fallback_models["secondary"]
 logger.info("Switched to secondary provider")

 # Step 3: Degrade quality where necessary
 self.config.update({
 "max_tokens": 256, # Smaller responses to save tokens
 "response_time_target_ms": 5000, # More lenient
 })

 # Step 4: Alert team
 self.alert_team("CRITICAL: Primary LLM provider down. Running on secondary.")

 # Step 5: Set up monitoring
 self.metrics.track_provider_status(provider_name)

 return True

 def verify_failover_successful(self):
 """Test that failover actually works"""
 assert self.current_provider == self.fallback_models["secondary"]
 assert self.call_model("test prompt") is not None
 logger.info("Failover verified successful")

Chaos Engineering Tools for AI

You don't need fancy infrastructure. Start simple: **For API failures**: Chaos Monkey, Gremlin, or even simple network blocking:

import subprocess

# Block OpenAI API temporarily
subprocess.run(["sudo", "iptables", "-A", "OUTPUT",
 "-d", "api.openai.com", "-j", "REJECT"])

# Run your tests
run_tests()

# Restore connectivity
subprocess.run(["sudo", "iptables", "-D", "OUTPUT",
 "-d", "api.openai.com", "-j", "REJECT"])

**For model failures**: Mock providers that randomly fail:

class ChaosModelClient:
 def __init__(self, real_client, failure_rate=0.1):
 self.client = real_client
 self.failure_rate = failure_rate

 def complete(self, prompt):
 if random.random() < self.failure_rate:
 raise Exception("Simulated model failure")
 return self.client.complete(prompt)

# Use in tests
chaos_client = ChaosModelClient(GPT4Client(), failure_rate=0.2)
result = chaos_client.complete("test") # 20% chance of failure

**For latency injection**: Delay responses:

import time

class LatencyInjectingClient:
 def __init__(self, real_client, latency_ms=5000):
 self.client = real_client
 self.latency = latency_ms / 1000.0

 def complete(self, prompt):
 time.sleep(self.latency)
 return self.client.complete(prompt)

Measuring Resilience

After you inject failures, measure: - **Availability**: What percentage of requests succeeded despite failures? - **Latency under failure**: Did response times degrade gracefully? - **Fallback quality**: When the primary failed, how good were the fallback responses? - **Recovery time**: How long until you switched back to primary? - **User impact**: How many users were affected? For how long? Track these metrics over time. Your resilience should improve with each game day.

Start Small, Test Often

You don't need a complex setup: 1. **Pick one failure mode** (e.g., API timeout) 2. **Build a test that injects that failure** 3. **Verify your fallback works** 4. **Run it weekly** 5. **Add one new failure mode each month** Within six months, your system will be dramatically more resilient. You'll sleep better knowing you're not one outage away from broken production.