update example

Author: yadavsahil197

README.md

@@ -170,17 +170,20 @@ async for chat_completion in stream:
 
 ## Realtime transcription
 
-Stream audio over WebSocket and receive interim and final transcripts as the
-audio plays. Requires the `realtime` extra:
+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
 pip install "together[realtime]"
 ```
 
-`client.realtime.transcription()` is the recommended surface: an
-auto-reconnecting session that buffers un-transcribed audio on the client and
-replays it after a connection drop, so transient WebSocket failures don't lose
-speech.
+Use `client.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
@@ -190,43 +193,50 @@ client = AsyncTogether()
 
 async with client.realtime.transcription(
     model="openai/whisper-large-v3",
-    sample_rate=16_000,  # validated against input_audio_format
+    sample_rate=16_000,
 ) as session:
-    await session.append(pcm_chunk)  # 16 kHz mono s16le PCM, any chunk size
+    # 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)
+            print("interim:", event.text)   # updates while a phrase is spoken
         elif isinstance(event, TranscriptCompleted):
-            print("final:", event.text)
-    transcript = await session.flush()
-```
-
-Notes:
-
-- Audio must be PCM signed 16-bit little-endian; the default format is
-  `pcm_s16le_16000` (16 kHz mono). Resample before appending.
-- Server-side voice activity detection is on by default: final transcripts
-  arrive per detected utterance without explicit commits. Pass
-  `turn_detection={"type": "none"}` to segment manually with
-  `await session.commit()`.
-- All server session parameters are supported: `language`, `prompt`,
-  `rolling_prompt`, `energy_gate_rms`, turn-detection tuning via
-  `turn_detection={...}` (`threshold`, `min_silence_duration_ms`,
-  `max_speech_duration_s`, `speech_pad_ms`, Deepgram `eot_*`), and
-  `session_params={...}` for engine-specific passthrough options.
-- On retryable failures (network errors, server hiccups, 429/5xx handshakes)
-  the session reconnects with jittered exponential backoff and replays buffered
-  audio from `max(head - max_replay_seconds, last transcribed position)`, where
-  the transcribed position comes from the `completed` event's `start + duration`
-  and `buffer={"max_replay_seconds": 5.0}` is the client-configurable cap
-  (`0` resumes live with no replay; `None` replays the full untranscribed
-  window). Events produced from replayed audio carry `replayed=True`.
-- Buffer gaps (audio dropped after outages exceeding the retention window) are
-  always surfaced as `BufferGap` events — never silent.
-- The sync client (`Together().realtime.transcription(...)`) mirrors this API
-  and runs the session on a background thread — intended for low session
-  counts; use the async client for high-concurrency workloads.
-- For full manual control (raw wire events, no retries), use
+            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().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.realtime.connect()`.
 
 See [`examples/realtime_transcription.py`](examples/realtime_transcription.py)

examples/realtime_failover.py

@@ -1,15 +1,13 @@
 #!/usr/bin/env python3
-"""Realtime transcription with endpoint failover.
+"""Realtime transcription with failover across multiple endpoints.
 
-Run against a primary endpoint; when it fails beyond recovery, resume on the
-next endpoint in the ring, carrying over any audio that was never transcribed
-so no speech is lost across the switch.
+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.
 
-Division of labor:
-  - The SDK handles transient failures on the CURRENT endpoint internally
-    (reconnect, backoff, replay) — you'll see Reconnecting/Reconnected events.
-  - When an endpoint is unrecoverable, SDK calls raise RealtimeConnectionError.
-    Moving to another endpoint is application code: the loop below.
+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:
     pip install "together[realtime]"
@@ -28,26 +26,19 @@
 from together.lib.realtime import (
     BufferGap,
     SessionStarted,
+    TranscriptDelta,
     TranscriptCompleted,
     RealtimeSessionEvent,
     RealtimeConnectionError,
 )
 
-# Independent deployments of the same model, addressed directly (per-cluster
-# URLs) so the two ring entries share no failure domain. Order = preference;
-# on failure the next entry takes over, wrapping around.
-#
-# base_url is the API root and must end at /v1 — the SDK appends the
-# /realtime path (and any other resource paths) itself.
+# 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://funkyfalcon.api.together.ai/pool/ipop-4-openai-whisper-large-v3-92f607e9/v1",
-        "together_sso/openai/whisper-large-v3-47df6eb0",
-    ),
-    (
-        "https://trickytrout.api.together.ai/pool/ipop-4-openai-whisper-large-v3-e6e3a747/v1",
-        "together_sso/openai/whisper-large-v3-e151bfbf",
-    ),
+    ("https://api.together.ai/v1", "openai/whisper-large-v3-endpoint1"),
+    ("https://api.together.ai/v1", "openai/whisper-large-v3-endpoint2"),
 ]
 
 SAMPLE_RATE = 16_000
@@ -56,18 +47,19 @@
 
 
 def on_event(event: RealtimeSessionEvent) -> None:
-    """Handle transcripts as they arrive (called for every session event)."""
+    """Handle session events as they arrive."""
     if isinstance(event, SessionStarted):
-        # Log the server-assigned session id — invaluable when correlating a
-        # stream with server-side logs or support requests.
+        # 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 — always announced, never silent.
+        # Audio dropped beyond recovery.
         print(f"[gap: {event.dropped_seconds:.1f}s lost]")
 
 
@@ -83,32 +75,29 @@ async def transcribe_with_failover(audio: bytes) -> str:
 
     for attempt in range(MAX_ATTEMPTS):
         client, model = clients[attempt % len(clients)]
