Phase 8: Evaluation, Testing & Production · 55 min · Python · Docker · Kubernetes
The Concept
Production Architecture
| Component | Purpose | Scaling Strategy |
|---|
| Load balancer | Distribute calls | Horizontal, auto-scaling |
| Agent instances | Process calls | Containerized, auto-scale |
| ASR provider | Speech-to-text | Multi-provider fallback |
| LLM provider | Response generation | Multi-provider + rate limit handling |
| TTS provider | Text-to-speech | Multi-provider fallback |
| Redis | Session state | Cluster, persistence |
| Monitoring | Health checks | Real-time alerts |
What's the difference between a fallback and graceful degradation?
Fallback = switching to a backup provider that provides the same function. Example: Deepgram ASR goes down → automatically switch to Whisper ASR. The caller doesn't notice — same quality, different provider. Graceful degradation = reducing functionality when no backup is available. Example: TTS provider goes down and backup is also down → agent responds with text displayed on screen (if browser) or plays a pre-recorded "please wait" message (if phone). The caller loses voice responses but the call doesn't crash. Fallback maintains full function with a different provider. Degradation maintains partial function when all providers fail. You need both: fallback chains for common failures, degradation for catastrophic failures.
Scaling Strategy
class AutoScaler:
def __init__(self, min_instances=2, max_instances=20):
self.min = min_instances
self.max = max_instances
self.current = min_instances
self.target_cpu = 60 # Scale up at 60% CPU
def evaluate(self, cpu_usage, active_calls, queue_depth):
"""Evaluate if scaling is needed."""
if cpu_usage > self.target_cpu or queue_depth > 10:
self.scale_up()
elif cpu_usage < 30 and self.current > self.min:
self.scale_down()
def scale_up(self):
"""Add instances."""
new_count = min(self.current + 2, self.max)
print(f"Scaling up: {self.current} → {new_count}")
self.current = new_count
Provider Fallback Chain
class ProviderFallbackChain:
def __init__(self, providers):
"""providers = [{"name": "deepgram", "priority": 1}, ...]"""
self.providers = sorted(providers, key=lambda p: p["priority"])
self.health = {p["name"]: True for p in self.providers}
async def call(self, function_name, *args, **kwargs):
"""Call function with fallback."""
for provider in self.providers:
if not self.health[provider["name"]]:
continue
try:
result = await provider[function_name](*args, **kwargs)
return result
except Exception as e:
print(f"Provider {provider['name']} failed: {e}")
self.health[provider["name"]] = False
continue
raise Exception("All providers failed")
Graceful Degradation
class GracefulDegradation:
def __init__(self):
self.degradation_level = 0 # 0=full, 1=reduced, 2=minimal
def check_health(self, provider_status):
"""Check provider health and set degradation level."""
if all(provider_status.values()):
self.degradation_level = 0 # All healthy
elif provider_status.get("tts", False):
self.degradation_level = 1 # TTS down, use text
else:
self.degradation_level = 2 # Major failure, minimal mode
async def respond(self, text):
"""Respond based on degradation level."""
if self.degradation_level == 0:
return await self.tts.synthesize(text) # Full voice
elif self.degradation_level == 1:
return {"type": "text", "content": text} # Text only
else:
return {"type": "pre-recorded", "audio": "please_wait.wav"} # Minimal
Rate Limit Handling
class RateLimitHandler:
def __init__(self, limits):
"""limits = {"openai": {"rpm": 500, "tpm": 150000}}"""
self.limits = limits
self.usage = {provider: {"rpm": 0, "tpm": 0} for provider in limits}
def can_call(self, provider, tokens=0):
"""Check if we can make a call."""
usage = self.usage[provider]
limits = self.limits[provider]
return usage["rpm"] < limits["rpm"] and usage["tpm"] + tokens < limits["tpm"]
def record_usage(self, provider, tokens):
"""Record API usage."""
self.usage[provider]["rpm"] += 1
self.usage[provider]["tpm"] += tokens
Health Checks
class HealthChecker:
def __init__(self):
self.checks = {}
def register_check(self, name, check_fn):
"""Register a health check."""
self.checks[name] = check_fn
async def run_checks(self):
"""Run all health checks."""
results = {}
for name, check_fn in self.checks.items():
try:
results[name] = await check_fn()
except Exception:
results[name] = False
return results