Production: Scaling and Monitoring

Updated November 2, 2026 — Audio models. Several models in this course are deprecated: tts-1, tts-1-hd, and gpt-4o-mini-tts will stop working on January 6, 2027 (replacement: gpt-realtime-2.1-mini, which the OpenAI TTS guide states is not supported by /v1/audio/speech: speech synthesis migration must go through the Realtime API); gpt-realtime, gpt-4o-realtime, and legacy audio models on January 20, 2027 (gpt-realtime-2.1, gpt-realtime-2.1-mini, gpt-audio-1.5); whisper-1, gpt-4o-transcribe, and /v1/audio/translations on February 26, 2027 (gpt-transcribe, gpt-live-transcribe, v1/realtime/translations). For full-duplex voice, GPT-Live 1 has been available in the API since September 10, 2026.

Updated November 3, 2026. Cost management previously fit in a single sentence; it now reflects the billing model described by OpenAI (billing per response, conversation returned every turn, audio counted in tokens, automatic caching, transcription billed separately) and truncation via retention_ratio, whose format was verified against the openai SDK types version 3.42.1. Source: Cost optimization.

Production: Scaling and Monitoring

Moving from a voice agent prototype to a reliable, scalable production service demands close attention to infrastructure, monitoring, and error handling. This final lesson covers the operational essentials for deploying a voice agent in real-world conditions.

Production Architecture

In production, the voice agent runs behind a load balancer and a pool of workers:

Managing Concurrent Sessions

Each WebSocket connection maintains a Realtime API session. The number of simultaneous sessions is limited by your OpenAI rate limits and your workers’ capacity:

import asyncio
from dataclasses import dataclass

@dataclass
class SessionManager:
    """Manages concurrent Realtime API sessions."""
    max_sessions: int = 50
    active_sessions: dict = None
    semaphore: asyncio.Semaphore = None

    def __post_init__(self):
        self.active_sessions = {}
        self.semaphore = asyncio.Semaphore(self.max_sessions)

    async def create_session(self, session_id: str, config: dict):
        """Creates a new session with concurrency control."""
        async with self.semaphore:
            if session_id in self.active_sessions:
                raise ValueError(f"Session {session_id} already active")

            self.active_sessions[session_id] = {
                "created_at": asyncio.get_event_loop().time(),
                "config": config,
                "status": "active"
            }
            return session_id

    async def close_session(self, session_id: str):
        """Properly closes a session."""
        if session_id in self.active_sessions:
            del self.active_sessions[session_id]

    @property
    def active_count(self) -> int:
        return len(self.active_sessions)

    @property
    def available_slots(self) -> int:
        return self.max_sessions - self.active_count

session_manager = SessionManager(max_sessions=50)

Structured Logging

Log every interaction for debugging and compliance:

import logging
import json
from datetime import datetime

class ConversationLogger:
    """Structured logger for voice conversations."""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.logger = logging.getLogger(f"voice.{session_id}")
        self.events = []

    def log_event(self, event_type: str, data: dict = None):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "session_id": self.session_id,
            "event_type": event_type,
            "data": data or {}
        }
        self.events.append(entry)
        self.logger.info(json.dumps(entry))

    def log_transcription(self, role: str, text: str):
        self.log_event("transcription", {
            "role": role,
            "text": text
        })

    def log_function_call(self, name: str, args: dict, result: dict, duration_ms: float):
        self.log_event("function_call", {
            "name": name,
            "arguments": args,
            "result": result,
            "duration_ms": round(duration_ms)
        })

    def log_error(self, error_type: str, message: str):
        self.log_event("error", {
            "type": error_type,
            "message": message
        })

    async def save_to_db(self, db_pool):
        """Persists the complete log to the database."""
        async with db_pool.acquire() as conn:
            await conn.execute(
                "INSERT INTO conversation_logs (session_id, events, created_at) "
                "VALUES ($1, $2, NOW())",
                self.session_id, json.dumps(self.events)
            )

Prometheus Metrics

Expose metrics for real-time monitoring:

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Key metrics
sessions_active = Gauge(
    "voice_sessions_active",
    "Number of active voice sessions"
)

session_duration = Histogram(
    "voice_session_duration_seconds",
    "Duration of voice sessions",
    buckets=[30, 60, 120, 300, 600, 1800]
)

response_latency = Histogram(
    "voice_response_latency_ms",
    "Response latency (end of speech → first audio)",
    buckets=[200, 400, 600, 800, 1000, 1500, 2000, 3000]
)

errors_total = Counter(
    "voice_errors_total",
    "Total number of errors",
    ["error_type"]
)

tokens_used = Counter(
    "voice_tokens_total",
    "Tokens consumed",
    ["type"]  # input_audio, output_audio, input_text, output_text
)

# Start the metrics server
start_http_server(9090)

Cost Management

According to OpenAI’s Cost optimization guide, a Realtime conversation session is billed per response, in input and output tokens for text, audio, and images. Three mechanisms drive most of the bill:

