我构建了一个语音AI平台，延迟仅442ms：完整架构如下

编程技术 2026-06-10 17:35:09 22

最近折腾了一个语音AI平台，从零开始搭了30天，最终做到了端到端延迟442ms。这篇把整个架构说清楚，有需要的直接抄。

大多数语音AI教程写到"调用ElevenLabs API"就结束了。

这不是平台，这是个demo。一旦ElevenLabs调价，整个系统说崩就崩。

我花了30天做了Mithivoices——一个开源TTS/STT平台，支持19+种神经网络音色、8种语言（包括印地语、马拉雅拉姆语、马拉地语），端到端延迟做到了442ms。核心设计决策：改一行配置就能在Piper、ElevenLabs、OpenAI、Coqui之间切换，代码零改动。

这是完整架构，直接复制使用。

云端语音AI的真正问题

每个生产级语音系统最终都会撞上同一堵墙：

问题	后果
阿里云TTS涨价3倍	你的单位成本一夜之间崩盘
百度语音接口改了参数	你的STT pipeline直接挂掉
客户需要印地语支持	你的英语优先服务商直接撂挑子
客户要离线部署	云端优先架构根本用不了

解决办法不是选"最好的"服务商。是搭一层抽象层，让服务商变成可替换的。

我们要做什么

一个provider无关的语音AI平台：

TTS：默认本地Piper TTS → 改配置切到阿里云/百度/腾讯
STT：默认本地Whisper → 改配置切到云端Whisper/腾讯云ASR
编排：FastAPI + Redis WebSocket引擎实现实时双向音频
记忆：LangGraph状态化Agent——对话上下文在多轮之间保持
目标：端到端延迟低于500ms（我们做到了442ms）

第一步：抽象层（最重要的部分）

先把这个搭好。后面所有东西都插进来。

# voice/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional @dataclass
class TTSConfig: provider: str # "piper" | "elevenlabs" | "openai" | "coqui" voice_id: str language: str speed: float = 1.0 @dataclass
class STTConfig: provider: str # "whisper_local" | "whisper_api" | "deepgram" model_size: str = "base" language: Optional[str] = None class TTSProvider(ABC): @abstractmethod async def synthesize(self, text: str, config: TTSConfig) -> bytes: """Returns raw audio bytes (WAV format)""" pass class STTProvider(ABC): @abstractmethod async def transcribe(self, audio_bytes: bytes, config: STTConfig) -> str: """Returns transcribed text""" pass

现在实现Piper作为本地默认：

# voice/providers/piper_tts.py
import asyncio
import tempfile
import os
from voice.base import TTSProvider, TTSConfig class PiperTTSProvider(TTSProvider): def __init__(self, model_dir: str = "./models/tts"): self.model_dir = model_dir async def synthesize(self, text: str, config: TTSConfig) -> bytes: model_path = f"{self.model_dir}/{config.language}/{config.voice_id}.onnx" with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: output_path = tmp.name try: cmd = [ "piper", "--model", model_path, "--output_file", output_path, "--sentence_silence", "0.1" ] proc = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) await proc.communicate(input=text.encode()) with open(output_path, "rb") as f: return f.read() finally: os.unlink(output_path)

然后阿里云TTS作为云端drop-in替换——同一套接口，代码零改动：

# voice/providers/elevenlabs_tts.py
import httpx
from voice.base import TTSProvider, TTSConfig class ElevenLabsTTSProvider(TTSProvider): def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.elevenlabs.io/v1" async def synthesize(self, text: str, config: TTSConfig) -> bytes: async with httpx.AsyncClient() as client: response = await client.post( f"{self.base_url}/text-to-speech/{config.voice_id}", headers={"xi-api-key": self.api_key}, json={ "text": text, "model_id": "eleven_multilingual_v2", "voice_settings": {"speed": config.speed} } ) return response.content

第二步：配置系统（一文件搞定所有）

