assemblyai-streaming

AssemblyAI 流式与实时转录技能

概述

使用此技能构建和维护与 AssemblyAI 以下服务交互的代码：

• 流式语音转文本（STT），通过 WebSocket（wss://streaming.assemblyai.com/v3/ws）
• 异步／预录制 STT，通过 REST（https://api.assemblyai.com/v2/transcript）
• LLM Gateway，用于将 Claude/GPT/Gemini 风格的模型应用于转录文本（https://llm-gateway.assemblyai.com）

本技能重点关注流式／实时转录、会议记录助手和语音 Agent，同时也涵盖异步工作流和后处理。

本技能假设使用可访问 Python（首选）和 Bash 的 Claude Code 环境。

---

适用场景

在以下情况使用此技能：

• 实现来自麦克风、电话流或音频文件的实时转录。
• 构建实时会议记录助手（Zoom/Teams/Meet），尤其是需要摘要、行动项和重点内容时。
• 实现语音 Agent，且延迟和自然轮换发言至关重要。
• 从其他 STT 服务商（OpenAI/Deepgram/Google/AWS 等）迁移到 AssemblyAI。
• 通过 LLM Gateway 将大模型应用于音频，用于生成摘要、问答、主题标注或自定义提示词。

以下情况不要使用此技能：

• 任务仅涉及通用 HTTP 客户端用法，不包含 AssemblyAI 特定逻辑。
• 请求明确针对其他 STT 服务商。
• 环境无法安全存储或使用 API 密钥。

---

AssemblyAI 概念模型

1. 需要了解的产品

1. 预录制语音转文本（异步）

   • REST API：POST /v2/transcript → GET /v2/transcript/{id}
   • 专为 URL、上传文件、S3 等来源的文件设计。
   • 支持额外模型：摘要、主题检测、情感分析、PII 脱敏、章节划分等。

2. 流式语音转文本

   • WebSocket：wss://streaming.assemblyai.com/v3/ws
   • 低延迟，转录结果不可变（约 300ms）。
   • 内置轮次检测，适用于语音 Agent 和实时字幕。

3. LLM Gateway

   • REST API：https://llm-gateway.assemblyai.com 上的 POST /v1/chat/completions
   • 统一访问多个大模型（Claude、GPT、Gemini 等）。
   • 专为“基于转录文本使用大模型”的工作流设计。

2. 关键模型参数（异步）

• speech_models：例如 ["slam-1", "universal"]
  • Slam-1：英语准确率最高，支持 keyterms_prompt，适合医疗／技术对话。
  • Universal：覆盖多种语言；语言未知时是不错的默认选择。
• language_code 与 language_detection：
  • 语言已知时使用 language_code。
  • 语言未知时使用 language_detection: true；也可设置 language_confidence_threshold。
• keyterms_prompt：
  • 用于提升识别效果的领域词汇／短语（医学术语、产品名称等）。
• 其他智能功能：summarization、iab_categories、content_safety、entity_detection、auto_chapters、sentiment_analysis、speaker_labels、auto_highlights、redact_pii 等。

3. 关键模型参数（流式）

连接 URL：

• 美国：wss://streaming.assemblyai.com/v3/ws
• 欧盟：wss://streaming.eu.assemblyai.com/v3/ws

重要查询参数：

• sample_rate（必填）：例如 16000
• format_turns（布尔值）：返回格式化后的最终转录文本；对于低延迟语音 Agent，请勿启用。
• speech_model：universal-streaming-english（默认）或 universal-streaming-multi。
• `keyterms_p

rompt：以 JSON 编码的术语列表，例如 ["AssemblyAI", "Slam-1", "Keanu Reeves"]`。

• 轮次检测：
  • end_of_turn_confidence_threshold（0.0–1.0，默认约为 0.4）
  • min_end_of_turn_silence_when_confident（毫秒，默认约为 400）
  • max_turn_silence（毫秒，默认约为 1280）

请求头：

• 使用 Authorization: <API_KEY>，或使用由后端签发的短期 token 查询参数。

消息：

• 客户端发送：
  • 二进制音频块（每块 50–1000 毫秒）。
  • 可选的 JSON 消息：{"type": "UpdateConfig", ...}、{"type": "Terminate"}、{"type": "ForceEndpoint"}。
• 服务端发送：
  • 包含 id、expires_at 的 Begin 事件。
  • Turn 事件，包含：
    • transcript（不可变的部分或最终转录文本），
    • utterance（完整的语义片段），
    • end_of_turn（布尔值），
    • turn_is_formatted（布尔值），
    • 包含时间戳和置信度的 words 数组。
  • 包含摘要统计信息的 Termination 事件。

4. 区域与数据驻留

• 异步：
  • 美国：https://api.assemblyai.com
  • 欧盟：https://api.eu.assemblyai.com
• 流式：
  • 美国：wss://streaming.assemblyai.com/v3/ws
  • 欧盟：wss://streaming.eu.assemblyai.com/v3/ws

同一项目始终使用一致的基础 URL；不要对同一数据混用美国和欧盟端点。

---

安全与 API 密钥

• 始终要求提供 AssemblyAI API 密钥，并确保在 Claude Code 的输出中不将密钥写入源代码：
  • 使用环境变量：ASSEMBLYAI_API_KEY。
  • 或在代码片段中使用占位符（"<YOUR_API_KEY>"）。
• 对于浏览器端/客户端代码：
  • 不要嵌入 API 密钥。
  • 指示用户在后端生成临时流式令牌，并且仅将令牌传入 WebSocket 连接。
• 切勿在日志或注释中打印真实密钥。

---

高级工作流模式

决策树

1. 音频是实时的吗？

   • 是 → 使用流式 STT。
   • 否 → 使用异步 STT。

2. 响应延迟是否至关重要（<1s）？

   • 是 → 使用流式处理，并设置 format_turns=false，同时仔细配置轮次检测。
   • 否 → 使用异步处理，然后进行摘要、章节划分等。

3. 转录文本是否会离开后端？

   • 是 → 分享前考虑使用 redact_pii（以及可选的 redact_pii_audio）。
   • 否 → 按需使用原始转录文本。

4. 是否需要基于 LLM 的处理（问答、结构化摘要）？

   • 是 → 通过 chat/completions 将转录文本传入LLM Gateway。

---

Claude 应如何使用此技能

一般原则

• 有官方 AssemblyAI SDK（Python/JS）时优先使用；只有在无法安装 SDK 时，才改用 requests/websocket-client。
• 始终：
  • 验证 HTTP 响应和 WebSocket 状态。
  • 显示有用的错误信息（转录 JSON 中的 status、error 字段）。
  • 遵守文档规定的音频块大小限制（每条二进制消息包含 50–1000 毫秒音频）。
• 对于语音 Agent 代码，应优化：
  • 不可变的部分转录文本（transcript）和 utterance 字段。
  • 尽可能降低延迟，避免额外的格式化处理。

---

配方 1 – 使用麦克风进行最简流式处理（Python SDK）

目标：将麦克风音频流式传输到 AssemblyAI，并实时打印转录文本。

当环境中已安装 Python、assemblyai 和 pyaudio，且用户希望快速体验流式处理时使用。

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()