Voice AI Agents: Building Real-Time Conversational Systems

┌─────────────────────────────────────────────────────────────────┐
│                     Voice AI Agent                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────┐    ┌───────────┐    ┌─────────┐    ┌──────────┐  │
│  │  Audio  │───▶│   Voice   │───▶│   LLM   │───▶│  Speech  │  │
│  │  Input  │    │  Activity │    │  Agent  │    │ Synthesis│  │
│  │(Microphone)  │ Detection │    │         │    │  (TTS)   │  │
│  └─────────┘    └───────────┘    └─────────┘    └──────────┘  │
│       │              │                │              │         │
│       │              ▼                ▼              │         │
│       │        ┌───────────┐    ┌─────────┐         │         │
│       │        │   Turn    │    │  Tool   │         │         │
│       │        │  Taking   │    │  Use    │         │         │
│       │        │  Manager  │    │         │         │         │
│       │        └───────────┘    └─────────┘         │         │
│       │                                             │         │
│       └─────────────────────────────────────────────┘         │
│                    Interruption Loop                           │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

import asyncio
import websockets
import json
import base64
import pyaudio

class RealtimeVoiceClient:
    """Client for OpenAI Realtime API."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.ws = None
        self.audio_buffer = []

        # Audio settings
        self.sample_rate = 24000
        self.channels = 1
        self.chunk_size = 1024

    async def connect(self):
        """Connect to Realtime API."""
        url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "OpenAI-Beta": "realtime=v1"
        }

        self.ws = await websockets.connect(url, extra_headers=headers)

        # Configure session
        await self.ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "You are a helpful voice assistant. Be concise and conversational.",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 500
                }
            }
        }))

        print("Connected to Realtime API")

    async def send_audio(self, audio_data: bytes):
        """Send audio chunk to API."""
        if self.ws:
            await self.ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_data).decode()
            }))

    async def receive_loop(self, on_audio: callable, on_transcript: callable):
        """Receive and process responses."""
        async for message in self.ws:
            event = json.loads(message)
            event_type = event.get("type")

            if event_type == "response.audio.delta":
                # Audio response chunk
                audio_data = base64.b64decode(event["delta"])
                await on_audio(audio_data)

            elif event_type == "response.audio_transcript.delta":
                # Transcript of AI response
                await on_transcript(event["delta"])

            elif event_type == "conversation.item.input_audio_transcription.completed":
                # User's speech transcribed
                print(f"User said: {event['transcript']}")

            elif event_type == "response.done":
                print("Response complete")

            elif event_type == "error":
                print(f"Error: {event['error']}")

    async def close(self):
        """Close connection."""
        if self.ws:
            await self.ws.close()

class RealtimeVoiceAgent(RealtimeVoiceClient):
    """Voice agent with tool use."""

    def __init__(self, api_key: str, tools: list[dict]):
        super().__init__(api_key)
        self.tools = tools
        self.tool_handlers = {}

    async def connect(self):
        """Connect with tools configured."""
        await super().connect()

        # Add tools to session
        await self.ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "tools": self.tools,
                "tool_choice": "auto"
            }
        }))

    def register_tool(self, name: str, handler: callable):
        """Register a tool handler."""
        self.tool_handlers[name] = handler

    async def receive_loop(self, on_audio: callable, on_transcript: callable):
        """Receive and process responses including tool calls."""
        async for message in self.ws:
            event = json.loads(message)
            event_type = event.get("type")

            if event_type == "response.audio.delta":
                audio_data = base64.b64decode(event["delta"])
                await on_audio(audio_data)

            elif event_type == "response.function_call_arguments.done":
                # Tool call requested
                await self._handle_tool_call(event)

            elif event_type == "response.audio_transcript.delta":
                await on_transcript(event["delta"])

    async def _handle_tool_call(self, event: dict):
        """Handle a tool call from the model."""
        call_id = event["call_id"]
        name = event["name"]
        arguments = json.loads(event["arguments"])

        print(f"Tool call: {name}({arguments})")

        if name in self.tool_handlers:
            try:
                result = await self.tool_handlers[name](**arguments)
                output = json.dumps(result)
            except Exception as e:
                output = json.dumps({"error": str(e)})
        else:
            output = json.dumps({"error": f"Unknown tool: {name}"})

        # Send tool result
        await self.ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": output
            }
        }))

        # Request continuation
        await self.ws.send(json.dumps({
            "type": "response.create"
        }))

# Define tools
weather_tool = {
    "type": "function",
    "name": "get_weather",
    "description": "Get current weather for a location",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "City name"
            }
        },
        "required": ["location"]
    }
}

# Usage
agent = RealtimeVoiceAgent(api_key, tools=[weather_tool])

async def get_weather(location: str):
    # Actual implementation would call weather API
    return {"temperature": 72, "condition": "sunny", "location": location}

agent.register_tool("get_weather", get_weather)

import asyncio
import pyaudio
import numpy as np
from queue import Queue
from threading import Thread

class VoiceApp:
    """Complete voice application."""

    def __init__(self, api_key: str):
        self.agent = RealtimeVoiceAgent(api_key, tools=[weather_tool])
        self.audio = pyaudio.PyAudio()
        self.input_queue = Queue()
        self.output_queue = Queue()
        self.running = False

    async def start(self):
        """Start the voice application."""
        await self.agent.connect()
        self.running = True

        # Register tools
        self.agent.register_tool("get_weather", self._get_weather)

        # Start audio threads
        input_thread = Thread(target=self._capture_audio)
        output_thread = Thread(target=self._play_audio)
        input_thread.start()
        output_thread.start()

        # Start send/receive tasks
        await asyncio.gather(
            self._send_audio_loop(),
            self.agent.receive_loop(
                on_audio=self._queue_output_audio,
                on_transcript=self._on_transcript
            )
        )

    def _capture_audio(self):
        """Capture audio from microphone."""
        stream = self.audio.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=24000,
            input=True,
            frames_per_buffer=1024
        )

        while self.running:
            data = stream.read(1024, exception_on_overflow=False)
            self.input_queue.put(data)

        stream.close()

    def _play_audio(self):
        """Play audio responses."""
        stream = self.audio.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=24000,
            output=True,
            frames_per_buffer=1024
        )

        while self.running:
            if not self.output_queue.empty():
                data = self.output_queue.get()
                stream.write(data)

        stream.close()

    async def _send_audio_loop(self):
        """Send captured audio to API."""
        while self.running:
            if not self.input_queue.empty():
                data = self.input_queue.get()
                await self.agent.send_audio(data)
            await asyncio.sleep(0.01)

    async def _queue_output_audio(self, audio_data: bytes):
        """Queue audio for playback."""
        self.output_queue.put(audio_data)

    async def _on_transcript(self, text: str):
        """Handle transcript updates."""
        print(f"Assistant: {text}", end="", flush=True)

    async def _get_weather(self, location: str):
        """Weather tool implementation."""
        return {"temperature": 72, "condition": "sunny", "location": location}

    async def stop(self):
        """Stop the application."""
        self.running = False
        await self.agent.close()
        self.audio.terminate()

# Run
async def main():
    app = VoiceApp(api_key="your-api-key")
    try:
        await app.start()
    except KeyboardInterrupt:
        await app.stop()

asyncio.run(main())

// voice-client.js
class VoiceClient {
  constructor(serverUrl) {
    this.serverUrl = serverUrl;
    this.ws = null;
    this.mediaStream = null;
    this.audioContext = null;
    this.processor = null;
    this.isListening = false;
  }

  async connect() {
    // Get microphone access
    this.mediaStream = await navigator.mediaDevices.getUserMedia({
      audio: {
        sampleRate: 24000,
        channelCount: 1,
        echoCancellation: true,
        noiseSuppression: true,
      }
    });

    // Set up audio processing
    this.audioContext = new AudioContext({ sampleRate: 24000 });
    const source = this.audioContext.createMediaStreamSource(this.mediaStream);

    // Create processor for sending audio
    await this.audioContext.audioWorklet.addModule('audio-processor.js');
    this.processor = new AudioWorkletNode(this.audioContext, 'audio-processor');

    this.processor.port.onmessage = (event) => {
      if (this.isListening && this.ws?.readyState === WebSocket.OPEN) {
        this.ws.send(event.data);
      }
    };

    source.connect(this.processor);

    // Connect to server
    this.ws = new WebSocket(this.serverUrl);
    this.ws.binaryType = 'arraybuffer';

    this.ws.onmessage = (event) => {
      this.handleMessage(event.data);
    };

    this.ws.onopen = () => {
      console.log('Connected to voice server');
      this.isListening = true;
    };
  }

  handleMessage(data) {
    if (data instanceof ArrayBuffer) {
      // Audio data - play it
      this.playAudio(data);
    } else {
      // JSON message
      const message = JSON.parse(data);
      this.handleEvent(message);
    }
  }

  async playAudio(audioData) {
    const audioBuffer = await this.audioContext.decodeAudioData(audioData);
    const source = this.audioContext.createBufferSource();
    source.buffer = audioBuffer;
    source.connect(this.audioContext.destination);
    source.start();
  }

  handleEvent(event) {
    switch (event.type) {
      case 'transcript':
        this.onTranscript?.(event.text, event.isFinal);
        break;
      case 'response_start':
        this.onResponseStart?.();
        break;
      case 'response_end':
        this.onResponseEnd?.();
        break;
      case 'tool_call':
        this.onToolCall?.(event.name, event.arguments);
        break;
    }
  }

  startListening() {
    this.isListening = true;
  }

  stopListening() {
    this.isListening = false;
  }

  disconnect() {
    this.isListening = false;
    this.processor?.disconnect();
    this.mediaStream?.getTracks().forEach(track => track.stop());
    this.ws?.close();
  }
}

// Audio processor worklet
// audio-processor.js
class AudioProcessor extends AudioWorkletProcessor {
  constructor() {
    super();
    this.buffer = [];
    this.bufferSize = 2400; // 100ms at 24kHz
  }

  process(inputs) {
    const input = inputs[0][0];
    if (input) {
      // Convert float32 to int16
      const int16 = new Int16Array(input.length);
      for (let i = 0; i < input.length; i++) {
        int16[i] = Math.max(-32768, Math.min(32767, input[i] * 32768));
      }

      this.buffer.push(...int16);

      if (this.buffer.length >= this.bufferSize) {
        const chunk = new Int16Array(this.buffer.splice(0, this.bufferSize));
        this.port.postMessage(chunk.buffer);
      }
    }
    return true;
  }
}

registerProcessor('audio-processor', AudioProcessor);

// VoiceUI.jsx
import React, { useState, useEffect, useRef } from 'react';

const VoiceUI = ({ serverUrl }) => {
  const [isConnected, setIsConnected] = useState(false);
  const [isListening, setIsListening] = useState(false);
  const [isSpeaking, setIsSpeaking] = useState(false);
  const [transcript, setTranscript] = useState('');
  const [response, setResponse] = useState('');

  const clientRef = useRef(null);

  useEffect(() => {
    const client = new VoiceClient(serverUrl);

    client.onTranscript = (text, isFinal) => {
      setTranscript(prev => isFinal ? text : prev + text);
    };

    client.onResponseStart = () => {
      setIsSpeaking(true);
      setResponse('');
    };

    client.onResponseEnd = () => {
      setIsSpeaking(false);
    };

    clientRef.current = client;

    return () => {
      client.disconnect();
    };
  }, [serverUrl]);

  const handleConnect = async () => {
    await clientRef.current.connect();
    setIsConnected(true);
    setIsListening(true);
  };

  const handleDisconnect = () => {
    clientRef.current.disconnect();
    setIsConnected(false);
    setIsListening(false);
  };

  const toggleListening = () => {
    if (isListening) {
      clientRef.current.stopListening();
    } else {
      clientRef.current.startListening();
    }
    setIsListening(!isListening);
  };

  return (
    <div className="voice-ui">
      <div className="status">
        <span className={`indicator ${isConnected ? 'connected' : ''}`} />
        {isConnected ? 'Connected' : 'Disconnected'}
      </div>

      <div className="visualizer">
        {isSpeaking && <div className="speaking-animation" />}
        {isListening && !isSpeaking && <div className="listening-animation" />}
      </div>

      <div className="transcript">
        <p><strong>You:</strong> {transcript}</p>
        <p><strong>Assistant:</strong> {response}</p>
      </div>

      <div className="controls">
        {!isConnected ? (
          <button onClick={handleConnect}>Connect</button>
        ) : (
          <>
            <button onClick={toggleListening}>
              {isListening ? 'Mute' : 'Unmute'}
            </button>
            <button onClick={handleDisconnect}>Disconnect</button>
          </>
        )}
      </div>
    </div>
  );
};

export default VoiceUI;

import numpy as np
from collections import deque

class VoiceActivityDetector:
    """Detect voice activity for turn-taking."""

    def __init__(
        self,
        sample_rate: int = 24000,
        frame_duration_ms: int = 30,
        threshold: float = 0.02,
        speech_pad_ms: int = 300,
        silence_duration_ms: int = 500
    ):
        self.sample_rate = sample_rate
        self.frame_size = int(sample_rate * frame_duration_ms / 1000)
        self.threshold = threshold
        self.speech_pad_frames = int(speech_pad_ms / frame_duration_ms)
        self.silence_frames = int(silence_duration_ms / frame_duration_ms)

        self.is_speaking = False
        self.silence_count = 0
        self.speech_count = 0
        self.ring_buffer = deque(maxlen=self.speech_pad_frames)

    def process_frame(self, audio_frame: np.ndarray) -> dict:
        """Process an audio frame and return VAD state."""
        # Calculate RMS energy
        rms = np.sqrt(np.mean(audio_frame.astype(np.float32) ** 2))
        is_speech = rms > self.threshold

        self.ring_buffer.append(is_speech)

        result = {
            "is_speech": is_speech,
            "rms": rms,
            "state_changed": False,
            "event": None
        }

        if self.is_speaking:
            if is_speech:
                self.silence_count = 0
            else:
                self.silence_count += 1

            if self.silence_count >= self.silence_frames:
                # End of speech
                self.is_speaking = False
                self.silence_count = 0
                result["state_changed"] = True
                result["event"] = "speech_end"
        else:
            if is_speech:
                self.speech_count += 1
                if self.speech_count >= 3:  # Require consecutive speech frames
                    self.is_speaking = True
                    self.speech_count = 0
                    result["state_changed"] = True
                    result["event"] = "speech_start"
            else:
                self.speech_count = 0

        result["is_speaking"] = self.is_speaking
        return result


class TurnManager:
    """Manage conversation turns between user and agent."""

    def __init__(self):
        self.vad = VoiceActivityDetector()
        self.current_turn = "agent"  # "user" or "agent"
        self.agent_speaking = False
        self.user_buffer = []
        self.pending_interruption = False

    def process_user_audio(self, audio: np.ndarray) -> dict:
        """Process user audio and manage turns."""
        vad_result = self.vad.process_frame(audio)

        result = {
            "action": None,
            "audio": None
        }

        if vad_result["event"] == "speech_start":
            if self.agent_speaking:
                # User interrupting agent
                self.pending_interruption = True
                result["action"] = "interrupt"
            else:
                # User starting to speak
                self.current_turn = "user"
                self.user_buffer = []
                result["action"] = "user_start"

        if self.current_turn == "user" and vad_result["is_speaking"]:
            self.user_buffer.append(audio)

        if vad_result["event"] == "speech_end" and self.current_turn == "user":
            # User finished speaking
            result["action"] = "user_end"
            result["audio"] = np.concatenate(self.user_buffer)
            self.user_buffer = []
            self.current_turn = "agent"

        return result

    def start_agent_response(self):
        """Mark agent as speaking."""
        self.agent_speaking = True
        self.current_turn = "agent"

    def end_agent_response(self):
        """Mark agent as done speaking."""
        self.agent_speaking = False

    def handle_interruption(self) -> bool:
        """Check and handle pending interruption."""
        if self.pending_interruption:
            self.pending_interruption = False
            self.agent_speaking = False
            self.current_turn = "user"
            return True
        return False

class InterruptibleVoiceAgent:
    """Voice agent that handles interruptions gracefully."""

    def __init__(self, realtime_client: RealtimeVoiceAgent):
        self.client = realtime_client
        self.turn_manager = TurnManager()
        self.response_buffer = []
        self.is_generating = False

    async def handle_audio(self, audio: np.ndarray):
        """Handle incoming user audio."""
        result = self.turn_manager.process_user_audio(audio)

        if result["action"] == "interrupt":
            await self._handle_interruption()

        elif result["action"] == "user_end":
            # Send complete utterance for processing
            await self._process_user_utterance(result["audio"])

    async def _handle_interruption(self):
        """Handle user interruption."""
        if self.is_generating:
            # Cancel current response
            await self.client.ws.send(json.dumps({
                "type": "response.cancel"
            }))

            self.is_generating = False
            self.response_buffer = []

            # Acknowledge interruption
            print("Response cancelled due to interruption")

    async def _process_user_utterance(self, audio: np.ndarray):
        """Process complete user utterance."""
        # Send audio to API
        await self.client.send_audio(audio.tobytes())

        # Commit the audio buffer
        await self.client.ws.send(json.dumps({
            "type": "input_audio_buffer.commit"
        }))

        # Request response
        await self.client.ws.send(json.dumps({
            "type": "response.create"
        }))

        self.is_generating = True
        self.turn_manager.start_agent_response()

    async def handle_response_chunk(self, audio_chunk: bytes):
        """Handle response audio chunk."""
        # Check for interruption before playing
        if self.turn_manager.handle_interruption():
            return

        self.response_buffer.append(audio_chunk)
        # Play audio...

    async def handle_response_end(self):
        """Handle end of response."""
        self.is_generating = False
        self.turn_manager.end_agent_response()
        self.response_buffer = []

from fastapi import FastAPI, WebSocket, Request
from fastapi.responses import Response
import asyncio

app = FastAPI()

# Store active calls
active_calls = {}

@app.post("/incoming-call")
async def handle_incoming_call(request: Request):
    """Handle incoming Twilio call."""
    form_data = await request.form()
    call_sid = form_data.get("CallSid")

    # Return TwiML to connect to WebSocket
    twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
    <Response>
        <Say>Welcome to the AI assistant. How can I help you today?</Say>
        <Connect>
            <Stream url="wss://{request.headers['host']}/media-stream/{call_sid}" />
        </Connect>
    </Response>"""

    return Response(content=twiml, media_type="application/xml")

