Scaling Realtime Agent Sessions: Connections, Queues, and State

A voice agent handling ten calls is a prototype. One handling a thousand simultaneous calls is a distributed system with sticky sessions, connection limits, queue back-pressure, and graceful drain. The transition from ten to a thousand is where most teams ship an outage.

The Scaling Challenge

Why Voice Agents Are Different

# Traditional API: Stateless, short-lived
request → process → response → done

# Voice Agent: Stateful, long-lived
session_start → streaming_audio → conversation → tool_calls → session_end
              (minutes to hours)

Voice agents have unique scaling requirements:

• Long-lived connections: Sessions last minutes to hours
• Stateful interactions: Each turn depends on conversation history
• Real-time streaming: Sub-second latency requirements
• Resource-heavy: LLM calls block for seconds

Capacity Planning

class CapacityPlanner:
    def __init__(self):
        self.pod_capacity = {
            'fastapi': 20-40,  # Concurrent Realtime sessions
            'websocket': 100-300,  # WebSocket connections
            'http': 1000,  # Short-lived HTTP requests
        }
        
    def calculate_pods_needed(self, expected_concurrent_calls: int) -> int:
        """Calculate pods needed for expected load."""
        calls_per_pod = self.pod_capacity['fastapi']
        pods_needed = expected_concurrent_calls / calls_per_pod
        
        # Add 20% headroom
        return int(pods_needed * 1.2)

Architecture Patterns

1. Stateless Design with External State

class StatelessVoiceAgent:
    def __init__(self, redis_client, db_client):
        self.redis = redis_client
        self.db = db_client
        
    async def handle_session(self, session_id: str, audio_chunk: bytes):
        """Handle audio without maintaining local state."""
        # Read state from Redis
        state = await self.redis.get(f"session:{session_id}")
        
        # Process audio
        response = await self.process_audio(audio_chunk, state)
        
        # Write state back to Redis
        await self.redis.set(
            f"session:{session_id}",
            response['new_state'],
            ex=3600  # 1 hour TTL
        )
        
        return response['audio']

2. Sticky Sessions for WebSocket

# Kubernetes Ingress with session affinity
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: voice-agent-ingress
  annotations:
    nginx.ingress.kubernetes.io/affinity: "cookie"
    nginx.ingress.kubernetes.io/session-cookie-name: "VOICE_AGENT_SESSION"
    nginx.ingress.kubernetes.io/session-cookie-max-age: "3600"
spec:
  rules:
  - host: voice.example.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: voice-agent-service
            port:
              number: 8080

3. Connection Pooling

class ConnectionPool:
    def __init__(self, max_connections: int = 20):
        self.max_connections = max_connections
        self.semaphore = asyncio.Semaphore(max_connections)
        self.connections = []
        
    async def get_connection(self):
        """Get connection from pool."""
        async with self.semaphore:
            if self.connections:
                return self.connections.pop()
            return await self.create_connection()
    
    async def release_connection(self, conn):
        """Return connection to pool."""
        if len(self.connections) < self.max_connections:
            self.connections.append(conn)
        else:
            await conn.close()

Queue Management

Message Queue for Agent Communication

import redis.asyncio as redis

class AgentMessageQueue:
    def __init__(self):
        self.redis = redis.Redis()
        self.stream_name = "agent_messages"
        
    async def enqueue_message(self, agent_id: str, message: dict):
        """Add message to agent's queue."""
        await self.redis.xadd(
            self.stream_name,
            {
                "agent_id": agent_id,
                "message": json.dumps(message),
                "timestamp": time.time()
            }
        )
    
    async def dequeue_messages(self, agent_id: str, count: int = 10):
        """Get messages for specific agent."""
        messages = await self.redis.xread(
            {self.stream_name: "$"},
            count=count,
            block=100  # Wait 100ms for messages
        )
        
        return [
            json.loads(msg[b"message"])
            for msg in messages.get(self.stream_name, [])
        ]

Priority Queue for Task Scheduling

class PriorityTaskQueue:
    def __init__(self):
        self.redis = redis.Redis()
        self.queue_key = "task_queue"
        
    async def add_task(self, task: dict, priority: int = 0):
        """Add task with priority (higher = more urgent)."""
        score = priority * 1000000 + time.time()
        await self.redis.zadd(
            self.queue_key,
            {json.dumps(task): score}
        )
    
    async def get_next_task(self) -> Optional[dict]:
        """Get highest priority task."""
        result = await self.redis.zpopmin(self.queue_key)
        if result:
            task_json, _ = result[0]
            return json.loads(task_json)
        return None

Auto-Scaling Configuration

Kubernetes Horizontal Pod Autoscaler

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: voice-agent-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: voice-agent
  minReplicas: 10
  maxReplicas: 200
  metrics:
  - type: Pods
    pods:
      metric:
        name: active_calls
      target:
        type: AverageValue
        averageValue: "25"
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 30
      policies:
      - type: Pods
        value: 5
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Pods
        value: 2
        periodSeconds: 120

