OpenAI Realtime API Turn Detection & Interruption Errors

import json


# ✅ CONVERSATIONAL — snappy responses, clear speakers
CONVERSATIONAL_VAD = {
    "type": "server_vad",
    "threshold": 0.5,               # standard sensitivity
    "prefix_padding_ms": 300,       # 300ms of audio before speech kept (preserves first syllable)
    "silence_duration_ms": 200,     # 200ms silence = user done (snappy)
    "create_response": True,
    "interrupt_response": True,
}


# ✅ HESITANT / THOUGHTFUL SPEAKERS — allow pauses
HESITANT_VAD = {
    "type": "server_vad",
    "threshold": 0.5,
    "prefix_padding_ms": 400,
    "silence_duration_ms": 600,     # 600ms — lets users pause and continue
    "create_response": True,
    "interrupt_response": True,
}


# ✅ NOISY ENVIRONMENT — higher threshold to reject background noise
NOISY_VAD = {
    "type": "server_vad",
    "threshold": 0.7,               # only clear speech triggers
    "prefix_padding_ms": 300,
    "silence_duration_ms": 400,
    "create_response": True,
    "interrupt_response": True,
}


# ✅ TELEPHONY — narrower band, lower fidelity
TELEPHONY_VAD = {
    "type": "server_vad",
    "threshold": 0.5,
    "prefix_padding_ms": 200,       # phone lines add their own latency
    "silence_duration_ms": 500,     # phone users tend to pause more
    "create_response": True,
    "interrupt_response": True,
}


# ✅ Semantic VAD — model decides, slower but handles natural speech
SEMANTIC_VAD = {
    "type": "semantic_vad",
    "eagerness": "medium",          # low = wait longer, high = interrupt sooner
    "create_response": True,
    "interrupt_response": True,
}


# ✅ Manual (push-to-talk) — client controls turn boundaries
MANUAL_TURN = None                  # turn_detection: null


# ✅ Apply the right mode per session context
def session_for_user(user_profile: dict) -> dict:
    if user_profile.get("prefers_slow_pace"):
        return HESITANT_VAD
    if user_profile.get("channel") == "telephony":
        return TELEPHONY_VAD
    if user_profile.get("environment") == "noisy":
        return NOISY_VAD
    return CONVERSATIONAL_VAD


async def configure_session(ws, user_profile: dict):
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "audio": {"input": {"format": "pcm16"}, "output": {"format": "pcm16"}},
            "turn_detection": session_for_user(user_profile),
        },
    }))


# ✅ Adaptive tuning — watch for user complaints or cut-offs, adjust
class AdaptiveVAD:
    def __init__(self):
        self.silence_ms = 300
        self.cutoff_events = 0     # times user restarted after being cut off

    def on_user_restart_within_2s(self):
        """User immediately spoke again after model responded — likely cut off."""
        self.cutoff_events += 1
        if self.cutoff_events >= 3:
            self.silence_ms = min(self.silence_ms + 100, 800)
            self.cutoff_events = 0

    def current_config(self) -> dict:
        return {
            "type": "server_vad",
            "threshold": 0.5,
            "prefix_padding_ms": 300,
            "silence_duration_ms": self.silence_ms,
            "create_response": True,
            "interrupt_response": True,
        }

import json
import base64


class VoicePlaybackController:
    """Manages the user's speaker output; supports mid-playback cancellation."""

    def __init__(self, audio_sink):
        self.sink = audio_sink       # your platform-specific audio output
        self.playing = False
        self.pending_chunks: list[bytes] = []

    async def enqueue(self, audio_bytes: bytes):
        self.pending_chunks.append(audio_bytes)
        if not self.playing:
            self.playing = True
            await self._play_loop()

    async def _play_loop(self):
        while self.pending_chunks and self.playing:
            chunk = self.pending_chunks.pop(0)
            await self.sink.write(chunk)
        self.playing = False

    async def cancel(self):
        """Stop playing immediately — user is interrupting."""
        self.playing = False
        self.pending_chunks.clear()
        await self.sink.flush_and_stop()


