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,
 }