⚡

AssemblyAI流式实时转录技能Skill assemblyai-streaming

此技能用于通过 AssemblyAI 的 API 实现语音转文本功能，特别关注流式/实时转录、会议笔记记录和语音代理应用。支持异步和流式处理，集成 LLM Gateway 进行音频分析和人工智能模型应用。关键词：AssemblyAI, 语音转文本, 实时转录, 会议笔记, 语音代理, LLM Gateway, 流式处理, 人工智能, NLP

NLP 0 次安装 0 次浏览 更新于 3/9/2026

名称: assemblyai-streaming 描述: 此技能应用于使用 AssemblyAI 的语音转文本和 LLM Gateway APIs，特别适用于流式/实时转录、会议笔记记录和语音代理，这些应用需要低延迟转录和音频分析。 许可证: MIT 允许的工具:

  • 读取
  • 写入
  • 编辑
  • Grep
  • Glob
  • Bash
  • Python 元数据: 技能版本: “1.0.0” 上游文档: “https://www.assemblyai.com/docs” 重点: “流式-stt, 会议笔记记录, 语音代理, 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. 需要关注的产品

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

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

  2. 流式语音转文本

    • WebSocket: wss://streaming.assemblyai.com/v3/ws
    • 低延迟、不可变转录本（约300毫秒）。
    • 内置回合检测；适合语音代理和实时字幕。

  3. LLM Gateway

    • REST API: POST /v1/chat/completions 在 https://llm-gateway.assemblyai.com
    • 统一访问多个 LLMs（Claude、GPT、Gemini 等）。
    • 设计用于“转录本上的 LLM”工作流。

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 连接。
  • 永远不要在日志或评论中打印真实密钥。

高级工作流模式

决策树

  1. 音频是实时的吗？

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

  2. 响应延迟是否关键 (<1秒)？

    • 是 → 使用流式 STT，设置 format_turns=false 并小心回合检测。
    • 否 → 使用异步 STT，然后进行摘要/章节等。

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

    • 是 → 在分享前考虑使用 redact_pii (并可选择 redact_pii_audio)。
    • 否 → 根据需要使用原始转录本。

  4. 需要基于 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"[最终] {text}")
        else:
            print(f"[部分] {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()