# ✅ Full barge-in loop
async def voice_loop(ws, playback: VoicePlaybackController):
    model_is_speaking = False
    current_response_id = None

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

        if t == "response.created":
            model_is_speaking = True
            current_response_id = event["response"]["id"]

        elif t == "response.output_audio.delta":
            audio = base64.b64decode(event["delta"])
            await playback.enqueue(audio)

        elif t == "response.done":
            model_is_speaking = False

        elif t == "input_audio_buffer.speech_started":
            # USER IS TALKING — implement barge-in
            if model_is_speaking:
                # 1. Stop playing model audio to user
                await playback.cancel()

                # 2. Tell server how much audio actually reached the user
                # (server needs this to reconstruct what the user heard vs. what was cancelled)
                await ws.send(json.dumps({
                    "type": "response.cancel",
                }))

                model_is_speaking = False

        elif t == "input_audio_buffer.speech_stopped":
            # User done — server_vad auto-triggers response.create if create_response=true
            pass

        elif t == "conversation.item.input_audio_transcription.completed":
            # Whisper transcribed what the user said — good for logging
            print(f"[user said] {event['transcript']}")


# ✅ WebRTC handles barge-in largely automatically
# The browser's peer connection stops playing model audio when the local mic
# picks up user speech, IF you've wired the local stream to the peer connection.
# Just make sure interrupt_response=true in the session config.


# ✅ Handle response.cancelled cleanly
async def on_response_done(event, playback):
    if event.get("response", {}).get("status") == "cancelled":
        # User interrupted; ensure no leftover chunks in playback buffer
        await playback.cancel()

        # Optionally send conversation.item.truncate so the server knows
        # exactly how much audio the user heard
        item_id = event["response"]["output"][-1]["id"] if event["response"]["output"] else None
        if item_id:
            await ws.send(json.dumps({
                "type": "conversation.item.truncate",
                "item_id": item_id,
                "content_index": 0,
                "audio_end_ms": playback.total_played_ms,
            }))

# ✅ Semantic VAD — model decides based on content
SEMANTIC_VAD_SESSION = {
    "type": "session.update",
    "session": {
        "audio": {"input": {"format": "pcm16"}, "output": {"format": "pcm16"}},
        "turn_detection": {
            "type": "semantic_vad",
            "eagerness": "medium",     # "low" | "medium" | "high" | "auto"
            "create_response": True,
            "interrupt_response": True,
        },
    },
}


# eagerness values:
#   "low"    — waits longer; best for slow speakers, minimizes false cuts
#   "medium" — balanced; good default for most conversational apps
#   "high"   — cuts sooner; feels snappy but may interrupt thinkers
#   "auto"   — model picks based on session context


# ✅ Manual turn detection — push-to-talk, controlled environments
MANUAL_SESSION = {
    "type": "session.update",
    "session": {
        "audio": {"input": {"format": "pcm16"}, "output": {"format": "pcm16"}},
        "turn_detection": None,       # null = manual
    },
}


async def push_to_talk(ws, mic_stream):
    """User holds a button; we send audio; on release we commit."""
    await ws.send(json.dumps(MANUAL_SESSION))

    async for pcm_chunk in mic_stream:
        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(pcm_chunk).decode("ascii"),
        }))

    # User released the button
    await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
    await ws.send(json.dumps({"type": "response.create"}))


# ✅ Hybrid — start with semantic_vad, downgrade if latency complaints mount
class TurnDetectionAdapter:
    def __init__(self):
        self.mode = "semantic_vad"
        self.latency_samples: list[float] = []

    def record_turn_latency(self, seconds: float):
        self.latency_samples.append(seconds)
        # Keep last 20 samples
        self.latency_samples = self.latency_samples[-20:]

    def should_downgrade(self) -> bool:
        """If p50 latency > 1.5s, drop to server_vad."""
        if len(self.latency_samples) < 5:
            return False
        sorted_lat = sorted(self.latency_samples)
        p50 = sorted_lat[len(sorted_lat) // 2]
        return p50 > 1.5

    async def maybe_switch(self, ws):
        if self.mode == "semantic_vad" and self.should_downgrade():
            self.mode = "server_vad"
            await ws.send(json.dumps({
                "type": "session.update",
                "session": {
                    "turn_detection": {
                        "type": "server_vad",
                        "silence_duration_ms": 300,
                        "create_response": True,
                        "interrupt_response": True,
                    },
                },
            }))


# ✅ Common misconception: semantic_vad + interrupt_response
# Semantic VAD DOES support interrupt_response=true. The model can decide
# both "user is done speaking" AND "user is interrupting me" from content.
# It's not either/or.