DX-818: Add realtime transcription support with failover handling#450
DX-818: Add realtime transcription support with failover handling#450yadavsahil197 wants to merge 10 commits into
Conversation
Broly Security ScanNote Summary 6 actionable finding(s) in this PR
5 highest-priority actionable rows in the table below (critical/high first, then top medium).
Dismiss false positivesEach dismissable row has a Dismiss key (
Note Re-scan this PR anytime with
|
blainekasten
left a comment
blainekasten
left a comment
There was a problem hiding this comment.
This looks great!
As we discussed in slack, let's move this functionality under the beta namespace for now to give us some wiggle room if we decide to change the api interface in the near term.
85d5879 to
6d913ab
Compare
|
Moved to the beta namespace as discussed! |
yadavsahil197
changed the title
| for _held, reclaim in holders: | ||
| if deficit <= 0: | ||
| return | ||
| deficit -= reclaim(deficit) |
There was a problem hiding this comment.
I think there's a cross-thread mutation hazard here. charge() runs on whatever thread is appending, and when the pool is under pressure this ends up calling reclaim → AudioBuffer.trim_to on other sessions buffers. But those sessions are running on their own event loops so we can popleft from a deque while its owner's writer loop is mid-iteration in read_from which will be a error.
It would only fire under pool pressure with multiple sessions - could we marshal the reclaim onto the holder's loop via call_soon_threadsafe or put a lock around AudioBuffer?
| pool=pool, | ||
| ) | ||
| self._pool = pool | ||
| pool.register(self, lambda: self.state.buffer.size, self.state.reclaim) |
There was a problem hiding this comment.
Registering in init but only unregistering in close()/_fail() means a session that's constructed but never started (or just abandoned) stays strongly referenced by the client-scoped pool forever. Could we move registration into start(), or make _holders a WeakKeyDictionary so the pool never keeps a session alive on its own?
| # 300s no-append idle timeout | ||
| if self._keepalive_silence and self._last_append_at: | ||
| if now - self._last_append_at > 200.0: | ||
| await self.append(b"\x00" * int(self.state.bps * 0.1)) # 100ms of silence |
There was a problem hiding this comment.
This append() can raise a RealtimeBufferOverflowError when overflow="error" or a stored terminal failure racing in via _raise_if_failed() and _watchdog_loop only handles CancelledError. So the watchdog task dies with an unobserved exception and we lose echo probing + keepalive while the session looks healthy. Worth wrapping the loop body in an except Exception that at least logs.
| self._writer_wakeup.set() | ||
| for task in (self._watchdog_task, self._writer_task, self._reconnect_task, self._reader_task): | ||
| if task is not None: | ||
| task.cancel() |
There was a problem hiding this comment.
We cancel these but never await them, and in the sync the loop gets stopped right after so tasks can be destroyed while still pending (asyncio warnings, and the reader may not finish tearing down the socket). A gather(*tasks, return_exceptions=True) after the cancels would make shutdown deterministic.
| self._session = self._call(_create()) | ||
|
|
||
| def __enter__(self) -> "RealtimeTranscriptionSession": | ||
| self.start() |
There was a problem hiding this comment.
Probably want a try/except around this that calls close() before re-raising. If start() raises here (e.g. bad base_url → RealtimeConnectionError), exit never runs so the background loop thread we spawned in init lives until atexit.
6e2c0c7 to
7aa3dfc
Compare
|
|
||
| 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 |
There was a problem hiding this comment.
too high, should be max(2, len(endpoints)) imo -- retry once if one endpoint otherwise try each once
|
Design question: warm/standby WS for cross-endpoint failover Right now there's no pre-established connection anywhere — every WS is opened lazily when it's needed. The The consequence is on the failover path: the backup endpoint's client is just config ("nothing connects until a session starts"), so when the primary dies we can only start dialing the destination after we already know the source is dead. That failover pays a cold open — Worth considering a warm-standby option: pre-dial the next endpoint in the ring and hold it through Not blocking — just want it captured. |
|
Suggestion: expose per-token Noticed while reading the event types: our realtime backend (ipop, {
"type": "...completed", "item_id": "msg_7", "transcript": "...",
"logprobs": { "avg_logprob": -0.21, "token_logprobs": [...], "token_texts": [...] },
"tokens": [ { "token_id": 123, "text": " hi", "confidence": 0.94 }, ... ]
}Right now the SDK's wire models don't declare these, so a consumer can only reach them accidentally — they survive as untyped extras because I put together a small additive change to make them first-class (happy to fold in or you can take it):
Non-breaking, verified:
diffdiff --git a/src/together/lib/realtime/__init__.py b/src/together/lib/realtime/__init__.py
index f7911422..36eb1b9e 100644
--- a/src/together/lib/realtime/__init__.py
+++ b/src/together/lib/realtime/__init__.py
@@ -13,10 +13,12 @@ from ._types import (
BufferOptions,
SessionStarted,
TranscriptDelta,
+ RealtimeLogprobs,
ReconnectOptions,
TranscriptFailed,
EchoResponseEvent,
RealtimeErrorInfo,
+ TranscriptionToken,
TurnDetectionParam,
RealtimeServerEvent,
SessionCreatedEvent,
@@ -48,6 +50,8 @@ __all__ = [
"EchoResponseEvent",
"UnknownEvent",
"RealtimeErrorInfo",
+ "RealtimeLogprobs",
+ "TranscriptionToken",
# events (normalized session)
"RealtimeSessionEvent",
"SessionStarted",
diff --git a/src/together/lib/realtime/_session.py b/src/together/lib/realtime/_session.py
index 6394a110..6f8a3528 100644
--- a/src/together/lib/realtime/_session.py
+++ b/src/together/lib/realtime/_session.py
@@ -489,6 +489,8 @@ class AsyncRealtimeTranscriptionSession:
audio_start=seg.audio_start,
audio_end=self.state.to_seconds(self.state.write_head),
replayed=seg.replayed,
+ logprobs=event.logprobs,
+ tokens=event.tokens,
raw=event,
)
)
@@ -509,6 +511,8 @@ class AsyncRealtimeTranscriptionSession:
audio_start=seg.audio_start,
audio_end=self.state.to_seconds(self.state.anchor or self.state.write_head),
replayed=seg.replayed,
+ logprobs=event.logprobs,
+ tokens=event.tokens,
raw=event,
)
)
diff --git a/src/together/lib/realtime/_types.py b/src/together/lib/realtime/_types.py
index ff4d86a5..39120842 100644
--- a/src/together/lib/realtime/_types.py
+++ b/src/together/lib/realtime/_types.py
@@ -2,7 +2,7 @@ from __future__ import annotations
import json
import base64
-from typing import Any, Dict, Union, Mapping, Optional, cast
+from typing import Any, Dict, List, Union, Mapping, Optional, cast
from typing_extensions import Literal, Required, Annotated, TypeAlias, TypedDict
from ..._utils import PropertyInfo
@@ -15,6 +15,8 @@ __all__ = [
# wire (server -> client) events
"RealtimeErrorInfo",
"RealtimeSessionInfo",
+ "RealtimeLogprobs",
+ "TranscriptionToken",
"SessionCreatedEvent",
"TranscriptionDeltaEvent",
"TranscriptionCompletedEvent",
@@ -109,6 +111,24 @@ class SessionCreatedEvent(BaseModel):
session: Optional[RealtimeSessionInfo] = None
+# Per-segment/token quality signals some backends attach to delta/completed
+# frames (off the OpenAI spec). Populated only when the engine decodes with
+# beam>=2; greedy decode omits them. All fields optional so frames that don't
+# carry them still parse.
+class RealtimeLogprobs(BaseModel):
+ avg_logprob: Optional[float] = None
+ token_logprobs: Optional[List[float]] = None
+ """Parallel to token_texts: token_logprobs[i] is the logprob of token_texts[i]."""
+ token_texts: Optional[List[str]] = None
+
+
+class TranscriptionToken(BaseModel):
+ token_id: Optional[int] = None
+ text: Optional[str] = None
+ confidence: Optional[float] = None
+ """Softmax probability of the emitted token (exp of its logprob)."""
+
+
class TranscriptionDeltaEvent(BaseModel):
type: Literal["conversation.item.input_audio_transcription.delta"]
item_id: Optional[str] = None
@@ -117,6 +137,8 @@ class TranscriptionDeltaEvent(BaseModel):
"""Server-side consumed-speech clock — NOT a position on the appended-audio
timeline (silence is skipped); do not use for buffer accounting."""
duration: Optional[float] = None
+ logprobs: Optional[RealtimeLogprobs] = None
+ tokens: Optional[List[TranscriptionToken]] = None
class TranscriptionCompletedEvent(BaseModel):
@@ -126,6 +148,8 @@ class TranscriptionCompletedEvent(BaseModel):
start: Optional[float] = None
"""See TranscriptionDeltaEvent.start — informational only."""
duration: Optional[float] = None
+ logprobs: Optional[RealtimeLogprobs] = None
+ tokens: Optional[List[TranscriptionToken]] = None
class TranscriptionFailedEvent(BaseModel):
@@ -328,6 +352,8 @@ class TranscriptDelta(BaseModel):
audio_start: Optional[float] = None
audio_end: Optional[float] = None
replayed: bool = False
+ logprobs: Optional[RealtimeLogprobs] = None
+ tokens: Optional[List[TranscriptionToken]] = None
raw: Optional[TranscriptionDeltaEvent] = None
@@ -338,6 +364,8 @@ class TranscriptCompleted(BaseModel):
audio_start: Optional[float] = None
audio_end: Optional[float] = None
replayed: bool = False
+ logprobs: Optional[RealtimeLogprobs] = None
+ tokens: Optional[List[TranscriptionToken]] = None
raw: Optional[TranscriptionCompletedEvent] = None
diff --git a/src/together/realtime.py b/src/together/realtime.py
index e16a8917..fbd49b0c 100644
--- a/src/together/realtime.py
+++ b/src/together/realtime.py
@@ -18,10 +18,12 @@ from .lib.realtime import (
RealtimeError as RealtimeError,
SessionStarted as SessionStarted,
TranscriptDelta as TranscriptDelta,
+ RealtimeLogprobs as RealtimeLogprobs,
ReconnectOptions as ReconnectOptions,
TranscriptFailed as TranscriptFailed,
EchoResponseEvent as EchoResponseEvent,
RealtimeErrorInfo as RealtimeErrorInfo,
+ TranscriptionToken as TranscriptionToken,
TurnDetectionParam as TurnDetectionParam,
RealtimeServerEvent as RealtimeServerEvent,
SessionCreatedEvent as SessionCreatedEvent,
@@ -50,6 +52,8 @@ __all__ = [
"EchoResponseEvent",
"UnknownEvent",
"RealtimeErrorInfo",
+ "RealtimeLogprobs",
+ "TranscriptionToken",
# events (normalized session)
"RealtimeSessionEvent",
"SessionStarted", |
…ription events - Classify server `no_healthy_upstream` as a terminal RealtimeConnectionError (code="no_healthy_upstream"), raised immediately — bypasses same-endpoint reconnect — so an application failover ring rotates to another endpoint. - Lower default reconnect max_attempts 10 -> 2: ride out a transient websocket blip, then fail over fast instead of grinding a dead endpoint for ~a minute. - Expose logprobs (avg_logprob, token_logprobs/token_texts) and per-token confidence on both wire and normalized delta/completed events (RealtimeLogprobs, TranscriptionToken); add `start` to TranscriptionFailedEvent. - Document the failover contract (catch RealtimeConnectionError; code marks no_healthy_upstream) in examples/realtime_failover.py. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Pushed a commit ( What it adds (9 files, +113/−12, realtime pkg only):
Failover orchestration stays app-side (the ring example) — this is just the SDK support that makes it trigger on the new signal. Verified: 88/88 realtime unit tests, ruff/format/mypy clean. 🤖 Generated with Claude Code |
Describe endpoint failures in terms of the API contract (the server reported it cannot serve; code='no_healthy_upstream') rather than internal infrastructure. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
4 participants
No description provided.