From 02dee09497d10fa60d8d2a65f0813ffddc5df7eb Mon Sep 17 00:00:00 2001 From: AssemblyAI Date: Fri, 3 Jul 2026 06:55:57 -0700 Subject: [PATCH] Project import generated by Copybara. GitOrigin-RevId: e75527160d41d82d1fe4747445db14cd88af3b02 --- README.md | 38 +++++------ assemblyai/__version__.py | 2 +- assemblyai/client.py | 7 ++ assemblyai/streaming/v3/_base.py | 5 ++ assemblyai/streaming/v3/models.py | 9 +++ assemblyai/sync.py | 53 ++++++++++++++- assemblyai/sync_api.py | 1 + assemblyai/types.py | 3 + tests/unit/test_streaming.py | 100 +++++++++++++++++++++++++++++ tests/unit/test_streaming_async.py | 78 ++++++++++++++++++++++ tests/unit/test_sync.py | 74 +++++++++++++++++++++ 11 files changed, 349 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index ba44178..34b267d 100644 --- a/README.md +++ b/README.md @@ -104,7 +104,7 @@ aai.settings.api_key = "YOUR_API_KEY" audio_file = "./example.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, speaker_labels=True, ) @@ -130,7 +130,7 @@ aai.settings.api_key = "YOUR_API_KEY" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, speaker_labels=True, ) @@ -177,7 +177,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True ) @@ -214,7 +214,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True ) @@ -248,7 +248,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True ) @@ -280,7 +280,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True ) config.set_custom_spelling( @@ -324,7 +324,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True ) @@ -402,7 +402,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, ).set_redact_pii( policies=[ @@ -431,7 +431,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, ).set_redact_pii( policies=[ @@ -465,7 +465,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, auto_chapters=True ) @@ -493,7 +493,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, summarization=True, summary_model=aai.SummarizationModel.informative, @@ -531,7 +531,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, content_safety=True ) @@ -583,7 +583,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, sentiment_analysis=True ) @@ -626,7 +626,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, entity_detection=True ) @@ -655,7 +655,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, iab_categories=True ) @@ -690,7 +690,7 @@ aai.settings.api_key = "" audio_file = "https://assembly.ai/wildfires.mp3" config = aai.TranscriptionConfig( - speech_models=["universal-3-pro", "universal-2"], + speech_models=["universal-3-5-pro", "universal-2"], language_detection=True, auto_highlights=True ) @@ -745,7 +745,7 @@ client.on(StreamingEvents.Termination, on_terminated) client.on(StreamingEvents.Error, on_error) client.connect(StreamingParameters( - sample_rate=16000, speech_model="universal-3-5-pro", format_turns=True, + sample_rate=16000, speech_model="universal-3-5-pro", )) try: client.stream(aai.extras.stream_file(filepath="audio.wav", sample_rate=16000)) @@ -874,7 +874,7 @@ async def main(): async with AsyncStreamingClient(StreamingClientOptions(api_key="")) as client: client.on(StreamingEvents.Turn, on_turn) await client.connect(StreamingParameters( - sample_rate=16000, speech_model="universal-3-5-pro", format_turns=True, + sample_rate=16000, speech_model="universal-3-5-pro", )) await client.stream(stream_file_async("audio.wav", 16000)) @@ -1032,7 +1032,7 @@ transcriber = aai.Transcriber(config=config) transcriber.transcribe( "https://example.com/audio.mp3", # overrides the above configuration on the `Transcriber` with the following - config=aai.TranscriptionConfig(speech_models=["universal-3-pro", "universal-2"], multichannel=True, disfluencies=True) + config=aai.TranscriptionConfig(speech_models=["universal-3-5-pro", "universal-2"], multichannel=True, disfluencies=True) ) ``` diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index bffaa07..27729bb 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "0.64.21" +__version__ = "0.64.25" diff --git a/assemblyai/client.py b/assemblyai/client.py index e8c7c09..2da2bd5 100644 --- a/assemblyai/client.py +++ b/assemblyai/client.py @@ -47,10 +47,17 @@ def __init__( def _store_response(response): self._last_response = response + keepalive_expiry = self.settings.keepalive_expiry + limits = ( + httpx.Limits(keepalive_expiry=keepalive_expiry) + if keepalive_expiry is not None + else httpx.Limits() + ) self._http_client = httpx.Client( base_url=self.settings.base_url, headers=headers, timeout=self.settings.http_timeout, + limits=limits, event_hooks={"response": [_store_response]}, ) diff --git a/assemblyai/streaming/v3/_base.py b/assemblyai/streaming/v3/_base.py index e244ec1..012cdf1 100644 --- a/assemblyai/streaming/v3/_base.py +++ b/assemblyai/streaming/v3/_base.py @@ -135,6 +135,11 @@ def _emit_param_warnings(params: StreamingParameters) -> None: "`customer_support_audio_capture=True` will record session audio. " "Only enable this when explicitly coordinating with AssemblyAI support." ) + if params.language_code is not None: + logger.warning( + "[Deprecation Warning] `language_code` is deprecated and will be removed in a future release. " + "Please use `language_codes` instead." + ) def _build_uri(host: str, params: StreamingParameters) -> str: diff --git a/assemblyai/streaming/v3/models.py b/assemblyai/streaming/v3/models.py index f64dede..879e4a7 100644 --- a/assemblyai/streaming/v3/models.py +++ b/assemblyai/streaming/v3/models.py @@ -147,11 +147,18 @@ class StreamingSessionParameters(BaseModel): agent_context: Optional[str] = None interruption_delay: Optional[int] = None turn_left_pad_ms: Optional[int] = None + language_codes: Optional[List[str]] = None class Encoding(str, Enum): pcm_s16le = "pcm_s16le" pcm_mulaw = "pcm_mulaw" + # Raw Opus packets, one packet per binary WS message. `sample_rate` is + # optional and ignored — the Opus stream is self-describing. + opus = "opus" + # Ogg-encapsulated Opus byte stream (ffmpeg, gstreamer, opusenc, browser + # MediaRecorder output). `sample_rate` is optional and ignored. + ogg_opus = "ogg_opus" def __str__(self): return self.value @@ -259,6 +266,8 @@ class StreamingParameters(StreamingSessionParameters): sample_rate: int encoding: Optional[Encoding] = None speech_model: Optional[SpeechModel] = None + # Deprecated: use language_codes instead (pass a single-element list, e.g. + # ["es"], for the same behavior). Still supported for backward compatibility. language_code: Optional[str] = None language_detection: Optional[bool] = None domain: Optional[StreamingDomain] = None diff --git a/assemblyai/sync.py b/assemblyai/sync.py index c0a2d1e..2c9ed3d 100644 --- a/assemblyai/sync.py +++ b/assemblyai/sync.py @@ -2,9 +2,11 @@ import concurrent.futures import os -from typing import BinaryIO, Optional, Tuple, Union +from typing import Any, BinaryIO, Optional, Tuple, Union from urllib.parse import urlparse +import httpx + from . import client as _client from . import sync_api, types @@ -201,3 +203,52 @@ def transcribe_async( data=data, config=config, ) + + def warm(self) -> bool: + """ + Opens the connection to the sync API ahead of time. + + The sync API is a single request/response, so a `transcribe()` that + opens its connection on demand pays the full DNS + TCP + TLS handshake + on the critical path — one network round trip that, for a distant + client, can rival the transcription itself. Calling `warm()` as soon as + you know audio is coming — typically while the clip is still being + recorded — spends that setup concurrently: the next `transcribe()` + reuses the already-open connection. + + The warmed connection is reused while it stays in the HTTP pool — + `settings.keepalive_expiry` seconds (httpx's 5s default unless raised). + Call `warm()` shortly before `transcribe()`, or raise + `keepalive_expiry` (e.g. to 120, the sync audio cap) so a single call + covers a whole in-progress recording. `warm()` is idempotent and cheap, + so calling it again to refresh the connection is fine. + + Routing the same `config.model` as the eventual transcription ensures + the warmed connection lands on the right backend. + + Returns: + True once the connection is open (any HTTP response — even a + non-200 health probe — means the socket is established); False if + the connection could not be opened (transport error). + """ + settings = self._client.settings + url = settings.sync_base_url.rstrip("/") + sync_api.ENDPOINT_HEALTH + try: + self._client.http_client.get( + url, + headers={sync_api.MODEL_HEADER: self.config.model}, + timeout=min(settings.sync_http_timeout, 10.0), + ) + except httpx.HTTPError: + return False + return True + + def close(self) -> None: + """Shuts down the worker-thread pool used by `transcribe_async`.""" + self._executor.shutdown(wait=False) + + def __enter__(self) -> "SyncTranscriber": + return self + + def __exit__(self, *_exc: Any) -> None: + self.close() diff --git a/assemblyai/sync_api.py b/assemblyai/sync_api.py index 41d279e..1e978f3 100644 --- a/assemblyai/sync_api.py +++ b/assemblyai/sync_api.py @@ -6,6 +6,7 @@ from . import types ENDPOINT_TRANSCRIBE = "/transcribe" +ENDPOINT_HEALTH = "/healthz" MODEL_HEADER = "X-AAI-Model" diff --git a/assemblyai/types.py b/assemblyai/types.py index 7597481..cff8a3f 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -126,6 +126,9 @@ class Settings(BaseSettings): http_timeout: float = 30.0 "The default HTTP timeout for general requests" + keepalive_expiry: Optional[float] = None + "Seconds an idle connection is kept in the HTTP pool before being closed. None uses httpx's default (5s). Raise it (e.g. to 120) so a connection opened by `SyncTranscriber.warm()` stays reusable across an in-progress recording." + base_url: str = "https://api.assemblyai.com" "The base URL for the AssemblyAI API" diff --git a/tests/unit/test_streaming.py b/tests/unit/test_streaming.py index a1ad588..154b7e8 100644 --- a/tests/unit/test_streaming.py +++ b/tests/unit/test_streaming.py @@ -12,6 +12,7 @@ from assemblyai.streaming.v3 import ( BeginEvent, + Encoding, NoiseSuppressionModel, SpeakerRevisionEvent, SpeechModel, @@ -312,6 +313,105 @@ def mocked_websocket_connect( assert "language_code=es" in actual_url +def test_client_connect_with_language_codes(mocker: MockFixture): + # Given: client + language_codes parameter + 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_3_5_pro, + language_codes=["en", "es"], + ) + + # When: connect + client.connect(params) + + # Then: the language_codes wire param is forwarded as a JSON-encoded list + assert "language_codes=%5B%22en%22%2C+%22es%22%5D" in actual_url + + +def test_client_connect_with_language_code_logs_deprecation_warning( + mocker: MockFixture, caplog: pytest.LogCaptureFixture +): + # Given: client + the deprecated language_code parameter + def mocked_websocket_connect( + url: str, additional_headers: dict, open_timeout: float + ): + pass + + 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_3_5_pro, + language_code="es", + ) + + # When: connect + with caplog.at_level(logging.WARNING): + client.connect(params) + + # Then: a deprecation warning points callers to language_codes + assert any( + "language_code" in r.message + and "deprecated" in r.message + and "language_codes" in r.message + for r in caplog.records + ) + + +@pytest.mark.parametrize("encoding", [Encoding.opus, Encoding.ogg_opus]) +def test_client_connect_with_opus_encodings(mocker: MockFixture, encoding: Encoding): + # Given: client + one of the new Opus encodings + 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_3_5_pro, + encoding=encoding, + ) + + # When: connect + client.connect(params) + + # Then: the encoding wire param is forwarded + assert f"encoding={encoding.value}" in actual_url + + def test_noise_suppression_deprecated_alias_migrates_to_voice_focus( mocker: MockFixture, caplog: pytest.LogCaptureFixture ): diff --git a/tests/unit/test_streaming_async.py b/tests/unit/test_streaming_async.py index 81449f0..318e522 100644 --- a/tests/unit/test_streaming_async.py +++ b/tests/unit/test_streaming_async.py @@ -872,6 +872,84 @@ async def test_set_params_with_agent_context(mocker: MockFixture): await client.disconnect() +async def test_set_params_with_language_codes(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 language_codes mid-stream + await client.set_params(StreamingSessionParameters(language_codes=["en", "es"])) + + 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 language_codes 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["language_codes"] == ["en", "es"] + + await client.disconnect() + + +async def test_set_params_with_empty_language_codes_clears_steering( + 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 an empty language_codes list (the + # server-side "clear steering" signal) + await client.set_params(StreamingSessionParameters(language_codes=[])) + + 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: the empty list survives serialization (exclude_none must not drop + # it) so the server can clear steering + 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["language_codes"] == [] + + await client.disconnect() + + async def test_force_endpoint_enqueues_force_endpoint_frame(mocker: MockFixture): fake_ws = _FakeAsyncWebSocket() _patch_connect(mocker, fake_ws) diff --git a/tests/unit/test_sync.py b/tests/unit/test_sync.py index fb0efcc..78013eb 100644 --- a/tests/unit/test_sync.py +++ b/tests/unit/test_sync.py @@ -7,6 +7,7 @@ aai.settings.api_key = "test" TRANSCRIBE_URL = f"{aai.settings.sync_base_url}/transcribe" +HEALTH_URL = f"{aai.settings.sync_base_url}/healthz" _OK_RESPONSE = { "text": "hello world", @@ -356,3 +357,76 @@ def test_default_model_is_u3_sync_pro(): # Then it is the sync U3-Pro identifier assert aai.SyncTranscriptionConfig().model == "u3-sync-pro" assert aai.SyncSpeechModel.u3_sync_pro.value == "u3-sync-pro" + + +def test_warm_opens_connection_with_model_header(httpx_mock: HTTPXMock): + # Given a mocked health endpoint + httpx_mock.add_response(url=HEALTH_URL, method="GET", status_code=httpx.codes.OK) + + # When warming the transcriber + warmed = aai.SyncTranscriber().warm() + + # Then it returns True and routes the probe via X-AAI-Model + assert warmed is True + request = httpx_mock.get_requests()[0] + assert request.url == HEALTH_URL + assert request.method == "GET" + assert request.headers["X-AAI-Model"] == "u3-sync-pro" + + +def test_warm_uses_configured_model(httpx_mock: HTTPXMock): + # Given a transcriber pinned to a specific model + httpx_mock.add_response(url=HEALTH_URL, method="GET", status_code=httpx.codes.OK) + config = aai.SyncTranscriptionConfig(model="some-other-model") + + # When warming + aai.SyncTranscriber(config=config).warm() + + # Then the warm probe carries that model so it lands on the right backend + assert httpx_mock.get_requests()[0].headers["X-AAI-Model"] == "some-other-model" + + +def test_warm_returns_true_on_non_200(httpx_mock: HTTPXMock): + # Given a health route that the load balancer answers with a 404 + httpx_mock.add_response(url=HEALTH_URL, method="GET", status_code=404) + + # When warming, Then the socket is still established, so warm() is True + assert aai.SyncTranscriber().warm() is True + + +def test_warm_returns_false_on_transport_error(httpx_mock: HTTPXMock): + # Given the sync host is unreachable + httpx_mock.add_exception(httpx.ConnectError("connection refused")) + + # When warming, Then the failure is swallowed and reported as False + assert aai.SyncTranscriber().warm() is False + + +def test_context_manager_returns_self_and_closes(): + # Given a transcriber used as a context manager + with aai.SyncTranscriber() as transcriber: + # Then the bound value is the transcriber itself + assert isinstance(transcriber, aai.SyncTranscriber) + + # And leaving the block shuts the worker pool down + assert transcriber._executor._shutdown is True + + +def test_keepalive_expiry_defaults_to_httpx_default(): + # Given a default config + # When inspecting keepalive_expiry + # Then it is None, leaving httpx's own default in place + assert aai.Settings().keepalive_expiry is None + + +def test_client_accepts_custom_keepalive_expiry(): + # Given a client configured with a longer keepalive + from assemblyai import client as client_mod + + # When constructed, Then it builds cleanly (the value reaches httpx.Limits) + # and round-trips on settings + client = client_mod.Client( + settings=aai.Settings(api_key="k", keepalive_expiry=120.0) + ) + assert client.settings.keepalive_expiry == 120.0 + assert client.http_client is not None