assemblyai-streaming（装配 AI 流）

AssemblyAI Streaming & Live Transcription Skill

Overview

Use this skill to build and maintain code that talks to AssemblyAI’s:

• Streaming Speech-to-Text (STT) via WebSockets (wss://streaming.assemblyai.com/v3/ws)
• Async / pre-recorded STT via REST (https://api.assemblyai.com/v2/transcript)
• LLM Gateway for applying Claude/GPT/Gemini-style models to transcripts (https://llm-gateway.assemblyai.com)

The emphasis is on streaming/live transcription, meeting notetakers, and voice agents, while still covering async workflows and post-processing.

This skill assumes a Claude Code environment with access to Python (preferred) and Bash.

When to Use

Use this skill when:

• Implementing real-time transcription from a microphone, telephony stream, or audio file.
• Building a live meeting notetaker (Zoom/Teams/Meet), especially with summaries, action items, and highlights.
• Implementing a voice agent where latency and natural turn-taking matter.
• Migrating from other STT providers (OpenAI/Deepgram/Google/AWS/etc.) to AssemblyAI.
• Applying LLMs to audio via LLM Gateway for summaries, Q&A, topic tagging, or custom prompts.

Do not use this skill when:

• The task is generic HTTP client usage with no AssemblyAI-specific logic.
• The request clearly targets a different STT vendor.
• The environment cannot safely store or use an API key.

AssemblyAI Mental Model

1. Products to care about

1. Pre-recorded Speech-to-Text (Async)

- REST API: POST /v2/transcript → GET /v2/transcript/{id} - Designed for files from URLs, uploads, S3, etc. - Supports extra models: summarization, topic detection, sentiment, PII redaction, chapters, etc.

1. Streaming Speech-to-Text

- WebSocket: wss://streaming.assemblyai.com/v3/ws - Low-latency, immutable transcripts (~300ms). - Turn detection built in; fits voice agents and live captioning.

1. LLM Gateway

- REST API: POST /v1/chat/completions at https://llm-gateway.assemblyai.com - Unified access to multiple LLMs (Claude, GPT, Gemini, etc.). - Designed for “LLM over transcripts” workflows.

2. Key model knobs (Async)

• speech_models: ["slam-1", "universal"] etc.

- Slam-1: best English accuracy + keyterms_prompt, good for medical/technical conversations. - Universal: multilingual coverage; good default if language is unknown.

• language_code vs language_detection:

- Use language_code when the language is known. - Use language_detection: true when unknown; optionally set language_confidence_threshold.

• keyterms_prompt:

- Domain words/phrases to boost (med terms, product names, etc.).

• Extra intelligence: summarization, iab_categories, content_safety, entity_detection, auto_chapters, sentiment_analysis, speaker_labels, auto_highlights, redact_pii, etc.

3. Key model knobs (Streaming)

Connection URL:

• US: wss://streaming.assemblyai.com/v3/ws
• EU: wss://streaming.eu.assemblyai.com/v3/ws

Important query parameters:

• sample_rate (required): e.g. 16000
• format_turns (bool): return formatted final transcripts; avoid for low-latency voice agents.
• speech_model: universal-streaming-english (default) or universal-streaming-multi.
• `keyterms_p

rompt: JSON-encoded list of terms, e.g.["AssemblyAI", "Slam-1", "Keanu Reeves"]`.

• Turn detection:

- end_of_turn_confidence_threshold (0.0–1.0, default ~0.4) - min_end_of_turn_silence_when_confident (ms, default ~400) - max_turn_silence (ms, default ~1280)

Headers:

• Use either Authorization: <API_KEY> or a short-lived token query parameter issued by your backend.

Messages:

• Client sends:

- Binary audio chunks (50–1000ms each). - Optional JSON messages: {"type": "UpdateConfig",...}, {"type": "Terminate"}, {"type": "ForceEndpoint"}.

• Server sends:

- Begin event with id, expires_at. - Turn events with: - transcript (immutable partials/finals), - utterance (complete semantic chunk), - end_of_turn (bool), - turn_is_formatted (bool), - words array with timestamps/confidences. - Termination event with summary stats.

4. Regions and data residency

• Async:

- US: https://api.assemblyai.com - EU: https://api.eu.assemblyai.com

• Streaming:

- US: wss://streaming.assemblyai.com/v3/ws - EU: wss://streaming.eu.assemblyai.com/v3/ws

Always keep base URLs consistent per project; don’t mix US/EU endpoints for the same data.

Security & API Keys

• Always require an AssemblyAI API key and keep it out of source in Claude Code output:

- Use environment variables: ASSEMBLYAI_API_KEY. - Or placeholders ("<YOUR_API_KEY>") in snippets.

• For browser/client code:

- Do not embed the API key. - Instruct the user to generate temporary streaming tokens on their backend and pass only the token into the WebSocket connection.

• Never print real keys in logs or comments.

High-Level Workflow Patterns

Decision tree

1. Is the audio live?

- Yes → Use Streaming STT. - No → Use Async STT.

1. Is latency critical (<1s) for responses?

- Yes → Streaming with format_turns=false and careful turn detection. - No → Async, then Summarization/Chapters/etc.

1. Do transcripts leave the backend?

- Yes → Consider redact_pii (and optionally redact_pii_audio) before sharing. - No → Use raw transcripts as needed.

1. Need LLM-based processing (Q&A, structured summaries)?

- Yes → Pipe transcripts into LLM Gateway via chat/completions.

How Claude Should Work with This Skill

General principles

• Prefer official AssemblyAI SDKs (Python/JS) when available; fall back to requests/websocket-client only if SDK cannot be installed.
• Always:

- Validate HTTP responses and WebSocket status. - Surface useful error messages (status, error fields in transcript JSON). - Respect documented min/max chunk sizes (50–1000ms of audio per binary message).

• For voice-agent code, optimize for:

- Immutable partials (transcript) and utterance field. - Minimal latency, avoid extra formatting passes.

Recipe 1 – Minimal Streaming from Microphone (Python SDK)

Goal: Stream mic audio to AssemblyAI and print transcripts in real time.

Use this when the environment has Python and assemblyai + pyaudio installed, and the user wants a quick streaming demo.

import assemblyai as aai
from assemblyai.streaming import v3 as aai_stream
import pyaudio

API_KEY = "<YOUR_API_KEY>"

aai.settings.api_key = API_KEY

SAMPLE_RATE = 16000
CHUNK_MS = 50
FRAMES_PER_BUFFER = int(SAMPLE_RATE * (CHUNK_MS / 1000.0))

def main():
    client = aai_stream.StreamingClient(
        aai_stream.StreamingClientOptions(
            api_key=API_KEY,
            api_host="streaming.assemblyai.com",  # or "streaming.eu.assemblyai.com"
        )
    )

    def on_begin(_client, event: aai_stream.BeginEvent):
        print(f"Session started: {event.id}, expires at {event.expires_at}")

    def on_turn(_client, event: aai_stream.TurnEvent):
        # Use immutable transcript text
        text = (event.transcript or "").strip()
        if not text:
            return
        # Use formatted finals only for display; keep unformatted for LLMs
        if event.turn_is_formatted:
            print(f"[FINAL] {text}")
        else:
            print(f"[PARTIAL] {text}", end="\r")

    def on_terminated(_client, event: aai_stream.TerminationEvent):
        print(f"\nTerminated. Audio duration={event.audio_duration_seconds}s")

    def on_error(_client, error: aai_stream.StreamingError):
        print(f"\nStreaming error: {error}")

    client.on(aai_stream.StreamingEvents.Begin, on_begin)
    client.on(aai_stream.StreamingEvents.Turn, on_turn)
    client.on(aai_stream.StreamingEvents.Termination, on_terminated)
    client.on(aai_stream.StreamingEvents.Error, on_error)

    client.connect(
        aai_stream.StreamingParameters(
            sample_rate=SAMPLE_RATE,
            format_turns=False,  # better latency for voice agents
        )
    )

    pa = pyaudio.PyAudio()
    stream = pa.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=SAMPLE_RATE,
        input=True,
        frames_per_buffer=FRAMES_PER_BUFFER,
    )

    try:
        print("Speak into your microphone (Ctrl+C to stop)...")
        def audio_gen():
            while True:
                yield stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
        client.stream(audio_gen())
    except KeyboardInterrupt:
        pass
    finally:
        client.disconnect(terminate=True)
        stream.stop_stream()
        stream.close()
        pa.terminate()

if __name__ == "__main__":
    main()