Skip to content

Commit 862d166

Browse files
ccampbell-aaiclaude
andcommitted
feat(sync): push-style live sessions via open_live()
transcribe_live() takes an iterator, but most live audio arrives in a callback: sounddevice, PyAudio, WebRTC tracks, telephony media streams, websocket frame handlers. Every such caller had to build the same bridge (a queue, a stop flag, a drain generator and, on the sync side, a thread to run the blocking call in) before getting a transcript. SyncTranscriber.open_live() returns a LiveSession and AsyncSyncTranscriber.open_live() an AsyncLiveSession. The request starts the moment the session opens, on the transcriber's worker pool or as a loop task; write() queues audio without blocking (thread-safe on the sync side), close() ends it, result() waits for the transcript, and abort() drops the request. Context-manager exit closes on success and aborts if the block raised. Chunk types are validated at write() so a str shows up immediately rather than in result(). Both sessions are thin layers over transcribe_live(): no new transport code, same errors, same 120 s cap and silence-abort semantics. The iterator form stays for sources that already are iterators or file objects. README leads the live examples with the session now (sounddevice, and a websocket handler for asyncio), and keeps transcribe_live() for iterator and file-object sources. 19 tests. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 689c51f commit 862d166

7 files changed

Lines changed: 820 additions & 60 deletions

File tree

‎CLAUDE.md‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,8 @@ aai.settings.api_key = "your-key"
4444
- `assemblyai.prerecorded.v2` — Canonical module for both transcribers, matching the `/v2/transcript` API. The top-level `aai.*` names re-export it
4545
- `aai.TranscriptionConfig` — All transcription options: `speech_models`, `speaker_labels`, `sentiment_analysis`, `entity_detection`, `auto_chapters`, `content_safety`, `language_detection`, `summarization`, `word_boost`, `disfluencies`
4646
- `aai.Transcript` — Result object with `.text`, `.status`, `.utterances`, `.words`, `.chapters`, `.entities`, `.sentiment_analysis`. Methods: `get_sentences()`, `get_paragraphs()`, `export_subtitles_srt()`, `export_subtitles_vtt()`
47-
- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_live()`, `transcribe_async()`, `warm()`
48-
- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one
47+
- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_live()`, `open_live()`, `transcribe_async()`, `warm()`
48+
- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines; `open_live()` returns an `AsyncLiveSession`. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one
4949
- `aai.SyncTranscriptionConfig` — Sync options: `model` (default `universal-3-5-pro`), `prompt`, `keyterms_prompt`, `conversation_context`, `language_codes`, `timestamps`, `sample_rate`, `channels`
5050
- `aai.SyncTranscriptResponse` — Sync result: `.text`, `.words` (`SyncWord` with `confidence` always, `start`/`end` only when `timestamps=True`), `.confidence`, `.audio_duration_ms`, `.session_id`, `.request_time_ms`
5151
- `assemblyai.streaming.v3.RealTimeTranscriber` — Real-time streaming with event-based API (threaded)
@@ -213,7 +213,7 @@ result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config)
213213
not asyncio) for fanning out a handful of files. In asyncio code use
214214
`aai.AsyncSyncTranscriber` instead (see "Asyncio sync transcription" below).
215215

216-
**Live upload** (`transcribe_live()`): uploads audio as it is produced instead of
216+
**Live upload** (`open_live()` / `transcribe_live()`): uploads audio as it is produced instead of
217217
waiting for the whole clip. The request starts immediately, so authorization, the upload
218218
and every speech segment but the last resolve while the caller is still recording; what
219219
is left to wait for once they stop is the final segment. Takes an iterable of chunks or a
@@ -225,6 +225,17 @@ def mic_chunks():
225225

