语音 AI 引擎开发技能

使用异步工作线程流水线、流式转录、LLM 智能体和 TTS 合成,构建可投入生产的实时对话式 AI 语音引擎。

概述

本技能为构建语音 AI 引擎提供全面指导,使您能够在用户与 AI 智能体之间实现自然的双向对话。它涵盖了从音频输入到音频输出的完整架构,包括:

  • 异步工作线程流水线模式 - 基于队列通信的并发处理
  • 流式转录 - 实时语音转文本转换
  • LLM 驱动的智能体 - 具有上下文感知的对话式 AI
  • 文本转语音合成 - 自然语音生成
  • 打断处理 - 用户可以在机器人说到一半时打断
  • 多提供商支持 - 轻松在不同服务提供商之间切换

快速开始

# Use the skill in your AI assistant
@voice-ai-engine-development I need to build a voice assistant that can handle real-time conversations with interrupts

包含内容

主要技能文件

  • SKILL.md - 语音 AI 引擎开发的综合指南

示例

  • complete_voice_engine.py - 完整可运行的实现
  • gemini_agent_example.py - 带正确响应缓冲的 LLM 智能体
  • interrupt_system_example.py - 打断处理演示

模板

  • base_worker_template.py - 用于创建新工作线程的模板
  • multi_provider_factory_template.py - 多提供商工厂模式

参考资料

  • common_pitfalls.md - 常见问题与解决方案
  • provider_comparison.md - 转录、LLM 和 TTS 提供商的比较

关键概念

工作线程流水线模式

每个语音 AI 引擎都遵循这条流水线:

Audio In → Transcriber → Agent → Synthesizer → Audio Out
           (Worker 1)   (Worker 2)  (Worker 3)

每个工作线程:

  • 通过 asyncio 独立运行
  • 通过 asyncio.Queue 对象通信
  • 可以在流中停止以处理打断
  • 优雅地处理错误

关键实现细节

  1. 缓冲 LLM 响应 - 在发送给合成器之前,始终缓冲完整的 LLM 响应,以防止音频跳跃
  2. 静音转录器 - 当机器人说话时静音转录器,以防止回声/反馈回路
  3. 限速音频 - 以实时速度发送音频块以支持打断
  4. 正确清理 - 始终在 finally 块中清理资源,以防止内存泄漏

支持的提供商

转录

  • Deepgram(最快,最适合实时)
  • AssemblyAI(准确率最高)
  • Azure Speech(企业级)
  • Google Cloud Speech(多语言)

LLM

  • OpenAI GPT-4(质量最高)
  • Google Gemini(经济高效)
  • Anthropic Claude(注重安全)

TTS

  • ElevenLabs(最自然的声音)
  • Azure TTS(企业级)
  • Google Cloud TTS(经济高效)
  • Amazon Polly(AWS 集成)
  • Play.ht(声音克隆)

常见用例

  • 客户服务语音机器人
  • 语音助手
  • 电话自动化系统
  • 支持语音的应用
  • 交互式语音应答(IVR)系统
  • 基于语音的辅导系统

架构亮点

异步工作线程模式

class BaseWorker:
    async def _run_loop(self):
        while self.active:
            item = await self.input_queue.get()
            await self.process(item)

打断系统

# User interrupts bot mid-sentence
if stop_event.is_set():
    partial_message = get_message_up_to(seconds_spoken)
    return partial_message, True  # cut_off = True

多提供商工厂

factory = VoiceComponentFactory()
transcriber = factory.create_transcriber(config)  # Deepgram, AssemblyAI, etc.
agent = factory.create_agent(config)              # OpenAI, Gemini, etc.
synthesizer = factory.create_synthesizer(config)  # ElevenLabs, Azure, etc.

测试

本技能包含以下示例:

  • 独立单元测试工作线程
  • 集成测试完整流水线
  • 测试打断功能
  • 使用不同提供商进行测试

最佳实践

  1. ✅ 在每个阶段始终进行流式处理(转录、LLM、合成)
  2. ✅ 在合成前缓冲完整的 LLM 响应
  3. ✅ 在机器人说话期间静音转录器
  4. ✅ 对音频块限速以支持打断
  5. ✅ 维护对话历史以提供上下文
  6. ✅ 在工作线程循环中使用正确的错误处理
  7. ✅ 在 finally 块中清理资源
  8. ✅ 使用 16kHz 的 LINEAR16 PCM 音频

常见陷阱

有关以下问题的详细解决方案,请参阅 references/common_pitfalls.md:

  • 音频跳跃/截断
  • 回声/反馈回路
  • 打断不起作用
  • 内存泄漏
  • 对话上下文丢失
  • 高延迟
  • 音频质量差

贡献

本技能是 Agentic Awesome Skills 仓库的一部分。欢迎贡献!

相关技能

  • @websocket-patterns - WebSocket 实现
  • @async-python - Asyncio 模式
  • @streaming-apis - 流式 API 集成
  • @audio-processing - 音频格式转换

许可证

MIT 许可证 - 参见仓库 LICENSE 文件

资源


为 Antigravity 社区用 ❤️ 构建

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