feat(streaming): add acknowledge_silence param and Silence event

assemblyai/__version__.py

@@ -1 +1 @@
-__version__ = "1.5.4"
+__version__ = "1.5.5"

assemblyai/streaming/v3/__init__.py

@@ -26,6 +26,7 @@
     RealTimeSessionParameters,
     RealTimeTranscriberOptions,
     SessionConfiguration,
+    SilenceEvent,
     SpeakerRevisionEvent,
     SpeakerRevisionItem,
     SpeechModel,
@@ -66,6 +67,7 @@
     "RealTimeSessionParameters",
     "RealTimeTranscriber",
     "RealTimeTranscriberOptions",
+    "SilenceEvent",
     "SpeakerRevisionEvent",
     "SpeakerRevisionItem",
     "SpeechModel",

assemblyai/streaming/v3/_base.py

@@ -39,6 +39,7 @@
     RealTimeEvents,
     RealTimeParameters,
     RealTimeTranscriberOptions,
+    SilenceEvent,
     SpeakerRevisionEvent,
     SpeechStartedEvent,
     TerminationEvent,
@@ -284,6 +285,8 @@ def _parse_message(cls, data: Dict[str, Any]) -> Optional[EventMessage]:
                 return _parse_model(SpeakerRevisionEvent, data)
             elif event_type == RealTimeEvents.Heartbeat:
                 return _parse_model(HeartbeatEvent, data)
+            elif event_type == RealTimeEvents.Silence:
+                return _parse_model(SilenceEvent, data)
             elif event_type == RealTimeEvents.Error:
                 return _parse_model(ErrorEvent, data)
             elif event_type == RealTimeEvents.Warning:

assemblyai/streaming/v3/models.py

@@ -95,6 +95,21 @@ class HeartbeatEvent(BaseModel):
     max_speech_probability: float = 0.0
 
 
+class SilenceEvent(BaseModel):
+    """A stretch of audio the server transcribed no speech in.
+
+    Emitted roughly once per second while no speech is being transcribed, and
+    only when the session opted in with `acknowledge_silence` (U3Pro only).
+    """
+
+    type: Literal["Silence"] = "Silence"
+    # Session audio positions, on the same clock as word start/end timestamps.
+    start_ms: int
+    # A worker reconnect replays ~500 ms of audio, so a range can overlap
+    # audio already reported as speech; trim with a running max.
+    end_ms: int
+
+
 class LLMGatewayResponseEvent(BaseModel):
     type: Literal["LLMGatewayResponse"] = "LLMGatewayResponse"
     turn_order: int
@@ -135,6 +150,7 @@ class SpeakerRevisionEvent(BaseModel):
     ErrorEvent,
     WarningEvent,
     HeartbeatEvent,
+    SilenceEvent,
     LLMGatewayResponseEvent,
     SpeakerRevisionEvent,
 ]
@@ -169,6 +185,11 @@ class RealTimeSessionParameters(BaseModel):
     turn_left_pad_ms: Optional[int] = None
     language_codes: Optional[List[str]] = None
     session_heartbeat: Optional[bool] = None
+    # Opt in to Silence messages during silence (U3Pro only). When True, the
+    # server emits a Silence message roughly once per second while no speech is
+    # being transcribed, carrying `start_ms` and `end_ms`, the audio positions
+    # the silent stretch spans. When unset/False, no Silence messages are sent.
+    acknowledge_silence: Optional[bool] = None
 
 
 # Alias: the former name for `RealTimeSessionParameters`, bound to the same object.
@@ -446,6 +467,7 @@ class RealTimeEvents(Enum):
     Error = "Error"
     Warning = "Warning"
     Heartbeat = "Heartbeat"
+    Silence = "Silence"
     LLMGatewayResponse = "LLMGatewayResponse"
     SpeakerRevision = "SpeakerRevision"
 

tests/unit/test_streaming.py

@@ -32,6 +32,7 @@
 from assemblyai.streaming.v3.models import (
     HeartbeatEvent,
     KeepAlive,
+    SilenceEvent,
     TerminateSession,
     UpdateConfiguration,
 )
@@ -2294,3 +2295,99 @@ def test_heartbeat_event_dispatched_to_handler(mocker: MockFixture):
     assert received[0].total_duration_ms == 45205
     assert received[0].realtime_factor == 0.9964
     assert received[0].max_speech_probability == 0.999954