226226
result = aai.SyncTranscriber().transcribe_live(mic_chunks())
227227
```
228+
For callback-driven sources (most microphone libraries, WebRTC, telephony) use the
229+
push-style session instead; `write()` is thread-safe and never blocks:
230+
```python
231+
with transcriber.open_live(config) as session:
232+
start_capture(on_audio=session.write) # e.g. sounddevice callback
233+
wait_until_done()
234+
result = session.result() # raises SyncTranscriptError on failure
235+
```
236+
Leaving the block calls `close()`; an exception in the block calls `abort()` instead. The
237+
asyncio twin is `AsyncSyncTranscriber.open_live()` → `AsyncLiveSession` (`write()`/`close()`
238+
plain functions, `result()`/`abort()` coroutines; call `write()` on the loop thread).
228239
It pays off only when the audio is genuinely still being produced. Streaming a file
229240
already on disk is *slower* than `transcribe()`, which sends it in one piece — the saving
230241
comes from overlapping the recording, not from chunking. It also needs enough audio to

‎README.md‎

Lines changed: 60 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ aai.settings.api_key = f"{ASSEMBLYAI_API_KEY}"
102102
| --- | --- |
103103
| `assemblyai.prerecorded.v2.Transcriber` | Long-form audio, URLs, and the audio-intelligence features (speaker labels, chapters, sentiment, …), over the polled job API |
104104
| `assemblyai.prerecorded.v2.AsyncTranscriber` | The same, from asyncio code |
105-
| `assemblyai.sync.v1.SyncTranscriber` | Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency — from a file you already have (`transcribe()`) or uploaded while it is still being recorded (`transcribe_live()`) |
105+
| `assemblyai.sync.v1.SyncTranscriber` | Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency — from a file you already have (`transcribe()`) or uploaded while it is still being recorded (`open_live()`, `transcribe_live()`) |
106106
| `assemblyai.sync.v1.AsyncSyncTranscriber` | The same, from asyncio code |
107107
| `assemblyai.streaming.v3.RealTimeTranscriber` | Live audio (microphone, telephony, voice agents), transcribed as it arrives over a websocket session |
108108
| `assemblyai.streaming.v3.AsyncRealTimeTranscriber` | The same, from asyncio code |
@@ -519,107 +519,116 @@ with aai.SyncTranscriber() as transcriber:
519519
</details>
520520

521521
<details>
522-
<summary>Transcribe live audio as it is recorded (`transcribe_live()`)</summary>
522+
<summary>Transcribe live audio as it is recorded (`open_live()`)</summary>
523523

524-
`transcribe()` needs the whole clip before it can send anything. `transcribe_live()` starts the request immediately and uploads audio as your code produces it, so the upload and all but the last speech segment are done by the time the speaker stops. What remains is one final segment: on a paced 7 s clip that cut the wait after the last byte roughly in half.
524+
`transcribe()` needs the whole clip before it can send anything. A live session starts the request immediately and uploads audio as you produce it, so the upload and all but the last speech segment are done by the time the speaker stops. What remains is one final segment: on a paced 7 s clip that cut the wait after the last byte roughly in half.
525525

526-
It takes an iterable of `bytes` chunks or a binary file object read as it fills, and returns the same `SyncTranscriptResponse` as `transcribe()`. Pass raw S16LE PCM with `sample_rate` and `channels` set; that is what microphones and telephony hand you, and it needs no container header.
526+
Most audio sources hand you chunks in a callback. `open_live()` is built for that: `write()` from the callback, `close()` when the speaker stops, and `result()` for the transcript, which is the same `SyncTranscriptResponse` that `transcribe()` returns. The request runs on one of the transcriber's worker threads, so `write()` never blocks and is safe from any thread.
527527

528528
```python
529-
import queue
530-
import threading
531-
532529
import assemblyai as aai
533530
import sounddevice as sd # pip install sounddevice
534531

535532
aai.settings.api_key = "<YOUR_API_KEY>"
536533

537534
RATE = 16000
538-
config = aai.SyncTranscriptionConfig(sample_rate=RATE, channels=1)
535+
config = aai.SyncTranscriptionConfig(sample_rate=RATE, channels=1) # raw PCM needs both
539536