Custom Metrics for Scaling

class VoiceAgentMetrics:
    def __init__(self):
        self.metrics = {
            'active_calls': 0,
            'queue_depth': 0,
            'p95_latency_ms': 0
        }
    
    def record_call_start(self):
        self.metrics['active_calls'] += 1
        
    def record_call_end(self):
        self.metrics['active_calls'] -= 1
        
    def get_scaling_metric(self) -> float:
        """Return metric for HPA."""
        return self.metrics['active_calls']

Graceful Drain and Deployment

Rolling Updates

apiVersion: apps/v1
kind: Deployment
metadata:
  name: voice-agent
spec:
  replicas: 20
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      terminationGracePeriodSeconds: 660
      containers:
      - name: voice-agent
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 15"]

Drain Endpoint

class DrainableAgent:
    def __init__(self):
        self.is_draining = False
        self.active_sessions = set()
        
    async def drain(self):
        """Start draining process."""
        self.is_draining = True
        
        # Stop accepting new sessions
        # Wait for active sessions to complete
        while self.active_sessions:
            await asyncio.sleep(1)
            
        # Ready for shutdown
        return True
    
    async def handle_session(self, session_id: str):
        """Handle session with drain awareness."""
        if self.is_draining:
            return {"error": "Service is draining"}
            
        self.active_sessions.add(session_id)
        try:
            # Process session
            pass
        finally:
            self.active_sessions.discard(session_id)

State Management Patterns

Redis for Hot State

class RedisStateManager:
    def __init__(self):
        self.redis = redis.Redis()
        
    async def get_session_state(self, session_id: str) -> dict:
        """Get session state from Redis."""
        state_json = await self.redis.get(f"session:{session_id}")
        return json.loads(state_json) if state_json else {}
        
    async def update_session_state(self, session_id: str, 
                                  updates: dict):
        """Update session state atomically."""
        state = await self.get_session_state(session_id)
        state.update(updates)
        
        await self.redis.set(
            f"session:{session_id}",
            json.dumps(state),
            ex=3600
        )

PostgreSQL for Durable State

class PostgresStateManager:
    def __init__(self):
        self.engine = create_async_engine(
            "postgresql+asyncpg://...",
            pool_size=20,
            max_overflow=10
        )
        
    async def save_conversation(self, session_id: str, 
                               messages: list):
        """Save conversation history to Postgres."""
        async with self.engine.begin() as conn:
            await conn.execute(
                insert(conversations).values(
                    session_id=session_id,
                    messages=json.dumps(messages),
                    updated_at=datetime.utcnow()
                )
            )

Monitoring and Alerting

Key Metrics to Monitor

class ProductionMetrics:
    def __init__(self):
        self.metrics = {
            'active_sessions': 0,
            'queue_depth': 0,
            'p95_latency_ms': 0,
            'error_rate': 0,
            'memory_usage_mb': 0
        }
        
    def check_alerts(self) -> list:
        """Check for alert conditions."""
        alerts = []
        
        if self.metrics['queue_depth'] > 100:
            alerts.append('HIGH_QUEUE_DEPTH')
            
        if self.metrics['p95_latency_ms'] > 500:
            alerts.append('HIGH_LATENCY')
            
        if self.metrics['error_rate'] > 0.05:
            alerts.append('HIGH_ERROR_RATE')
            
        return alerts

Best Practices

1. Right-Size Your Pods

# Per-pod capacity planning
pod_limits = {
    'cpu': '2000m',  # 2 CPU cores
    'memory': '4Gi',  # 4GB RAM
    'concurrent_sessions': 30,  # Target
    'memory_per_session': '100MB'  # Budget
}

# Calculate pod count
pods_needed = total_concurrent_sessions / pod_limits['concurrent_sessions']

2. Use Connection Pooling

# Don't create new connections per session
class BadPattern:
    async def handle_session(self):
        db = await create_connection()  # Bad: new connection each time
        await db.query(...)

class GoodPattern:
    def __init__(self):
        self.pool = ConnectionPool(max_size=20)
        
    async def handle_session(self):
        async with self.pool.get_connection() as db:  # Good: reuse
            await db.query(...)

3. Implement Circuit Breakers

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0
        
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
    def record_success(self):
        self.failure_count = 0
        
    def should_allow_request(self) -> bool:
        if self.failure_count < self.failure_threshold:
            return True
            
        if time.time() - self.last_failure_time > self.recovery_timeout:
            return True
            
        return False

Conclusion

Scaling voice agents to 1000+ concurrent sessions requires:

1. Externalize state: Use Redis for hot state, Postgres for durability
2. Sticky sessions: For WebSocket streaming, not entire lifecycle
3. Custom metrics: Scale on active calls, not CPU
4. Graceful drain: Don’t kill live calls on deployment
5. Connection pooling: Never create new connections per session

The transition from prototype to production is where most teams fail. By implementing these patterns—stateless design, proper queue management, and careful scaling—you can handle thousands of concurrent sessions while maintaining sub-second latency.