From ba6e56dc07d3815734c31ca78d1e533871bee2dc Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Thu, 24 Sep 2026 10:23:53 -0600 Subject: [PATCH 1/2] feat(streaming): add speaker_labels_revision_interval_ms param Connect-time cadence for mid-stream SpeakerRevision events when speaker_labels is enabled. Mirrors Node SDK #188. Version 1.6.1. Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 2 +- assemblyai/__version__.py | 2 +- assemblyai/streaming/v3/models.py | 11 +++- tests/unit/test_streaming.py | 96 ++++++++++++++++++++++++++++++ tests/unit/test_streaming_async.py | 28 +++++++++ 5 files changed, 135 insertions(+), 4 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 51592ca2..7e055617 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -336,7 +336,7 @@ The former `Streaming*` names (`StreamingClient`, `AsyncStreamingClient`, `Strea - `Warning` → `WarningEvent(warning_code: int, warning: str)` - `Error` → `RealTimeError` (an `Exception` subclass) with `.code: int | None`; `str(error)` is the message. Server-side errors come through `on_error` rather than being raised, and the payload is a `RealTimeError`, **not** the wire `ErrorEvent` class. - `LLMGatewayResponse` → `LLMGatewayResponseEvent(turn_order: int, transcript: str, data: Any)` -- `SpeakerRevision` → `SpeakerRevisionEvent(revisions: list[SpeakerRevisionItem])` — diarization-only. Sent once per offline-recluster resolve. Each `SpeakerRevisionItem(turn_order: int, speaker_label: str | None, words: list[Word])` is an earlier Turn whose labels changed (unchanged turns are omitted). For each item, match by `turn_order` against the original Turn and replace its per-word `speaker` (and the turn-level `speaker_label`) with the revision's values. Text and word timestamps are unchanged. +- `SpeakerRevision` → `SpeakerRevisionEvent(revisions: list[SpeakerRevisionItem])` — diarization-only. Sent once per offline-recluster resolve. Each `SpeakerRevisionItem(turn_order: int, speaker_label: str | None, words: list[Word])` is an earlier Turn whose labels changed (unchanged turns are omitted). For each item, match by `turn_order` against the original Turn and replace its per-word `speaker` (and the turn-level `speaker_label`) with the revision's values. Text and word timestamps are unchanged. By default only the end-of-stream revision is sent; set `speaker_labels_revision_interval_ms` on `RealTimeParameters` (ms of audio time; values below 120_000 are raised to 120_000 server-side, larger values honored, 300_000 recommended) to also get them mid-stream — the first can only arrive after ~120 s of speech, and one is sent only when a label actually changed. **Sync streaming:** ```python diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index e4adfb83..f49459c7 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "1.6.0" +__version__ = "1.6.1" diff --git a/assemblyai/streaming/v3/models.py b/assemblyai/streaming/v3/models.py index 2ab88d9d..ca0be1de 100644 --- a/assemblyai/streaming/v3/models.py +++ b/assemblyai/streaming/v3/models.py @@ -4,12 +4,12 @@ try: # pydantic v2 import - from pydantic import BaseModel, model_validator + from pydantic import BaseModel, Field, model_validator pydantic_v2 = True except ImportError: # pydantic v1 import (fallback for Python < 3.14) - from pydantic import BaseModel, root_validator + from pydantic import BaseModel, Field, root_validator pydantic_v2 = False @@ -332,6 +332,13 @@ class RealTimeParameters(RealTimeSessionParameters): webhook_auth_header_value: Optional[str] = None llm_gateway: Optional[LLMGatewayConfig] = None speaker_labels: Optional[bool] = None + # Cadence, in ms of audio time, at which the server emits mid-stream + # SpeakerRevision events (plus the final one at end of stream). Unset or + # 0 means end-of-stream revision only. Values below 120_000 are raised + # to 120_000 server-side; larger values are honored, and 300_000 is the + # recommended cadence. The server rejects values outside 0..86_400_000 + # (24 h) at connect. Only used when speaker_labels is enabled. + speaker_labels_revision_interval_ms: Optional[int] = Field(None, ge=0) max_speakers: Optional[int] = None voice_focus: Optional[NoiseSuppressionModel] = None voice_focus_threshold: Optional[float] = None diff --git a/tests/unit/test_streaming.py b/tests/unit/test_streaming.py index 1bb2a5d7..70049a47 100644 --- a/tests/unit/test_streaming.py +++ b/tests/unit/test_streaming.py @@ -867,6 +867,102 @@ def mocked_websocket_connect( assert "max_speakers=3" in actual_url +def test_client_connect_with_speaker_labels_revision_interval(mocker: MockFixture): + # Given: speaker_labels plus a mid-stream revision cadence + actual_url = None + + def mocked_websocket_connect( + url: str, additional_headers: dict, open_timeout: float + ): + nonlocal actual_url + actual_url = url + + mocker.patch( + "assemblyai.streaming.v3.client.websocket_connect", + new=mocked_websocket_connect, + ) + + _disable_rw_threads(mocker) + + client = StreamingClient( + StreamingClientOptions(api_key="test", api_host="api.example.com") + ) + + # When: connecting + client.connect( + StreamingParameters( + sample_rate=16000, + speech_model=SpeechModel.universal_streaming_english, + speaker_labels=True, + speaker_labels_revision_interval_ms=120_000, + ) + ) + + # Then: the GA (un-prefixed) query param carries the value verbatim + assert "speaker_labels=True" in actual_url + assert "speaker_labels_revision_interval_ms=120000" in actual_url + assert "_speaker_labels_revision_interval_ms" not in actual_url + + +def test_build_uri_keeps_explicit_zero_revision_interval(): + # Given: the interval explicitly set to 0 (end-of-stream revision only) + params = StreamingParameters( + sample_rate=16000, + speech_model=SpeechModel.universal_streaming_english, + speaker_labels=True, + speaker_labels_revision_interval_ms=0, + ) + + # When: the connection URI is built + uri = _build_uri("api.example.com", params) + + # Then: 0 is sent, not dropped like an unset value + assert "speaker_labels_revision_interval_ms=0" in uri + + +def test_build_uri_omits_unset_revision_interval(): + # Given: speaker_labels without a revision interval + params = StreamingParameters( + sample_rate=16000, + speech_model=SpeechModel.universal_streaming_english, + speaker_labels=True, + ) + + # When: the connection URI is built + uri = _build_uri("api.example.com", params) + + # Then: the param is absent so the server default applies + assert "speaker_labels_revision_interval_ms" not in uri + + +def test_negative_revision_interval_is_rejected(): + # Given/When: a negative cadence, which the server also rejects at connect + # Then: the SDK refuses it before a connection is attempted + with pytest.raises(ValueError, match="speaker_labels_revision_interval_ms"): + StreamingParameters( + sample_rate=16000, + speech_model=SpeechModel.universal_streaming_english, + speaker_labels=True, + speaker_labels_revision_interval_ms=-1, + ) + + +def test_large_revision_interval_is_not_clamped_client_side(): + # Given: a cadence above the server's 300_000 default, which the server + # honors rather than clamps + params = StreamingParameters( + sample_rate=16000, + speech_model=SpeechModel.universal_streaming_english, + speaker_labels=True, + speaker_labels_revision_interval_ms=600_000, + ) + + # When/Then: the SDK forwards it unchanged + assert "speaker_labels_revision_interval_ms=600000" in _build_uri( + "api.example.com", params + ) + + def test_client_connect_with_continuous_partials(mocker: MockFixture): # Given: client + continuous_partials=True (U3-Pro steady-partials mode) actual_url = None diff --git a/tests/unit/test_streaming_async.py b/tests/unit/test_streaming_async.py index 396308a8..77a455c8 100644 --- a/tests/unit/test_streaming_async.py +++ b/tests/unit/test_streaming_async.py @@ -137,6 +137,34 @@ async def test_client_connect_with_token(mocker: MockFixture): await client.disconnect() +async def test_client_connect_with_speaker_labels_revision_interval( + mocker: MockFixture, +): + # Given: speaker_labels plus a mid-stream revision cadence + fake_ws = _FakeAsyncWebSocket() + fake_connect = _patch_connect(mocker, fake_ws) + + client = AsyncStreamingClient( + StreamingClientOptions(api_key="test", api_host="api.example.com") + ) + + # When: connecting + await client.connect( + StreamingParameters( + sample_rate=16000, + speech_model=SpeechModel.universal_streaming_english, + speaker_labels=True, + speaker_labels_revision_interval_ms=120_000, + ) + ) + + # Then: the query string carries the GA param name and value + assert "speaker_labels=True" in fake_connect.uri + assert "speaker_labels_revision_interval_ms=120000" in fake_connect.uri + + await client.disconnect() + + async def test_stream_bytes_writes_to_socket(mocker: MockFixture): fake_ws = _FakeAsyncWebSocket() _patch_connect(mocker, fake_ws) From 2307dc485ddfcced8b8f63ebb63c34fc71a088b1 Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Thu, 24 Sep 2026 11:15:29 -0600 Subject: [PATCH 2/2] Remove comment on model --- assemblyai/streaming/v3/models.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/assemblyai/streaming/v3/models.py b/assemblyai/streaming/v3/models.py index ca0be1de..b278dee4 100644 --- a/assemblyai/streaming/v3/models.py +++ b/assemblyai/streaming/v3/models.py @@ -332,12 +332,6 @@ class RealTimeParameters(RealTimeSessionParameters): webhook_auth_header_value: Optional[str] = None llm_gateway: Optional[LLMGatewayConfig] = None speaker_labels: Optional[bool] = None - # Cadence, in ms of audio time, at which the server emits mid-stream - # SpeakerRevision events (plus the final one at end of stream). Unset or - # 0 means end-of-stream revision only. Values below 120_000 are raised - # to 120_000 server-side; larger values are honored, and 300_000 is the - # recommended cadence. The server rejects values outside 0..86_400_000 - # (24 h) at connect. Only used when speaker_labels is enabled. speaker_labels_revision_interval_ms: Optional[int] = Field(None, ge=0) max_speakers: Optional[int] = None voice_focus: Optional[NoiseSuppressionModel] = None