Vision Agents 的 AssemblyAI 流式语音识别插件：Universal-3 Pro 实时 STT 集成指南

Vision Agents 的 AssemblyAI 流式语音识别插件：Universal-3 Pro 实时 STT 集成指南

【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Stream's edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents

AssemblyAI 插件（位于 plugins/assemblyai/README.md）为 Vision Agents 提供了基于 AssemblyAI Universal-3 Pro 模型的流式语音转写（Streaming STT）能力，通过异步 WebSocket 将实时音频转成文本，并原生支持基于标点的轮次检测、流式说话人分离与断线重连。读完本文，你将掌握该插件的安装方式、全部配置参数的含义与取值约束，并能基于源码级原理把它正确集成进自己的语音 Agent 管线中。

插件定位与核心特性

该插件是对vision_agents.core.stt.STT抽象基类的一个具体实现，类名为assemblyai.STT，其核心文件为 plugins/assemblyai/vision_agents/plugins/assemblyai/stt.py。与常见的 REST 批量转写不同，它面向的是实时双向语音 Agent 场景，官方文档宣称的实用能力包括：

• 实时流式转写：基于异步 WebSocket（wss://streaming.assemblyai.com/v3/ws）持续上传音频、持续接收转写结果，而非等待整段音频结束；
• 内置基于标点的轮次检测：可通过min_turn_silence/max_turn_silence两个阈值微调“用户是否说完话”的判断时机；
• 流式说话人分离：开启speaker_labels后可在混合音频流中实时区分说话人，每个转写事件携带独立的participant标识；
• 原生SpeechStarted事件支持：语音开始即触发TurnStarted，让 Agent 能提前感知用户开口；
• 自定义提示词与关键词增强：支持prompt与keyterms_prompt两种方式提升特定领域或专有名词的识别率；
• 内置断线重连：采用带指数退避（exponential backoff）的有限次数自动重连，避免短暂网络抖动导致会话中断。

从源码结构看，上述能力均可在 stt.py 中找到对应实现：_build_ws_url负责组装 URL 参数，_receive_loop/_send_loop负责双工收发，_reconnect实现退避重连，_handle_turn负责轮次消息到统一事件模型的转换。

安装

推荐使用uv添加依赖，有两种等价方式：

# 方式一：通过 vision-agents 的 extra 安装 uv add "vision-agents[assemblyai]" # 方式二：直接安装独立插件包 uv add vision-agents-plugins-assemblyai

插件包以vision-agents-plugins-assemblyai命名发布，其元数据定义在 plugins/assemblyai/pyproject.toml，要求requires-python = ">=3.10"，运行时依赖为vision-agents与aiohttp（当前 pyproject 中约束为aiohttp>=3.13.3，README 中标注的下限为 3.9.0，安装时以锁定的实际版本为准）。

快速开始：基础用法

最简用法只需指定模型与采样率（两者均有合理默认值）：

