OpenAI Realtime API

┌─────────────┐     ┌──────────────────┐     ┌─────────────────┐
│   Twilio    │────▶│   Audio Bridge   │────▶│  OpenAI Realtime│
│  (μ-law)    │     │  (PCM converter) │     │  (gpt-4o)       │
└─────────────┘     └──────────────────┘     └─────────────────┘
                           │                          │
                           ▼                          ▼
                    Conversation Brain        Whisper Transcripts
                    (Intent Analysis)        (Accurate proper nouns)

from agenticai.openai.realtime_handler import OpenAIRealtimeHandler

handler = OpenAIRealtimeHandler(
    api_key=api_key,
    voice="echo",  # Warm, conversational
    system_instruction="You are a helpful assistant."
)

class OpenAIRealtimeHandler:
    def __init__(
        self,
        api_key: str,
        model: str = "gpt-4o-realtime-preview-2024-12-17",
        voice: str = "alloy",
        system_instruction: str = "You are a helpful AI assistant.",
    ):
        """Initialize OpenAI Realtime handler."""

    def set_callbacks(
        self,
        on_audio: Callable[[bytes], Awaitable[None]] | None = None,
        on_transcript: Callable[[str, bool], Awaitable[None]] | None = None,
        on_user_transcript: Callable[[str], Awaitable[None]] | None = None,
        on_turn_complete: Callable[[], Awaitable[None]] | None = None,
    ):
        """Set event callbacks."""

    async def connect(self, initial_prompt: str | None = None):
        """Connect to OpenAI Realtime API."""

    async def send_audio(self, audio_data: bytes):
        """Send PCM 16-bit audio to OpenAI.
        
        Args:
            audio_data: PCM 16-bit 24kHz audio bytes
        """

    async def send_text(self, text: str, end_of_turn: bool = True):
        """Send text for OpenAI to respond to.
        
        Used to inject ClawdBot responses for the AI to speak.
        """

    async def get_audio(self) -> bytes:
        """Get audio chunk from OpenAI.
        
        Returns:
            PCM 16-bit 24kHz audio bytes
        """

    async def disconnect():
        """Disconnect from OpenAI Realtime API."""

import asyncio
from agenticai.openai.realtime_handler import OpenAIRealtimeHandler

async def main():
    handler = OpenAIRealtimeHandler(
        api_key="sk-proj-xxx",
        voice="alloy",
    )

    # Set callbacks
    handler.set_callbacks(
        on_audio=lambda audio: print(f"Received {len(audio)} bytes"),
        on_user_transcript=lambda text: print(f"User: {text}"),
        on_transcript=lambda text, is_final: print(f"AI: {text}"),
    )

    # Connect
    await handler.connect(initial_prompt="Hello! How can I help you today?")

    # Send audio chunks
    while True:
        audio_chunk = await get_audio_from_microphone()  # Your audio source
        await handler.send_audio(audio_chunk)

        # Receive AI audio
        try:
            ai_audio = await asyncio.wait_for(handler.get_audio(), timeout=0.1)
            await play_audio(ai_audio)  # Your audio output
        except asyncio.TimeoutError:
            pass

asyncio.run(main())

async def on_audio(audio_bytes: bytes):
    """Called when AI generates audio.
    
    Args:
        audio_bytes: PCM 16-bit 24kHz audio
    """

async def on_transcript(text: str, is_final: bool):
    """Called when AI speaks (transcript of audio).
    
    Args:
        text: Transcript text
        is_final: True if complete, False if partial
    """

async def on_user_transcript(text: str):
    """Called when user speech is transcribed (Whisper).
    
    Args:
        text: Accurate transcription with proper nouns
    """

async def on_turn_complete():
    """Called when AI finishes speaking."""

async def on_user_turn_complete():
    """Called when user finishes speaking (VAD detected)."""

config = {
    "type": "session.update",
    "session": {
        "modalities": ["text", "audio"],
        "instructions": system_instruction,
        "voice": voice,
        "input_audio_format": "pcm16",      # 16-bit PCM
        "output_audio_format": "pcm16",     # 16-bit PCM
        "input_audio_transcription": {
            "model": "whisper-1"              # Built-in Whisper
        },
        "turn_detection": {
            "type": "server_vad",             # Server-side VAD
            "threshold": 0.6,                 # Sensitivity
            "prefix_padding_ms": 200,         # Pre-speech buffer
            "silence_duration_ms": 300,       # End-of-speech timeout
        },
    }
}

User starts speaking  → input_audio_buffer.speech_started
User stops speaking   → input_audio_buffer.speech_stopped
Transcription ready   → conversation.item.input_audio_transcription.completed

elif event_type == "conversation.item.input_audio_transcription.completed":
    # User transcript (Whisper)
    transcript = event.get("transcript", "")
    if transcript:
        print(f"=== OPENAI USER TRANSCRIPT: {transcript} ===")
        if self._on_user_transcript:
            await self._on_user_transcript(transcript)