Phase 8: Evaluation, Testing & Production · 55 min · Python · Docker · Kubernetes

Production Deployment: Scaling, Fallback, Graceful Degradation

Production voice agents must handle hundreds of concurrent calls with fallback providers and zero data loss.

Hiring signal: Production deployment and scaling knowledge is tested in voice AI system design interviews.

What you will learn

Deploy voice agents with Docker and Kubernetes for concurrent call management
Implement auto-scaling based on active call count, CPU, memory
Configure fallback providers: ASR and TTS failover
Implement graceful degradation when providers fail

The Problem

A voice agent works perfectly in development with 1 call at a time. In production, you get 100 concurrent calls, Deepgram goes down for 5 minutes, OpenAI rate-limits you, and a regional outage takes out your TTS. Without scaling, fallback, and graceful degradation, your agent crashes — and callers hear silence.

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 switches to a backup provider (Deepgram → Whisper). Graceful degradation reduces functionality (if TTS is down, agent responds via text instead of voice). Fallback maintains full function; degradation maintains partial function

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

Unlock the full lesson

You've read the first 2 sections. The rest of this lesson covers Build It, Use It, Ship It, Evaluation, Key Terms, Common Pitfalls, Interview Framing — plus a hands-on lab, quiz, and project artifact.

Create a free account to unlock Phase 0 and Phase 1 of every course — no credit card.

Key terms in this lesson

Agent