Index

realtime

Realtime duplex voice sessions (Gemini Live / OpenAI Realtime).

Classes

RealtimeVoiceSession

Bases: ABC

Provider-agnostic duplex voice session.

Methods:

connect abstractmethod async

connect() -> None

Open the provider connection and finish setup.

Source code in src/diapason/speech/realtime/base.py

@abstractmethod
async def connect(self) -> None:
    """Open the provider connection and finish setup."""

send_audio abstractmethod async

send_audio(pcm16: bytes) -> None

Send a chunk of mono PCM16 at input_sample_rate.

Source code in src/diapason/speech/realtime/base.py

@abstractmethod
async def send_audio(self, pcm16: bytes) -> None:
    """Send a chunk of mono PCM16 at ``input_sample_rate``."""

send_text abstractmethod async

send_text(text: str) -> None

Optional text turn (debug / typed barge).

Source code in src/diapason/speech/realtime/base.py

@abstractmethod
async def send_text(self, text: str) -> None:
    """Optional text turn (debug / typed barge)."""

interrupt abstractmethod async

interrupt() -> None

Cancel current model speech (barge-in).

Source code in src/diapason/speech/realtime/base.py

@abstractmethod
async def interrupt(self) -> None:
    """Cancel current model speech (barge-in)."""

events abstractmethod

events() -> AsyncIterator[SessionEvent]

Async iterator of normalized session events.

Source code in src/diapason/speech/realtime/base.py

@abstractmethod
def events(self) -> AsyncIterator[SessionEvent]:
    """Async iterator of normalized session events."""

close abstractmethod async

close() -> None

Tear down the provider connection.

Source code in src/diapason/speech/realtime/base.py

@abstractmethod
async def close(self) -> None:
    """Tear down the provider connection."""

SessionEvent dataclass

SessionEvent(
    kind: EventKind,
    text: str = "",
    role: str = "",
    final: bool = False,
    replace: bool = False,
    audio_b64: str = "",
    sample_rate: int = 24000,
    detail: str = "",
    tool_name: str = "",
    tool_ok: bool = False,
    raw: Optional[dict[str, Any]] = None,
)

Normalized event emitted by a provider session toward the UI.

Functions:

create_realtime_session

create_realtime_session(
    provider: str,
    *,
    model: str = "",
    voice: str = "",
    instructions: str = "",
    language: str = "",
    api_key: Optional[str] = None,
    enable_tools: bool = True,
    max_tool_steps: int = 12,
    allowed_tools: Optional[Sequence[str]] = None,
    sur_echange: Optional[
        Callable[[str, str], None]
    ] = None,
) -> RealtimeVoiceSession

Create a provider session. Raises ValueError for unknown providers.

Raises :class:LocalOnlyError under [privacy] local_only: every realtime provider is remote, so there is no honest degradation here.

Source code in src/diapason/speech/realtime/factory.py

def create_realtime_session(
    provider: str,
    *,
    model: str = "",
    voice: str = "",
    instructions: str = "",
    language: str = "",
    api_key: Optional[str] = None,
    enable_tools: bool = True,
    max_tool_steps: int = 12,
    allowed_tools: Optional[Sequence[str]] = None,
    # Appelé (question, réponse) à chaque échange abouti — le raccord vers
    # la mémoire vivante. Fournisseur LOCAL seulement : les sessions cloud
    # ne journalisent rien côté serveur.
    sur_echange: Optional[Callable[[str, str], None]] = None,
) -> RealtimeVoiceSession:
    """Create a provider session. Raises ``ValueError`` for unknown providers.

    Raises :class:`LocalOnlyError` under ``[privacy] local_only``: every
    realtime provider is remote, so there is no honest degradation here.
    """
    # The gravest path in the codebase: a REMOTE realtime session streams RAW
    # MICROPHONE PCM to Gemini or OpenAI continuously — not a finished
    # sentence, everything the microphone hears for as long as the socket is
    # open. And the provider is chosen by the CLIENT (a query parameter or the
    # `start` frame in server/voice_live_routes.py), so a guard placed on the
    # caller's default would be bypassed by anyone passing ?provider=openai.
    #
    # The guard therefore sits on the factory, which no provider can avoid,
    # and fires before the session object exists — hence before any API key is
    # read from the environment by a provider constructor.
    #
    # The LOCAL provider passes: Whisper, Ollama and Kokoro all run on this
    # machine, and letting it through is precisely the honest degradation the
    # guard used to say did not exist.
    name = (provider or "").strip().lower()

    from diapason.core.local_mode import REFUSAL_HINT, LocalOnlyError, local_only

    if local_only() and name not in ("local", "local_voice"):
        raise LocalOnlyError(
            "Realtime voice with a remote provider streams the microphone off "
            "this machine, so it was refused. Use the 'local' provider "
            f"instead. {REFUSAL_HINT}"
        )
    common = dict(
        api_key=api_key,
        instructions=instructions,
        language=language,
        enable_tools=enable_tools,
        max_tool_steps=max_tool_steps,
        allowed_tools=allowed_tools,
    )
    if name in ("gemini", "gemini_live", "google"):
        from diapason.speech.realtime.gemini_live import GeminiLiveSession

        return GeminiLiveSession(
            model=model or "gemini-2.0-flash-live-001",
            voice=voice or "Zephyr",
            **common,
        )
    if name in ("openai", "openai_realtime", "gpt-realtime"):
        from diapason.speech.realtime.openai_realtime import OpenAIRealtimeSession

        return OpenAIRealtimeSession(
            model=model or "gpt-4o-realtime-preview",
            voice=voice or "alloy",
            **common,
        )
    if name in ("local", "local_voice"):
        from diapason.speech.realtime.local_voice import LocalVoiceSession

        return LocalVoiceSession(
            model=model, voice=voice, sur_echange=sur_echange, **common
        )
    raise ValueError(
        f"Unknown realtime voice provider: {provider!r} "
        "(expected 'gemini', 'openai' or 'local')"
    )