@app.websocket("/media-stream/{call_sid}")
async def media_stream(websocket: WebSocket, call_sid: str):
    """Handle Twilio media stream."""
    await websocket.accept()

    # Create voice agent for this call
    agent = TwilioVoiceAgent(call_sid)
    active_calls[call_sid] = agent

    try:
        await agent.connect()

        async for message in websocket.iter_text():
            data = json.loads(message)
            event = data.get("event")

            if event == "media":
                # Incoming audio from caller
                audio_payload = data["media"]["payload"]
                audio_bytes = base64.b64decode(audio_payload)

                # Convert mulaw to PCM
                pcm_audio = audioop.ulaw2lin(audio_bytes, 2)

                # Process with agent
                await agent.process_audio(pcm_audio)

            elif event == "start":
                stream_sid = data["start"]["streamSid"]
                agent.stream_sid = stream_sid

            elif event == "stop":
                break

    finally:
        await agent.disconnect()
        del active_calls[call_sid]


class TwilioVoiceAgent:
    """Voice agent for Twilio calls."""

    def __init__(self, call_sid: str):
        self.call_sid = call_sid
        self.stream_sid = None
        self.realtime_client = None
        self.websocket = None

    async def connect(self):
        """Connect to OpenAI Realtime API."""
        self.realtime_client = RealtimeVoiceClient(api_key)
        await self.realtime_client.connect()

    async def process_audio(self, pcm_audio: bytes):
        """Process incoming audio from Twilio."""
        # Resample from 8kHz (Twilio) to 24kHz (OpenAI)
        resampled = self._resample(pcm_audio, 8000, 24000)
        await self.realtime_client.send_audio(resampled)

    async def send_audio_to_twilio(self, audio: bytes):
        """Send audio response back to Twilio."""
        # Resample from 24kHz to 8kHz
        resampled = self._resample(audio, 24000, 8000)

        # Convert PCM to mulaw
        mulaw_audio = audioop.lin2ulaw(resampled, 2)

        # Send via WebSocket
        message = {
            "event": "media",
            "streamSid": self.stream_sid,
            "media": {
                "payload": base64.b64encode(mulaw_audio).decode()
            }
        }
        await self.websocket.send_text(json.dumps(message))

    def _resample(self, audio: bytes, from_rate: int, to_rate: int) -> bytes:
        """Resample audio."""
        import audioop
        return audioop.ratecv(audio, 2, 1, from_rate, to_rate, None)[0]

    async def disconnect(self):
        """Disconnect from APIs."""
        if self.realtime_client:
            await self.realtime_client.close()