-        print(f"--- streaming to {model} via {client.base_url.host}")
+        print(f"--- streaming to {model}")
 
         session = client.realtime.transcription(
             model=model,
             sample_rate=SAMPLE_RATE,
             event_callback=on_event,
-            # Fail fast: hand a dead endpoint straight to this loop instead of
-            # retrying it. Raise max_attempts to retry in place before switching.
+            # 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:
-                    # One append, no pacing: 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.
+                    # 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])
+                    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.

examples/realtime_transcription.py

@@ -1,14 +1,14 @@
 #!/usr/bin/env python3
-"""Realtime transcription over WebSocket with automatic reconnection.
+"""Realtime transcription of live audio.
 
-Streams a 16 kHz mono PCM WAV (or raw PCM) file to the Together realtime API
-and prints interim and final transcripts. The session survives connection
-drops: buffered audio is replayed from the last confirmed transcript.
+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:
     pip install "together[realtime]"
     export TOGETHER_API_KEY=...
-    python examples/realtime_transcription.py audio.wav
+    python examples/realtime_transcription.py audio.wav   # 16 kHz mono s16le WAV
 """
 
 from __future__ import annotations
@@ -20,70 +20,59 @@
 
 from together import AsyncTogether
 from together.lib.realtime import (
-    BufferGap,
-    Reconnected,
-    Reconnecting,
+    SessionStarted,
     TranscriptDelta,
     TranscriptCompleted,
+    RealtimeSessionEvent,
 )
 
-CHUNK_MS = 40
+MODEL = "openai/whisper-large-v3"
+
 SAMPLE_RATE = 16_000
-BYTES_PER_SECOND = SAMPLE_RATE * 2
+CHUNK_BYTES = SAMPLE_RATE * 2 // 10  # 100 ms per append, like a live source
 
 
-def load_pcm(path: Path) -> bytes:
-    if path.suffix.lower() != ".wav":
-        return path.read_bytes()
-    with wave.open(str(path), "rb") as wav:
-        if wav.getnchannels() != 1 or wav.getsampwidth() != 2 or wav.getframerate() != SAMPLE_RATE:
-            raise SystemExit("expected mono 16-bit 16 kHz WAV (resample first)")
-        return wav.readframes(wav.getnframes())
+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 main() -> None:
-    audio = load_pcm(Path(sys.argv[1]))
+async def transcribe(audio: bytes) -> str:
     client = AsyncTogether()
 
     async with client.realtime.transcription(
-        # model="together_sso/openai/whisper-large-v3-47df6eb0",
-        model="together_sso/openai/whisper-large-v3-e151bfbf",
+        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
+
 
-        async def feed() -> None:
-            chunk_bytes = BYTES_PER_SECOND * CHUNK_MS // 1000
-            for start in range(0, len(audio), chunk_bytes):
-                await session.append(audio[start : start + chunk_bytes])
-                await asyncio.sleep(CHUNK_MS / 1000)  # pace like a live mic
-            print("\n[audio fully sent — draining transcripts]")
-
-        feeder = asyncio.create_task(feed())
-
-        async def consume() -> None:
-            async for event in session:
-                if isinstance(event, TranscriptDelta):
-                    # print(f"\r… {event.text}", end="", flush=True)
-                    pass
-                elif isinstance(event, TranscriptCompleted):
-                    # marker = " (replayed)" if event.replayed else ""
-                    # print(f"\rfinal: {event.text}{marker}")
-                    print(event.text)
-                elif isinstance(event, Reconnecting):
-                    # print(f"\n[reconnecting: {event.reason} (attempt {event.attempt})]")
-                    pass
-                elif isinstance(event, Reconnected):
-                    # print(f"[reconnected; replayed {event.replayed_seconds:.1f}s of audio]")
-                    pass
-                elif isinstance(event, BufferGap):
-                    # print(f"[warning: {event.dropped_seconds:.1f}s of audio lost]")
-                    pass
-
-        consumer = asyncio.create_task(consume())
-        await feeder
-        transcript = await session.flush()
-        consumer.cancel()
-        print(f"\nfull transcript: {transcript}")
+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__":

src/together/lib/realtime/_session.py

@@ -218,7 +218,17 @@ async def __aexit__(self, exc_type: object, *_exc: object) -> None:
     async def start(self) -> None:
         if self._connection is not None:
             return
-        connection = await self._open_connection()
+        try:
+            connection = await self._open_connection()
+        except Exception as exc:
+            status = handshake_status_of(exc)
+            detail = f"HTTP {status}" if status is not None else exc.__class__.__name__
+            raise RealtimeConnectionError(
+                f"could not open realtime connection to {self._client.base_url} ({detail}); "
+                "check that base_url / TOGETHER_BASE_URL points at an API root that "
+                "serves /realtime, e.g. https://api.together.ai/v1",
+                cause=exc,
+            ) from exc
         created = await self._await_session_created(connection)
         self.state.begin_epoch(self.state.write_head)
         self._sent_offset = self.state.write_head
@@ -300,14 +310,18 @@ async def flush(self, *, quiescence: float = 2.0, timeout: float = 30.0) -> str:
         if self.state.write_head > (self.state.anchor or 0):
             await self.commit()
         self._raise_if_failed()
-        deadline = self._now() + timeout
+        started = self._now()
+        deadline = started + timeout
         while self._now() < deadline:
             if self._failure is not None:
                 raise self._failure
             if self._closed:
                 break
             fully_sent = self._sent_offset >= self.state.write_head and not self._pending_controls
-            quiet_for = self._now() - max(self._last_transcript_event, self._connected_at)
+            # the quiescence clock starts at flush time so the commit sent
+            # above always gets a response window: a final for the tail
+            # utterance may land seconds after the last transcript event
+            quiet_for = self._now() - max(self._last_transcript_event, self._connected_at, started)
             if fully_sent and quiet_for >= quiescence and self._reconnect_task is None:
                 break
             await asyncio.sleep(min(0.1, quiescence / 4))