AI语音助手开发实战完全教程
从架构设计到生产部署，手把手构建一个支持多轮对话、多语言的AI语音助手系统。
目录
- 一、语音助手架构设计
- 二、ASR语音识别集成
- 三、LLM对话引擎
- 四、TTS语音合成
- 五、实时语音交互（WebSocket）
- 六、多轮对话管理
- 七、语音唤醒与VAD
- 八、多语言支持
- 九、智能家居/车载场景实战
- 十、性能优化与部署
一、语音助手架构设计
1.1 整体架构概览
一个完整的AI语音助手系统由以下核心模块组成：
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 音频输入 │───→│ ASR语音识别 │───→│ LLM对话引擎 │───→│ TTS语音合成 │
│ (麦克风/流) │ │ (Whisper等) │ │ (GPT-4等) │ │ (Edge TTS等) │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
↑ │
│ ┌─────────────┐ │
└──────────────│ 播放输出 │←──────────────────────────┘
│ (扬声器) │
└─────────────┘
1.2 技术栈选型
| 模块 | 推荐方案 | 备选方案 | 说明 |
|---|---|---|---|
| ASR | OpenAI Whisper | FunASR, Paraformer | Whisper多语言能力最强 |
| LLM | GPT-4o | Claude, Qwen, DeepSeek | 根据场景选择 |
| TTS | Edge TTS | CosyVoice, Fish Speech | Edge TTS免费且质量高 |
| 通信 | WebSocket | HTTP SSE, gRPC | 实时交互用WebSocket |
| 唤醒 | Porcupine | Whisper关键词检测 | 唤醒词专用引擎 |
| VAD | Silero VAD | WebRTC VAD | Silero准确率更高 |
1.3 项目结构
voice-assistant/
├── main.py # 主入口
├── config.py # 配置管理
├── requirements.txt
├── core/
│ ├── __init__.py
│ ├── asr.py # 语音识别模块
│ ├── llm.py # 对话引擎模块
│ ├── tts.py # 语音合成模块
│ ├── vad.py # 语音活动检测
│ └── wake_word.py # 唤醒词检测
├── session/
│ ├── __init__.py
│ └── manager.py # 会话管理
├── transport/
│ ├── __init__.py
│ └── websocket_server.py # WebSocket服务
├── utils/
│ ├── __init__.py
│ └── audio.py # 音频工具函数
└── tests/
├── test_asr.py
├── test_tts.py
└── test_session.py
二、ASR语音识别集成
2.1 Whisper本地部署
OpenAI Whisper是目前最流行的开源语音识别模型，支持99种语言：
# core/asr.py
import whisper
import numpy as np
from typing import Optional
class ASREngine:
"""基于Whisper的语音识别引擎"""
def __init__(self, model_size: str = "base", device: str = "cpu"):
"""
初始化ASR引擎
Args:
model_size: 模型大小 tiny/base/small/medium/large
device: 运行设备 cpu/cuda
"""
self.model = whisper.load_model(model_size, device=device)
self.sample_rate = 16000 # Whisper要求16kHz
def transcribe(
self,
audio_data: np.ndarray,
language: Optional[str] = None,
task: str = "transcribe",
) -> dict:
"""
转录音频数据
Args:
audio_data: 音频数据（float32, 16kHz, 单声道）
language: 语言代码，None为自动检测
task: "transcribe"转录 / "translate"翻译为英文
Returns:
{"text": str, "language": str, "segments": list}
"""
# 确保音频格式正确
if audio_data.dtype != np.float32:
audio_data = audio_data.astype(np.float32)
# 归一化
if audio_data.max() > 1.0:
audio_data = audio_data / 32768.0
result = self.model.transcribe(
audio_data,
language=language,
task=task,
fp16=False, # CPU模式下关闭fp16
)
return {
"text": result["text"].strip(),
"language": result.get("language", "unknown"),
"segments": result.get("segments", []),
}
def transcribe_file(self, file_path: str, language: Optional[str] = None) -> dict:
"""转录音频文件"""
result = self.model.transcribe(file_path, language=language)
return {
"text": result["text"].strip(),
"language": result.get("language", "unknown"),
"segments": result.get("segments", []),
}
# 使用示例
if __name__ == "__main__":
asr = ASREngine(model_size="base")
result = asr.transcribe_file("test.wav")
print(f"识别结果: {result['text']}")
print(f"检测语言: {result['language']}")
2.2 流式识别方案
对于实时对话场景，需要边听边识别的流式方案：
# core/streaming_asr.py
import numpy as np
from collections import deque
import threading
import time
class StreamingASR:
"""流式语音识别，支持实时转录"""
def __init__(self, asr_engine, chunk_duration=2.0, overlap=0.5):
"""
Args:
asr_engine: ASREngine实例
chunk_duration: 每次识别的音频长度（秒）
overlap: 重叠长度（秒），用于上下文连贯
"""
self.asr = asr_engine
self.chunk_samples = int(chunk_duration * 16000)
self.overlap_samples = int(overlap * 16000)
self.buffer = deque(maxlen=self.chunk_samples * 3)
self.result_callback = None
self._running = False
def set_callback(self, callback):
"""设置识别结果回调"""
self.result_callback = callback
def feed_audio(self, audio_chunk: np.ndarray):
"""喂入音频数据"""
self.buffer.extend(audio_chunk)
if len(self.buffer) >= self.chunk_samples:
# 取出一个chunk进行识别
chunk = np.array(list(self.buffer)[:self.chunk_samples])
# 保留overlap部分用于上下文
for _ in range(self.chunk_samples - self.overlap_samples):
self.buffer.popleft()
# 异步识别
threading.Thread(
target=self._process_chunk,
args=(chunk,),
daemon=True,
).start()
def _process_chunk(self, chunk: np.ndarray):
"""处理单个音频块"""
result = self.asr.transcribe(chunk)
if result["text"] and self.result_callback:
self.result_callback(result)
2.3 FunASR阿里方案（中文优化）
# 适合中文场景的FunASR方案
from funasr import AutoModel
class FunASREngine:
"""基于FunASR的中文优化语音识别"""
def __init__(self):
# 语音识别模型
self.model = AutoModel(
model="paraformer-zh",
vad_model="fsmn-vad", # 内置VAD
punc_model="ct-punc", # 自动标点
device="cpu",
)
def transcribe(self, audio_path: str) -> dict:
result = self.model.generate(input=audio_path)
return {
"text": result[0]["text"] if result else "",
"timestamp": result[0].get("timestamp", []),
}
# 使用
asr = FunASREngine()
result = asr.transcribe("recording.wav")
print(result["text"])
三、LLM对话引擎
3.1 对话引擎核心实现
# core/llm.py
import openai
from typing import List, Dict, Optional
from dataclasses import dataclass, field
@dataclass
class ConversationState:
"""对话状态"""
messages: List[Dict] = field(default_factory=list)
system_prompt: str = ""
context: Dict = field(default_factory=dict)
turn_count: int = 0
class LLMEngine:
"""LLM对话引擎"""
def __init__(
self,
model: str = "gpt-4o-mini",
api_key: Optional[str] = None,
base_url: Optional[str] = None,
):
self.client = openai.OpenAI(
api_key=api_key,
base_url=base_url,
)
self.model = model
self.default_system_prompt = """你是一个友好的AI语音助手。
你的回答将被转换为语音播放，因此请遵循以下规则：
1. 回答简洁明了，避免过长的段落
2. 不要使用Markdown格式、代码块或特殊符号
3. 使用自然口语化的表达
4. 数字用中文读法（如"一百二十三"而非"123"）
5. 如需列举，用"第一、第二"而非"1. 2. 3."
6. 单次回答控制在150字以内"""
def chat(
self,
user_input: str,
conversation: ConversationState,
stream: bool = True,
) -> str:
"""
处理用户输入，返回回复文本
Args:
user_input: 用户的文本输入（ASR识别结果）
conversation: 对话状态
stream: 是否流式输出
"""
# 添加用户消息
conversation.messages.append({
"role": "user",
"content": user_input,
})
# 构建完整消息列表
full_messages = [
{"role": "system", "content": conversation.system_prompt or self.default_system_prompt},
*conversation.messages[-20:], # 保留最近20轮对话
]
if stream:
return self._stream_chat(full_messages, conversation)
else:
return self._sync_chat(full_messages, conversation)
def _sync_chat(self, messages: List[Dict], conversation: ConversationState) -> str:
"""同步调用"""
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=500,
temperature=0.7,
)
reply = response.choices[0].message.content
conversation.messages.append({"role": "assistant", "content": reply})
conversation.turn_count += 1
return reply
def _stream_chat(self, messages: List[Dict], conversation: ConversationState) -> str:
"""流式调用，逐句返回"""
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=500,
temperature=0.7,
stream=True,
)
full_reply = ""
for chunk in stream:
if chunk.choices[0].delta.content:
token = chunk.choices[0].delta.content
full_reply += token
conversation.messages.append({"role": "assistant", "content": full_reply})
conversation.turn_count += 1
return full_reply
3.2 工具调用（Function Calling）
让语音助手能够执行实际操作：
# core/tools.py
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'",
},
"date": {
"type": "string",
"description": "日期，格式YYYY-MM-DD，空值表示今天",
},
},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "control_device",
"description": "控制智能家居设备",
"parameters": {
"type": "object",
"properties": {
"device": {
"type": "string",
"description": "设备名称，如'客厅灯'、'空调'",
},
"action": {
"type": "string",
"enum": ["on", "off", "set"],
"description": "操作类型",
},
"value": {
"type": "string",
"description": "设置值，如温度、亮度百分比",
},
},
"required": ["device", "action"],
},
},
},
]
def execute_tool_call(tool_name: str, arguments: dict) -> str:
"""执行工具调用"""
if tool_name == "get_weather":
# 实际项目中对接天气API
return f"{arguments['city']}今天晴，气温25度，适合出行"
elif tool_name == "control_device":
device = arguments["device"]
action = arguments["action"]
value = arguments.get("value", "")
if action == "on":
return f"已打开{device}"
elif action == "off":
return f"已关闭{device}"
else:
return f"已将{device}设置为{value}"
return "未知操作"
四、TTS语音合成
4.1 Edge TTS方案（免费高质量）
# core/tts.py
import edge_tts
import asyncio
import io
from typing import Optional
class TTSEngine:
"""基于Edge TTS的语音合成引擎"""
# 可用的中文语音
VOICES = {
"zh-female-warm": "zh-CN-XiaoxiaoNeural", # 温暖女声
"zh-female-sweet": "zh-CN-XiaoyiNeural", # 甜美女声
"zh-male-calm": "zh-CN-YunxiNeural", # 沉稳男声
"zh-male-lively": "zh-CN-YunjianNeural", # 活力男声
"en-female": "en-US-JennyNeural", # 英文女声
"en-male": "en-US-GuyNeural", # 英文男声
"ja-female": "ja-JP-NanamiNeural", # 日文女声
}
def __init__(self, voice: str = "zh-CN-XiaoxiaoNeural", rate: str = "+0%", volume: str = "+0%"):
"""
Args:
voice: 语音名称或预设别名
rate: 语速调节，如"+20%"、"-10%"
volume: 音量调节
"""
self.voice = self.VOICES.get(voice, voice)
self.rate = rate
self.volume = volume
async def synthesize(self, text: str) -> bytes:
"""
合成语音，返回WAV格式的字节数据
Args:
text: 要合成的文本
Returns:
MP3格式的音频字节数据
"""
communicate = edge_tts.Communicate(
text=text,
voice=self.voice,
rate=self.rate,
volume=self.volume,
)
audio_data = b""
async for chunk in communicate.stream():
if chunk["type"] == "audio":
audio_data += chunk["data"]
return audio_data
def synthesize_sync(self, text: str) -> bytes:
"""同步版本的语音合成"""
return asyncio.run(self.synthesize(text))
async def synthesize_to_file(self, text: str, output_path: str):
"""合成并保存到文件"""
communicate = edge_tts.Communicate(
text=text,
voice=self.voice,
rate=self.rate,
volume=self.volume,
)
await communicate.save(output_path)
@staticmethod
async def list_voices(language: str = "zh") -> list:
"""列出可用语音"""
voices = await edge_tts.list_voices()
return [v for v in voices if v["Locale"].startswith(language)]
# 使用示例
if __name__ == "__main__":
tts = TTSEngine(voice="zh-female-warm", rate="+10%")
audio = tts.synthesize_sync("你好，我是你的AI语音助手，有什么可以帮你的吗？")
with open("output.mp3", "wb") as f:
f.write(audio)
print(f"语音已保存，大小: {len(audio)} bytes")
4.2 流式TTS（首包延迟优化）
# core/streaming_tts.py
import edge_tts
import asyncio
from typing import AsyncGenerator
class StreamingTTSEngine:
"""流式语音合成，降低首包延迟"""
def __init__(self, voice: str = "zh-CN-XiaoxiaoNeural", rate: str = "+0%"):
self.voice = voice
self.rate = rate
async def stream_synthesize(self, text: str) -> AsyncGenerator[bytes, None]:
"""
流式合成语音，逐块yield音频数据
适合WebSocket场景：边生成文本边播放语音
"""
communicate = edge_tts.Communicate(
text=text,
voice=self.voice,
rate=self.rate,
)
async for chunk in communicate.stream():
if chunk["type"] == "audio":
yield chunk["data"]
async def synthesize_sentences(self, text: str) -> AsyncGenerator[bytes, None]:
"""
按句子分割，逐句合成
适合长文本：每句话单独合成，降低整体延迟
"""
sentences = self._split_sentences(text)
for sentence in sentences:
if sentence.strip():
communicate = edge_tts.Communicate(
text=sentence.strip(),
voice=self.voice,
rate=self.rate,
)
async for chunk in communicate.stream():
if chunk["type"] == "audio":
yield chunk["data"]
def _split_sentences(self, text: str) -> list:
"""将文本按句子分割"""
import re
# 支持中英文标点
sentences = re.split(r'([。！？；\.\!\?\;])', text)
result = []
for i in range(0, len(sentences) - 1, 2):
result.append(sentences[i] + sentences[i + 1])
if len(sentences) % 2 == 1 and sentences[-1]:
result.append(sentences[-1])
return result
4.3 CosyVoice方案（本地部署）
# 适合需要本地部署、低延迟的场景
# 需要先安装 cosyvoice: pip install cosyvoice
from cosyvoice import CosyVoice
import torchaudio
class CosyVoiceTTS:
"""基于CosyVoice的本地TTS"""
def __init__(self, model_path: str = "pretrained_models/CosyVoice-300M"):
self.model = CosyVoice(model_path)
def synthesize(
self,
text: str,
speaker: str = "中文女",
speed: float = 1.0,
) -> tuple:
"""
合成语音
Returns:
(audio_tensor, sample_rate)
"""
output = self.model.inference_sft(text, speaker, speed=speed)
return output["tts_speech"], 22050
def synthesize_with_clone(
self,
text: str,
reference_audio: str,
reference_text: str,
) -> tuple:
"""
声音克隆模式：用参考音频的声音说新文本
"""
output = self.model.inference_cross_lingual(
text, reference_audio, reference_text
)
return output["tts_speech"], 22050
五、实时语音交互（WebSocket）
5.1 WebSocket服务端
# transport/websocket_server.py
import asyncio
import websockets
import json
import numpy as np
from typing import Dict, Set
class VoiceAssistantServer:
"""语音助手WebSocket服务"""
def __init__(self, asr, llm, tts, session_manager):
self.asr = asr
self.llm = llm
self.tts = tts
self.sessions = session_manager
self.clients: Set = set()
async def handler(self, websocket, path=None):
"""处理WebSocket连接"""
session_id = str(id(websocket))
self.clients.add(websocket)
conversation = self.sessions.create_session(session_id)
print(f"Client connected: {session_id}")
try:
async for message in websocket:
await self._process_message(websocket, session_id, message)
except websockets.exceptions.ConnectionClosed:
print(f"Client disconnected: {session_id}")
finally:
self.clients.discard(websocket)
self.sessions.remove_session(session_id)
async def _process_message(self, websocket, session_id: str, raw_message):
"""处理客户端消息"""
if isinstance(raw_message, bytes):
# 音频数据
await self._handle_audio(websocket, session_id, raw_message)
else:
# JSON控制消息
data = json.loads(raw_message)
msg_type = data.get("type")
if msg_type == "config":
await self._handle_config(websocket, session_id, data)
elif msg_type == "text":
await self._handle_text(websocket, session_id, data)
elif msg_type == "end_of_speech":
await self._handle_eos(websocket, session_id)
async def _handle_audio(self, websocket, session_id: str, audio_bytes: bytes):
"""处理音频流数据"""
# 转换为numpy数组
audio_array = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32) / 32768.0
session = self.sessions.get_session(session_id)
session["audio_buffer"].extend(audio_array)
# 通知客户端正在处理
await websocket.send(json.dumps({
"type": "status",
"status": "listening",
}))
async def _handle_eos(self, websocket, session_id: str):
"""处理语音结束信号"""
session = self.sessions.get_session(session_id)
audio_buffer = np.array(session["audio_buffer"], dtype=np.float32)
session["audio_buffer"] = []
if len(audio_buffer) < 16000 * 0.5: # 少于0.5秒
await websocket.send(json.dumps({
"type": "status",
"status": "too_short",
}))
return
# ASR识别
await websocket.send(json.dumps({
"type": "status",
"status": "processing",
}))
asr_result = self.asr.transcribe(audio_buffer)
user_text = asr_result["text"]
if not user_text:
await websocket.send(json.dumps({
"type": "status",
"status": "no_speech",
}))
return
# 发送识别结果
await websocket.send(json.dumps({
"type": "asr_result",
"text": user_text,
"language": asr_result["language"],
}))
# LLM生成回复
conversation = self.sessions.get_conversation(session_id)
reply = self.llm.chat(user_text, conversation)
# 发送文本回复
await websocket.send(json.dumps({
"type": "llm_result",
"text": reply,
}))
# TTS合成并发送
await websocket.send(json.dumps({
"type": "status",
"status": "speaking",
}))
audio_data = await self.tts.synthesize(reply)
# 分块发送音频（每块4KB）
chunk_size = 4096
for i in range(0, len(audio_data), chunk_size):
chunk = audio_data[i:i + chunk_size]
await websocket.send(chunk)
# 发送播放结束信号
await websocket.send(json.dumps({
"type": "audio_end",
}))
async def _handle_text(self, websocket, session_id: str, data: dict):
"""处理文本输入（跳过ASR）"""
user_text = data["text"]
conversation = self.sessions.get_conversation(session_id)
reply = self.llm.chat(user_text, conversation)
await websocket.send(json.dumps({
"type": "llm_result",
"text": reply,
}))
audio_data = await self.tts.synthesize(reply)
await websocket.send(audio_data)
await websocket.send(json.dumps({"type": "audio_end"}))
def start(self, host: str = "0.0.0.0", port: int = 8765):
"""启动服务"""
print(f"Voice Assistant Server starting on ws://{host}:{port}")
start_server = websockets.serve(self.handler, host, port)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
5.2 客户端实现
# client/voice_client.py
import asyncio
import websockets
import json
import pyaudio
import threading
class VoiceClient:
"""语音助手客户端"""
def __init__(self, server_url: str = "ws://localhost:8765"):
self.server_url = server_url
self.audio = pyaudio.PyAudio()
self.is_recording = False
self.ws = None
async def connect(self):
"""连接到服务端"""
self.ws = await websockets.connect(self.server_url)
print("Connected to voice assistant")
# 启动接收和录音
await asyncio.gather(
self._receive_loop(),
self._record_loop(),
)
async def _record_loop(self):
"""录音循环"""
stream = self.audio.open(
format=pyaudio.paInt16,
channels=1,
rate=16000,
input=True,
frames_per_buffer=1024,
)
print("🎤 按回车开始录音...")
input()
print("🔴 录音中... 按回车停止")
self.is_recording = True
# 在后台线程中等待停止信号
def wait_for_stop():
input()
self.is_recording = False
threading.Thread(target=wait_for_stop, daemon=True).start()
while self.is_recording:
data = stream.read(1024, exception_on_overflow=False)
await self.ws.send(data)
await asyncio.sleep(0.01)
stream.stop_stream()
stream.close()
# 发送语音结束信号
await self.ws.send(json.dumps({"type": "end_of_speech"}))
async def _receive_loop(self):
"""接收服务端消息"""
async for message in self.ws:
if isinstance(message, bytes):
# 音频数据，直接播放
self._play_audio(message)
else:
data = json.loads(message)
self._handle_message(data)
def _handle_message(self, data: dict):
"""处理控制消息"""
msg_type = data.get("type")
if msg_type == "status":
status = data["status"]
if status == "listening":
print("👂 正在聆听...")
elif status == "processing":
print("🧠 正在思考...")
elif status == "speaking":
print("🔊 正在回答...")
elif msg_type == "asr_result":
print(f"📝 你说: {data['text']}")
elif msg_type == "llm_result":
print(f"🤖 回复: {data['text']}")
elif msg_type == "audio_end":
print("✅ 回答完毕")
print("\n🎤 按回车开始下一轮对话...")
# 重新开始录音
asyncio.ensure_future(self._record_loop())
def _play_audio(self, audio_data: bytes):
"""播放音频"""
stream = self.audio.open(
format=pyaudio.paInt16,
channels=1,
rate=24000,
output=True,
)
stream.write(audio_data)
stream.stop_stream()
stream.close()
# 运行客户端
if __name__ == "__main__":
client = VoiceClient("ws://localhost:8765")
asyncio.run(client.connect())
六、多轮对话管理
6.1 会话管理器
# session/manager.py
import uuid
import time
from typing import Dict, Optional
from dataclasses import dataclass, field
from core.llm import ConversationState
@dataclass
class Session:
"""会话数据"""
session_id: str
conversation: ConversationState
created_at: float = field(default_factory=time.time)
last_active: float = field(default_factory=time.time)
audio_buffer: list = field(default_factory=list)
metadata: Dict = field(default_factory=dict)
class SessionManager:
"""会话管理器，支持多用户并发"""
def __init__(self, max_sessions: int = 100, timeout: int = 1800):
"""
Args:
max_sessions: 最大并发会话数
timeout: 会话超时时间（秒）
"""
self.sessions: Dict[str, Session] = {}
self.max_sessions = max_sessions
self.timeout = timeout
def create_session(self, session_id: Optional[str] = None) -> Session:
"""创建新会话"""
self._cleanup_expired()
if len(self.sessions) >= self.max_sessions:
self._remove_oldest()
sid = session_id or str(uuid.uuid4())
session = Session(
session_id=sid,
conversation=ConversationState(),
)
self.sessions[sid] = session
return session
def get_session(self, session_id: str) -> Optional[Session]:
"""获取会话"""
session = self.sessions.get(session_id)
if session:
session.last_active = time.time()
return session
def get_conversation(self, session_id: str) -> ConversationState:
"""获取对话状态"""
session = self.get_session(session_id)
if not session:
session = self.create_session(session_id)
return session.conversation
def remove_session(self, session_id: str):
"""移除会话"""
self.sessions.pop(session_id, None)
def _cleanup_expired(self):
"""清理过期会话"""
now = time.time()
expired = [
sid for sid, session in self.sessions.items()
if now - session.last_active > self.timeout
]
for sid in expired:
self.sessions.pop(sid)
def _remove_oldest(self):
"""移除最旧的会话"""
if self.sessions:
oldest = min(self.sessions, key=lambda s: self.sessions[s].last_active)
self.sessions.pop(oldest)
6.2 上下文记忆增强
# core/memory.py
from typing import List, Dict
import json
class ConversationMemory:
"""对话记忆管理，支持长期记忆"""
def __init__(self, max_short_term: int = 10, max_long_term: int = 50):
self.short_term: List[Dict] = [] # 短期记忆（当前对话）
self.long_term: List[Dict] = [] # 长期记忆（跨会话）
self.entity_memory: Dict[str, str] = {} # 实体记忆
self.max_short_term = max_short_term
self.max_long_term = max_long_term
def add_turn(self, role: str, content: str):
"""添加一轮对话"""
self.short_term.append({
"role": role,
"content": content,
"timestamp": time.time(),
})
# 超出容量时转移到长期记忆
if len(self.short_term) > self.max_short_term:
overflow = self.short_term[:len(self.short_term) - self.max_short_term]
self.short_term = self.short_term[-self.max_short_term:]
self.long_term.extend(overflow)
# 长期记忆也有上限
if len(self.long_term) > self.max_long_term:
self.long_term = self.long_term[-self.max_long_term:]
def extract_entities(self, text: str):
"""从文本中提取关键实体（简化版）"""
# 实际项目中使用NER模型
import re
# 提取可能的名字
names = re.findall(r'我叫(\w+)|我是(\w+)|我的名字是(\w+)', text)
for match in names:
name = next(m for m in match if m)
self.entity_memory["user_name"] = name
def get_context_messages(self) -> List[Dict]:
"""获取用于LLM的上下文消息"""
messages = []
# 添加长期记忆摘要
if self.long_term:
summary = self._summarize_history(self.long_term[-10:])
messages.append({
"role": "system",
"content": f"之前的对话摘要：{summary}",
})
# 添加实体记忆
if self.entity_memory:
entities = ", ".join(f"{k}是{v}" for k, v in self.entity_memory.items())
messages.append({
"role": "system",
"content": f"已知用户信息：{entities}",
})
# 添加短期记忆
for turn in self.short_term:
messages.append({
"role": turn["role"],
"content": turn["content"],
})
return messages
def _summarize_history(self, history: List[Dict]) -> str:
"""摘要历史对话"""
# 简化版：实际项目中用LLM生成摘要
topics = set()
for turn in history:
topics.add(turn["content"][:50])
return "讨论了：" + "、".join(list(topics)[:5])
七、语音唤醒与VAD
7.1 VAD语音活动检测
# core/vad.py
import torch
import numpy as np
class SileroVAD:
"""基于Silero的语音活动检测"""
def __init__(self, threshold: float = 0.5, sample_rate: int = 16000):
self.model, _ = torch.hub.load(
repo_or_dir='snakers4/silero-vad',
model='silero_vad',
force_reload=False,
)
self.threshold = threshold
self.sample_rate = sample_rate
self.model.eval()
def is_speech(self, audio_chunk: np.ndarray) -> bool:
"""
检测音频块是否包含语音
Args:
audio_chunk: 音频数据，float32格式
Returns:
True表示检测到语音
"""
if len(audio_chunk) < 512:
return False
tensor = torch.from_numpy(audio_chunk).float()
if len(tensor.shape) == 1:
tensor = tensor.unsqueeze(0)
with torch.no_grad():
prob = self.model(tensor, self.sample_rate).item()
return prob > self.threshold
def process_stream(self, audio_stream, chunk_size: int = 512):
"""
处理音频流，yield语音片段
用法:
for speech_chunk in vad.process_stream(audio_stream):
process_speech(speech_chunk)
"""
speech_buffer = []
silence_count = 0
max_silence = 15 # 连续静音块数（约0.5秒）
for chunk in audio_stream:
if self.is_speech(chunk):
speech_buffer.append(chunk)
silence_count = 0
else:
if speech_buffer:
silence_count += 1
speech_buffer.append(chunk)
if silence_count >= max_silence:
# 语音结束
yield np.concatenate(speech_buffer)
speech_buffer = []
silence_count = 0
7.2 唤醒词检测
# core/wake_word.py
import pvporcupine
import pyaudio
import struct
class WakeWordDetector:
"""基于Porcupine的唤醒词检测"""
# 内置唤醒词
BUILTIN_KEYWORDS = [
"hey google", "alexa", "hey siri",
"jarvis", "computer", "picovoice",
]
def __init__(
self,
access_key: str,
keywords: list = None,
sensitivities: list = None,
):
"""
Args:
access_key: Picovoice访问密钥
keywords: 唤醒词列表
sensitivities: 灵敏度列表（0-1）
"""
keywords = keywords or ["computer"]
sensitivities = sensitivities or [0.5] * len(keywords)
self.porcupine = pvporcupine.create(
access_key=access_key,
keywords=keywords,
sensitivities=sensitivities,
)
self.audio = pyaudio.PyAudio()
self.stream = self.audio.open(
rate=self.porcupine.sample_rate,
channels=1,
format=pyaudio.paInt16,
input=True,
frames_per_buffer=self.porcupine.frame_length,
)
def listen(self, callback=None):
"""
持续监听唤醒词
Args:
callback: 唤醒时的回调函数
"""
print("🎧 等待唤醒词...")
try:
while True:
pcm = self.stream.read(self.porcupine.frame_length)
pcm = struct.unpack_from("h" * self.porcupine.frame_length, pcm)
keyword_index = self.porcupine.process(pcm)
if keyword_index >= 0:
print(f"✨ 检测到唤醒词！(index={keyword_index})")
if callback:
callback(keyword_index)
except KeyboardInterrupt:
print("停止监听")
finally:
self.cleanup()
def cleanup(self):
"""释放资源"""
self.stream.close()
self.audio.terminate()
self.porcupine.delete()
# 简易唤醒方案（不需要Porcupine）
class SimpleWakeWord:
"""基于关键词匹配的简易唤醒检测"""
def __init__(self, asr_engine, wake_phrase: str = "你好助手"):
self.asr = asr_engine
self.wake_phrase = wake_phrase.lower()
def check(self, audio_data: np.ndarray) -> bool:
"""检查音频中是否包含唤醒词"""
result = self.asr.transcribe(audio_data, language="zh")
text = result["text"].lower()
return self.wake_phrase in text
八、多语言支持
8.1 语言检测与路由
# core/multilingual.py
from typing import Optional
class MultilingualRouter:
"""多语言路由器：自动检测语言并选择对应的ASR/TTS"""
# 语言代码映射
LANG_MAP = {
"zh": {"name": "中文", "tts_voice": "zh-CN-XiaoxiaoNeural", "asr_lang": "zh"},
"en": {"name": "English", "tts_voice": "en-US-JennyNeural", "asr_lang": "en"},
"ja": {"name": "日本語", "tts_voice": "ja-JP-NanamiNeural", "asr_lang": "ja"},
"ko": {"name": "한국어", "tts_voice": "ko-KR-SunHiNeural", "asr_lang": "ko"},
"fr": {"name": "Français", "tts_voice": "fr-FR-DeniseNeural", "asr_lang": "fr"},
"de": {"name": "Deutsch", "tts_voice": "de-DE-KatjaNeural", "asr_lang": "de"},
"es": {"name": "Español", "tts_voice": "es-ES-ElviraNeural", "asr_lang": "es"},
}
def __init__(self, default_lang: str = "zh"):
self.current_lang = default_lang
self.default_lang = default_lang
def detect_and_route(self, asr_result: dict) -> dict:
"""
根据ASR检测到的语言，返回对应的TTS配置
Args:
asr_result: ASR识别结果，包含language字段
Returns:
包含语言配置的字典
"""
detected_lang = asr_result.get("language", self.default_lang)
# 语言切换逻辑
if detected_lang in self.LANG_MAP:
self.current_lang = detected_lang
config = self.LANG_MAP.get(self.current_lang, self.LANG_MAP[self.default_lang])
return {
"language": self.current_lang,
"language_name": config["name"],
"tts_voice": config["tts_voice"],
"asr_language": config["asr_lang"],
"system_prompt_suffix": self._get_lang_prompt(self.current_lang),
}
def _get_lang_prompt(self, lang: str) -> str:
"""获取对应语言的系统提示词后缀"""
prompts = {
"zh": "请用中文回答。",
"en": "Please respond in English.",
"ja": "日本語で回答してください。",
"ko": "한국어로 대답해 주세요.",
"fr": "Veuillez répondre en français.",
"de": "Bitte antworten Sie auf Deutsch.",
"es": "Por favor responda en español.",
}
return prompts.get(lang, prompts[self.default_lang])
def set_language(self, lang: str):
"""手动设置语言"""
if lang in self.LANG_MAP:
self.current_lang = lang
8.2 多语言对话示例
# 多语言使用示例
router = MultilingualRouter(default_lang="zh")
# 用户说中文
asr_result = {"text": "今天天气怎么样？", "language": "zh"}
config = router.detect_and_route(asr_result)
# → {"language": "zh", "tts_voice": "zh-CN-XiaoxiaoNeural", ...}
# 用户切换到英文
asr_result = {"text": "What's the weather like today?", "language": "en"}
config = router.detect_and_route(asr_result)
# → {"language": "en", "tts_voice": "en-US-JennyNeural", ...}
# 在LLM提示词中加入语言指令
system_prompt = f"""你是一个多语言AI助手。
{config['system_prompt_suffix']}
保持简洁口语化。"""
九、智能家居/车载场景实战
9.1 智能家居语音控制
# scenarios/smart_home.py
from typing import Dict, List
from dataclasses import dataclass
@dataclass
class Device:
"""智能家居设备"""
name: str
device_type: str # light, ac, tv, speaker, curtain, etc.
room: str
state: Dict # 当前状态
class SmartHomeController:
"""智能家居语音控制器"""
def __init__(self):
self.devices: Dict[str, Device] = {}
self._init_devices()
def _init_devices(self):
"""初始化示例设备"""
devices_config = [
{"name": "客厅灯", "type": "light", "room": "客厅", "state": {"on": True, "brightness": 80, "color": "暖白"}},
{"name": "卧室灯", "type": "light", "room": "卧室", "state": {"on": False, "brightness": 50, "color": "暖白"}},
{"name": "空调", "type": "ac", "room": "客厅", "state": {"on": True, "temp": 26, "mode": "制冷"}},
{"name": "电视", "type": "tv", "room": "客厅", "state": {"on": False, "channel": 1, "volume": 30}},
{"name": "窗帘", "type": "curtain", "room": "卧室", "state": {"open": True, "position": 100}},
]
for cfg in devices_config:
self.devices[cfg["name"]] = Device(
name=cfg["name"],
device_type=cfg["type"],
room=cfg["room"],
state=cfg["state"],
)
def get_tools_definition(self) -> List[Dict]:
"""获取工具定义（用于LLM Function Calling）"""
return [
{
"type": "function",
"function": {
"name": "control_device",
"description": "控制智能家居设备",
"parameters": {
"type": "object",
"properties": {
"device_name": {
"type": "string",
"description": f"设备名称，可选：{', '.join(self.devices.keys())}",
},
"action": {
"type": "string",
"enum": ["turn_on", "turn_off", "set_brightness", "set_temp", "set_color"],
},
"value": {
"type": "string",
"description": "设置值（亮度0-100、温度16-30、颜色名称）",
},
},
"required": ["device_name", "action"],
},
},
},
{
"type": "function",
"function": {
"name": "query_device",
"description": "查询设备状态",
"parameters": {
"type": "object",
"properties": {
"device_name": {
"type": "string",
"description": f"设备名称，可选：{', '.join(self.devices.keys())}",
},
},
"required": ["device_name"],
},
},
},
]
def execute(self, tool_name: str, arguments: dict) -> str:
"""执行设备控制"""
if tool_name == "control_device":
return self._control_device(arguments)
elif tool_name == "query_device":
return self._query_device(arguments)
return "未知操作"
def _control_device(self, args: dict) -> str:
"""控制设备"""
name = args["device_name"]
action = args["action"]
value = args.get("value")
device = self.devices.get(name)
if not device:
return f"未找到设备：{name}"
if action == "turn_on":
device.state["on"] = True
return f"已打开{name}"
elif action == "turn_off":
device.state["on"] = False
return f"已关闭{name}"
elif action == "set_brightness" and device.device_type == "light":
device.state["brightness"] = int(value)
return f"已将{name}亮度设置为{value}%"
elif action == "set_temp" and device.device_type == "ac":
device.state["temp"] = int(value)
return f"已将{name}温度设置为{value}度"
return f"操作完成"
def _query_device(self, args: dict) -> str:
"""查询设备状态"""
name = args["device_name"]
device = self.devices.get(name)
if not device:
return f"未找到设备：{name}"
status_parts = []
if "on" in device.state:
status_parts.append("开启" if device.state["on"] else "关闭")
if "brightness" in device.state:
status_parts.append(f"亮度{device.state['brightness']}%")
if "temp" in device.state:
status_parts.append(f"温度{device.state['temp']}度")
if "open" in device.state:
status_parts.append("已打开" if device.state["open"] else "已关闭")
return f"{name}当前状态：{'，'.join(status_parts)}"
# 使用示例
controller = SmartHomeController()
# 模拟LLM工具调用
result = controller.execute("control_device", {
"device_name": "客厅灯",
"action": "set_brightness",
"value": "60",
})
print(result) # "已将客厅灯亮度设置为60%"
9.2 车载语音助手
# scenarios/car_assistant.py
CAR_SYSTEM_PROMPT = """你是一个车载AI语音助手。你的职责包括：
1. 导航指引：帮助驾驶员查找目的地、规划路线
2. 车辆控制：调节空调、音乐、车窗等
3. 信息查询：天气、新闻、日程提醒
4. 安全提醒：疲劳驾驶提醒、超速提醒
安全规则：
- 驾驶时回复必须极简（20字以内）
- 不在驾驶时可以详细回复
- 涉及安全问题时优先提醒
- 紧急情况建议靠边停车"""
class CarVoiceAssistant:
"""车载语音助手"""
def __init__(self, llm_engine, tts_engine):
self.llm = llm_engine
self.tts = tts_engine
self.is_driving = True # 驾驶状态
self.speed = 0 # 车速
async def handle_command(self, user_text: str) -> str:
"""处理车载语音命令"""
# 根据驾驶状态调整回复策略
if self.is_driving:
max_tokens = 80 # 驾驶时限制回复长度
else:
max_tokens = 300
# 特殊命令处理
if "导航" in user_text or "怎么走" in user_text:
return await self._handle_navigation(user_text)
if "空调" in user_text or "温度" in user_text:
return await self._handle_climate(user_text)
if "音乐" in user_text or "播放" in user_text:
return await self._handle_media(user_text)
# 通用对话
response = self.llm.chat(
user_text,
self.llm.create_conversation(system_prompt=CAR_SYSTEM_PROMPT),
)
return response
async def _handle_navigation(self, text: str) -> str:
"""处理导航请求"""
# 实际项目中对接地图API
return "已为您规划路线，预计到达时间30分钟"
async def _handle_climate(self, text: str) -> str:
"""处理空调控制"""
if "调高" in text or "热" in text:
return "已将温度调高至27度"
elif "调低" in text or "冷" in text:
return "已将温度调低至24度"
return "请问您想调节到多少度？"
async def _handle_media(self, text: str) -> str:
"""处理媒体控制"""
return "正在为您播放音乐"
十、性能优化与部署
10.1 延迟优化策略
# 优化策略一览
optimization_strategies = """
1. ASR优化：
- 使用较小的Whisper模型（base而非large）
- 流式识别，边听边转录
- 指定语言避免检测耗时
2. LLM优化：
- 使用流式输出（stream=True）
- 控制max_tokens避免过长回复
- 使用更快的模型（GPT-4o-mini而非GPT-4o）
- 缓存系统提示词的tokenization
3. TTS优化：
- 流式合成，边生成边播放
- 按句子分割，逐句合成
- 预合成常用回复（如"好的"、"明白了"）
4. 网络优化：
- 使用WebSocket持久连接
- 音频数据压缩（opus编码）
- 就近部署，减少网络延迟
"""
10.2 完整的优化Pipeline
# core/optimized_pipeline.py
import asyncio
import time
class OptimizedVoicePipeline:
"""优化的语音处理流水线"""
def __init__(self, asr, llm, tts, vad):
self.asr = asr
self.llm = llm
self.tts = tts
self.vad = vad
# 预缓存常用回复
self.cache = {}
self._precache_common_responses()
def _precache_common_responses(self):
"""预缓存常用短回复"""
common_phrases = [
"好的", "明白了", "请稍等", "没问题",
"已为您完成", "请问还有什么需要帮助的吗？",
]
for phrase in common_phrases:
self.cache[phrase] = self.tts.synthesize_sync(phrase)
async def process_audio_optimized(self, audio_data, conversation):
"""
优化的音频处理流程
关键优化点：
1. ASR和VAD并行
2. LLM流式输出 + TTS逐句合成
3. 缓存常用回复
"""
# 阶段1：ASR识别
t0 = time.time()
asr_result = self.asr.transcribe(audio_data)
t_asr = time.time() - t0
user_text = asr_result["text"]
if not user_text:
return None
# 检查缓存
if user_text in self.cache:
return {
"text": user_text,
"reply": user_text,
"audio": self.cache[user_text],
"latency": {"asr": t_asr, "llm": 0, "tts": 0},
}
# 阶段2：LLM流式生成
t1 = time.time()
reply = self.llm.chat(user_text, conversation, stream=True)
t_llm = time.time() - t1
# 阶段3：流式TTS合成
t2 = time.time()
# 按句子分割，边合成边发送
sentences = self._split_sentences(reply)
audio_chunks = []
for sentence in sentences:
if sentence.strip():
chunk = await self.tts.synthesize(sentence.strip())
audio_chunks.append(chunk)
# 可以在这里yield chunk给客户端
full_audio = b"".join(audio_chunks)
t_tts = time.time() - t2
return {
"text": user_text,
"reply": reply,
"audio": full_audio,
"latency": {
"asr": round(t_asr * 1000),
"llm": round(t_llm * 1000),
"tts": round(t_tts * 1000),
"total": round((t_asr + t_llm + t_tts) * 1000),
},
}
def _split_sentences(self, text):
import re
return re.split(r'(?<=[。！？；\.\!\?\;])', text)
10.3 Docker部署
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
portaudio19-dev \
ffmpeg \
&& rm -rf /var/lib/apt/lists/*
# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 下载Whisper模型
RUN python -c "import whisper; whisper.load_model('base')"
EXPOSE 8765
CMD ["python", "main.py"]
# docker-compose.yml
version: '3.8'
services:
voice-assistant:
build: .
ports:
- "8765:8765"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- ASR_MODEL=base
- TTS_VOICE=zh-CN-XiaoxiaoNeural
- LLM_MODEL=gpt-4o-mini
volumes:
- ./data:/app/data
restart: unless-stopped
deploy:
resources:
limits:
memory: 4G
cpus: '2.0'
10.4 性能基准
| 环节 | 目标延迟 | 优化手段 |
|---|---|---|
| VAD检测 | < 10ms | Silero模型，GPU加速 |
| ASR识别 | < 1s | Whisper base，指定语言 |
| LLM回复 | < 2s（首token） | 流式输出，GPT-4o-mini |
| TTS合成 | < 500ms（首包） | 流式合成，按句分割 |
| 端到端 | < 4s | 全链路并行优化 |
10.5 监控与日志
# utils/monitor.py
import time
import logging
from functools import wraps
from dataclasses import dataclass, field
from collections import defaultdict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("voice_assistant")
@dataclass
class Metrics:
"""性能指标收集"""
latency_history: dict = field(default_factory=lambda: defaultdict(list))
error_count: dict = field(default_factory=lambda: defaultdict(int))
request_count: int = 0
def record_latency(self, stage: str, latency_ms: float):
"""记录延迟"""
self.latency_history[stage].append(latency_ms)
# 只保留最近1000条
if len(self.latency_history[stage]) > 1000:
self.latency_history[stage] = self.latency_history[stage][-1000:]
def record_error(self, stage: str):
"""记录错误"""
self.error_count[stage] += 1
def get_stats(self) -> dict:
"""获取统计信息"""
stats = {}
for stage, latencies in self.latency_history.items():
if latencies:
stats[stage] = {
"avg_ms": round(sum(latencies) / len(latencies), 1),
"p50_ms": round(sorted(latencies)[len(latencies) // 2], 1),
"p95_ms": round(sorted(latencies)[int(len(latencies) * 0.95)], 1),
"p99_ms": round(sorted(latencies)[int(len(latencies) * 0.99)], 1),
"count": len(latencies),
}
stats["errors"] = dict(self.error_count)
stats["total_requests"] = self.request_count
return stats
# 全局指标实例
metrics = Metrics()
def track_latency(stage: str):
"""延迟追踪装饰器"""
def decorator(func):
@wraps(func)
async def async_wrapper(*args, **kwargs):
start = time.time()
try:
result = await func(*args, **kwargs)
latency = (time.time() - start) * 1000
metrics.record_latency(stage, latency)
logger.info(f"[{stage}] {latency:.1f}ms")
return result
except Exception as e:
metrics.record_error(stage)
raise
@wraps(func)
def sync_wrapper(*args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
latency = (time.time() - start) * 1000
metrics.record_latency(stage, latency)
return result
except Exception as e:
metrics.record_error(stage)
raise
return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
return decorator
总结
本教程从架构设计到生产部署，完整覆盖了AI语音助手开发的全链路。核心要点回顾：
- 架构清晰：ASR → LLM → TTS 三段式架构，模块解耦
- 流式优先：无论是ASR、LLM还是TTS，都采用流式处理降低延迟
- 多轮管理：通过SessionManager和ConversationMemory实现上下文保持
- 场景适配：智能家居和车载场景有不同的交互策略
- 性能优化：预缓存、并行处理、按句合成等手段将端到端延迟控制在4秒内
随着语音技术的快速发展，建议持续关注以下方向：
- 端到端语音模型（如GPT-4o的语音模式）可能简化整体架构
- 本地小模型（如Qwen2-Audio）提供离线能力
- 情感合成让语音助手更有"人味"
- 多模态融合（语音+视觉）拓展应用场景
📅 最后更新：2025年
📝 作者：AI教程系列
🔗 相关资源：OpenAI Whisper | Edge TTS | Silero VAD