from twilio.rest import Client

class OutboundCaller:
    """Make outbound calls with voice AI."""

    def __init__(self, account_sid: str, auth_token: str, from_number: str):
        self.client = Client(account_sid, auth_token)
        self.from_number = from_number

    def make_call(self, to_number: str, webhook_url: str) -> str:
        """Initiate an outbound call."""
        call = self.client.calls.create(
            to=to_number,
            from_=self.from_number,
            url=webhook_url,
            status_callback=f"{webhook_url}/status",
            status_callback_event=["initiated", "ringing", "answered", "completed"]
        )
        return call.sid

    def create_greeting_twiml(self, message: str, stream_url: str) -> str:
        """Create TwiML for outbound call."""
        return f"""<?xml version="1.0" encoding="UTF-8"?>
        <Response>
            <Say voice="Polly.Amy">{message}</Say>
            <Connect>
                <Stream url="{stream_url}" />
            </Connect>
        </Response>"""

# Usage
caller = OutboundCaller(
    account_sid="ACxxxxx",
    auth_token="xxxxx",
    from_number="+1234567890"
)

call_sid = caller.make_call(
    to_number="+0987654321",
    webhook_url="https://your-server.com/outbound-call"
)

User speaks → Silence detected → Audio sent → STT → LLM → TTS → Audio plays
   |              |                |           |      |      |         |
   0ms          ~500ms          ~600ms      ~800ms ~1500ms ~1800ms   ~2000ms