537+
with aai.SyncTranscriber() as transcriber:
538+
with transcriber.open_live(config) as session:
539+
microphone = sd.RawInputStream(
540+
samplerate=RATE, channels=1, dtype="int16",
541+
callback=lambda data, *_: session.write(bytes(data)),
542+
)
543+
with microphone:
544+
input("Recording, press Enter to stop... ")
545+
# leaving the block ends the audio; the final segment is all that is left
546+
print(session.result().text)
547+
```
540548

541-
def microphone(stop: threading.Event):
542-
"""Yields PCM chunks from the default microphone until `stop` is set."""
543-
chunks: queue.Queue[bytes] = queue.Queue()
544-
with sd.RawInputStream(
545-
samplerate=RATE, channels=1, dtype="int16",
546-
callback=lambda data, *_: chunks.put(bytes(data)),
547-
):
548-
while not stop.is_set():
549-
try:
550-
yield chunks.get(timeout=0.1)
551-
except queue.Empty:
552-
pass
549+
Leaving the `with` block calls `close()`. If the block raises, the session is aborted instead and `result()` raises. Call `abort()` yourself to drop a recording you no longer want transcribed.
553550

551+
This is **not** the streaming API. A live session returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking.
554552

555-
stop = threading.Event()
556-
threading.Thread(target=lambda: (input("Recording, press Enter to stop... "), stop.set()), daemon=True).start()
553+
Use it only when the audio is genuinely still being produced. Streaming a file that is already on disk is slower than `transcribe()`, which sends it in one piece. Short clips gain less: below about a minute the saving is the elided upload only. The same 120 s cap applies, and the server aborts an upload that goes silent for long, so keep writing until you are done rather than pausing.
557554

558-
result = aai.SyncTranscriber().transcribe_live(microphone(stop), config=config)
559-
print(result.text)
560-
```
555+
Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload; `result()` raises them as `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in, so `warm()` before you start recording if you want to fail fast.
561556

562-
Any producer works, not only a microphone: a file object on a pipe or socket, chunks arriving over a websocket, a telephony media stream. End the iterator to finish. The server aborts an upload that goes silent for long, so keep sending until you are done rather than pausing.
557+
</details>
563558

564-
This is **not** the streaming API. `transcribe_live()` returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking.
559+
<details>
560+
<summary>Transcribe live audio from an iterator or file object (`transcribe_live()`)</summary>
565561

566-
Use it only when the audio is genuinely still being produced. Streaming a file that is already on disk is slower than `transcribe()`, which sends it in one piece. Short clips gain less: below about a minute the saving is the elided upload only. The same 120 s cap applies.
562+
When the audio already comes as something you can iterate, or as a file object that fills over time, hand it to `transcribe_live()` directly. It takes an iterable of `bytes` chunks or a binary file object, blocks until the audio ends, and returns the transcript. `open_live()` is this method with a queue in front of it.
567563

568564
```python
569565
import subprocess
566+
import assemblyai as aai
570567

571-
# Audio you already hold whole: use transcribe(), which is faster for it.
572-
result = aai.SyncTranscriber().transcribe("./call.wav")
568+
aai.settings.api_key = "<YOUR_API_KEY>"
569+
config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)
573570

574571
# A file object that fills over time, e.g. the stdout of a recorder (sox):
575-
proc = subprocess.Popen(["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"], stdout=subprocess.PIPE)
576-
result = aai.SyncTranscriber().transcribe_live(proc.stdout, config=config)
572+
recorder = subprocess.Popen(
573+
["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"],
574+
stdout=subprocess.PIPE,
575+
)
576+
result = aai.SyncTranscriber().transcribe_live(recorder.stdout, config=config)
577+
578+
# Or any generator of chunks:
579+
def from_call(call):
580+
while call.active:
581+
yield call.read_frame()
582+
583+
result = aai.SyncTranscriber().transcribe_live(from_call(call), config=config)
584+
585+
# Audio you already hold whole belongs in transcribe(), which is faster for it:
586+
result = aai.SyncTranscriber().transcribe("./call.wav")
577587
```
578588

579-
Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload as a `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in, so `warm()` before you start recording if you want to fail fast.
589+
End the iterator to finish. Anything the producer raises propagates unchanged and drops the upload.
580590

581591
</details>
582592

583593
<details>
584594
<summary>Transcribe live audio from asyncio</summary>
585595