• The entire conversation is sent back to the model with every response: later turns cost more than earlier ones.
• Audio is counted in tokens: 1 token per 100 ms of user audio, 1 token per 50 ms of assistant audio.
• Prompt caching applies automatically: as long as the beginning of the conversation doesn’t change, it’s billed at the reduced cached token rate. Modifying or deleting an older element “breaks” the cache from that point onward.

Input transcription, if enabled, is billed separately at its own model’s rate: gpt-transcribe is charged by duration, read from the conversation.item.input_audio_transcription.completed event. Consumption for each response appears in response.done, under response.usage (total_tokens, input_tokens, output_tokens, plus the breakdown for text, audio, image, and cache).

To cap context size—and thus the cost of later turns—the session accepts a truncation rule: beyond the limit, the oldest messages are dropped. With retention_ratio set below 1, the server removes slightly more than necessary, which spaces out truncations and preserves the cache:

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "type": "realtime",
        "truncation": {
            "type": "retention_ratio",
            "retention_ratio": 0.8,
            "token_limits": {"post_instructions": 8000},
        },
    },
}))

The trade-off: an agent that retains less context forgets faster. Complement this with session limits in your code:

class CostController:
    """Controls costs for a Realtime session."""

    def __init__(self, max_duration_s: int = 600, max_tokens: int = 50000):
        self.max_duration_s = max_duration_s
        self.max_tokens = max_tokens
        self.total_tokens = 0
        self.start_time = None

    def start(self):
        self.start_time = asyncio.get_event_loop().time()

    def update_tokens(self, usage: dict):
        self.total_tokens += usage.get("total_tokens", 0)

    @property
    def elapsed_seconds(self) -> float:
        if not self.start_time:
            return 0
        return asyncio.get_event_loop().time() - self.start_time

    @property
    def should_terminate(self) -> bool:
        if self.elapsed_seconds > self.max_duration_s:
            return True
        if self.total_tokens > self.max_tokens:
            return True
        return False

    @property
    def warning_message(self) -> str:
        remaining = self.max_duration_s - self.elapsed_seconds
        if remaining < 60:
            return f"This session will end in {int(remaining)} seconds."
        return ""

Production Error Handling

async def production_event_loop(ws, logger, cost_ctrl):
    """Robust event loop for production."""
    cost_ctrl.start()

    try:
        async for raw in ws:
            event = json.loads(raw)
            t = event["type"]

            # Check cost limits
            if cost_ctrl.should_terminate:
                logger.log_event("session_limit_reached")
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {
                        "type": "message",
                        "role": "user",
                        "content": [{"type": "input_text",
                                     "text": "The session is approaching its duration limit."}]
                    }
                }))
                break

            if t == "response.done":
                usage = event["response"].get("usage", {})
                cost_ctrl.update_tokens(usage)
                tokens_used.labels(type="total").inc(usage.get("total_tokens", 0))

            elif t == "error":
                error = event["error"]
                logger.log_error(error["type"], error["message"])
                errors_total.labels(error_type=error["type"]).inc()

                if error["type"] == "rate_limit_error":
                    await asyncio.sleep(1)

    except websockets.exceptions.ConnectionClosed as e:
        logger.log_error("connection_closed", str(e))
    except Exception as e:
        logger.log_error("unexpected", str(e))
    finally:
        session_duration.observe(cost_ctrl.elapsed_seconds)
        sessions_active.dec()

Key Takeaways

• Production architecture separates load balancer, asyncio workers, database, and monitoring
• A SessionManager with a semaphore controls the number of simultaneous sessions
• JSON structured logging enables debugging and compliance (conversation recording)
• Prometheus metrics (TTFA, duration, tokens, errors) are essential for daily operations
• The CostController limits session duration and tokens to prevent budget overruns

Test Your Knowledge

Five questions covering the entire voice chain, from WebSocket to production deployment.

1. Why does the Realtime API use WebSocket instead of HTTP requests?

Answer: Because a persistent, bidirectional connection lets you stream microphone audio and receive responses continuously, without paying the overhead of establishing a request for every exchange. This is the prerequisite for low-latency conversation.

2. What must you send immediately after `session.created`, and why?

Answer: session.update, immediately: this is where you configure the voice, instructions, tools, and audio format. Until this message is sent, the session runs with default values.

3. Why is a proxy server essential in production?

Answer: It prevents exposing the API key in the client, and it lets you authenticate the user, cap consumption, and log calls—three things impossible if the browser talks directly to the API.

4. What is the purpose of the `prompt` parameter in the transcription API?

Answer: To steer recognition toward expected vocabulary: proper nouns, brands, domain jargon. Terms you list there are transcribed much more accurately.

5. What is the main lever for perceived latency in a voice agent?

Answer: End-to-end streaming: playing audio as soon as the first fragments arrive instead of waiting for the full response, and handling interruption so the user can cut in at any time.