# config.yml — 改这个文件就能切换provider tts: provider: piper # 切到: elevenlabs | openai | coqui voice_id: en_US-lessac-medium language: en_US speed: 1.0 stt: provider: whisper_local # 切到: whisper_api | deepgram model_size: base # tiny | base | small | medium | large language: null # null = 自动检测 elevenlabs: api_key: ${ELEVENLABS_API_KEY} # 支持8种语言
languages: - en_US - hi_IN # 印地语 - ml_IN # 马拉雅拉姆语 - mr_IN # 马拉地语 - ta_IN # 泰米尔语 - te_IN # 泰卢固语 - bn_IN # 孟加拉语 - gu_IN # 古吉拉特语

Provider工厂——根据配置加载对应类：

# voice/factory.py
import yaml
from voice.providers.piper_tts import PiperTTSProvider
from voice.providers.elevenlabs_tts import ElevenLabsTTSProvider
from voice.providers.openai_tts import OpenAITTSProvider # 同样接口实现 def get_tts_provider(config: dict): provider = config["tts"]["provider"] if provider == "piper": return PiperTTSProvider(model_dir="./models/tts") elif provider == "elevenlabs": return ElevenLabsTTSProvider(api_key=config["elevenlabs"]["api_key"]) elif provider == "openai": return OpenAITTSProvider(api_key=config["openai"]["api_key"]) else: raise ValueError(f"Unknown TTS provider: {provider}")

每个provider（openai_tts.py、coqui_tts.py等）都是独立文件，实现同一套TTSProvider接口。完整实现在repo的backend/目录下。

第三步：实时音频引擎（FastAPI + Redis）

这是延迟生死攸关的地方。WebSocket做双向流，Redis在handler和pipeline之间做缓冲。

# backend/app/main.py
import json
import time
import redis.asyncio as aioredis
import yaml
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from voice.factory import get_tts_provider, get_stt_provider
from agents.conversation import build_conversation_graph app = FastAPI() with open("config.yml") as f: config = yaml.safe_load(f) tts = get_tts_provider(config)
stt = get_stt_provider(config)
redis_client = aioredis.Redis(host='localhost', port=6379, decode_responses=False)
agent_graph = build_conversation_graph(redis_client) @app.websocket("/voice")
async def voice_endpoint(websocket: WebSocket): await websocket.accept() session_id = str(id(websocket)) try: while True: audio_data = await websocket.receive_bytes() t_start = time.perf_counter() # STT → LangGraph agent → TTS transcript = await stt.transcribe(audio_data, config["stt"]) result = await agent_graph.ainvoke( {"messages": [{"role": "user", "content": transcript}], "session_id": session_id, "last_action": ""}, config={"configurable": {"thread_id": session_id}} ) response_text = result["messages"][-1]["content"] audio_response = await tts.synthesize(response_text, config["tts"]) latency_ms = (time.perf_counter() - t_start) * 1000 await websocket.send_bytes(audio_response) await websocket.send_text(json.dumps({ "transcript": transcript, "latency_ms": round(latency_ms, 1) })) except WebSocketDisconnect: pass

为什么用Redis而不是内存队列？我一开始用的内存队列。实际负载一上来，帧就丢了。Redis多~3ms延迟，但能让系统在并发会话下稳定运行。

第四步：LangGraph状态化记忆

没有这个，每轮对话都是全新的。有了这个，Agent记住整个会话的上下文。

# agents/conversation.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.aioredis import AsyncRedisSaver
from typing import TypedDict, List, Literal class ConversationState(TypedDict): messages: List[dict] session_id: str last_action: str async def understand_intent(state: ConversationState) -> ConversationState: # 分类用户意图——回复或执行动作（比如调API） last_msg = state["messages"][-1]["content"].lower() action = "act" if any(w in last_msg for w in ["book", "schedule", "find", "search"]) else "respond" return {**state, "last_action": action} async def generate_response(state: ConversationState) -> ConversationState: # 这里调LLM——推荐用Groq，延迟最低 # groq_client.chat.completions.create(...) reply = {"role": "assistant", "content": "Response from LLM"} return {**state, "messages": state["messages"] + [reply]} async def execute_action(state: ConversationState) -> ConversationState: # 调外部API、查数据库、预约等 result = {"role": "tool", "content": "Action completed"} return {**state, "messages": state["messages"] + [result]} def route_intent(state: ConversationState) -> Literal["respond", "act"]: return state["last_action"] if state["last_action"] in ("respond", "act") else "respond" def build_conversation_graph(redis_client): checkpointer = AsyncRedisSaver(redis_client) # 记忆存在Redis里 graph = StateGraph(ConversationState) graph.add_node("understand", understand_intent) graph.add_node("respond", generate_response) graph.add_node("act", execute_action) graph.add_conditional_edges("understand", route_intent, {"respond": "respond", "act": "act"}) graph.add_edge("act", "respond") graph.add_edge("respond", END) graph.set_entry_point("understand") return graph.compile(checkpointer=checkpointer)

