DX-818: Add realtime transcription support with failover handling

README.md

@@ -168,6 +168,80 @@ async for chat_completion in stream:
     print(chat_completion.choices)
 ```
 
+## Realtime transcription
+
+Transcribe live audio — a microphone, a phone call, a meeting — as it is being
+spoken. You stream PCM audio over a WebSocket and receive interim text within
+moments and a finalized transcript for each utterance. Requires the `realtime`
+extra:
+
+```sh
+uv add "together[realtime]"
+```
+
+Use `client.beta.realtime.transcription()`: a session built for live sources. If
+the connection drops mid-conversation, the session holds on to the speech the
+server hadn't transcribed yet, reconnects automatically, and picks up where
+the transcript left off — words spoken during the outage still come back as
+text.
+
+```python
+from together import AsyncTogether
+from together.realtime import TranscriptDelta, TranscriptCompleted
+
+client = AsyncTogether()
+
+async with client.beta.realtime.transcription(
+    model="openai/whisper-large-v3",
+    sample_rate=16_000,
+) as session:
+    # feed audio from your capture source as it arrives (any chunk size)
+    await session.append(pcm_chunk)  # 16 kHz mono 16-bit PCM
+
+    async for event in session:
+        if isinstance(event, TranscriptDelta):
+            print("interim:", event.text)   # updates while a phrase is spoken
+        elif isinstance(event, TranscriptCompleted):
+            print("final:", event.text)     # one per finished utterance
+
+    transcript = await session.flush()  # finalize whatever was said last
+```
+
+What to know before integrating:
+
+- **Audio in**: 16 kHz mono 16-bit PCM (`pcm_s16le_16000`). Resample your
+  source before appending — passing `sample_rate=` lets the SDK reject a
+  mismatch loudly instead of incorrecly transcribing different sample rate .
+  `append()` never blocks on network state, so it is safe to call from a capture loop.
+- **Utterance boundaries**: detected server-side by default — final
+  transcripts arrive on their own as the speaker pauses. To control
+  segmentation yourself, pass `turn_detection={"type": "none"}` and call
+  `await session.commit()` when each segment ends.
+- **Tuning**: `language`, `prompt`, and turn-detection parameters
+  (`min_silence_duration_ms`, `max_speech_duration_s`, ...) are passed
+  through to the service.
+- **When the connection drops**: The session emits
+  `Reconnecting`/`Reconnected` events and tries to retry the connection.
+   Transcripts recomputed from speech that was carried across the reconnect are marked
+  `replayed=True` and may overlap text you already received; voice agents
+  that act on each final can set `buffer={"max_replay_seconds": 0}` to
+  resume live with no re-emission instead.
+- **If audio is ever dropped** (an outage longer than the retention window),
+  the session tells you with a `BufferGap` event.
+- **If an endpoint fails for good**, calls raise `RealtimeConnectionError`.
+  To keep a conversation alive across endpoint outages, run a failover ring —
+  see [`examples/realtime_failover.py`](examples/realtime_failover.py):
+  on failure, `session.pending_audio()` hands you the un-transcribed speech
+  to seed a new session on another endpoint.
+- **Sync applications**: `Together().beta.realtime.transcription(...)` mirrors the
+  async API on a background thread — good for a handful of sessions; use the
+  async client for high concurrency.
+- **Full manual control** (raw wire events, no automatic recovery):
+  `client.beta.realtime.connect()`.
+
+See [`examples/realtime_transcription.py`](examples/realtime_transcription.py)
+for a runnable end-to-end example.
+
 ## Using types
 
 Nested request parameters are [TypedDicts](https://docs.python.org/3/library/typing.html#typing.TypedDict). Responses are [Pydantic models](https://docs.pydantic.dev) which also provide helper methods for things like:

examples/realtime_failover.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Realtime transcription with failover across multiple endpoints.
+
+Stream live audio to a primary endpoint; when it fails beyond recovery,
+resume on the next endpoint in the ring, carrying over any speech that was
+never transcribed so nothing is lost across the switch.
+
+The SDK handles transient failures on a single endpoint internally.
+When an endpoint is unrecoverable, SDK calls raise RealtimeConnectionError.
+Moving to another endpoint is application code: the loop below handles failover.
+
+Usage:
+    uv add "together[realtime]"
+    export TOGETHER_API_KEY=...
+    uv run ./examples/realtime_failover.py audio.wav   # 16 kHz mono s16le WAV
+"""
+
+from __future__ import annotations
+
+import sys
+import wave
+import asyncio
+from pathlib import Path
+
+from together import AsyncTogether
+from together.realtime import (
+    BufferGap,
+    SessionStarted,
+    TranscriptDelta,
+    TranscriptCompleted,
+    RealtimeSessionEvent,
+    RealtimeConnectionError,
+)
+
+# Independent deployments of the same model. Order = preference; on failure
+# the next entry takes over, wrapping around. Each entry is (base_url, model);
+# base_url is the API root and may point at region-specific hosts so the ring
+# entries share no failure domain.
+ENDPOINTS = [
+    ("https://api.together.ai/v1", "openai/whisper-large-v3-endpoint1"),
+    ("https://api.together.ai/v1", "openai/whisper-large-v3-endpoint2"),
+]
+
+SAMPLE_RATE = 16_000
+CHUNK_BYTES = SAMPLE_RATE * 2 // 10  # 100 ms per append, like a live source
+MAX_ATTEMPTS = 3 * len(ENDPOINTS)  # give up after several full ring cycles
+
+
+def on_event(event: RealtimeSessionEvent) -> None:
+    """Handle session events as they arrive."""
+    if isinstance(event, SessionStarted):
+        # Log the server-assigned session id — useful for correlating a session.
+        print(f"session {event.session_id} started on {event.model}")
+    elif isinstance(event, TranscriptDelta):
+        print(f"interim: {event.text}", end="\r")
+    elif isinstance(event, TranscriptCompleted):
+        # `replayed` marks re-transcriptions of carried-over audio; they may
+        # overlap text you already received.
+        marker = " (replayed)" if event.replayed else ""
+        print(f"final: {event.text}{marker}")
+    elif isinstance(event, BufferGap):
+        # Audio dropped beyond recovery.
+        print(f"[gap: {event.dropped_seconds:.1f}s lost]")
+
+
+async def transcribe_with_failover(audio: bytes) -> str:
+    # One client per endpoint: base_url is client-level configuration.
+    # Clients are lightweight config objects — nothing connects until a
+    # session starts.
+    clients = [(AsyncTogether(base_url=base_url), model) for base_url, model in ENDPOINTS]
+
+    transcripts: list[str] = []
+    carry_over = b""  # audio the failed endpoint received but never transcribed
+    position = 0  # progress through the source; survives endpoint switches
+
+    for attempt in range(MAX_ATTEMPTS):
+        client, model = clients[attempt % len(clients)]
+        print(f"--- streaming to {model}")
+
+        session = client.beta.realtime.transcription(
+            model=model,
+            sample_rate=SAMPLE_RATE,
+            event_callback=on_event,
+            # Switch endpoints immediately on failure. Raise max_attempts to
+            # let the SDK retry the same endpoint first.
+            reconnect={"max_attempts": 0},
+        )
+        try:
+            async with session:
+                if carry_over:
+                    # The un-transcribed speech from the failed endpoint is transmitted
+                    # as fast as the connection allows, so the new endpoint catches up to live immediately.
+                    await session.append(carry_over)
+
+                while position < len(audio):
+                    await session.append(audio[position : position + CHUNK_BYTES])
+                    position += CHUNK_BYTES
+                    await asyncio.sleep(0.1)  # simulate a live capture cadence
+                transcripts.append(await session.flush())
+                return " ".join(t for t in transcripts if t)
+        except RealtimeConnectionError as exc:
+            # Endpoint is unrecoverable: keep its transcripts, take back the
+            # audio it never transcribed, and move to the next endpoint.
+            transcripts.extend(session.transcripts)
+            carry_over = session.pending_audio()
+            print(f"--- {model} failed ({exc}); carrying over {len(carry_over) / (SAMPLE_RATE * 2):.1f}s of audio")
+
+    raise SystemExit("all endpoints failed repeatedly; giving up")
+
+
+def load_pcm(path: Path) -> bytes:
+    with wave.open(str(path), "rb") as w:
+        if (w.getnchannels(), w.getsampwidth(), w.getframerate()) != (1, 2, SAMPLE_RATE):
+            raise SystemExit(f"expected mono 16-bit {SAMPLE_RATE} Hz WAV")
+        return w.readframes(w.getnframes())
+
+
+async def main() -> None:
+    if len(sys.argv) != 2:
+        raise SystemExit(__doc__)
+    transcript = await transcribe_with_failover(load_pcm(Path(sys.argv[1])))
+    print(f"\nfull transcript: {transcript}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/realtime_single_endpoint_failover.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""Realtime transcription that rides out failures on a single endpoint.
+
+The SDK recovers from transient failures on its own: if the connection drops
+mid-conversation, it reconnects with backoff and replays the speech the
+server never transcribed — the application just keeps feeding audio. You'll
+see Reconnecting/Reconnected events; there is nothing to handle.
+
+Only when the endpoint stays unrecoverable past the retry budget do SDK
+calls raise RealtimeConnectionError. To keep a conversation alive across
+endpoint outages, add an endpoint ring on top — see
+examples/realtime_failover.py.
+
+Usage:
+    uv add "together[realtime]"
+    export TOGETHER_API_KEY=...
+    uv run ./examples/realtime_single_endpoint_failover.py audio.wav   # 16 kHz mono s16le WAV
+"""
+
+from __future__ import annotations
+
+import sys
+import wave
+import asyncio
+from pathlib import Path
+
+from together import AsyncTogether
+from together.lib.realtime import (
+    BufferGap,
+    Reconnected,
+    Reconnecting,
+    SessionStarted,
+    TranscriptDelta,
+    TranscriptFailed,
+    TranscriptCompleted,
+    RealtimeSessionEvent,
+)
+
+MODEL = "openai/whisper-large-v3-endpoint1"
+
+SAMPLE_RATE = 16_000
+CHUNK_BYTES = SAMPLE_RATE * 2 // 10  # 100 ms per append, like a live source
+
+
+def on_event(event: RealtimeSessionEvent) -> None:
+    """Handle session events as they arrive."""
+    if isinstance(event, SessionStarted):
+        # Log the server-assigned session id — useful when correlating a
+        # stream with server-side logs or support requests.
+        print(f"session {event.session_id} started on {event.model}")
+    elif isinstance(event, TranscriptDelta):
+        print(f"interim: {event.text}", end="\r")
+    elif isinstance(event, TranscriptCompleted):
+        # `replayed` marks re-transcriptions of speech replayed after a
+        # reconnect; they may overlap text you already received.
+        marker = " (replayed)" if event.replayed else ""
+        print(f"final: {event.text}{marker}")
+    elif isinstance(event, Reconnecting):
+        # Transient failure: the SDK is recovering on its own — no action
+        # needed, shown here only to make the recovery visible.
+        print(f"[reconnecting, attempt {event.attempt}: {event.reason}]")
+    elif isinstance(event, Reconnected):
+        print(f"[reconnected; {event.replayed_seconds:.1f}s of speech replayed]")
+    elif isinstance(event, TranscriptFailed):
+        # One utterance failed server-side; the session continues.
+        print(f"[utterance failed: {event.message}]")
+    elif isinstance(event, BufferGap):
+        # Audio dropped beyond recovery — always announced, never silent.
+        print(f"[gap: {event.dropped_seconds:.1f}s lost]")
+
+
+async def transcribe(audio: bytes) -> str:
+    client = AsyncTogether()
+
+    # Default reconnect settings retry this endpoint with backoff; failures
+    # you see as Reconnecting/Reconnected events are handled entirely by the
+    # SDK, with un-transcribed speech replayed after each reconnect.
+    async with client.beta.realtime.transcription(
+        model=MODEL,
+        sample_rate=SAMPLE_RATE,
+        event_callback=on_event,
+    ) as session:
+        position = 0
+        while position < len(audio):
+            await session.append(audio[position : position + CHUNK_BYTES])
+            position += CHUNK_BYTES
+            await asyncio.sleep(0.1)  # simulate a live capture cadence
+
+        return await session.flush()
+
+
+def load_pcm(path: Path) -> bytes:
+    with wave.open(str(path), "rb") as w:
+        if (w.getnchannels(), w.getsampwidth(), w.getframerate()) != (1, 2, SAMPLE_RATE):
+            raise SystemExit(f"expected mono 16-bit {SAMPLE_RATE} Hz WAV")
+        return w.readframes(w.getnframes())
+
+
+async def main() -> None:
+    if len(sys.argv) != 2:
+        raise SystemExit(__doc__)
+    transcript = await transcribe(load_pcm(Path(sys.argv[1])))
+    print(f"\nfull transcript: {transcript}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/realtime_transcription.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Realtime transcription of live audio.
+
+Stream PCM audio as it is captured and receive interim text while a phrase
+is being spoken, plus a finalized transcript per utterance. The WAV file
+here stands in for a live source (microphone, phone call, meeting audio).
+
+Usage:
+    uv add "together[realtime]"
+    export TOGETHER_API_KEY=...
+    uv run ./examples/realtime_transcription.py audio.wav   # 16 kHz mono s16le WAV
+"""
+
+from __future__ import annotations
+
+import sys
+import wave
+import asyncio
+from pathlib import Path
+
+from together import AsyncTogether
+from together.realtime import (
+    SessionStarted,
+    TranscriptDelta,
+    TranscriptCompleted,
+    RealtimeSessionEvent,
+)
+
+MODEL = "openai/whisper-large-v3"
+
+SAMPLE_RATE = 16_000
+CHUNK_BYTES = SAMPLE_RATE * 2 // 10  # 100 ms per append, like a live source
+
+
+def on_event(event: RealtimeSessionEvent) -> None:
+    """Handle session events as they arrive."""
+    if isinstance(event, SessionStarted):
+        # Log the server-assigned session id — useful when correlating a
+        # stream with server-side logs or support requests.
+        print(f"session {event.session_id} started on {event.model}")
+    elif isinstance(event, TranscriptDelta):
+        print(f"interim: {event.text}", end="\r")
+    elif isinstance(event, TranscriptCompleted):
+        print(f"final: {event.text}")
+
+
+async def transcribe(audio: bytes) -> str:
+    client = AsyncTogether()
+
+    async with client.beta.realtime.transcription(
+        model=MODEL,
+        sample_rate=SAMPLE_RATE,
+        event_callback=on_event,
+    ) as session:
+        position = 0
+        while position < len(audio):
+            await session.append(audio[position : position + CHUNK_BYTES])
+            position += CHUNK_BYTES
+            await asyncio.sleep(0.1)  # simulate a live capture cadence
+
+        return await session.flush()  # finalize whatever was said last
+
+
+def load_pcm(path: Path) -> bytes:
+    with wave.open(str(path), "rb") as w:
+        if (w.getnchannels(), w.getsampwidth(), w.getframerate()) != (1, 2, SAMPLE_RATE):
+            raise SystemExit(f"expected mono 16-bit {SAMPLE_RATE} Hz WAV")
+        return w.readframes(w.getnframes())
+
+
+async def main() -> None:
+    if len(sys.argv) != 2:
+        raise SystemExit(__doc__)
+    transcript = await transcribe(load_pcm(Path(sys.argv[1])))
+    print(f"\nfull transcript: {transcript}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())