尧图网站建设 
Vision-Agents 集成 Deepgram基于 Flux 模型的实时语音识别与流式语音合成实战【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Streams edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents导读本文讲解如何在 Vision-Agents 中通过官方插件vision-agents-plugins-deepgram接入 Deepgram 的语音能力一方面使用 Deepgram Flux 模型实现带内置端点检测Turn Detection的高质量 Speech-to-TextSTT另一方面使用 Flux TTS 通过 WebSocket 长连接实现低延迟的流式 Text-to-SpeechTTS。读完本文你将掌握插件的安装方式、STT/TTS 的全部核心参数及其含义、底层实现原理连接管理、事件处理、中断机制并能借助仓库自带的完整示例快速跑通一个语音识别 → LLM → 语音合成的实时语音 Agent。一、插件定位与安装Deepgram 插件为 Vision-Agents 提供了两大语音能力源码位于 plugins/deepgramSTT基于 Deepgram Flux 模型的语音识别内置端点检测可输出部分/最终转写结果并触发 turn 相关事件TTS基于 Deepgram Flux TTS/v2/speakWebSocket 接口的低延迟语音合成支持长连接复用与流式输出。安装方式有两种推荐使用 uvuv add vision-agents[deepgram] # 或直接安装插件包 uv add vision-agents-plugins-deepgram从 plugins/deepgram/pyproject.toml 可以看到插件运行时依赖vision-agents与deepgram-sdk7.7.0,7.8.0要求 Python 3.10。SDK 版本被严格锁定在 7.7.x 是因为实现中同时兼容了 SDK 7.7 的类型化事件模型与旧版 7.x 的纯 dict 事件两种消息形态见 deepgram_stt.py。二、Speech-to-TextFlux 实时转写2.1 快速上手from vision_agents.plugins import deepgram stt deepgram.STT( modelflux-general-en, # 默认模型 eager_turn_detectionTrue, # 开启急切式端点检测提前预判说话结束 )2.2 完整参数解析STT构造函数定义于 deepgram_stt.py各参数说明如下参数默认值说明api_keyNoneDeepgram API Key不传时自动读取环境变量DEEPGRAM_API_KEYmodelflux-general-en用于转写的 Flux 模型languageNone语言代码如en、es不传时启用自动检测并会出现在转写结果的language元数据中eager_turn_detectionFalse是否开启急切式端点检测Eager End-of-Turneot_thresholdNone端点检测灵敏度阈值对应 Deepgram v2 的eot_threshold参数eager_eot_thresholdNone急切式端点检测阈值开启eager_turn_detection但未显式指定时自动取0.5clientNone可选预配置好的AsyncDeepgramClient实例便于复用已有客户端2.3 内置端点检测无需额外 turn 检测器这是 Deepgram Flux 实现的关键设计Deepgram 在服务端内部完成了端点检测因此 Vision-Agents 侧的 turn detection 是可选的、不需要的。这一点在源码注释中明确说明deepgram_stt.pyeot_threshold控制回合结束判定的灵敏度eager_eot_threshold控制急切式回合结束——即在真正说完之前提前触发让 LLM 可以先开始准备回复从而降低端到端延迟。因此STT类的类属性turn_detection Truedeepgram_stt.py向框架表明该 STT 已自带端点检测能力。2.4 底层工作流程从实现看STT 的完整链路为deepgram_stt.pystart()通过client.listen.v2.connect(...)建立 Deepgram v2 listen WebSocket连接建立带 10 秒超时连接参数包含model、encodinglinear16、sample_rate16000并根据配置追加eot_threshold/eager_eot_threshold随后注册OPEN、MESSAGE、ERROR、CLOSE四类事件处理器并启动监听任务。process_audio()将收到的 PCM 音频自动重采样为 16kHz 单声道Deepgram 推荐规格转为字节后通过 WebSocket 的send_media发送内部用time.perf_counter()记录首个音频块到达时间用于计算音频输入 → 转写产出的处理延迟。事件处理_on_message()收到TurnInfo消息后StartOfTurn/Update视为部分转写partialEndOfTurn视为最终转写final并触发turn_ended事件EagerEndOfTurn触发急切回合结束事件eagerTrue从消息中提取逐词置信度并求平均作为整体置信度同时携带language、audio_duration_ms、model_name、processing_time_ms等元数据生成TranscriptResponse。由于 Deepgram 返回的是更新后的完整转写而非增量差异插件以modefinal或modereplacement两种模式广播转写事件deepgram_stt.py。close()先发送CloseStream消息优雅关闭再退出连接上下文最后通过 SDK 内部 httpx client 关闭底层 HTTP 连接。2.5 测试对行为的验证test_stt.py 中的集成测试证实了上述行为向 STT 发送 48kHz 的音频后再发送 2 秒静音以触发回合结束会得到 final 转写Transcript.final True与TurnEnded事件单元测试则覆盖了 dict 与类型化ListenV2TurnInfo两种消息格式都能正确产出 hello world 的最终转写且非TurnInfo消息会被安全忽略test_stt.py。close()也会正确关闭底层 httpx 客户端避免资源泄漏test_stt.py。三、Text-to-SpeechFlux TTS 流式合成3.1 快速上手from vision_agents.plugins import deepgram tts deepgram.TTS( modelflux-haley-en, # 默认音色 sample_rate16000, # 音频采样率 speed1.0, # 可选语速倍率0.85–1.15 )3.2 完整参数解析TTS构造函数定义于 tts.py参数默认值说明api_keyNoneDeepgram API Key不传时读取环境变量DEEPGRAM_API_KEYmodelflux-haley-enFlux 音色模型命名格式为flux-{voice}-ensample_rate16000音频采样率仅支持8000 / 16000 / 24000 / 32000 / 44100 / 48000其他值会抛出ValueErrorspeedNone可选语速倍率合法范围为 0.85–1.15按 0.05 步进clientNone可选预配置好的AsyncDeepgramClient实例两点硬性约束构造函数即校验见 tts.pyAura 系列模型字符串不受支持传入以aura开头的模型名会直接抛出ValueError提示必须使用 Flux 模型非法采样率直接报错如sample_rate12345会抛出ValueError并列出全部合法采样率对应测试见 test_tts.py。3.3 可用音色Deepgram Flux 音色统一使用flux-{voice}-en命名格式Aura 模型字符串不支持flux-haley-en—— 默认主打音色flux-kit-en更多音色请参考 Flux TTS voices 官方列表自定义音色只需替换model参数tts deepgram.TTS(modelflux-haley-en) # 默认主打音色 tts deepgram.TTS(modelflux-kit-en)3.4 底层工作流程与设计要点Flux TTS 实现的核心设计是跨多次合成调用复用同一条 WebSocket 长连接tts.py从而避免每次合成都重新建连的开销并消除音频之间的不连续感。测试test_connection_reused_across_calls明确断言了两次send_iter调用后tts._socket是同一个对象test_tts.py。关键流程tts.pystart()调用_ensure_connection()通过client.speak.v2.connect(model..., encodinglinear16, sample_rate..., speed...)建立/v2/speakWebSocket若连接中途被关闭下一次合成会自动重连。stream_audio(text)发送SpeakV2Speak(texttext)与send_flush()触发合成随后进入_receive_audio异步迭代器逐条读取 WebSocket 消息bytes消息直接封装为 16kHz/单声道/S16 的PcmData音频块逐块产出收到SpeechMetadata或SpeechInterrupted则结束本轮SpeakV2Warning仅记日志。stop_audio()支持中断——设置停止事件并发送send_interrupt()取消服务端正在进行的合成中断后残留的旧消息通过_drain()用短超时0.05s快速消费掉避免污染下一轮输出。close()发送send_close()关闭长连接并清理AsyncExitStack同时向基类广播断开事件。由于实现接受文本增量streaming True见 tts.pyTTS 可以直接消费 LLM 的部分输出流实现边说边想的低延迟体验。四、环境变量插件统一通过环境变量注入凭证DEEPGRAM_API_KEY设置方式二选一# 方式一环境变量 export DEEPGRAM_API_KEYyour_key_here # 方式二构造函数传入 stt deepgram.STT(api_keyyour_key_here) tts deepgram.TTS(api_keyyour_key_here)五、完整可运行示例插件自带一个完整的端到端示例位于 plugins/deepgram/example创建了一个同时使用 Deepgram STT TTS、GetStream 边缘通信、Gemini 大模型的实时语音 Agent。1. 配置环境变量参考 deepgram_tts_example.py 顶部的说明# 需要设置以下环境变量 DEEPGRAM_API_KEY # Deepgram API Key STREAM_API_KEY # Stream API Key STREAM_API_SECRET # Stream API Secret GOOGLE_API_KEY # Google API Key用于 Gemini LLM2. 安装依赖uv sync示例的 pyproject.toml 声明依赖vision-agents-plugins-deepgram、vision-agents-plugins-getstream、vision-agents-plugins-gemini与vision-agents并通过[tool.uv.sources]以 editable 方式指向仓库本地路径。3. 运行示例uv run python deepgram_tts_example.py run --call-type audio_room --call-id test4. Agent 构建核心代码deepgram_tts_example.pyfrom vision_agents.core import Agent, Runner, User from vision_agents.core.agents import AgentLauncher from vision_agents.plugins import deepgram, gemini, getstream async def create_agent(**kwargs) - Agent: agent Agent( edgegetstream.Edge(), agent_userUser(nameDeepgram Agent, idagent), instructionsYoure a helpful voice AI assistant. Keep replies short and conversational., ttsdeepgram.TTS(), # Deepgram Flux TTS 合成语音 sttdeepgram.STT(), # Deepgram Flux 语音识别内置端点检测 llmgemini.LLM(), ) return agentAgent 加入通话后会先通过agent.simple_response(textHello! How can I help you today?)播报开场白随后进入agent.finish()持续服务直到通话结束deepgram_tts_example.py。六、常用组合技巧追求更低延迟STT 开启eager_turn_detectionTrue让EagerEndOfTurn提前触发 LLM 准备回复TTS 依赖streamingTrue消费 LLM 增量文本两级流水线叠加可显著缩短用户说完 → 听到回应的往返时间。控制打断行为Agent 需要抢话/打断能力时TTS 的stop_audio()已内置中断语义框架层可直接调用以取消当前合成并立即播报新内容。服务端转写 vs 增量转写Deepgram 返回的是整段更新后的转写而非增量插件已按replacement模式处理上层只需消费Transcript事件即可无需自行做文本拼接。连接生命周期STT 与 TTS 都维护了长连接Agent 结束时务必调用close()——STT 会发送CloseStream并关闭底层 httpx 客户端deepgram_stt.pyTTS 会发送send_close()并清理连接栈tts.py。总结Deepgram 插件以极小的接入成本为 Vision-Agents 补齐了实时语音闭环STT 侧利用 Flux 服务端内置端点检测免去额外的 turn detection 组件并通过eager机制压低延迟TTS 侧借助/v2/speakWebSocket 长连接与流式输出实现低延迟、可中断的语音合成。配合仓库内置的完整示例plugins/deepgram/example与集成测试test_stt.py、test_tts.py你可以快速验证接入效果并将其套用到任意语音识别 → 大模型 → 语音合成的实时语音 Agent 场景中。【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Streams edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Vision-Agents 集成 Deepgram：基于 Flux 模型的实时语音识别与流式语音合成实战
Vision-Agents 集成 Deepgram基于 Flux 模型的实时语音识别与流式语音合成实战【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Streams edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents导读本文讲解如何在 Vision-Agents 中通过官方插件vision-agents-plugins-deepgram接入 Deepgram 的语音能力一方面使用 Deepgram Flux 模型实现带内置端点检测Turn Detection的高质量 Speech-to-TextSTT另一方面使用 Flux TTS 通过 WebSocket 长连接实现低延迟的流式 Text-to-SpeechTTS。读完本文你将掌握插件的安装方式、STT/TTS 的全部核心参数及其含义、底层实现原理连接管理、事件处理、中断机制并能借助仓库自带的完整示例快速跑通一个语音识别 → LLM → 语音合成的实时语音 Agent。一、插件定位与安装Deepgram 插件为 Vision-Agents 提供了两大语音能力源码位于 plugins/deepgramSTT基于 Deepgram Flux 模型的语音识别内置端点检测可输出部分/最终转写结果并触发 turn 相关事件TTS基于 Deepgram Flux TTS/v2/speakWebSocket 接口的低延迟语音合成支持长连接复用与流式输出。安装方式有两种推荐使用 uvuv add vision-agents[deepgram] # 或直接安装插件包 uv add vision-agents-plugins-deepgram从 plugins/deepgram/pyproject.toml 可以看到插件运行时依赖vision-agents与deepgram-sdk7.7.0,7.8.0要求 Python 3.10。SDK 版本被严格锁定在 7.7.x 是因为实现中同时兼容了 SDK 7.7 的类型化事件模型与旧版 7.x 的纯 dict 事件两种消息形态见 deepgram_stt.py。二、Speech-to-TextFlux 实时转写2.1 快速上手from vision_agents.plugins import deepgram stt deepgram.STT( modelflux-general-en, # 默认模型 eager_turn_detectionTrue, # 开启急切式端点检测提前预判说话结束 )2.2 完整参数解析STT构造函数定义于 deepgram_stt.py各参数说明如下参数默认值说明api_keyNoneDeepgram API Key不传时自动读取环境变量DEEPGRAM_API_KEYmodelflux-general-en用于转写的 Flux 模型languageNone语言代码如en、es不传时启用自动检测并会出现在转写结果的language元数据中eager_turn_detectionFalse是否开启急切式端点检测Eager End-of-Turneot_thresholdNone端点检测灵敏度阈值对应 Deepgram v2 的eot_threshold参数eager_eot_thresholdNone急切式端点检测阈值开启eager_turn_detection但未显式指定时自动取0.5clientNone可选预配置好的AsyncDeepgramClient实例便于复用已有客户端2.3 内置端点检测无需额外 turn 检测器这是 Deepgram Flux 实现的关键设计Deepgram 在服务端内部完成了端点检测因此 Vision-Agents 侧的 turn detection 是可选的、不需要的。这一点在源码注释中明确说明deepgram_stt.pyeot_threshold控制回合结束判定的灵敏度eager_eot_threshold控制急切式回合结束——即在真正说完之前提前触发让 LLM 可以先开始准备回复从而降低端到端延迟。因此STT类的类属性turn_detection Truedeepgram_stt.py向框架表明该 STT 已自带端点检测能力。2.4 底层工作流程从实现看STT 的完整链路为deepgram_stt.pystart()通过client.listen.v2.connect(...)建立 Deepgram v2 listen WebSocket连接建立带 10 秒超时连接参数包含model、encodinglinear16、sample_rate16000并根据配置追加eot_threshold/eager_eot_threshold随后注册OPEN、MESSAGE、ERROR、CLOSE四类事件处理器并启动监听任务。process_audio()将收到的 PCM 音频自动重采样为 16kHz 单声道Deepgram 推荐规格转为字节后通过 WebSocket 的send_media发送内部用time.perf_counter()记录首个音频块到达时间用于计算音频输入 → 转写产出的处理延迟。事件处理_on_message()收到TurnInfo消息后StartOfTurn/Update视为部分转写partialEndOfTurn视为最终转写final并触发turn_ended事件EagerEndOfTurn触发急切回合结束事件eagerTrue从消息中提取逐词置信度并求平均作为整体置信度同时携带language、audio_duration_ms、model_name、processing_time_ms等元数据生成TranscriptResponse。由于 Deepgram 返回的是更新后的完整转写而非增量差异插件以modefinal或modereplacement两种模式广播转写事件deepgram_stt.py。close()先发送CloseStream消息优雅关闭再退出连接上下文最后通过 SDK 内部 httpx client 关闭底层 HTTP 连接。2.5 测试对行为的验证test_stt.py 中的集成测试证实了上述行为向 STT 发送 48kHz 的音频后再发送 2 秒静音以触发回合结束会得到 final 转写Transcript.final True与TurnEnded事件单元测试则覆盖了 dict 与类型化ListenV2TurnInfo两种消息格式都能正确产出 hello world 的最终转写且非TurnInfo消息会被安全忽略test_stt.py。close()也会正确关闭底层 httpx 客户端避免资源泄漏test_stt.py。三、Text-to-SpeechFlux TTS 流式合成3.1 快速上手from vision_agents.plugins import deepgram tts deepgram.TTS( modelflux-haley-en, # 默认音色 sample_rate16000, # 音频采样率 speed1.0, # 可选语速倍率0.85–1.15 )3.2 完整参数解析TTS构造函数定义于 tts.py参数默认值说明api_keyNoneDeepgram API Key不传时读取环境变量DEEPGRAM_API_KEYmodelflux-haley-enFlux 音色模型命名格式为flux-{voice}-ensample_rate16000音频采样率仅支持8000 / 16000 / 24000 / 32000 / 44100 / 48000其他值会抛出ValueErrorspeedNone可选语速倍率合法范围为 0.85–1.15按 0.05 步进clientNone可选预配置好的AsyncDeepgramClient实例两点硬性约束构造函数即校验见 tts.pyAura 系列模型字符串不受支持传入以aura开头的模型名会直接抛出ValueError提示必须使用 Flux 模型非法采样率直接报错如sample_rate12345会抛出ValueError并列出全部合法采样率对应测试见 test_tts.py。3.3 可用音色Deepgram Flux 音色统一使用flux-{voice}-en命名格式Aura 模型字符串不支持flux-haley-en—— 默认主打音色flux-kit-en更多音色请参考 Flux TTS voices 官方列表自定义音色只需替换model参数tts deepgram.TTS(modelflux-haley-en) # 默认主打音色 tts deepgram.TTS(modelflux-kit-en)3.4 底层工作流程与设计要点Flux TTS 实现的核心设计是跨多次合成调用复用同一条 WebSocket 长连接tts.py从而避免每次合成都重新建连的开销并消除音频之间的不连续感。测试test_connection_reused_across_calls明确断言了两次send_iter调用后tts._socket是同一个对象test_tts.py。关键流程tts.pystart()调用_ensure_connection()通过client.speak.v2.connect(model..., encodinglinear16, sample_rate..., speed...)建立/v2/speakWebSocket若连接中途被关闭下一次合成会自动重连。stream_audio(text)发送SpeakV2Speak(texttext)与send_flush()触发合成随后进入_receive_audio异步迭代器逐条读取 WebSocket 消息bytes消息直接封装为 16kHz/单声道/S16 的PcmData音频块逐块产出收到SpeechMetadata或SpeechInterrupted则结束本轮SpeakV2Warning仅记日志。stop_audio()支持中断——设置停止事件并发送send_interrupt()取消服务端正在进行的合成中断后残留的旧消息通过_drain()用短超时0.05s快速消费掉避免污染下一轮输出。close()发送send_close()关闭长连接并清理AsyncExitStack同时向基类广播断开事件。由于实现接受文本增量streaming True见 tts.pyTTS 可以直接消费 LLM 的部分输出流实现边说边想的低延迟体验。四、环境变量插件统一通过环境变量注入凭证DEEPGRAM_API_KEY设置方式二选一# 方式一环境变量 export DEEPGRAM_API_KEYyour_key_here # 方式二构造函数传入 stt deepgram.STT(api_keyyour_key_here) tts deepgram.TTS(api_keyyour_key_here)五、完整可运行示例插件自带一个完整的端到端示例位于 plugins/deepgram/example创建了一个同时使用 Deepgram STT TTS、GetStream 边缘通信、Gemini 大模型的实时语音 Agent。1. 配置环境变量参考 deepgram_tts_example.py 顶部的说明# 需要设置以下环境变量 DEEPGRAM_API_KEY # Deepgram API Key STREAM_API_KEY # Stream API Key STREAM_API_SECRET # Stream API Secret GOOGLE_API_KEY # Google API Key用于 Gemini LLM2. 安装依赖uv sync示例的 pyproject.toml 声明依赖vision-agents-plugins-deepgram、vision-agents-plugins-getstream、vision-agents-plugins-gemini与vision-agents并通过[tool.uv.sources]以 editable 方式指向仓库本地路径。3. 运行示例uv run python deepgram_tts_example.py run --call-type audio_room --call-id test4. Agent 构建核心代码deepgram_tts_example.pyfrom vision_agents.core import Agent, Runner, User from vision_agents.core.agents import AgentLauncher from vision_agents.plugins import deepgram, gemini, getstream async def create_agent(**kwargs) - Agent: agent Agent( edgegetstream.Edge(), agent_userUser(nameDeepgram Agent, idagent), instructionsYoure a helpful voice AI assistant. Keep replies short and conversational., ttsdeepgram.TTS(), # Deepgram Flux TTS 合成语音 sttdeepgram.STT(), # Deepgram Flux 语音识别内置端点检测 llmgemini.LLM(), ) return agentAgent 加入通话后会先通过agent.simple_response(textHello! How can I help you today?)播报开场白随后进入agent.finish()持续服务直到通话结束deepgram_tts_example.py。六、常用组合技巧追求更低延迟STT 开启eager_turn_detectionTrue让EagerEndOfTurn提前触发 LLM 准备回复TTS 依赖streamingTrue消费 LLM 增量文本两级流水线叠加可显著缩短用户说完 → 听到回应的往返时间。控制打断行为Agent 需要抢话/打断能力时TTS 的stop_audio()已内置中断语义框架层可直接调用以取消当前合成并立即播报新内容。服务端转写 vs 增量转写Deepgram 返回的是整段更新后的转写而非增量插件已按replacement模式处理上层只需消费Transcript事件即可无需自行做文本拼接。连接生命周期STT 与 TTS 都维护了长连接Agent 结束时务必调用close()——STT 会发送CloseStream并关闭底层 httpx 客户端deepgram_stt.pyTTS 会发送send_close()并清理连接栈tts.py。总结Deepgram 插件以极小的接入成本为 Vision-Agents 补齐了实时语音闭环STT 侧利用 Flux 服务端内置端点检测免去额外的 turn detection 组件并通过eager机制压低延迟TTS 侧借助/v2/speakWebSocket 长连接与流式输出实现低延迟、可中断的语音合成。配合仓库内置的完整示例plugins/deepgram/example与集成测试test_stt.py、test_tts.py你可以快速验证接入效果并将其套用到任意语音识别 → 大模型 → 语音合成的实时语音 Agent 场景中。【免费下载链接】Vision-AgentsOpen Vision Agents by Stream. Build voice and vision agents quickly with any model or video provider. Uses Streams edge network for ultra-low latency.项目地址: https://gitcode.com/GitHub_Trending/vi/Vision-Agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考