Target: <1000ms total latency for responsive conversation

class LatencyOptimizedAgent:
    """Voice agent optimized for low latency."""

    def __init__(self):
        # Use server-side VAD to minimize latency
        self.use_server_vad = True

        # Reduce silence detection threshold
        self.silence_duration_ms = 300  # Faster turn detection

        # Enable response prefetching
        self.prefetch_enabled = True

        # Use streaming TTS
        self.streaming_tts = True

    async def configure_session(self, ws):
        """Configure low-latency session."""
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.4,  # Lower threshold for faster detection
                    "prefix_padding_ms": 200,  # Less padding
                    "silence_duration_ms": 300  # Faster end-of-turn
                },
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                # Request shorter responses for speed
                "instructions": "Be very concise. Answer in 1-2 sentences."
            }
        }))

    def optimize_network(self):
        """Network-level optimizations."""
        return {
            # Use regional endpoints
            "endpoint": "wss://api.openai.com/v1/realtime",

            # Keep connection alive
            "keepalive": True,

            # Disable Nagle's algorithm for lower latency
            "tcp_nodelay": True,

            # Use binary WebSocket frames when possible
            "binary_frames": True
        }

class AudioStreamOptimizer:
    """Optimize audio streaming for low latency."""

    def __init__(self, sample_rate: int = 24000):
        self.sample_rate = sample_rate
        self.buffer_duration_ms = 20  # Small chunks
        self.buffer_size = int(sample_rate * self.buffer_duration_ms / 1000)

    def create_optimized_stream(self):
        """Create optimized audio stream."""
        return {
            "format": "pcm16",
            "sample_rate": self.sample_rate,
            "buffer_size": self.buffer_size,
            # Use low-latency audio APIs
            "latency": "low",
            # Disable audio processing that adds latency
            "echo_cancellation": True,  # Keep this for quality
            "noise_suppression": False,  # Disable for speed
            "auto_gain_control": False  # Disable for speed
        }

class ResponseStreamer:
    """Stream responses with minimal buffering."""

    def __init__(self, min_chunk_ms: int = 50):
        self.min_chunk_size = int(24000 * min_chunk_ms / 1000) * 2  # bytes

    async def stream_response(self, audio_generator, play_func):
        """Stream audio with minimal delay."""
        buffer = bytearray()

        async for chunk in audio_generator:
            buffer.extend(chunk)

            # Play as soon as we have minimum chunk
            while len(buffer) >= self.min_chunk_size:
                to_play = bytes(buffer[:self.min_chunk_size])
                del buffer[:self.min_chunk_size]
                await play_func(to_play)
                # Don't await completion - fire and forget for overlap

import time
from dataclasses import dataclass
from collections import deque

@dataclass
class LatencyMeasurement:
    timestamp: float
    vad_latency_ms: float
    stt_latency_ms: float
    llm_latency_ms: float
    tts_latency_ms: float
    total_latency_ms: float