+
+
+def test_client_connect_with_acknowledge_silence(mocker: MockFixture):
+    # Given: client + acknowledge_silence=True
+    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")
+    )
+    params = StreamingParameters(
+        sample_rate=16000,
+        speech_model=SpeechModel.universal_streaming_english,
+        acknowledge_silence=True,
+    )
+
+    # When: connect
+    client.connect(params)
+
+    # Then: the acknowledge_silence wire param is forwarded
+    assert "acknowledge_silence=True" in actual_url
+
+
+def test_acknowledge_silence_defaults_to_none():
+    # Given: params/update-config with no acknowledge_silence set
+    params = StreamingParameters(sample_rate=16000)
+    update = UpdateConfiguration()
+
+    # Then: the field defaults to None on both StreamingSessionParameters
+    # subclasses and is omitted from the connection querystring when unset.
+    assert params.acknowledge_silence is None
+    assert update.acknowledge_silence is None
+    uri = _build_uri("wss://example.com/v3/ws", params)
+    assert "acknowledge_silence" not in uri
+
+
+def test_silence_event_parses_from_wire_message():
+    # Given: a raw Silence wire message
+    data = {
+        "type": "Silence",
+        "start_ms": 12000,
+        "end_ms": 13000,
+    }
+
+    # When: routed through the shared inbound-message dispatch
+    event = StreamingClient._parse_message(data)
+
+    # Then: a fully-populated SilenceEvent is returned
+    assert isinstance(event, SilenceEvent)
+    assert event.type == "Silence"
+    assert event.start_ms == 12000
+    assert event.end_ms == 13000
+
+
+def test_silence_event_dispatched_to_handler(mocker: MockFixture):
+    # Given: a Silence frame on the wire and a handler registered
+    silence_json = json.dumps(
+        {
+            "type": "Silence",
+            "start_ms": 12000,
+            "end_ms": 13000,
+        }
+    )
+    fake_ws = _FakeWebSocket(recv_script=[silence_json])
+    mocker.patch(
+        "assemblyai.streaming.v3.client.websocket_connect",
+        return_value=fake_ws,
+    )
+    received = []
+    client = StreamingClient(
+        StreamingClientOptions(api_key="test", api_host="api.example.com")
+    )
+    client.on(StreamingEvents.Silence, lambda _c, event: received.append(event))
+
+    # When: the client reads the frame
+    client.connect(_default_params())
+    deadline = time.monotonic() + 2.0
+    while time.monotonic() < deadline and not received:
+        time.sleep(0.02)
+    client.disconnect(terminate=False)
+
+    # Then: the handler is invoked with a parsed SilenceEvent
+    assert len(received) == 1
+    assert isinstance(received[0], SilenceEvent)
+    assert received[0].start_ms == 12000
+    assert received[0].end_ms == 13000

tests/unit/test_streaming_async.py

@@ -1031,6 +1031,83 @@ def on_heartbeat(_client, event):
     await client.disconnect()
 
 
+async def test_set_params_with_acknowledge_silence(mocker: MockFixture):
+    # Given: a connected async streaming client
+    fake_ws = _FakeAsyncWebSocket()
+    _patch_connect(mocker, fake_ws)
+
+    client = AsyncStreamingClient(
+        StreamingClientOptions(api_key="test", api_host="api.example.com")
+    )
+    await client.connect(_default_params())
+
+    from assemblyai.streaming.v3.models import (
+        StreamingSessionParameters,
+    )
+
+    # When: set_params is called with acknowledge_silence mid-stream
+    await client.set_params(StreamingSessionParameters(acknowledge_silence=True))
+
+    for _ in range(100):
+        update_frames = [
+            s for s in fake_ws.sent if isinstance(s, str) and "UpdateConfiguration" in s
+        ]
+        if update_frames:
+            break
+        await asyncio.sleep(0.01)
+
+    # Then: an UpdateConfiguration frame carrying acknowledge_silence is sent
+    update_frames = [
+        s for s in fake_ws.sent if isinstance(s, str) and "UpdateConfiguration" in s
+    ]
+    assert len(update_frames) == 1
+    payload = json.loads(update_frames[0])
+    assert payload["type"] == "UpdateConfiguration"
+    assert payload["acknowledge_silence"] is True
+
+    await client.disconnect()
+
+
+async def test_silence_event_dispatched_to_handler(mocker: MockFixture):
+    fake_ws = _FakeAsyncWebSocket()
+    _patch_connect(mocker, fake_ws)
+
+    received = []
+
+    def on_silence(_client, event):
+        received.append(event)
+
+    client = AsyncStreamingClient(
+        StreamingClientOptions(api_key="test", api_host="api.example.com")
+    )
+    client.on(StreamingEvents.Silence, on_silence)
+    await client.connect(_default_params())
+
+    fake_ws.push_message(
+        json.dumps(
+            {
+                "type": "Silence",
+                "start_ms": 12000,
+                "end_ms": 13000,
+            }
+        )
+    )
+
+    for _ in range(50):
+        if received:
+            break
+        await asyncio.sleep(0.01)
+
+    from assemblyai.streaming.v3.models import SilenceEvent
+
+    assert len(received) == 1
+    assert isinstance(received[0], SilenceEvent)
+    assert received[0].start_ms == 12000
+    assert received[0].end_ms == 13000
+
+    await client.disconnect()
+
+
 async def test_force_endpoint_enqueues_force_endpoint_frame(mocker: MockFixture):
     fake_ws = _FakeAsyncWebSocket()
     _patch_connect(mocker, fake_ws)