name: assemblyai-streaming description: 该技能应在使用AssemblyAI的语音转文本和LLM Gateway API时使用，特别适用于流式/实时转录、会议笔记记录和需要低延迟转录与音频分析的语音代理。 license: MIT allowed-tools:

Read
Write
Edit
Grep
Glob
Bash
Python metadata: skill-version: “1.0.0” upstream-docs: “https://www.assemblyai.com/docs” focus: “streaming-stt, meeting-notetaker, voice-agent, llm-gateway”

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)

重点在于流式/实时转录、会议笔记记录和语音代理，同时涵盖异步工作流和后处理。

此技能假设Claude Code环境可访问Python（首选）和Bash。

使用场景

在以下情况使用此技能：

从麦克风、电话流或音频文件实施实时转录。
构建实时会议笔记记录器（Zoom/Teams/Meet），特别是带摘要、行动项和亮点。
实施语音代理，其中延迟和自然轮流对话很重要。
从其他STT提供商（OpenAI/Deepgram/Google/AWS等）迁移到AssemblyAI。
通过LLM Gateway将LLMs应用于音频进行摘要、问答、主题标记或自定义提示。

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

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

AssemblyAI 心智模型

1. 需关注的产品

预录制语音转文本 (异步)
- REST API: POST /v2/transcript → GET /v2/transcript/{id}
- 设计用于来自URL、上传、S3等的文件。
- 支持额外模型：摘要、主题检测、情感、PII编辑、章节等。
流式语音转文本
- WebSocket: wss://streaming.assemblyai.com/v3/ws
- 低延迟、不可变转录（约300ms）。
- 内置轮流检测；适合语音代理和实时字幕。
LLM Gateway
- REST API: POST /v1/chat/completions 在 https://llm-gateway.assemblyai.com
- 统一访问多个LLMs（Claude、GPT、Gemini等）。
- 设计用于“LLM over transcripts”工作流。

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 (布尔): 返回格式化最终转录；避免用于低延迟语音代理。
speech_model: universal-streaming-english (默认) 或 universal-streaming-multi。
keyterms_prompt: 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"}。
服务器发送:
- Begin 事件带 id, expires_at。
- 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连接。
切勿在日志或注释中打印真实密钥。

高级工作流模式

决策树

音频是实时的吗？
- 是 → 使用流式 STT。
- 否 → 使用异步 STT。
延迟对于响应是否关键 (<1秒)？
- 是 → 流式带 format_turns=false 和谨慎的轮流检测。
- 否 → 异步，然后摘要/章节等。
转录是否离开后端？
- 是 → 考虑在分享前使用 redact_pii (和可选 redact_pii_audio)。
- 否 → 根据需要使用原始转录。
需要基于LLM的处理（问答、结构化摘要）？
- 是 → 通过 chat/completions 将转录传输到LLM Gateway。

Claude 应如何使用此技能

一般原则

优先使用官方AssemblyAI SDKs (Python/JS)；仅在无法安装SDK时回退到 requests/websocket-client。
始终:
- 验证HTTP响应和WebSocket状态。
- 展示有用的错误消息（转录JSON中的 status, error 字段）。
- 遵守文档中的最小/最大块大小（每个二进制消息50–1000毫秒音频）。
对于语音代理代码，优化:
- 不可变部分 (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",  # 或 "streaming.eu.assemblyai.com"
        )
    )

    def on_begin(_client, event: aai_stream.BeginEvent):
        print(f"会话开始: {event.id}, 到期于 {event.expires_at}")

    def on_turn(_client, event: aai_stream.TurnEvent):
        # 使用不可变转录文本
        text = (event.transcript or "").strip()
        if not text:
            return
        # 仅用于显示时使用格式化最终；为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"
终止。音频时长={event.audio_duration_seconds}秒")

    def on_error(_client, error: aai_stream.StreamingError):
        print(f"
流式错误: {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,  # 语音代理的更好延迟
        )
    )

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

    try:
        print("请对着麦克风说话（Ctrl+C停止）...")
        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()