Voice Agents and Speech Interfaces

import asyncio
import websockets
import json
import base64
REALTIME_URL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
async def voice_agent():
    """Connect to OpenAI Realtime API for speech-to-speech interaction."""
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1",
        }
    async with websockets.connect(REALTIME_URL, extra_headers=headers) as ws:
        # Configure the session with tools and voice
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
            "modalities": ["text", "audio"],
            "voice": "alloy",
            "instructions": (
            "You are a helpful customer service agent for Acme Corp. "
            "You can look up orders, check shipping status, and process "
            "returns. Be concise and friendly."
            ),
            "tools": [
            {
            "type": "function",
            "name": "lookup_order",
            "description": "Look up an order by order ID or customer email",
            "parameters": {
            "type": "object",
            "properties": {
            "order_id": {"type": "string"},
            "email": {"type": "string"},
            },
            },
            },
            {
            "type": "function",
            "name": "check_shipping",
            "description": "Check shipping status for an order",
            "parameters": {
            "type": "object",
            "properties": {
            "order_id": {"type": "string"},
            },
            "required": ["order_id"],
            },
            },
            ],
            "turn_detection": {
            "type": "server_vad",
            "threshold": 0.5,
            "silence_duration_ms": 500,
            },
            },
            }))
        # Main event loop: process audio and tool calls as they arrive
        async for message in ws:
            event = json.loads(message)
            if event["type"] == "response.audio.delta":
                # Stream audio chunks to the speaker
                audio_bytes = base64.b64decode(event["delta"])
                await play_audio_chunk(audio_bytes)
            elif event["type"] == "response.function_call_arguments.done":
                # Execute tool call and return result
                result = await execute_tool(
                    event["name"],
                    json.loads(event["arguments"]),
                    )
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {
                    "type": "function_call_output",
                    "call_id": event["call_id"],
                    "output": json.dumps(result),
                    },
                    }))
                # Trigger the model to continue speaking
                await ws.send(json.dumps({"type": "response.create"}))

from livekit.agents import AgentSession, Agent, RoomInputOptions
from livekit.agents.llm import function_tool
from livekit.plugins import deepgram, openai, cartesia
class CustomerServiceAgent(Agent):
    """Voice agent with tool-calling capabilities."""
    def __init__(self):
        super().__init__(
            instructions=(
            "You are a customer service agent for Acme Corp. "
            "Help users with order lookups and shipping inquiries. "
            "Keep responses concise since this is a voice conversation."
            ),
            )
@function_tool()
async def lookup_order(self, order_id: str) -> str:
    """Look up order details by order ID."""
    order = await db.get_order(order_id)
    if not order:
        return f"No order found with ID {order_id}."
        return (
            f"Order {order_id}: {order.item_count} items, "
            f"total ${order.total:.2f}, status: {order.status}."
            )
        @function_tool()
        async def check_shipping(self, order_id: str) -> str:
            """Check the current shipping status of an order."""
            tracking = await shipping_api.get_status(order_id)
            return (
                f"Order {order_id} shipped via {tracking.carrier}. "
                f"Tracking: {tracking.number}. "
                f"Estimated delivery: {tracking.eta}."
                )
            async def entrypoint(ctx):
                """LiveKit agent entrypoint."""
                session = AgentSession(
                    stt=deepgram.STT(model="nova-3"),
                    llm=openai.LLM(model="gpt-4o"),
                    tts=cartesia.TTS(voice="customer-service-female"),
                    )
                await session.start(
                    room=ctx.room,
                    agent=CustomerServiceAgent(),
                    room_input_options=RoomInputOptions(),
                    )

import asyncio
import time
class LatencyOptimizedVoiceAgent:
    """Voice agent with aggressive latency optimization."""
    def __init__(self, stt, llm, tts):
        self.stt = stt
        self.llm = llm
        self.tts = tts
        self.metrics = LatencyMetrics()
    async def handle_utterance(self, audio_stream):
        """Process user speech with overlapping pipeline stages."""
        start = time.monotonic()
        # Stage 1: Streaming STT (partial results as user speaks)
        transcript = ""
        async for partial in self.stt.stream_transcribe(audio_stream):
            transcript = partial.text
            if partial.is_final:
                break
                stt_done = time.monotonic()
                self.metrics.record("stt_latency", stt_done - start)
                # Stage 2: Check if this needs a tool call
                needs_tool = await self._quick_intent_check(transcript)
                if needs_tool:
                    # Start filler speech immediately (non-blocking)
                    filler_task = asyncio.create_task(
                        self.tts.speak("One moment, let me look that up.")
                        )
                    # Execute tool in parallel with filler speech
                    tool_result = await self._execute_tool(transcript)
                    await filler_task # Ensure filler finishes before response
                    llm_input = f"User asked: {transcript}\nTool result: {tool_result}"
                else:
                    llm_input = transcript
                    # Stage 3: Stream LLM response directly to TTS
                    llm_start = time.monotonic()
                    token_buffer = ""
                    first_audio = False
                    async for token in self.llm.stream(llm_input):
                        token_buffer += token
                        # Send to TTS in sentence-sized chunks for natural speech
                        if self._is_sentence_boundary(token_buffer):
                            tts_task = asyncio.create_task(
                                self.tts.speak_streamed(token_buffer)
                                )
                            if not first_audio:
                                self.metrics.record("ttfb", time.monotonic() - start)
                                first_audio = True
                                token_buffer = ""
                                # Flush remaining tokens
                                if token_buffer:
                                    await self.tts.speak_streamed(token_buffer)
                                    total = time.monotonic() - start
                                    self.metrics.record("total_latency", total)
                                def _is_sentence_boundary(self, text: str) -> bool:
                                    """Detect natural speech boundaries for TTS chunking."""
                                    return any(text.rstrip().endswith(p) for p in [".", "!", "?", ",", ":"])