586-
`AsyncSyncTranscriber.transcribe_live()` takes an async iterable of chunks as well as a file object or plain iterable. A file object is read in a worker thread, so a blocking `read()` is fine; a plain iterable is consumed inline and must not block the loop.
596+
`AsyncSyncTranscriber.open_live()` returns an `AsyncLiveSession`: `write()` and `close()` are plain functions so a callback can call them, and `result()` and `abort()` are coroutines. The request runs as a task on the event loop. `write()` must be called on the loop's thread; from an audio library's capture thread, schedule it with `loop.call_soon_threadsafe(session.write, chunk)`.
587597

588598
```python
589599
import asyncio
590600
import assemblyai as aai
591601

592602
aai.settings.api_key = "<YOUR_API_KEY>"
593-
594603
config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)
595604

596605

597-
async def microphone(frames: asyncio.Queue):
598-
"""Yields PCM chunks pushed onto `frames` by a capture callback; None ends it."""
599-
while (chunk := await frames.get()) is not None:
600-
yield chunk
601-
606+
async def handle(websocket):
607+
"""Transcribes the PCM frames a browser sends over one websocket."""
608+
async with aai.AsyncSyncTranscriber() as transcriber:
609+
async with transcriber.open_live(config) as session:
610+
async for frame in websocket:
611+
session.write(frame)
612+
result = await session.result()
613+
await websocket.send(result.text)
614+
```
602615

603-
async def main():
604-
frames: asyncio.Queue = asyncio.Queue()
605-
start_capture(lambda pcm: frames.put_nowait(pcm)) # your audio callback
616+
`AsyncSyncTranscriber.transcribe_live()` is the pull-style twin: it takes an async iterable of chunks as well as a file object or plain iterable. A file object is read in a worker thread, so a blocking `read()` is fine; a plain iterable is consumed inline and must not block the loop.
606617

607-
async with aai.AsyncSyncTranscriber() as transcriber:
608-
task = asyncio.create_task(transcriber.transcribe_live(microphone(frames), config=config))
609-
await asyncio.sleep(5) # ... or until the user stops speaking
610-
frames.put_nowait(None) # end the iterator; the upload finishes
611-
result = await task
612-
print(result.text)
618+
```python
619+
async def frames(websocket):
620+
async for frame in websocket:
621+
yield frame
613622

614-
asyncio.run(main())
623+
result = await transcriber.transcribe_live(frames(websocket), config=config)
615624
```
616625

617626
</details>
618627

619628
<details>
620629
<summary>Use it from asyncio (`AsyncSyncTranscriber`)</summary>
621630

622-
`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()`, `transcribe_live()` and `warm()` as coroutines. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded `transcribe()` would block the event loop and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable.
631+
`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()`, `transcribe_live()` and `warm()` as coroutines and `open_live()` returning an `AsyncLiveSession`. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded `transcribe()` would block the event loop and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable.
623632

624633
```python
625634
import asyncio

‎assemblyai/sync/v1/__init__.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,16 @@
77
)
88
from ._base import AudioInput
99
from ._multipart import AsyncAudioChunks, AudioChunks
10-
from .async_client import AsyncSyncTranscriber
11-
from .client import SyncTranscriber
10+
from .async_client import AsyncLiveSession, AsyncSyncTranscriber
11+
from .client import LiveSession, SyncTranscriber
1212

1313
__all__ = [
1414
"AsyncAudioChunks",
15+
"AsyncLiveSession",
1516
"AsyncSyncTranscriber",
1617
"AudioChunks",
1718
"AudioInput",
19+
"LiveSession",
1820
"SyncSpeechModel",
1921
"SyncTranscriber",
2022
"SyncTranscriptError",

‎assemblyai/sync/v1/_base.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,15 @@ def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]:
122122
return data or None
123123

124124

125+
class _Aborted(Exception):
126+
"""
127+
Raised inside a live session's producer when the session is aborted.
128+
129+
Propagates out of the transport, which drops the connection, and is
130+
swallowed by `abort()`; it never reaches the caller.
131+
"""
132+
133+
125134
def check_chunks(data: object, *, allow_async: bool = False) -> None:
126135
"""
127136
Raises unless `data` can be streamed.

0 commit comments

Comments
 (0)