class LatencyMonitor:
    """Monitor and report voice latency metrics."""

    def __init__(self, window_size: int = 100):
        self.measurements = deque(maxlen=window_size)
        self.current_measurement = {}

    def start_turn(self):
        """Start measuring a new turn."""
        self.current_measurement = {
            "speech_start": time.time()
        }

    def mark_vad_complete(self):
        """Mark VAD detection complete."""
        self.current_measurement["vad_complete"] = time.time()

    def mark_stt_complete(self):
        """Mark speech-to-text complete."""
        self.current_measurement["stt_complete"] = time.time()

    def mark_llm_start(self):
        """Mark LLM processing start."""
        self.current_measurement["llm_start"] = time.time()

    def mark_llm_first_token(self):
        """Mark first token from LLM."""
        self.current_measurement["llm_first_token"] = time.time()

    def mark_tts_first_audio(self):
        """Mark first audio output."""
        self.current_measurement["tts_first_audio"] = time.time()

    def end_turn(self):
        """Complete the measurement."""
        m = self.current_measurement
        now = time.time()

        measurement = LatencyMeasurement(
            timestamp=now,
            vad_latency_ms=(m.get("vad_complete", now) - m.get("speech_start", now)) * 1000,
            stt_latency_ms=(m.get("stt_complete", now) - m.get("vad_complete", now)) * 1000,
            llm_latency_ms=(m.get("llm_first_token", now) - m.get("llm_start", now)) * 1000,
            tts_latency_ms=(m.get("tts_first_audio", now) - m.get("llm_first_token", now)) * 1000,
            total_latency_ms=(m.get("tts_first_audio", now) - m.get("speech_start", now)) * 1000
        )

        self.measurements.append(measurement)
        return measurement

    def get_stats(self) -> dict:
        """Get latency statistics."""
        if not self.measurements:
            return {}

        total_latencies = [m.total_latency_ms for m in self.measurements]

        return {
            "count": len(self.measurements),
            "avg_total_ms": sum(total_latencies) / len(total_latencies),
            "p50_total_ms": sorted(total_latencies)[len(total_latencies) // 2],
            "p90_total_ms": sorted(total_latencies)[int(len(total_latencies) * 0.9)],
            "p99_total_ms": sorted(total_latencies)[int(len(total_latencies) * 0.99)],
            "breakdown": {
                "avg_vad_ms": sum(m.vad_latency_ms for m in self.measurements) / len(self.measurements),
                "avg_stt_ms": sum(m.stt_latency_ms for m in self.measurements) / len(self.measurements),
                "avg_llm_ms": sum(m.llm_latency_ms for m in self.measurements) / len(self.measurements),
                "avg_tts_ms": sum(m.tts_latency_ms for m in self.measurements) / len(self.measurements)
            }
        }

class ResilientVoiceAgent:
    """Voice agent with error recovery."""

    def __init__(self, config: dict):
        self.config = config
        self.connection_attempts = 0
        self.max_reconnects = 5
        self.backoff_base = 1.0

    async def connect_with_retry(self):
        """Connect with exponential backoff retry."""
        while self.connection_attempts < self.max_reconnects:
            try:
                await self._connect()
                self.connection_attempts = 0
                return True
            except Exception as e:
                self.connection_attempts += 1
                wait_time = self.backoff_base * (2 ** self.connection_attempts)
                print(f"Connection failed ({e}), retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)

        return False

    async def handle_error(self, error: Exception):
        """Handle various error types."""
        error_type = type(error).__name__

        if "WebSocket" in error_type or "Connection" in error_type:
            # Connection error - try to reconnect
            await self._reconnect()

        elif "Timeout" in error_type:
            # Timeout - retry the operation
            await self._retry_last_operation()

        elif "RateLimit" in error_type:
            # Rate limited - back off
            await asyncio.sleep(60)
            await self._reconnect()

        else:
            # Unknown error - log and continue
            print(f"Unknown error: {error}")

    async def _reconnect(self):
        """Reconnect to the API."""
        print("Reconnecting...")
        await self._disconnect()
        success = await self.connect_with_retry()
        if success:
            await self._notify_user("Connection restored. Please continue.")

    async def _notify_user(self, message: str):
        """Notify user of system state via TTS."""
        # Play a system message
        pass

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class ConversationState(Enum):
    IDLE = "idle"
    LISTENING = "listening"
    PROCESSING = "processing"
    SPEAKING = "speaking"
    WAITING_FOR_TOOL = "waiting_for_tool"
    ERROR = "error"

@dataclass
class VoiceConversation:
    """Manage voice conversation state."""

    session_id: str
    state: ConversationState = ConversationState.IDLE
    turn_count: int = 0
    context: dict = field(default_factory=dict)
    history: list[dict] = field(default_factory=list)
    pending_tool_calls: list[dict] = field(default_factory=list)

    def add_user_turn(self, transcript: str):
        """Add user turn to history."""
        self.history.append({
            "role": "user",
            "content": transcript,
            "timestamp": time.time()
        })
        self.turn_count += 1

    def add_assistant_turn(self, transcript: str):
        """Add assistant turn to history."""
        self.history.append({
            "role": "assistant",
            "content": transcript,
            "timestamp": time.time()
        })

    def get_context_window(self, max_turns: int = 10) -> list[dict]:
        """Get recent conversation for context."""
        return self.history[-max_turns:]

    def set_context(self, key: str, value: any):
        """Set conversation context."""
        self.context[key] = value

    def get_context(self, key: str) -> Optional[any]:
        """Get conversation context."""
        return self.context.get(key)

class ConversationManager:
    """Manage multiple voice conversations."""

    def __init__(self):
        self.conversations: dict[str, VoiceConversation] = {}

    def create_conversation(self, session_id: str) -> VoiceConversation:
        """Create a new conversation."""
        conv = VoiceConversation(session_id=session_id)
        self.conversations[session_id] = conv
        return conv

    def get_conversation(self, session_id: str) -> Optional[VoiceConversation]:
        """Get existing conversation."""
        return self.conversations.get(session_id)

    def end_conversation(self, session_id: str):
        """End and archive a conversation."""
        if session_id in self.conversations:
            conv = self.conversations.pop(session_id)
            self._archive(conv)

    def _archive(self, conversation: VoiceConversation):
        """Archive conversation for analytics."""
        # Store conversation history, metrics, etc.
        pass

# Store state for 24-hour session persistence
await redis.setex(f"voice:session:{session_id}", 86400, json.dumps(state))

import numpy as np
from dataclasses import dataclass
from typing import Optional
from enum import Enum

class Emotion(str, Enum):
    NEUTRAL = "neutral"
    HAPPY = "happy"
    SAD = "sad"
    ANGRY = "angry"
    FRUSTRATED = "frustrated"
    CONFUSED = "confused"
    EXCITED = "excited"

@dataclass
class EmotionAnalysis:
    primary_emotion: Emotion
    confidence: float
    valence: float  # -1 (negative) to 1 (positive)
    arousal: float  # 0 (calm) to 1 (excited)
    frustration_level: float  # 0 to 1

class VoiceEmotionAnalyzer:
    """Analyze emotions from voice characteristics."""

    def __init__(self):
        self.emotion_history = []
        self.baseline_pitch = None
        self.baseline_energy = None

    def analyze_audio_features(self, audio: np.ndarray, sample_rate: int = 24000) -> dict:
        """Extract emotion-relevant audio features."""
        import librosa

        # Extract features
        pitch, _ = librosa.piptrack(y=audio.astype(float), sr=sample_rate)
        pitch_mean = np.mean(pitch[pitch > 0]) if np.any(pitch > 0) else 0

        energy = np.mean(librosa.feature.rms(y=audio.astype(float)))
        tempo, _ = librosa.beat.beat_track(y=audio.astype(float), sr=sample_rate)

        # Speech rate (syllables per second approximation)
        zero_crossings = librosa.feature.zero_crossing_rate(y=audio.astype(float))
        speech_rate = np.mean(zero_crossings) * sample_rate

        # Spectral features
        spectral_centroid = np.mean(librosa.feature.spectral_centroid(y=audio.astype(float), sr=sample_rate))
        spectral_rolloff = np.mean(librosa.feature.spectral_rolloff(y=audio.astype(float), sr=sample_rate))

        return {
            "pitch_mean": pitch_mean,
            "pitch_variance": np.var(pitch[pitch > 0]) if np.any(pitch > 0) else 0,
            "energy": energy,
            "tempo": tempo,
            "speech_rate": speech_rate,
            "spectral_centroid": spectral_centroid,
            "spectral_rolloff": spectral_rolloff
        }

    def detect_emotion(self, audio: np.ndarray, transcript: str = "") -> EmotionAnalysis:
        """Detect emotion from audio and optional transcript."""
        features = self.analyze_audio_features(audio)

        # Set baseline on first utterance
        if self.baseline_pitch is None:
            self.baseline_pitch = features["pitch_mean"]
            self.baseline_energy = features["energy"]

        # Calculate relative changes
        pitch_delta = (features["pitch_mean"] - self.baseline_pitch) / max(self.baseline_pitch, 1)
        energy_delta = (features["energy"] - self.baseline_energy) / max(self.baseline_energy, 0.01)

        # Heuristic emotion detection
        valence = 0.0
        arousal = 0.0

        # High pitch + high energy = excited/angry
        if pitch_delta > 0.2 and energy_delta > 0.3:
            arousal = 0.8

        # Low pitch + low energy = sad
        if pitch_delta < -0.1 and energy_delta < -0.2:
            valence = -0.5

        # High speech rate = frustration or excitement
        if features["speech_rate"] > 150:
            arousal = max(arousal, 0.6)

        # Determine primary emotion
        if arousal > 0.6 and valence < -0.3:
            emotion = Emotion.ANGRY
        elif arousal > 0.6 and valence > 0.3:
            emotion = Emotion.EXCITED
        elif valence < -0.3:
            emotion = Emotion.SAD
        elif valence > 0.3:
            emotion = Emotion.HAPPY
        elif arousal > 0.5:
            emotion = Emotion.FRUSTRATED
        else:
            emotion = Emotion.NEUTRAL

        # Calculate frustration from patterns
        frustration = self._detect_frustration(features, transcript)

        analysis = EmotionAnalysis(
            primary_emotion=emotion,
            confidence=0.6,  # Audio-only has limited confidence
            valence=valence,
            arousal=arousal,
            frustration_level=frustration
        )

        self.emotion_history.append(analysis)
        return analysis

    def _detect_frustration(self, features: dict, transcript: str) -> float:
        """Detect user frustration level."""
        frustration_score = 0.0

        # High energy variance suggests frustration
        if features.get("energy", 0) > self.baseline_energy * 1.5:
            frustration_score += 0.3

        # Repeated phrases in recent history
        if len(self.emotion_history) >= 3:
            recent_negative = sum(
                1 for e in self.emotion_history[-3:]
                if e.valence < 0
            )
            frustration_score += recent_negative * 0.15

        # Text indicators
        frustration_words = ["again", "already", "still", "why", "doesn't work", "not working"]
        if transcript:
            for word in frustration_words:
                if word in transcript.lower():
                    frustration_score += 0.2

        return min(frustration_score, 1.0)

class EmotionAwareVoiceAgent:
    """Voice agent that adapts to user emotions."""

    def __init__(self, base_agent, emotion_analyzer: VoiceEmotionAnalyzer):
        self.agent = base_agent
        self.analyzer = emotion_analyzer
        self.escalation_threshold = 0.7

    async def process_with_emotion(self, audio: np.ndarray, transcript: str) -> dict:
        """Process user input with emotion awareness."""
        emotion = self.analyzer.detect_emotion(audio, transcript)

        # Adapt response strategy based on emotion
        response_modifiers = self._get_response_modifiers(emotion)

        # Check for escalation
        if emotion.frustration_level > self.escalation_threshold:
            return await self._handle_escalation(transcript, emotion)

        # Process with emotion context
        result = await self.agent.process(
            transcript,
            emotion_context=emotion,
            **response_modifiers
        )

        return {
            "response": result,
            "emotion_detected": emotion,
            "response_strategy": response_modifiers
        }

    def _get_response_modifiers(self, emotion: EmotionAnalysis) -> dict:
        """Get response modifiers based on emotion."""
        modifiers = {}

        if emotion.primary_emotion == Emotion.FRUSTRATED:
            modifiers["tone"] = "empathetic"
            modifiers["acknowledge_difficulty"] = True
            modifiers["offer_alternatives"] = True

        elif emotion.primary_emotion == Emotion.CONFUSED:
            modifiers["tone"] = "patient"
            modifiers["simplify_language"] = True
            modifiers["provide_examples"] = True

        elif emotion.primary_emotion == Emotion.ANGRY:
            modifiers["tone"] = "calm_professional"
            modifiers["acknowledge_feelings"] = True
            modifiers["prioritize_resolution"] = True

        elif emotion.primary_emotion in [Emotion.HAPPY, Emotion.EXCITED]:
            modifiers["tone"] = "enthusiastic"
            modifiers["match_energy"] = True

        return modifiers

    async def _handle_escalation(self, transcript: str, emotion: EmotionAnalysis) -> dict:
        """Handle escalation to human agent."""
        return {
            "action": "escalate",
            "reason": f"User frustration level: {emotion.frustration_level:.2f}",
            "transcript": transcript,
            "emotion": emotion,
            "message": "I understand this has been frustrating. Let me connect you with a specialist who can help."
        }

from dataclasses import dataclass
from typing import Optional
import numpy as np

@dataclass
class SpeakerSegment:
    speaker_id: str
    start_time: float
    end_time: float
    audio: np.ndarray
    transcript: Optional[str] = None
    confidence: float = 0.0

class SpeakerDiarizer:
    """Identify and separate multiple speakers."""

    def __init__(self):
        self.speaker_embeddings = {}
        self.speaker_count = 0

    def process_audio_segment(
        self,
        audio: np.ndarray,
        timestamp: float,
        sample_rate: int = 24000
    ) -> SpeakerSegment:
        """Process audio segment and identify speaker."""
        # Extract speaker embedding
        embedding = self._extract_embedding(audio, sample_rate)

        # Match to existing speaker or create new
        speaker_id, confidence = self._match_speaker(embedding)

        if speaker_id is None:
            # New speaker
            speaker_id = f"speaker_{self.speaker_count}"
            self.speaker_count += 1
            self.speaker_embeddings[speaker_id] = [embedding]
            confidence = 1.0
        else:
            # Update embedding history
            self.speaker_embeddings[speaker_id].append(embedding)
            # Keep only recent embeddings
            self.speaker_embeddings[speaker_id] = self.speaker_embeddings[speaker_id][-10:]

        return SpeakerSegment(
            speaker_id=speaker_id,
            start_time=timestamp,
            end_time=timestamp + len(audio) / sample_rate,
            audio=audio,
            confidence=confidence
        )

    def _extract_embedding(self, audio: np.ndarray, sample_rate: int) -> np.ndarray:
        """Extract speaker embedding from audio."""
        # Simplified embedding extraction using MFCC
        import librosa

        mfcc = librosa.feature.mfcc(
            y=audio.astype(float),
            sr=sample_rate,
            n_mfcc=20
        )

        # Use mean of MFCCs as simple embedding
        embedding = np.mean(mfcc, axis=1)
        return embedding / np.linalg.norm(embedding)

    def _match_speaker(self, embedding: np.ndarray) -> tuple[Optional[str], float]:
        """Match embedding to existing speakers."""
        if not self.speaker_embeddings:
            return None, 0.0

        best_match = None
        best_similarity = 0.0

        for speaker_id, embeddings in self.speaker_embeddings.items():
            # Compare to mean of recent embeddings
            mean_embedding = np.mean(embeddings, axis=0)
            similarity = np.dot(embedding, mean_embedding)

            if similarity > best_similarity:
                best_similarity = similarity
                best_match = speaker_id

        # Threshold for matching
        if best_similarity > 0.7:
            return best_match, best_similarity

        return None, 0.0

class MultiSpeakerVoiceAgent:
    """Handle multi-speaker conversations."""

    def __init__(self, base_agent, diarizer: SpeakerDiarizer):
        self.agent = base_agent
        self.diarizer = diarizer
        self.speaker_contexts = {}  # Per-speaker conversation state

    async def process_multi_speaker(
        self,
        audio: np.ndarray,
        timestamp: float
    ) -> dict:
        """Process audio from potentially multiple speakers."""
        # Identify speaker
        segment = self.diarizer.process_audio_segment(audio, timestamp)

        # Get or create speaker context
        if segment.speaker_id not in self.speaker_contexts:
            self.speaker_contexts[segment.speaker_id] = {
                "history": [],
                "preferences": {},
                "turn_count": 0
            }

        context = self.speaker_contexts[segment.speaker_id]
        context["turn_count"] += 1

        # Process with speaker-specific context
        result = await self.agent.process(
            audio=segment.audio,
            speaker_context=context,
            speaker_id=segment.speaker_id
        )

        # Update history
        context["history"].append({
            "timestamp": timestamp,
            "transcript": result.get("transcript"),
            "response": result.get("response")
        })

        return {
            "speaker_id": segment.speaker_id,
            "speaker_confidence": segment.confidence,
            "result": result,
            "total_speakers": self.diarizer.speaker_count
        }

    def get_speaker_summary(self) -> dict:
        """Get summary of all speakers in conversation."""
        return {
            speaker_id: {
                "turn_count": ctx["turn_count"],
                "history_length": len(ctx["history"])
            }
            for speaker_id, ctx in self.speaker_contexts.items()
        }

from dataclasses import dataclass
from datetime import datetime, timedelta
from collections import defaultdict

@dataclass
class VoiceCostMetrics:
    audio_minutes: float
    api_calls: int
    input_tokens: int
    output_tokens: int
    estimated_cost_usd: float

class VoiceCostTracker:
    """Track and optimize voice API costs."""

    # Pricing (example rates)
    PRICING = {
        "openai_realtime": {
            "audio_input_per_minute": 0.06,
            "audio_output_per_minute": 0.24,
            "text_input_per_1k_tokens": 0.0025,
            "text_output_per_1k_tokens": 0.01
        },
        "elevenlabs": {
            "characters_per_1k": 0.30
        },
        "whisper": {
            "audio_per_minute": 0.006
        }
    }

    def __init__(self):
        self.sessions = {}
        self.daily_costs = defaultdict(float)

    def start_session(self, session_id: str):
        """Start tracking a voice session."""
        self.sessions[session_id] = {
            "start_time": datetime.now(),
            "audio_input_seconds": 0,
            "audio_output_seconds": 0,
            "input_tokens": 0,
            "output_tokens": 0,
            "api_calls": 0
        }

    def track_audio_input(self, session_id: str, duration_seconds: float):
        """Track audio input duration."""
        if session_id in self.sessions:
            self.sessions[session_id]["audio_input_seconds"] += duration_seconds
            self.sessions[session_id]["api_calls"] += 1

    def track_audio_output(self, session_id: str, duration_seconds: float):
        """Track audio output duration."""
        if session_id in self.sessions:
            self.sessions[session_id]["audio_output_seconds"] += duration_seconds

    def track_tokens(self, session_id: str, input_tokens: int, output_tokens: int):
        """Track token usage."""
        if session_id in self.sessions:
            self.sessions[session_id]["input_tokens"] += input_tokens
            self.sessions[session_id]["output_tokens"] += output_tokens

    def end_session(self, session_id: str) -> VoiceCostMetrics:
        """End session and calculate costs."""
        if session_id not in self.sessions:
            return VoiceCostMetrics(0, 0, 0, 0, 0)

        session = self.sessions.pop(session_id)

        # Calculate costs
        pricing = self.PRICING["openai_realtime"]

        audio_input_cost = (session["audio_input_seconds"] / 60) * pricing["audio_input_per_minute"]
        audio_output_cost = (session["audio_output_seconds"] / 60) * pricing["audio_output_per_minute"]
        text_input_cost = (session["input_tokens"] / 1000) * pricing["text_input_per_1k_tokens"]
        text_output_cost = (session["output_tokens"] / 1000) * pricing["text_output_per_1k_tokens"]

        total_cost = audio_input_cost + audio_output_cost + text_input_cost + text_output_cost

        # Track daily costs
        today = datetime.now().strftime("%Y-%m-%d")
        self.daily_costs[today] += total_cost

        total_audio_minutes = (session["audio_input_seconds"] + session["audio_output_seconds"]) / 60

        return VoiceCostMetrics(
            audio_minutes=total_audio_minutes,
            api_calls=session["api_calls"],
            input_tokens=session["input_tokens"],
            output_tokens=session["output_tokens"],
            estimated_cost_usd=total_cost
        )

    def get_daily_summary(self, date: str = None) -> dict:
        """Get daily cost summary."""
        date = date or datetime.now().strftime("%Y-%m-%d")
        return {
            "date": date,
            "total_cost_usd": self.daily_costs.get(date, 0)
        }

    def estimate_monthly_cost(self, daily_sessions: int, avg_session_minutes: float) -> dict:
        """Estimate monthly costs based on usage patterns."""
        pricing = self.PRICING["openai_realtime"]

        # Assume 50/50 input/output split
        input_minutes = avg_session_minutes * 0.5
        output_minutes = avg_session_minutes * 0.5

        per_session_cost = (
            input_minutes * pricing["audio_input_per_minute"] +
            output_minutes * pricing["audio_output_per_minute"]
        )

        daily_cost = per_session_cost * daily_sessions
        monthly_cost = daily_cost * 30

        return {
            "per_session_cost": per_session_cost,
            "daily_cost": daily_cost,
            "monthly_cost": monthly_cost,
            "assumptions": {
                "daily_sessions": daily_sessions,
                "avg_session_minutes": avg_session_minutes
            }
        }

class CostOptimizedVoiceAgent:
    """Voice agent with cost optimization strategies."""

    def __init__(self, agent, cost_tracker: VoiceCostTracker):
        self.agent = agent
        self.tracker = cost_tracker
        self.optimization_enabled = True

    async def process_with_optimization(
        self,
        session_id: str,
        audio: np.ndarray
    ) -> dict:
        """Process with cost optimization strategies."""
        optimizations_applied = []

        # Strategy 1: Skip silence
        if self._is_silence(audio):
            return {"skipped": True, "reason": "silence"}

        # Strategy 2: Batch short utterances
        if len(audio) < 24000 * 0.5:  # Less than 0.5 seconds
            buffered = await self._buffer_short_audio(session_id, audio)
            if buffered:
                optimizations_applied.append("buffered_short_audio")
                return {"buffered": True}

        # Strategy 3: Use text mode for simple responses
        if self._should_use_text_mode(session_id):
            result = await self._process_text_mode(audio)
            optimizations_applied.append("text_mode")
        else:
            result = await self.agent.process(audio)

        # Track costs
        self.tracker.track_audio_input(session_id, len(audio) / 24000)

        return {
            "result": result,
            "optimizations": optimizations_applied
        }

    def _is_silence(self, audio: np.ndarray, threshold: float = 0.01) -> bool:
        """Check if audio is silence."""
        rms = np.sqrt(np.mean(audio.astype(float) ** 2))
        return rms < threshold

    async def _buffer_short_audio(self, session_id: str, audio: np.ndarray) -> bool:
        """Buffer short audio for batching."""
        # Implementation would buffer and return True if buffered
        return False

    def _should_use_text_mode(self, session_id: str) -> bool:
        """Determine if text mode would be more cost-effective."""
        # Use text mode for FAQ-type questions
        return False

    async def _process_text_mode(self, audio: np.ndarray) -> dict:
        """Process using text mode (STT -> LLM -> TTS)."""
        # Cheaper for some use cases
        pass

class BudgetManager:
    """Manage voice API budgets."""

    def __init__(
        self,
        daily_budget_usd: float = 100,
        session_budget_usd: float = 5,
        alert_threshold: float = 0.8
    ):
        self.daily_budget = daily_budget_usd
        self.session_budget = session_budget_usd
        self.alert_threshold = alert_threshold
        self.cost_tracker = VoiceCostTracker()
        self.alert_callbacks = []

    def register_alert(self, callback):
        """Register alert callback."""
        self.alert_callbacks.append(callback)

    def check_budget(self, session_id: str) -> dict:
        """Check if session/daily budget allows continuing."""
        daily_summary = self.cost_tracker.get_daily_summary()
        daily_used = daily_summary["total_cost_usd"]

        session = self.cost_tracker.sessions.get(session_id, {})
        session_used = self._calculate_session_cost(session)

        # Check limits
        daily_remaining = self.daily_budget - daily_used
        session_remaining = self.session_budget - session_used

        result = {
            "can_continue": True,
            "daily_used": daily_used,
            "daily_remaining": daily_remaining,
            "session_used": session_used,
            "session_remaining": session_remaining
        }

        # Check for alerts
        if daily_used > self.daily_budget * self.alert_threshold:
            self._send_alert("daily_threshold", daily_used)
            result["alerts"] = ["approaching_daily_limit"]

        if daily_remaining <= 0:
            result["can_continue"] = False
            result["reason"] = "daily_budget_exceeded"

        if session_remaining <= 0:
            result["can_continue"] = False
            result["reason"] = "session_budget_exceeded"

        return result

    def _calculate_session_cost(self, session: dict) -> float:
        """Calculate current session cost."""
        if not session:
            return 0

        pricing = VoiceCostTracker.PRICING["openai_realtime"]
        return (
            (session.get("audio_input_seconds", 0) / 60) * pricing["audio_input_per_minute"] +
            (session.get("audio_output_seconds", 0) / 60) * pricing["audio_output_per_minute"]
        )

    def _send_alert(self, alert_type: str, value: float):
        """Send budget alert."""
        for callback in self.alert_callbacks:
            callback(alert_type, value)

┌─────────────────────────────────────────────────────────────┐
│                    Budget Enforcement                       │
├──────────────┬──────────────┬──────────────┬───────────────┤
│ Per-Session  │   Per-User   │    Daily     │   Monthly     │
│ ($5 max)     │ ($50/day)    │ ($1000 org)  │ ($20k org)    │
└──────────────┴──────────────┴──────────────┴───────────────┘

class AccessibleVoiceAgent:
    """Voice agent with accessibility features."""

    def __init__(self, base_agent):
        self.agent = base_agent
        self.accessibility_mode = False
        self.speech_rate = 1.0
        self.confirmation_mode = True

    async def process_accessible(self, audio: np.ndarray) -> dict:
        """Process with accessibility accommodations."""
        result = await self.agent.process(audio)

        if self.accessibility_mode:
            result = await self._enhance_for_accessibility(result)

        return result

    async def _enhance_for_accessibility(self, result: dict) -> dict:
        """Enhance response for accessibility."""
        response = result.get("response", "")

        # Add structure announcements
        enhanced = self._add_structure_announcements(response)

        # Slow down speech rate
        result["speech_rate"] = self.speech_rate

        # Add confirmation prompts
        if self.confirmation_mode:
            enhanced += " Is there anything you'd like me to clarify?"

        result["response"] = enhanced
        return result

    def _add_structure_announcements(self, text: str) -> str:
        """Add announcements for structure (lists, sections)."""
        # Announce list items
        lines = text.split("\n")
        enhanced_lines = []

        list_count = 0
        for line in lines:
            if line.strip().startswith(("-", "*", "•")):
                list_count += 1
                enhanced_lines.append(f"Item {list_count}: {line.strip()[1:].strip()}")
            else:
                if list_count > 0:
                    # End of list
                    list_count = 0
                enhanced_lines.append(line)

        return "\n".join(enhanced_lines)

    def set_speech_rate(self, rate: float):
        """Set speech rate (0.5 to 2.0)."""
        self.speech_rate = max(0.5, min(2.0, rate))

    def toggle_confirmation_mode(self):
        """Toggle confirmation prompts."""
        self.confirmation_mode = not self.confirmation_mode