import asyncio
from enum import Enum
class TurnState(Enum):
    LISTENING = "listening"
    PROCESSING = "processing"
    SPEAKING = "speaking"
class TurnManager:
    """Manage conversational turn-taking with interruption support."""
    def __init__(self, vad, agent):
        self.vad = vad
        self.agent = agent
        self.state = TurnState.LISTENING
        self._speaking_task: asyncio.Task = None
        self._audio_queue = asyncio.Queue()
async def on_audio_frame(self, frame):
    """Process incoming audio frames for turn detection."""
    speech_detected = await self.vad.process_frame(frame)
    if self.state == TurnState.SPEAKING and speech_detected:
        # User is interrupting; stop speaking immediately
        await self._handle_interruption()
    elif self.state == TurnState.LISTENING and speech_detected:
        self._audio_queue.put_nowait(frame)
    elif self.state == TurnState.LISTENING and not speech_detected:
        # Check if user finished their turn (silence after speech)
        if not self._audio_queue.empty():
            silence_duration = await self.vad.get_silence_duration()
            if silence_duration > 0.5: # 500ms silence threshold
                await self._end_of_turn()
async def _handle_interruption(self):
    """Handle user interruption during agent speech."""
    # Cancel current speech output
    if self._speaking_task and not self._speaking_task.done():
        self._speaking_task.cancel()
        try:
            await self._speaking_task
        except asyncio.CancelledError:
            pass
            # Clear any queued audio
            await self.agent.tts.flush()
            # Switch to listening mode
            self.state = TurnState.LISTENING
            self._audio_queue = asyncio.Queue()
async def _end_of_turn(self):
    """Process the user's completed utterance."""
    self.state = TurnState.PROCESSING
    # Collect all audio frames from the queue
    frames = []
    while not self._audio_queue.empty():
        frames.append(self._audio_queue.get_nowait())
        # Process and respond
        self._speaking_task = asyncio.create_task(
            self.agent.handle_utterance(frames)
            )
        self.state = TurnState.SPEAKING

from livekit.agents import AutoSubscribe, WorkerOptions, cli
from livekit.plugins import deepgram, openai, elevenlabs
# Production configuration for a telephony voice agent
async def entrypoint(ctx):
    """Production voice agent with telephony support."""
    # Select TTS voice based on call metadata
    caller_language = ctx.room.metadata.get("language", "en")
    voice_id = VOICE_MAP.get(caller_language, "default-english")
    session = AgentSession(
        stt=deepgram.STT(
        model="nova-3",
        language=caller_language,
        # Telephony-optimized settings
        encoding="mulaw",
        sample_rate=8000,
        channels=1,
        endpointing=300, # Faster endpointing for phone calls
        ),
        llm=openai.LLM(
        model="gpt-4o-mini", # Faster model for voice latency
        temperature=0.7,
        ),
        tts=elevenlabs.TTS(
        voice_id=voice_id,
        model="eleven_turbo_v2_5",
        output_format="pcm_mulaw", # Telephony codec
        ),
        )
    # Configure session with call context
    session.on("agent_started_speaking", lambda: log_event("agent_speaking"))
    session.on("agent_stopped_speaking", lambda: log_event("agent_silent"))
    session.on("user_started_speaking", lambda: log_event("user_speaking"))
    await session.start(
        room=ctx.room,
        agent=CustomerServiceAgent(),
        )
    # Worker configuration for production scaling
    if __name__ == "__main__":
        cli.run_app(
            WorkerOptions(
            entrypoint_fnc=entrypoint,
            auto_subscribe=AutoSubscribe.AUDIO_ONLY,
            max_retry=3,
            )
            )

from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class VoiceSessionMetrics:
    """Metrics for monitoring voice agent quality."""
    session_id: str
    start_time: float
    turns: int = 0
    total_stt_latency_ms: float = 0
    total_llm_latency_ms: float = 0
    total_tts_latency_ms: float = 0
    interruptions: int = 0
    tool_calls: int = 0
    errors: int = 0
    user_satisfaction: Optional[float] = None
    @property
    def avg_turn_latency_ms(self) -> float:
        if self.turns == 0:
            return 0
            total = self.total_stt_latency_ms + self.total_llm_latency_ms + self.total_tts_latency_ms
            return total / self.turns
            @property
            def duration_seconds(self) -> float:
                return time.time() - self.start_time
            def to_otel_attributes(self) -> dict:
                """Export metrics as OTel span attributes."""
                return {
                    "voice.session_id": self.session_id,
                    "voice.turns": self.turns,
                    "voice.avg_turn_latency_ms": self.avg_turn_latency_ms,
                    "voice.interruptions": self.interruptions,
                    "voice.tool_calls": self.tool_calls,
                    "voice.errors": self.errors,
                    "voice.duration_s": self.duration_seconds,
                    }