AssemblyAI Python SDK

Use ChannelStreamer to handle multiple live audio sources (e.g., microphone and system audio) in a single streaming session.

Key Concepts:

• Mixing: The SDK mixes multiple PCM channels into a single mono stream over one WebSocket. It keeps channels aligned by consuming the shortest buffer; if a source ends, you must call mixer.close_channel(name) to pad it with silence and prevent stalling.
• Attribution: ChannelStreamer delivers an enriched DualChannelTurnEvent. This allows you to identify the physical source of words via turn.channel and word.channel.
• Independence: Channel attribution (word.channel) is independent of speaker diarization (word.speaker).
• Async Support: AsyncChannelStreamer provides an asyncio-native interface with await mixer.stream(...) and await mixer.close_channel(...).

Workflow:

1. Initialize RealTimeTranscriber.
2. Create a ChannelStreamer instance, declaring named channels and the session sample rate.
3. Register handlers on the mixer (e.g., mixer.on(RealTimeEvents.Turn, handler)).
4. Connect the client.
5. Feed PCM data to each channel using mixer.stream(channel_name, pcm_data).
6. Call mixer.flush() to push trailing audio.

from assemblyai.streaming.v3 import (
    ChannelStreamer, RealTimeTranscriber, RealTimeTranscriberOptions,
    RealTimeEvents, RealTimeParameters,
)

def on_turn(client, event):   # event is a DualChannelTurnEvent
    print(f"[{event.channel}] {event.transcript}")
    for w in event.words:
        print(f"  {w.text!r} -> channel={w.channel} speaker={w.speaker}")

client = RealTimeTranscriber(RealTimeTranscriberOptions(api_key="<YOUR_API_KEY>"))

# Declare the channels and the session sample rate (must be pcm_s16le).
mixer = ChannelStreamer(client, channels=["mic", "system"], sample_rate=16000)
# Register handlers on the mixer: Turn handlers receive the enriched event, 
# other events (Begin/Error/…) are forwarded to the client.
mixer.on(RealTimeEvents.Turn, on_turn)
client.connect(RealTimeParameters(
    sample_rate=16000, speech_model="universal-3-5-pro", speaker_labels=True,
))

# Feed each source separately — e.g. from two capture callbacks. Send
# continuous PCM for every channel (silence as zeros), at the same rate.
mixer.stream("mic", mic_pcm)
mixer.stream("system", system_pcm)

mixer.flush()                    # push trailing buffered audio
client.disconnect(terminate=True)

aai.DictationTranscriber is designed for short dictations (up to 120s). It transcribes audio as it is spoken and can optionally use an LLM to clean up or reformat the text. It uses a single live-upload connection and returns a DictationResponse.

DictationResponse fields:

• text: The raw transcript.
• llm_response: The rewritten text (if an LLM pass was performed).
• final_text: The text to show the user (the rewrite if available, otherwise the raw transcript).

Constraints:

• Audio must be WAV or raw 16-bit PCM.
• Maximum 120 seconds per request.
• URLs are not supported; use aai.Transcriber for URL ingestion.

import assemblyai as aai
import sounddevice as sd

aai.settings.api_key = "<YOUR_API_KEY>"

RATE = 16000
config = aai.DictationConfig(sample_rate=RATE, channels=1)

with aai.DictationTranscriber() as transcriber:
    with transcriber.open_live(config) as session:
        microphone = sd.RawInputStream(
            samplerate=RATE, channels=1, dtype="int16",
            callback=lambda data, *_: session.write(bytes(data)),
        )
        with microphone:
            input("Dictating, press Enter to stop... ")
    print(session.result().final_text)

The SDK provides several transcriber classes depending on your use case (audio length, latency requirements, and features needed). Versioned paths (e.g., assemblyai.prerecorded.v2.Transcriber) are preferred for new code, though top-level shortcuts like aai.Transcriber are available.

In 1.0, it is preferred to import transcribers from their specific versioned subpackages to explicitly pin the product and API version.

Mapping:

• assemblyai.prerecorded.v2: Provides Transcriber, AsyncTranscriber, Transcript, AsyncTranscript, TranscriptGroup, and TranscriptionConfig.
• assemblyai.sync.v1: Provides SyncTranscriber, AsyncSyncTranscriber, SyncTranscriptionConfig, SyncTranscriptError, and friends.
• assemblyai.streaming.v3: Provides RealTime* classes and event types.

Note: Cross-cutting names like TranscriptStatus, TranscriptError, Settings, Client, AsyncClient, and the global settings should still be imported from the top-level assemblyai package.

from assemblyai.prerecorded.v2 import Transcriber
from assemblyai.sync.v1 import SyncTranscriber

transcriber = Transcriber(api_key="YOUR_API_KEY")
sync_transcriber = SyncTranscriber(api_key="YOUR_API_KEY")