Agent可以跳出对话（预约、查数据库）再带着结果回来——整个过程通过Redis checkpointing保持完整上下文。

第五步：如何做到442ms

完整延迟拆解：

阶段	耗时	方法
音频接收 + Redis缓冲	~10ms	完全避免磁盘I/O
Whisper STT（本地）	~150ms	`base`模型，不用`medium`
LLM响应（Groq）	~180ms	Groq LPU——最快的推理硬件
Piper TTS合成	~80ms	ONNX模型 + GPU加速运算
WebSocket发送	~22ms	压缩分块，不发原始WAV
总计	~442ms

两个最关键的决定：

1. 用Groq做LLM推理。他们的LPU硬件比标准GPU推理快10-20倍。Groq和普通云LLM端点的差距大约是300ms。这是从"像机器人"到"像真人"的关键差距。

2. Whisper用base模型，不用medium。准确率降~4%，但延迟降~110ms。对于实时对话，这个tradeoff永远是正确选择。离线处理转录文本的时候可以用medium跑一遍。

项目结构

ai-voice-platform/
├── backend/
│ ├── app/
│ │ └── main.py # FastAPI + WebSocket引擎
│ ├── tts.py # Piper TTS集成
│ └── llm/ # LLM支持
├── frontend/ # React + Vite UI
├── models/
│ └── tts/ # Piper ONNX音色模型（单独下载）
├── voice_assets/
├── docs/
│ ├── PRD.md
│ └── TRD.md
├── requirements.txt
├── download_models.py # 从Hugging Face下载音色模型
├── start_backend.bat # Windows：只启动后端
└── start_all.bat # Windows：启动全部

运行方式

# 克隆
git clone https://github.com/mithivoices/ai-voice-platform
cd ai-voice-platform # Python依赖
pip install -r requirements.txt # 前端依赖
cd frontend && npm install && cd .. # 下载音色模型（~570MB——不在repo里）
python download_models.py # Windows——启动全部：
start_all.bat # Linux/Mac——开两个终端：
# 终端1：
python -m uvicorn backend.app.main:app --port 8000
# 终端2：
cd frontend && npm run dev

前端：http://localhost:5173 · API：http://localhost:8000

可用端点：

端点	方法	功能
`/health`	GET	服务端状态
`/api/voices`	GET	列出可用音色
`/api/languages`	GET	列出支持语言
`/api/tts/generate`	POST	生成音频
`/voice`	WebSocket	实时双向音频

下一步做什么

跑起来之后可以做的：

加个印地语语音Agent——改config里language: hi_IN，印地语模型自动下载
离线部署——Piper + Whisper本地跑，不需要互联网。树莓派5上都能跑
流式TTS——合成的时候直接流式输出音频块，不用等完整响应
加个"act"节点——把LangGraph Agent接日历API、数据库或预约系统

三件事做之前就想告诉自己的

先抽象，后实现。我第一次尝试把Piper直接写进WebSocket handler。拆掉重做抽象层花了3天。先搭接口，再做实现。

延迟会叠加。在一个阶段省50ms不只是省50ms——会改变整个对话的体验。600ms和442ms的差距，就是"像个机器人"和"像个真人"的差距。

Redis对音频pipeline是必需的。内存队列测试的时候挺好使。并发会话一上来就丢帧。Redis多~3ms，但能让系统生产环境稳定。

完整源码、PRD、TRD文档在github.com/mithivoices/ai-voice-platform。

如果你用这个做了东西——IVR系统、印地语语音助手、离线 kiosk——扔到评论区。想看看大家实际能做出什么来。

你在语音AI pipeline里遇到的最难的延迟瓶颈是什么？