from vision_agents.plugins import assemblyai stt = assemblyai.STT( speech_model="u3-rt-pro", # 默认模型 sample_rate=16000, # 默认采样率 16kHz )

插件默认使用u3-rt-pro（Universal-3 Pro 的实时版本），采样率默认为 16000 Hz。插件基类 agents-core/vision_agents/core/stt/stt.py 中定义的turn_detection标志默认为False，而 AssemblyAI 插件将其覆写为True（见 stt.py 第 30 行），表明该 STT 自带轮次检测能力，无需额外挂接独立的 turn detection 组件。

在测试 test_assemblyai_stt.py 中，test_default_configuration验证了默认配置：_speech_model == "u3-rt-pro"、_sample_rate == 16000、turn_detection is True、provider_name == "assemblyai"，可作为默认行为的直接证据。

流式说话人分离（Streaming Diarization）

开启speaker_labels即可在多人混合音频中实时区分说话人：

stt = assemblyai.STT( speaker_labels=True, max_speakers=2, # 可选提示，取值范围 1-10 )

开启后，每个Turn转写事件都会携带该说话人对应的独立participant，同时原始标签可通过response.other["speaker_label"]获取。其底层实现在_handle_turn与_resolve_participant（stt.py）：

• 当未开启说话人分离或消息中没有speaker_label时，事件归属当前活跃的participant；
• 开启后，插件会为每个speaker_label创建并缓存一个合成 Participant：user_id形如speaker_A，id形如speaker_A_<session_id前8位>，同一个标签复用同一对象，不同标签互不相同；
• 组装响应时，response.other会被设置为{"speaker_label": ...}，方便上层消费原始标签。

对应的测试用例覆盖了合成 Participant 的创建与缓存（test_resolve_participant_creates_synthetic、test_resolve_participant_caches_synthetic、test_resolve_participant_distinct_per_label），以及三人轮流对话时speaker_A/speaker_B/speaker_A的正确归属（test_handle_turn_multi_speaker_conversation）。

需要注意参数校验：max_speakers只能在speaker_labels=True时使用，否则构造时会抛出ValueError("max_speakers requires speaker_labels=True")（见 stt.py 第 70-71 行，对应测试test_max_speakers_without_speaker_labels_raises）。

关键词增强（Keyterms Boosting）与自定义提示词

对于品牌名、专有名词等冷僻词，可通过关键词增强提升识别准确率：

stt = assemblyai.STT( keyterms_prompt=["AssemblyAI", "Vision Agents"], )

或者使用更灵活的自定义转写提示词：

stt = assemblyai.STT( prompt="This is a conversation between two developers discussing SDKs.", )

约束：prompt与keyterms_prompt二者互斥，同时传入会抛出ValueError("prompt and keyterms_prompt cannot be used together")（stt.py 第 67-68 行）。测试test_prompt_and_keyterms_exclusive对此有专门验证。从_build_ws_url可以看出，keyterms_prompt在发送前会经过json.dumps序列化，而prompt直接作为 URL 查询参数传递。

自定义轮次静音阈值

Vision Agents 的轮次检测决定“何时判定用户说完话、交出话语权”。AssemblyAI 提供两个与 API 对齐的静音阈值参数：

stt = assemblyai.STT( min_turn_silence=100, # 触发"推测性轮次结束"检查前的静音毫秒数 max_turn_silence=1200, # 超过此静音时长强制结束当前轮次 )

• min_turn_silence：在该静音时长之前不做“可能已说完”的推测，用于避免把短暂停顿误判为轮次结束；
• max_turn_silence：静音超过该上限后强制判定轮次结束，防止冷场时长时间占用。

两者默认值均为None，即交给 AssemblyAI 服务端使用其 API 默认值；传入时才会出现在 WebSocket URL 的查询参数中（见_build_ws_url第 105-108 行）。

服务端按轮次返回结果时，Turn消息中的end_of_turn标志是插件判断轮次归属的依据：为True时以mode="final"发射最终转写并附带TurnEnded事件，否则以mode="replacement"发射增量/替换型中间结果（stt.py 第 316-323 行）。Transcript事件与TurnEnded/TurnStarted事件的定义分别位于 core/stt/stt.py 与 core/turn_detection/turn_detection.py。

配置参数一览

关于采样率的实现细节：process_audio会先通过pcm_data.resample(self._sample_rate, 1)将任意输入采样率统一重采样为配置的sample_rate，再进入发送队列（stt.py 第 342-343 行），因此上游设备哪怕以 48kHz 推流也能被正确处理——集成测试test_transcribe_mia_audio_48khz正是用 48kHz 的mia_audio_48khz分块喂入并断言最终转写包含 "forgotten treasures"。

环境变量与密钥管理

插件优先使用构造参数api_key，未传时自动读取ASSEMBLYAI_API_KEY环境变量（stt.py 第 73 行）。推荐将密钥写入.env文件，配合python-dotenv加载：

# .env ASSEMBLYAI_API_KEY=your_key_here

完整示例 plugins/assemblyai/example/assemblyai_stt_example.py 展示了密钥的组织方式：除ASSEMBLYAI_API_KEY外，还依赖STREAM_API_KEY、STREAM_API_SECRET（GetStream 边缘网络）、GOOGLE_API_KEY（Gemini LLM）与CARTESIA_API_KEY（Cartesia TTS），并在入口处通过load_dotenv()统一加载。

源码原理：WebSocket 会话与消息处理

连接建立与 URL 构建

插件连接的目标为wss://streaming.assemblyai.com/v3/ws，请求头携带Authorization: <api_key>与AssemblyAI-Version: 2025-05-12（stt.py 第 17-18、134-138 行）。_build_ws_url负责把配置项编码为查询参数：sample_rate、speech_model必传；min_turn_silence、max_turn_silence、prompt、keyterms_prompt按需传入；speaker_labels=true与max_speakers仅在开启分离时追加。测试test_build_ws_url_*系列对这几类 URL 组合做了精确断言。

start()在建立连接后会等待_connection_ready事件，最长 10 秒，超时则抛出TimeoutError("Failed to connect to AssemblyAI within 10 seconds")（第 119-128 行）。

音频分块与发送

AssemblyAI 服务端要求每条消息 50-1000ms 的音频，插件按100ms 的 int16 PCM分块：chunk_size = sample_rate * 2 // 10（16kHz 下即 3200 字节）。process_audio将重采样后的字节追加进_audio_buffer，攒够一块就放入_audio_queue，由独立的_send_loop任务以send_bytes异步写出，实现发送与接收互不阻塞（第 96-98、230-242、325-354 行）。

消息类型与轮次处理

_receive_loop解析服务端下发的 JSON，按type字段分发（第 244-271 行）：

转写事件以统一结构Transcript（含text、mode、participant、response等字段）推入基类的output流，最终转写还会触发MetricsCollector.on_stt_transcript指标记录，方便接入观测体系（core/stt/stt.py 第 124-147 行）。

断线重连与指数退避

当连接被关闭、正在关闭或出错时，_receive_loop会转入_reconnect：先清理旧连接与说话人缓存，再以reconnect_backoff_initial_s起步、每次翻倍、封顶reconnect_backoff_max_s的退避策略重试，最多max_reconnect_attempts次；只有WSServerHandshakeError、TimeoutError、OSError这类瞬时故障才会触发重连，重连成功即返回（第 171-204 行）。测试test_reconnect_defaults与test_custom_reconnect_config分别验证了默认值（3 次 / 0.5s / 4.0s）与自定义值（5 次 / 1.0s / 8.0s）的生效。

close()会先冲刷残留音频缓冲、发送{"type": "Terminate"}优雅终止消息，再关闭 WebSocket 与 HTTP 会话，确保资源彻底释放（第 356-379 行）。

与完整 Agent 的集成示例

插件不是孤立组件，而是 Vision Agents 管线中stt=插槽的一等公民。仓库示例 assemblyai_stt_example.py 展示了如何将它与其他插件拼装成一个可运行的语音 Agent：

from vision_agents.core import Agent, Runner, User from vision_agents.core.agents import AgentLauncher from vision_agents.plugins import assemblyai, cartesia, gemini, getstream async def create_agent(**kwargs) -> Agent: agent = Agent( edge=getstream.Edge(), # GetStream 边缘实时通信 agent_user=User(name="AssemblyAI Agent", id="agent"), instructions="You're a helpful voice AI assistant. Keep replies short and conversational.", stt=assemblyai.STT(), # 本插件：流式 STT tts=cartesia.TTS(), # Cartesia 语音合成 llm=gemini.LLM(), # Gemini 大模型 ) return agent

示例还展示了以Runner(AgentLauncher(create_agent=..., join_call=...)).cli()启动 CLI、通过agent.join(call)加入通话、用agent.simple_response(...)播报开场白并agent.finish()收尾的完整生命周期。该示例的依赖声明见 plugins/assemblyai/example/pyproject.toml，除 AssemblyAI 插件外还组合了getstream、gemini、vision-agents三个包。

依赖与版本要求

环境要求：Python>=3.10；运行时需可用的ASSEMBLYAI_API_KEY；真实推流场景还需配合边缘传输（如 GetStream）与 LLM / TTS 插件才能构成完整闭环。

测试验证

插件质量由 plugins/assemblyai/tests/test_assemblyai_stt.py 中的 17 个测试用例保障，覆盖：

• 真实集成转写：test_transcribe_mia_audio_48khz（标注@pytest.mark.integration）用 48kHz 音频分块喂入，断言最终文本包含关键词且首条转写的 participant 归属正确；
• 参数互斥校验：prompt与keyterms_prompt互斥、max_speakers依赖speaker_labels；
• 默认与自定义配置：模型、采样率、重连参数、说话人分离开关的默认值及自定义值；
• URL 构建：各参数在 WebSocket URL 中的出现与缺失；
• Participant 解析：合成 Participant 的创建、缓存、去重与回退逻辑；
• 轮次事件发射：final / replacement 两种模式、response.other中speaker_label的携带，以及多人轮流对话的归属正确性。

这些用例既是插件行为的契约，也是二次开发时的参考样例。

结语

AssemblyAI 插件为 Vision Agents 提供了一条开箱即用的实时语音转写通道：默认u3-rt-pro模型、100ms 音频分块、基于end_of_turn的轮次事件、流式说话人分离与有限次指数退避重连，全部封装在单一assemblyai.STT类中。无论是构建客服机器人、会议纪要助手还是多说话人场景的语音应用，都可以按照本文的配置表与源码分析快速落地，并借助仓库内的测试与示例完成验证。

【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Stream's edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents