AI实时语音交互应用开发完全教程
从语音识别到端到端Speech-to-Speech，全面掌握AI语音交互核心技术
教程简介
随着GPT-4o Realtime API、Gemini Live、Qwen2.5-Omni等多模态大模型的发布，AI实时语音交互正从实验室走向大规模商用。本教程将系统性地讲解AI语音交互的完整技术栈，从传统的级联式架构（VAD→ASR→LLM→TTS）到前沿的端到端Speech-to-Speech方案，帮助开发者构建低延迟、高自然度的语音AI系统。
你将学到：
- 语音AI技术架构的演进与选型（级联式 vs 端到端）
- 主流语音大模型的能力对比与实战接入
- 语音识别（ASR）与语音合成（TTS）核心技术
- 实时流式语音处理（WebSocket/WebRTC）
- 语音Agent架构设计与工程实现
- 低延迟优化、语音RAG、场景落地等高级话题
第一章：语音AI技术架构概述
1.1 级联式架构（Cascaded Pipeline）
传统的语音交互系统采用级联式架构，将语音处理分为多个独立模块串联执行：
用户语音 → VAD → ASR → LLM → TTS → 播放语音
核心模块：
- VAD（Voice Activity Detection）：检测用户是否在说话，决定何时开始/停止录音
- ASR（Automatic Speech Recognition）：将语音转为文本
- LLM（Large Language Model）：理解文本含义并生成回复
- TTS（Text-to-Speech）：将文本回复转为语音输出
优势： 各模块独立优化，技术成熟度高，可替换性强 劣势： 端到端延迟较高（通常1-3秒），信息在文本转换中丢失韵律、情感等副语言信息
1.2 端到端架构（End-to-End Speech-to-Speech）
新一代端到端架构直接在语音模态上进行理解和生成：
用户语音 → Speech-to-Speech Model → AI语音回复
代表方案： GPT-4o Realtime API、Gemini Live API、Qwen2.5-Omni
优势： 延迟极低（<500ms），保留语音中的情感和韵律信息，支持更自然的对话 劣势： 技术较新，可控性相对较弱，成本较高
1.3 混合架构实践
在实际工程中，常采用混合架构平衡延迟与可控性：
# 混合架构示意
class HybridVoiceAgent:
def __init__(self):
self.vad = SileroVAD()
self.asr = WhisperASR()
self.llm = GPT4oRealtime() # 或传统LLM
self.tts = CosyVoiceTTS()
self.mode = "cascade" # cascade / e2e / hybrid
async def process(self, audio_stream):
if self.mode == "e2e":
# 端到端模式：直接处理语音流
return await self.llm.stream_audio(audio_stream)
elif self.mode == "cascade":
# 级联模式：分步处理
text = await self.asr.transcribe(audio_stream)
response = await self.llm.generate(text)
audio = await self.tts.synthesize(response)
return audio
else:
# 混合模式：简单问题用端到端，复杂问题用级联
if self.is_simple_query(audio_stream):
return await self.llm.stream_audio(audio_stream)
else:
text = await self.asr.transcribe(audio_stream)
response = await self.llm.generate(text)
return await self.tts.synthesize(response)
第二章：主流语音大模型对比与实战
2.1 GPT-4o Realtime API
OpenAI的GPT-4o Realtime API是目前最成熟的端到端语音交互方案，支持双向音频流。
核心特性：
- 原生语音理解与生成，无需ASR/TTS
- 支持WebSocket实时双向通信
- 支持function calling（工具调用）
- 支持多种语音音色（alloy, echo, shimmer等）
- 典型延迟300-500ms
接入示例：
import asyncio
import websockets
import json
import base64
class GPT4oRealtimeClient:
def __init__(self, api_key):
self.api_key = api_key
self.ws_url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
self.ws = None
async def connect(self):
self.ws = await websockets.connect(
self.ws_url,
extra_headers={
"Authorization": f"Bearer {self.api_key}",
"OpenAI-Beta": "realtime=v1"
}
)
# 配置session
await self.ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"instructions": "你是一个友好的AI助手，用中文回答问题。",
"voice": "alloy",
"input_audio_format": "pcm16",
"output_audio_format": "pcm16",
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500
}
}
}))
async def send_audio(self, audio_bytes: bytes):
"""发送音频数据（PCM16格式）"""
audio_b64 = base64.b64encode(audio_bytes).decode()
await self.ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": audio_b64
}))
async def listen(self):
"""监听服务端响应"""
async for message in self.ws:
event = json.loads(message)
if event["type"] == "response.audio.delta":
# 收到音频片段，实时播放
audio_chunk = base64.b64decode(event["delta"])
yield audio_chunk
elif event["type"] == "response.text.delta":
# 收到文本片段（用于显示字幕）
print(event["delta"], end="", flush=True)
elif event["type"] == "response.done":
print("\n[回复完成]")
elif event["type"] == "error":
print(f"[错误] {event['error']['message']}")
# 使用示例
async def main():
client = GPT4oRealtimeClient("your-api-key")
await client.connect()
# 发送音频
with open("user_input.pcm", "rb") as f:
while chunk := f.read(4096):
await client.send_audio(chunk)
# 接收回复
async for audio_chunk in client.listen():
# 将audio_chunk写入播放缓冲区
pass
asyncio.run(main())
2.2 Gemini Live API
Google的Gemini Live API提供了类似的实时语音交互能力，特点是支持视频输入和更长的上下文。
import asyncio
from google import genai
class GeminiLiveClient:
def __init__(self, api_key):
self.client = genai.Client(api_key=api_key)
async def start_session(self):
config = {
"response_modalities": ["AUDIO"],
"system_instruction": "你是一个知识渊博的AI助手。",
"speech_config": {
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Aoede"
}
}
}
}
async with self.client.aio.live.connect(
model="gemini-2.0-flash-live-001",
config=config
) as session:
# 发送音频
await session.send(input={
"audio": audio_bytes # 16kHz PCM16
}, end_of_turn=True)
# 接收回复
async for response in session.receive():
if response.audio:
yield response.audio.data
2.3 Qwen2.5-Omni
阿里巴巴的Qwen2.5-Omni是开源的端到端多模态模型，支持本地部署。
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
import soundfile as sf
class QwenOmniClient:
def __init__(self, model_path="Qwen/Qwen2.5-Omni-7B"):
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
self.tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=True
)
def chat_with_audio(self, audio_path: str, text_prompt: str = ""):
"""输入音频+文本，获取文本+语音回复"""
messages = [
{
"role": "system",
"content": "你是一个有帮助的AI助手。"
},
{
"role": "user",
"content": [
{"type": "audio", "audio_url": audio_path},
{"type": "text", "text": text_prompt or "请回答音频中的问题。"}
]
}
]
text, audio = self.model.chat(
messages=messages,
tokenizer=self.tokenizer
)
# 保存回复音频
sf.write("response.wav", audio, samplerate=24000)
return text, audio
2.4 模型选型对比
| 特性 | GPT-4o Realtime | Gemini Live | Qwen2.5-Omni |
|---|---|---|---|
| 部署方式 | 云端API | 云端API | 本地/云端 |
| 延迟 | 300-500ms | 400-600ms | 500-1000ms |
| 多语言 | 优秀 | 优秀 | 良好 |
| Function Calling | 支持 | 支持 | 有限 |
| 视频输入 | 不支持 | 支持 | 支持 |
| 成本 | 较高 | 中等 | 可自控 |
| 中文能力 | 优秀 | 良好 | 优秀 |
| 开源 | 否 | 否 | 是 |
第三章：语音识别（ASR）核心技术
3.1 Whisper — OpenAI开源语音识别
Whisper是OpenAI开源的通用语音识别模型，支持99种语言，是目前最流行的开源ASR方案。
import whisper
import numpy as np
class WhisperASR:
def __init__(self, model_size="medium"):
"""
model_size: tiny, base, small, medium, large, turbo
中文推荐medium或large-v3
"""
self.model = whisper.load_model(model_size)
def transcribe(self, audio_path: str, language: str = "zh") -> dict:
"""完整转录"""
result = self.model.transcribe(
audio_path,
language=language,
task="transcribe",
fp16=False # CPU环境设为False
)
return {
"text": result["text"],
"segments": result["segments"],
"language": result["language"]
}
def transcribe_streaming(self, audio_chunks, language="zh"):
"""流式转录（简化实现）"""
buffer = np.array([], dtype=np.float32)
for chunk in audio_chunks:
buffer = np.append(buffer, chunk)
# 每积累2秒音频处理一次
if len(buffer) >= 16000 * 2:
result = self.model.transcribe(
buffer,
language=language,
fp16=False
)
yield result["text"]
buffer = np.array([], dtype=np.float32)
# 处理剩余音频
if len(buffer) > 0:
result = self.model.transcribe(buffer, language=language, fp16=False)
yield result["text"]
# 使用示例
asr = WhisperASR(model_size="medium")
result = asr.transcribe("meeting_recording.wav")
print(f"识别结果: {result['text']}")
3.2 Paraformer — 阿里达摩院高效ASR
Paraformer是阿里达摩院提出的非自回归语音识别模型，推理速度比Whisper快10倍以上。
# 使用FunASR框架（集成了Paraformer）
from funasr import AutoModel
class ParaformerASR:
def __init__(self):
# 加载Paraformer-large模型，支持标点恢复和时间戳
self.model = AutoModel(
model="paraformer-zh",
vad_model="fsmn-vad", # 语音活动检测
punc_model="ct-punc", # 标点恢复
spk_model="cam++" # 说话人识别
)
def transcribe(self, audio_path: str) -> dict:
result = self.model.generate(input=audio_path)
return {
"text": result[0]["text"],
"timestamp": result[0].get("timestamp", []),
"sentences": result[0].get("sentence_info", [])
}
def transcribe_with_speaker(self, audio_path: str) -> list:
"""带说话人分离的转录"""
result = self.model.generate(
input=audio_path,
batch_size_s=300 # 批处理大小（秒）
)
sentences = result[0].get("sentence_info", [])
return [
{
"speaker": f"Speaker_{s.get('spk', 0)}",
"text": s["text"],
"start": s["start"],
"end": s["end"]
}
for s in sentences
]
# 使用示例
asr = ParaformerASR()
result = asr.transcribe("meeting.wav")
print(f"识别结果: {result['text']}")
# 说话人分离
segments = asr.transcribe_with_speaker("meeting.wav")
for seg in segments:
print(f"[{seg['speaker']}] ({seg['start']}-{seg['end']}ms): {seg['text']}")
3.3 流式ASR实现
在实时语音交互中，流式ASR至关重要。以下是基于WebSocket的流式识别方案：
import asyncio
import websockets
import numpy as np
from collections import deque
class StreamingASR:
"""流式语音识别服务"""
def __init__(self, model_size="base"):
import whisper
self.model = whisper.load_model(model_size)
self.sample_rate = 16000
self.chunk_duration = 1.0 # 每秒处理一次
self.context_duration = 3.0 # 上下文窗口3秒
self.buffer = deque(maxlen=int(self.context_duration * self.sample_rate))
def process_chunk(self, audio_chunk: np.ndarray) -> str:
"""处理一个音频块"""
self.buffer.extend(audio_chunk)
# 使用VAD判断是否有语音
if self._has_speech(list(self.buffer)):
audio_array = np.array(list(self.buffer), dtype=np.float32)
# Whisper需要30秒输入，不足部分补零
if len(audio_array) < self.sample_rate * 30:
padding = np.zeros(self.sample_rate * 30 - len(audio_array))
audio_array = np.concatenate([audio_array, padding])
result = self.model.transcribe(
audio_array,
language="zh",
fp16=False,
no_speech_threshold=0.6
)
text = result["text"].strip()
if text and result.get("segments"):
# 只返回最后一个segment（最新内容）
return result["segments"][-1]["text"].strip()
return ""
def _has_speech(self, audio_list):
"""简单的能量检测VAD"""
if len(audio_list) < 1600:
return False
audio = np.array(audio_list[-1600:]) # 最近100ms
energy = np.mean(audio ** 2)
return energy > 0.001
# WebSocket流式ASR服务器
async def asr_websocket_handler(websocket):
asr = StreamingASR(model_size="base")
async for message in websocket:
if isinstance(message, bytes):
# 接收PCM16音频数据
audio = np.frombuffer(message, dtype=np.int16).astype(np.float32) / 32768.0
text = asr.process_chunk(audio)
if text:
await websocket.send(json.dumps({
"type": "transcription",
"text": text,
"is_final": False
}))
第四章：语音合成（TTS）核心技术
4.1 VITS — 端到端语音合成
VITS（Variational Inference with adversarial learning for end-to-end Text-to-Speech）是端到端的TTS模型，质量高且速度快。
import torch
import soundfile as sf
from vits import SynthesizerTrn, get_hparams_from_file
class VITSTTS:
def __init__(self, model_path, config_path):
hps = get_hparams_from_file(config_path)
self.model = SynthesizerTrn(
len(hps.symbols),
hps.data.filter_length // 2 + 1,
hps.train.segment_size // hps.data.hop_length,
**hps.model
)
self.model.load_state_dict(torch.load(model_path)["model"])
self.model.eval()
self.hps = hps
def synthesize(self, text: str, speaker_id: int = 0) -> np.ndarray:
"""文本转语音"""
from vits import text_to_sequence, symbols
# 文本转音素序列
sequence = text_to_sequence(text, [self.hps.data.text_cleaners])
sequence = torch.LongTensor(sequence).unsqueeze(0)
with torch.no_grad():
audio = self.model.infer(
sequence,
torch.LongTensor([speaker_id]),
noise_scale=0.667,
noise_scale_w=0.8,
length_scale=1.0
)[0][0].numpy()
return audio
def save(self, audio: np.ndarray, path: str, sample_rate: int = 22050):
sf.write(path, audio, sample_rate)
# 使用示例
tts = VITSTTS("model.pth", "config.json")
audio = tts.synthesize("你好，欢迎使用AI语音助手！")
tts.save(audio, "output.wav")
4.2 XTTS — 多语言声音克隆TTS
Coqui XTTS支持17种语言和零样本声音克隆，只需几秒参考音频即可克隆任意声音。
from TTS.api import TTS
import torch
class XTTSCloneTTS:
def __init__(self):
self.tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to("cuda")
def clone_and_synthesize(
self,
text: str,
speaker_wav: str,
language: str = "zh",
output_path: str = "output.wav"
):
"""
使用参考音频克隆声音并合成
Args:
text: 要合成的文本
speaker_wav: 参考说话人音频文件路径（10-30秒最佳）
language: 语言代码 (zh, en, ja, ko, fr, de, etc.)
output_path: 输出音频路径
"""
self.tts.tts_to_file(
text=text,
speaker_wav=speaker_wav,
language=language,
file_path=output_path
)
print(f"合成完成: {output_path}")
def stream_synthesize(self, text: str, speaker_wav: str, language: str = "zh"):
"""流式合成（返回音频chunk）"""
for chunk in self.tts.tts_stream(
text=text,
speaker_wav=speaker_wav,
language=language
):
yield chunk
# 使用示例
tts = XTTSCloneTTS()
tts.clone_and_synthesize(
text="这是一段使用克隆声音合成的语音。",
speaker_wav="reference_speaker.wav",
language="zh",
output_path="cloned_output.wav"
)
4.3 CosyVoice — 阿里语音合成
CosyVoice是阿里开源的高保真语音合成系统，支持中文零样本声音克隆和情感控制。
from cosyvoice import CosyVoice
import soundfile as sf
class CosyVoiceTTS:
def __init__(self, model_dir="pretrained_models/CosyVoice-300M"):
self.model = CosyVoice(model_dir)
def synthesize(self, text: str, mode: str = "sft", **kwargs) -> tuple:
"""
语音合成
Args:
text: 输入文本
mode:
- "sft": 预训练音色（speaker参数选择音色）
- "zero_shot": 零样本声音克隆（需要prompt_wav和prompt_text）
- "cross_lingual": 跨语言合成
- "instruct": 指令式合成（描述想要的语音风格）
"""
if mode == "sft":
speaker = kwargs.get("speaker", "中文女")
for result in self.model.inference_sft(text, speaker):
return result["tts_speech"], 22050
elif mode == "zero_shot":
prompt_wav = kwargs["prompt_wav"] # 参考音频路径
prompt_text = kwargs["prompt_text"] # 参考音频对应的文本
for result in self.model.inference_zero_shot(
text, prompt_text, prompt_wav
):
return result["tts_speech"], 22050
elif mode == "instruct":
instruct = kwargs.get("instruct", "用温柔的声音朗读")
speaker = kwargs.get("speaker", "中文女")
for result in self.model.inference_instruct(text, instruct, speaker):
return result["tts_speech"], 22050
# 使用示例
tts = CosyVoiceTTS()
# 预训练音色合成
audio, sr = tts.synthesize("你好，欢迎来到AI语音助手的世界。", mode="sft", speaker="中文女")
sf.write("sft_output.wav", audio.numpy(), sr)
# 零样本声音克隆
audio, sr = tts.synthesize(
"用你的声音说出这段话。",
mode="zero_shot",
prompt_wav="speaker_ref.wav",
prompt_text="这是参考音频的文本内容。"
)
sf.write("clone_output.wav", audio.numpy(), sr)
# 指令式情感合成
audio, sr = tts.synthesize(
"今天天气真好啊！",
mode="instruct",
instruct="用开心活泼的语气说",
speaker="中文女"
)
sf.write("emotion_output.wav", audio.numpy(), sr)
4.4 F5-TTS — 非自回归零样本TTS
F5-TTS采用非自回归架构，推理速度极快，适合实时场景。
import torch
import soundfile as sf
from f5_tts.api import F5TTS
class F5TTSEngine:
def __init__(self):
self.model = F5TTS(device="cuda")
def synthesize(
self,
text: str,
ref_audio: str,
ref_text: str = "",
output_path: str = "output.wav"
):
"""
零样本语音合成
Args:
text: 要合成的文本
ref_audio: 参考音频路径（3-10秒）
ref_text: 参考音频对应文本（可选，留空自动推断）
output_path: 输出路径
"""
audio, sr = self.model.infer(
ref_file=ref_audio,
ref_text=ref_text,
gen_text=text,
speed=1.0
)
sf.write(output_path, audio, sr)
return audio, sr
# 使用示例
engine = F5TTSEngine()
engine.synthesize(
text="这是一段使用F5-TTS合成的高质量语音。",
ref_audio="reference.wav",
ref_text="参考音频的文字内容。",
output_path="f5_output.wav"
)
第五章：实时流式语音处理
5.1 WebSocket实时通信
WebSocket是语音实时交互的基础协议，支持全双工通信。
import asyncio
import websockets
import json
import numpy as np
from typing import AsyncGenerator
class VoiceStreamingServer:
"""基于WebSocket的语音流式处理服务器"""
def __init__(self, asr_model, llm_client, tts_model):
self.asr = asr_model
self.llm = llm_client
self.tts = tts_model
self.sample_rate = 16000
async def handle_client(self, websocket):
"""处理单个客户端连接"""
print(f"新客户端连接: {websocket.remote_address}")
audio_buffer = bytearray()
conversation_history = []
try:
async for message in websocket:
if isinstance(message, bytes):
# 音频数据
audio_buffer.extend(message)
# 检查是否积累了足够的音频（2秒）
if len(audio_buffer) >= self.sample_rate * 2 * 2: # 16bit = 2 bytes/sample
# 转为numpy数组
audio = np.frombuffer(bytes(audio_buffer), dtype=np.int16)
audio_float = audio.astype(np.float32) / 32768.0
# 语音识别
text = await self.asr.transcribe_async(audio_float)
audio_buffer.clear()
if text.strip():
# 发送识别结果
await websocket.send(json.dumps({
"type": "asr_result",
"text": text
}))
# 生成LLM回复
conversation_history.append({"role": "user", "content": text})
response_text = ""
async for token in self.llm.stream_generate(conversation_history):
response_text += token
# 流式发送文本
await websocket.send(json.dumps({
"type": "llm_token",
"token": token
}))
conversation_history.append({
"role": "assistant",
"content": response_text
})
# 语音合成并发送
audio_chunks = await self.tts.synthesize_streaming(response_text)
for chunk in audio_chunks:
await websocket.send(chunk.tobytes())
# 发送完成信号
await websocket.send(json.dumps({"type": "turn_complete"}))
elif isinstance(message, str):
# 控制消息
data = json.loads(message)
if data.get("type") == "config":
# 更新配置
pass
elif data.get("type") == "interrupt":
# 用户打断，清空缓冲区
audio_buffer.clear()
except websockets.exceptions.ConnectionClosed:
print(f"客户端断开: {websocket.remote_address}")
# 启动服务器
async def main():
server = VoiceStreamingServer(asr_model, llm_client, tts_model)
async with websockets.serve(
server.handle_client,
"0.0.0.0",
8765,
max_size=10 * 1024 * 1024, # 10MB
ping_interval=20,
ping_timeout=10
):
print("语音流式服务器启动在 ws://0.0.0.0:8765")
await asyncio.Future() # 永远运行
asyncio.run(main())
5.2 WebRTC低延迟方案
WebRTC提供更低延迟的实时通信，特别适合浏览器端语音交互。
# 使用aiortc库实现WebRTC语音处理
import asyncio
from aiortc import RTCPeerConnection, MediaStreamTrack
from aiortc.contrib.media import MediaRelay
import numpy as np
class AudioProcessorTrack(MediaStreamTrack):
"""处理WebRTC音频流的自定义轨道"""
kind = "audio"
def __init__(self, track, asr, llm, tts):
super().__init__()
self.track = track
self.asr = asr
self.llm = llm
self.tts = tts
self.audio_buffer = []
self.sample_rate = 48000 # WebRTC默认48kHz
async def recv(self):
frame = await self.track.recv()
# 转为numpy处理
audio = frame.to_ndarray()
self.audio_buffer.append(audio)
# 积累足够音频后处理
total_samples = sum(len(a) for a in self.audio_buffer)
if total_samples >= self.sample_rate * 2: # 2秒
# 合并音频
full_audio = np.concatenate(self.audio_buffer, axis=-1)
self.audio_buffer.clear()
# 降采样到16kHz用于ASR
audio_16k = self._resample(full_audio, 48000, 16000)
# ASR -> LLM -> TTS处理
text = await self.asr.transcribe_async(audio_16k)
if text.strip():
response = await self.llm.generate_async(text)
tts_audio = await self.tts.synthesize_async(response)
# 上采样回48kHz
audio_48k = self._resample(tts_audio, 22050, 48000)
return self._numpy_to_frame(audio_48k)
return frame # 无语音时返回静音帧
def _resample(self, audio, from_rate, to_rate):
"""简单的重采样"""
if from_rate == to_rate:
return audio
ratio = to_rate / from_rate
new_length = int(len(audio) * ratio)
indices = np.linspace(0, len(audio) - 1, new_length)
return np.interp(indices, np.arange(len(audio)), audio).astype(np.float32)
5.3 音频格式处理
实时语音交互中，音频格式处理是关键环节：
import numpy as np
import struct
import io
import wave
class AudioFormatUtils:
"""音频格式转换工具"""
@staticmethod
def pcm16_to_float(pcm_data: bytes) -> np.ndarray:
"""PCM16字节转float32数组"""
return np.frombuffer(pcm_data, dtype=np.int16).astype(np.float32) / 32768.0
@staticmethod
def float_to_pcm16(audio: np.ndarray) -> bytes:
"""float32数组转PCM16字节"""
audio = np.clip(audio, -1.0, 1.0)
pcm = (audio * 32767).astype(np.int16)
return pcm.tobytes()
@staticmethod
def resample(audio: np.ndarray, from_rate: int, to_rate: int) -> np.ndarray:
"""高质量重采样（使用线性插值）"""
if from_rate == to_rate:
return audio
duration = len(audio) / from_rate
new_length = int(duration * to_rate)
old_indices = np.arange(len(audio))
new_indices = np.linspace(0, len(audio) - 1, new_length)
return np.interp(new_indices, old_indices, audio).astype(audio.dtype)
@staticmethod
def add_wav_header(audio: bytes, sample_rate: int, channels: int = 1, bits: int = 16) -> bytes:
"""添加WAV头"""
buffer = io.BytesIO()
with wave.open(buffer, 'wb') as wf:
wf.setnchannels(channels)
wf.setsampwidth(bits // 8)
wf.setframerate(sample_rate)
wf.writeframes(audio)
return buffer.getvalue()
@staticmethod
def split_audio_chunks(audio: np.ndarray, chunk_duration_ms: int, sample_rate: int) -> list:
"""将音频分割为固定时长的块"""
chunk_size = int(sample_rate * chunk_duration_ms / 1000)
chunks = []
for i in range(0, len(audio), chunk_size):
chunks.append(audio[i:i + chunk_size])
return chunks
第六章：语音Agent架构设计
6.1 VAD（语音活动检测）
VAD是语音Agent的第一个组件，决定何时开始/停止录音。
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.reset_states()
def is_speech(self, audio_chunk: np.ndarray) -> bool:
"""判断音频块是否包含语音"""
if len(audio_chunk) < 512:
return False
audio_tensor = torch.from_numpy(audio_chunk).float()
if len(audio_tensor.shape) == 1:
audio_tensor = audio_tensor.unsqueeze(0)
prob = self.model(audio_tensor, self.sample_rate).item()
return prob > self.threshold
def detect_speech_segments(self, audio: np.ndarray, min_speech_ms: int = 250) -> list:
"""检测语音片段"""
chunk_size = 512 # Silero VAD要求512采样点
segments = []
current_start = None
min_speech_samples = int(self.sample_rate * min_speech_ms / 1000)
for i in range(0, len(audio), chunk_size):
chunk = audio[i:i + chunk_size]
if len(chunk) < chunk_size:
chunk = np.pad(chunk, (0, chunk_size - len(chunk)))
is_speech = self.is_speech(chunk)
position = i
if is_speech and current_start is None:
current_start = position
elif not is_speech and current_start is not None:
if position - current_start >= min_speech_samples:
segments.append((current_start, position))
current_start = None
# 处理最后一段
if current_start is not None and len(audio) - current_start >= min_speech_samples:
segments.append((current_start, len(audio)))
return segments
6.2 完整语音Agent实现
将所有组件整合为一个完整的语音Agent：
import asyncio
import numpy as np
from dataclasses import dataclass, field
from typing import Optional, Callable
from enum import Enum
class AgentState(Enum):
IDLE = "idle"
LISTENING = "listening"
THINKING = "thinking"
SPEAKING = "speaking"
@dataclass
class VoiceAgentConfig:
asr_model: str = "medium"
tts_model: str = "cosyvoice"
llm_model: str = "gpt-4o"
vad_threshold: float = 0.5
silence_duration_ms: int = 800
max_recording_duration_s: int = 30
language: str = "zh"
voice_id: str = "中文女"
system_prompt: str = "你是一个友好的AI助手。"
class VoiceAgent:
"""完整的语音Agent"""
def __init__(self, config: VoiceAgentConfig):
self.config = config
self.state = AgentState.IDLE
self.conversation_history = []
# 初始化各组件
self.vad = SileroVAD(threshold=config.vad_threshold)
self.asr = self._init_asr()
self.tts = self._init_tts()
self.llm = self._init_llm()
# 状态回调
self.on_state_change: Optional[Callable] = None
self.on_transcription: Optional[Callable] = None
self.on_response_text: Optional[Callable] = None
self.on_audio_chunk: Optional[Callable] = None
def _init_asr(self):
if self.config.asr_model.startswith("whisper"):
return WhisperASR(model_size="medium")
else:
return ParaformerASR()
def _init_tts(self):
if self.config.tts_model == "cosyvoice":
return CosyVoiceTTS()
elif self.config.tts_model == "xtts":
return XTTSCloneTTS()
else:
return VITSTTS("model.pth", "config.json")
def _init_llm(self):
# 返回LLM客户端
from openai import OpenAI
return OpenAI()
def _set_state(self, new_state: AgentState):
self.state = new_state
if self.on_state_change:
self.on_state_change(new_state)
async def process_audio_stream(self, audio_generator):
"""处理音频流的核心逻辑"""
self._set_state(AgentState.LISTENING)
audio_buffer = []
silence_count = 0
silence_threshold_chunks = int(
self.config.silence_duration_ms / 1000 * 16000 / 512
)
async for audio_chunk in audio_generator:
# VAD检测
if self.vad.is_speech(audio_chunk):
audio_buffer.append(audio_chunk)
silence_count = 0
else:
if audio_buffer: # 已经开始录音
silence_count += 1
audio_buffer.append(audio_chunk) # 保留部分静音
if silence_count >= silence_threshold_chunks:
# 用户说完，开始处理
full_audio = np.concatenate(audio_buffer)
audio_buffer.clear()
silence_count = 0
await self._process_speech(full_audio)
async def _process_speech(self, audio: np.ndarray):
"""处理一段完整的语音输入"""
# ASR
self._set_state(AgentState.THINKING)
text = self.asr.transcribe(audio, language=self.config.language)
if not text.strip():
self._set_state(AgentState.IDLE)
return
if self.on_transcription:
self.on_transcription(text)
# LLM生成
self.conversation_history.append({"role": "user", "content": text})
response_text = ""
for token in self._stream_llm():
response_text += token
if self.on_response_text:
self.on_response_text(token)
self.conversation_history.append({
"role": "assistant",
"content": response_text
})
# TTS合成并播放
self._set_state(AgentState.SPEAKING)
audio_result = self.tts.synthesize(response_text)
if self.on_audio_chunk:
if isinstance(audio_result, tuple):
audio_data, sr = audio_result
self.on_audio_chunk(audio_data, sr)
else:
self.on_audio_chunk(audio_result, 22050)
self._set_state(AgentState.IDLE)
def _stream_llm(self):
"""流式LLM生成"""
response = self.llm.chat.completions.create(
model=self.config.llm_model,
messages=[
{"role": "system", "content": self.config.system_prompt},
*self.conversation_history[-10:] # 保留最近10轮
],
stream=True,
max_tokens=500
)
for chunk in response:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
# 使用示例
config = VoiceAgentConfig(
asr_model="whisper",
tts_model="cosyvoice",
llm_model="gpt-4o",
system_prompt="你是一个专业的AI健康顾问，用温和的语气回答问题。"
)
agent = VoiceAgent(config)
agent.on_state_change = lambda s: print(f"[状态] {s.value}")
agent.on_transcription = lambda t: print(f"[用户] {t}")
agent.on_response_text = lambda t: print(t, end="", flush=True)
6.3 打断（Barge-in）支持
语音交互中，用户经常需要打断AI的回复。实现打断检测：
class BargeInDetector:
"""打断检测器"""
def __init__(self, vad: SileroVAD, energy_threshold: float = 0.02):
self.vad = vad
self.energy_threshold = energy_threshold
self.is_ai_speaking = False
self.user_speech_start_time = None
self.min_interrupt_ms = 300 # 至少300ms语音才触发打断
def should_interrupt(self, audio_chunk: np.ndarray) -> bool:
"""判断是否应该打断AI"""
if not self.is_ai_speaking:
return False
# 检查能量
energy = np.sqrt(np.mean(audio_chunk ** 2))
if energy < self.energy_threshold:
self.user_speech_start_time = None
return False
# 检查是否为语音
if not self.vad.is_speech(audio_chunk):
self.user_speech_start_time = None
return False
import time
current_time = time.time() * 1000
if self.user_speech_start_time is None:
self.user_speech_start_time = current_time
return False
# 持续说话超过阈值，触发打断
if current_time - self.user_speech_start_time > self.min_interrupt_ms:
self.user_speech_start_time = None
return True
return False
def on_ai_start_speaking(self):
self.is_ai_speaking = True
def on_ai_stop_speaking(self):
self.is_ai_speaking = False
self.user_speech_start_time = None
第七章：语音情感识别与表达
7.1 语音情感识别（SER）
识别用户语音中的情感，可以帮助AI更好地理解用户意图。
import torch
import numpy as np
from transformers import pipeline
class SpeechEmotionRecognizer:
"""语音情感识别"""
# 常见情感标签
EMOTIONS = ["neutral", "happy", "sad", "angry", "fear", "surprise", "disgust"]
EMOTION_ZH = {
"neutral": "平静", "happy": "开心", "sad": "悲伤",
"angry": "生气", "fear": "恐惧", "surprise": "惊讶", "disgust": "厌恶"
}
def __init__(self):
# 使用HuggingFace的情感识别模型
self.classifier = pipeline(
"audio-classification",
model="ehcalabres/wav2vec2-lg-xlsr-en-speech-emotion-recognition",
device=0 if torch.cuda.is_available() else -1
)
def recognize(self, audio: np.ndarray, sample_rate: int = 16000) -> dict:
"""识别音频中的情感"""
results = self.classifier(audio, sampling_rate=sample_rate)
# 返回所有情感的置信度
emotions = {r["label"]: r["score"] for r in results}
top_emotion = results[0]
return {
"dominant_emotion": top_emotion["label"],
"dominant_emotion_zh": self.EMOTION_ZH.get(top_emotion["label"], top_emotion["label"]),
"confidence": top_emotion["score"],
"all_emotions": emotions
}
def get_emotion_prompt(self, emotion_result: dict) -> str:
"""根据识别的情感生成系统提示"""
emotion = emotion_result["dominant_emotion"]
confidence = emotion_result["confidence"]
if confidence < 0.5:
return ""
emotion_instructions = {
"happy": "用户语气开心，回复时也保持积极愉快的语调。",
"sad": "用户语气低落，回复时请表达理解和关怀，语气温和。",
"angry": "用户语气生气，请保持冷静和专业，不要反驳，先表示理解。",
"fear": "用户语气紧张恐惧，请给予安慰和安全感。",
"surprise": "用户语气惊讶，可以适当表示共鸣。",
}
return emotion_instructions.get(emotion, "")
7.2 情感化语音合成
根据对话情感调整TTS的输出风格：
class EmotionAwareTTS:
"""情感感知的语音合成"""
def __init__(self, base_tts):
self.base_tts = base_tts
def synthesize_with_emotion(
self,
text: str,
target_emotion: str = "neutral",
reference_audio: str = None
) -> tuple:
"""
根据目标情感合成语音
Args:
text: 输入文本
target_emotion: 目标情感 (happy, sad, calm, excited, etc.)
reference_audio: 参考音频（用于声音克隆）
"""
# 情感控制提示
emotion_prompts = {
"happy": "用开心、活泼的语气说",
"sad": "用低沉、缓慢的语气说",
"calm": "用平静、温和的语气说",
"excited": "用兴奋、充满活力的语气说",
"serious": "用严肃、正式的语气说",
"gentle": "用温柔、体贴的语气说"
}
instruct = emotion_prompts.get(target_emotion, "用自然的语气说")
if hasattr(self.base_tts, 'synthesize') and 'instruct' in self.base_tts.synthesize.__code__.co_varnames:
# CosyVoice支持指令式合成
return self.base_tts.synthesize(
text,
mode="instruct",
instruct=instruct
)
else:
# 其他TTS通过SSML或语速控制实现
return self.base_tts.synthesize(text)
def auto_emotion_synthesize(self, text: str, context: list) -> tuple:
"""根据上下文自动决定合成情感"""
# 简单规则：根据LLM回复内容判断情感
emotion = self._detect_text_emotion(text)
return self.synthesize_with_emotion(text, emotion)
def _detect_text_emotion(self, text: str) -> str:
"""从文本中检测情感倾向"""
happy_words = ["开心", "高兴", "太好了", "哈哈", "恭喜", "棒"]
sad_words = ["抱歉", "遗憾", "可惜", "难过", "对不起"]
excited_words = ["太棒了", "厉害", "amazing", "哇", "真的吗"]
text_lower = text.lower()
happy_count = sum(1 for w in happy_words if w in text_lower)
sad_count = sum(1 for w in sad_words if w in text_lower)
excited_count = sum(1 for w in excited_words if w in text_lower)
if excited_count > 0:
return "excited"
elif happy_count > sad_count:
return "happy"
elif sad_count > 0:
return "gentle"
else:
return "calm"
第八章：多语言语音交互
8.1 多语言ASR实现
import whisper
class MultilingualASR:
"""多语言语音识别"""
LANGUAGE_MAP = {
"zh": "Chinese",
"en": "English",
"ja": "Japanese",
"ko": "Korean",
"fr": "French",
"de": "German",
"es": "Spanish",
"ru": "Russian",
"ar": "Arabic",
"pt": "Portuguese",
}
def __init__(self, model_size="large-v3"):
self.model = whisper.load_model(model_size)
def detect_language(self, audio_path: str) -> tuple:
"""自动检测语言"""
audio = whisper.load_audio(audio_path)
audio = whisper.pad_or_trim(audio)
mel = whisper.log_mel_spectrogram(audio).to(self.model.device)
_, probs = self.model.detect_language(mel)
detected = max(probs, key=probs.get)
return detected, probs[detected]
def transcribe(self, audio_path: str, language: str = None) -> dict:
"""多语言转录，语言可选自动检测"""
options = {"fp16": False}
if language:
options["language"] = language
result = self.model.transcribe(audio_path, **options)
return {
"text": result["text"],
"language": result["language"],
"segments": [
{
"text": seg["text"],
"start": seg["start"],
"end": seg["end"],
"language": seg.get("language", result["language"])
}
for seg in result["segments"]
]
}
# 使用示例
asr = MultilingualASR()
# 自动检测语言
lang, confidence = asr.detect_language("audio.wav")
print(f"检测到语言: {lang} (置信度: {confidence:.2f})")
# 指定语言转录
result = asr.transcribe("chinese_audio.wav", language="zh")
print(f"中文识别: {result['text']}")
# 自动语言转录
result = asr.transcribe("multilingual_audio.wav")
print(f"语言: {result['language']}, 内容: {result['text']}")
8.2 实时翻译语音交互
class RealtimeTranslator:
"""实时语音翻译系统"""
def __init__(self):
self.asr = MultilingualASR()
self.llm = None # LLM客户端
self.tts = CosyVoiceTTS()
async def translate_speech(
self,
audio: np.ndarray,
source_lang: str,
target_lang: str
) -> tuple:
"""
语音翻译：输入语言A的语音，输出语言B的语音
Returns:
(原文文本, 翻译文本, 翻译音频)
"""
# 1. 语音识别
text = self.asr.transcribe(audio, language=source_lang)["text"]
# 2. LLM翻译
translation = await self._llm_translate(text, source_lang, target_lang)
# 3. 语音合成
audio_result = self.tts.synthesize(translation)
if isinstance(audio_result, tuple):
synth_audio, sr = audio_result
else:
synth_audio, sr = audio_result, 22050
return text, translation, synth_audio
async def _llm_translate(self, text: str, from_lang: str, to_lang: str) -> str:
"""使用LLM进行翻译"""
lang_names = {"zh": "中文", "en": "英文", "ja": "日文", "ko": "韩文"}
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": f"你是一个专业的翻译。将以下{lang_names.get(from_lang, from_lang)}文本翻译为{lang_names.get(to_lang, to_lang)}，保持自然口语化。只输出翻译结果。"
},
{"role": "user", "content": text}
],
max_tokens=500
)
return response.choices[0].message.content
第九章：端到端Speech-to-Speech实现
9.1 基于GPT-4o的端到端实现
import asyncio
import websockets
import json
import base64
import pyaudio
import threading
class SpeechToSpeechApp:
"""基于GPT-4o Realtime API的端到端语音对话应用"""
def __init__(self, api_key: str):
self.api_key = api_key
self.ws_url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
self.is_running = False
self.is_ai_speaking = False
# 音频配置
self.sample_rate = 24000
self.chunk_size = 1024
self.format = pyaudio.paInt16
self.channels = 1
async def start(self):
"""启动语音对话"""
self.is_running = True
# 连接WebSocket
self.ws = await websockets.connect(
self.ws_url,
extra_headers={
"Authorization": f"Bearer {self.api_key}",
"OpenAI-Beta": "realtime=v1"
}
)
# 配置session
await self._configure_session()
# 同时运行录音和播放
await asyncio.gather(
self._record_and_send(),
self._receive_and_play(),
self._handle_events()
)
async def _configure_session(self):
await self.ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"instructions": (
"你是一个友好的AI助手，名叫小智。"
"你善于用自然、生动的语气和用户交流。"
"回答简洁明了，适合语音对话场景。"
),
"voice": "alloy",
"input_audio_format": "pcm16",
"output_audio_format": "pcm16",
"input_audio_transcription": {
"model": "whisper-1"
},
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500
},
"temperature": 0.8,
"max_response_output_tokens": 4096
}
}))
async def _record_and_send(self):
"""录音并发送到服务器"""
p = pyaudio.PyAudio()
stream = p.open(
format=self.format,
channels=self.channels,
rate=self.sample_rate,
input=True,
frames_per_buffer=self.chunk_size
)
print("🎤 开始录音，按Ctrl+C停止...")
try:
while self.is_running:
audio_data = stream.read(self.chunk_size, exception_on_overflow=False)
audio_b64 = base64.b64encode(audio_data).decode()
await self.ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": audio_b64
}))
await asyncio.sleep(0.01) # 控制发送频率
finally:
stream.stop_stream()
stream.close()
p.terminate()
async def _receive_and_play(self):
"""接收服务器音频并播放"""
p = pyaudio.PyAudio()
stream = p.open(
format=self.format,
channels=self.channels,
rate=self.sample_rate,
output=True
)
try:
async for message in self.ws:
event = json.loads(message)
if event["type"] == "response.audio.delta":
audio_chunk = base64.b64decode(event["delta"])
stream.write(audio_chunk)
self.is_ai_speaking = True
elif event["type"] == "response.audio.done":
self.is_ai_speaking = False
elif event["type"] == "response.audio_transcript.delta":
print(event["delta"], end="", flush=True)
elif event["type"] == "response.audio_transcript.done":
print()
finally:
stream.stop_stream()
stream.close()
p.terminate()
async def _handle_events(self):
"""处理其他事件"""
async for message in self.ws:
event = json.loads(message)
if event["type"] == "error":
print(f"错误: {event['error']['message']}")
elif event["type"] == "session.created":
print("✅ 会话已建立")
elif event["type"] == "input_audio_buffer.speech_started":
print("🎙️ 检测到语音输入...")
elif event["type"] == "input_audio_buffer.speech_stopped":
print("🔇 语音输入结束")
# 运行
async def main():
app = SpeechToSpeechApp("your-api-key")
try:
await app.start()
except KeyboardInterrupt:
print("\n会话结束")
asyncio.run(main())
9.2 开源端到端方案
对于需要本地部署的场景，可以使用Qwen2.5-Omni等开源模型：
import torch
import sounddevice as sd
import numpy as np
import queue
import threading
class LocalSpeechToSpeech:
"""基于开源模型的本地端到端语音对话"""
def __init__(self, model_path="Qwen/Qwen2.5-Omni-7B"):
from transformers import AutoModelForCausalLM, AutoTokenizer
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
self.tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=True
)
self.audio_queue = queue.Queue()
self.is_running = False
def start_conversation(self):
"""开始语音对话"""
self.is_running = True
# 启动录音线程
record_thread = threading.Thread(target=self._record_audio)
record_thread.daemon = True
record_thread.start()
print("🎤 语音对话已开始，按Ctrl+C停止...")
try:
while self.is_running:
# 收集音频
audio_chunks = []
while not self.audio_queue.empty():
audio_chunks.append(self.audio_queue.get())
if audio_chunks:
audio = np.concatenate(audio_chunks)
# 检查是否有足够的语音内容
if np.max(np.abs(audio)) > 0.01:
self._process_audio(audio)
import time
time.sleep(0.1)
except KeyboardInterrupt:
self.is_running = False
print("\n对话结束")
def _record_audio(self):
"""录音线程"""
def callback(indata, frames, time_info, status):
self.audio_queue.put(indata.copy())
with sd.InputStream(
samplerate=16000,
channels=1,
dtype='float32',
blocksize=4000,
callback=callback
):
while self.is_running:
sd.sleep(100)
def _process_audio(self, audio: np.ndarray):
"""处理音频输入"""
import soundfile as sf
import tempfile
# 保存临时音频文件
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
sf.write(f.name, audio, 16000)
audio_path = f.name
# 使用模型处理
messages = [
{"role": "user", "content": [
{"type": "audio", "audio_url": audio_path},
{"type": "text", "text": "请回答音频中的问题。"}
]}
]
text, response_audio = self.model.chat(
messages=messages,
tokenizer=self.tokenizer
)
print(f"AI: {text}")
# 播放回复音频
sd.play(response_audio.numpy(), samplerate=24000)
sd.wait()
第十章：低延迟优化技巧
10.1 延迟分析与优化策略
语音交互的延迟是用户体验的核心指标。以下是各环节的延迟分析和优化策略：
import time
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Dict
@dataclass
class LatencyMetrics:
"""延迟指标"""
vad_latency_ms: float = 0
asr_latency_ms: float = 0
llm_first_token_ms: float = 0
llm_total_ms: float = 0
tts_first_chunk_ms: float = 0
tts_total_ms: float = 0
total_e2e_ms: float = 0
class LatencyOptimizer:
"""延迟优化器"""
def __init__(self):
self.metrics_history = []
@contextmanager
def measure(self, name: str):
"""测量某环节延迟"""
start = time.perf_counter()
yield
elapsed = (time.perf_counter() - start) * 1000
print(f"[延迟] {name}: {elapsed:.1f}ms")
def optimize_asr(self):
"""ASR优化策略"""
return {
"1_使用流式识别": "边录边识别，不等用户说完",
"2_模型量化": "使用INT8量化模型，推理速度提升2-3倍",
"3_GPU加速": "确保ASR模型在GPU上运行",
"4_音频降采样": "16kHz足够，不需要更高采样率",
"5_并行处理": "录音和识别在不同线程/进程中执行",
}
def optimize_llm(self):
"""LLM优化策略"""
return {
"1_流式输出": "使用streaming API，不等全部生成完",
"2_减少上下文": "只保留最近几轮对话",
"3_使用小模型": "对话场景7B-14B模型足够",
"4_预测解码": "使用speculative decoding加速",
"5_减少max_tokens": "语音对话回复控制在100-200字",
}
def optimize_tts(self):
"""TTS优化策略"""
return {
"1_流式合成": "边生成文本边合成语音",
"2_句子级合成": "不等全部文本，按句子分批合成",
"3_音频缓存": "常用回复预合成缓存",
"4_轻量模型": "实时场景使用轻量TTS模型",
}
def optimize_network(self):
"""网络优化策略"""
return {
"1_音频压缩": "使用Opus编码，压缩比10:1",
"2_UDP传输": "WebRTC使用UDP，比TCP更实时",
"3_边缘部署": "将ASR/TTS部署到靠近用户的边缘节点",
"4_预连接": "建立WebSocket/RTC连接池",
}
10.2 流式TTS与LLM并行
关键优化：LLM生成文本的同时，TTS已开始合成前面的句子。
import asyncio
from typing import AsyncGenerator
class StreamingVoicePipeline:
"""流式语音管线：LLM和TTS并行执行"""
def __init__(self, llm, tts):
self.llm = llm
self.tts = tts
async def process(
self,
user_text: str,
history: list
) -> AsyncGenerator[bytes, None]:
"""
流式处理：LLM生成token的同时，TTS合成已完成的句子
"""
sentence_buffer = ""
tts_queue = asyncio.Queue()
# 启动TTS消费者
tts_task = asyncio.create_task(
self._tts_consumer(tts_queue)
)
# LLM流式生成
async for token in self._stream_llm(user_text, history):
sentence_buffer += token
# 检查是否构成完整句子
sentence = self._extract_sentence(sentence_buffer)
if sentence:
sentence_buffer = sentence_buffer[len(sentence):]
await tts_queue.put(sentence)
# 处理剩余文本
if sentence_buffer.strip():
await tts_queue.put(sentence_buffer.strip())
# 发送结束信号
await tts_queue.put(None)
# 等待TTS完成
async for audio_chunk in self._collect_tts_results(tts_queue):
yield audio_chunk
async def _stream_llm(self, text: str, history: list):
"""流式LLM生成"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "简洁回答，适合语音播放。"},
*history,
{"role": "user", "content": text}
],
stream=True,
max_tokens=300
)
for chunk in response:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
def _extract_sentence(self, text: str) -> str:
"""提取完整句子"""
sentence_endings = ["。", "！", "？", "!", "?", ".", "\n"]
for i, char in enumerate(text):
if char in sentence_endings:
return text[:i + 1]
# 如果缓冲区太长，强制分割
if len(text) > 50:
# 在逗号或分号处分割
for i in range(len(text) - 1, 0, -1):
if text[i] in ["，", ",", "；", ";"]:
return text[:i + 1]
return text[:50]
return ""
async def _tts_consumer(self, queue: asyncio.Queue):
"""TTS消费者任务"""
while True:
sentence = await queue.get()
if sentence is None:
await queue.put("DONE")
break
audio = await self._synthesize_async(sentence)
await queue.put(audio)
async def _synthesize_async(self, text: str) -> bytes:
"""异步TTS合成"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, self.tts.synthesize, text)
10.3 音频缓冲与播放优化
import numpy as np
import threading
import queue
import time
class AudioPlaybackBuffer:
"""音频播放缓冲区，确保平滑播放"""
def __init__(self, sample_rate: int = 24000, buffer_ms: int = 100):
self.sample_rate = sample_rate
self.buffer_size = int(sample_rate * buffer_ms / 1000)
self.play_queue = queue.Queue()
self.is_playing = False
self.play_thread = None
def start_playback(self):
"""开始播放"""
self.is_playing = True
self.play_thread = threading.Thread(target=self._play_loop)
self.play_thread.daemon = True
self.play_thread.start()
def add_audio(self, audio: np.ndarray):
"""添加音频到播放队列"""
self.play_queue.put(audio)
def stop(self):
"""停止播放"""
self.is_playing = False
self.play_queue.put(None)
def _play_loop(self):
"""播放循环"""
import pyaudio
p = pyaudio.PyAudio()
stream = p.open(
format=pyaudio.paFloat32,
channels=1,
rate=self.sample_rate,
output=True,
frames_per_buffer=self.buffer_size
)
# 预缓冲
pre_buffer = []
min_buffer_chunks = 3
try:
while self.is_playing:
try:
audio = self.play_queue.get(timeout=0.5)
if audio is None:
break
if len(pre_buffer) < min_buffer_chunks:
pre_buffer.append(audio)
if len(pre_buffer) >= min_buffer_chunks:
# 开始播放预缓冲
for chunk in pre_buffer:
stream.write(chunk.astype(np.float32).tobytes())
pre_buffer.clear()
else:
stream.write(audio.astype(np.float32).tobytes())
except queue.Empty:
continue
finally:
stream.stop_stream()
stream.close()
p.terminate()
第十一章：语音RAG系统
11.1 语音RAG架构
语音RAG将检索增强生成技术应用于语音场景，让AI语音助手能基于知识库回答问题。
from dataclasses import dataclass
from typing import List, Optional
import numpy as np
@dataclass
class VoiceRAGConfig:
asr_model: str = "paraformer"
embedding_model: str = "text-embedding-3-small"
llm_model: str = "gpt-4o"
tts_model: str = "cosyvoice"
top_k: int = 3
similarity_threshold: float = 0.7
class VoiceRAGSystem:
"""语音RAG系统"""
def __init__(self, config: VoiceRAGConfig):
self.config = config
self.asr = ParaformerASR()
self.tts = CosyVoiceTTS()
self.knowledge_base = []
self.embeddings = []
def add_documents(self, documents: List[dict]):
"""
添加文档到知识库
Args:
documents: [{"title": "...", "content": "...", "source": "..."}]
"""
from openai import OpenAI
client = OpenAI()
for doc in documents:
# 文档分块
chunks = self._split_document(doc["content"])
for chunk in chunks:
# 生成嵌入
response = client.embeddings.create(
model=self.config.embedding_model,
input=chunk
)
embedding = response.data[0].embedding
self.knowledge_base.append({
"title": doc["title"],
"content": chunk,
"source": doc.get("source", ""),
"embedding": embedding
})
self.embeddings.append(embedding)
self.embeddings = np.array(self.embeddings)
def _split_document(self, text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""文档分块"""
chunks = []
sentences = text.replace("。", "。\n").replace("！", "！\n").replace("？", "？\n").split("\n")
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) > chunk_size:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence
else:
current_chunk += sentence
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
def retrieve(self, query: str) -> List[dict]:
"""检索相关文档"""
from openai import OpenAI
client = OpenAI()
# 查询嵌入
response = client.embeddings.create(
model=self.config.embedding_model,
input=query
)
query_embedding = np.array(response.data[0].embedding)
# 计算相似度
similarities = np.dot(self.embeddings, query_embedding) / (
np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_embedding)
)
# 获取top_k
top_indices = np.argsort(similarities)[::-1][:self.config.top_k]
results = []
for idx in top_indices:
if similarities[idx] >= self.config.similarity_threshold:
results.append({
**self.knowledge_base[idx],
"score": float(similarities[idx])
})
return results
async def process_voice_query(self, audio: np.ndarray) -> tuple:
"""
处理语音查询
Returns:
(查询文本, 回复文本, 回复音频)
"""
# 1. 语音识别
query_text = self.asr.transcribe(audio, language="zh")["text"]
# 2. 检索相关文档
relevant_docs = self.retrieve(query_text)
# 3. 构建增强prompt
context = "\n".join([
f"[来源: {doc['title']}] {doc['content']}"
for doc in relevant_docs
])
prompt = f"""基于以下参考资料回答用户问题。如果资料中没有相关信息，请说明你不确定。
参考资料：
{context}
用户问题：{query_text}
请用简洁、自然的口语化方式回答："""
# 4. LLM生成回复
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model=self.config.llm_model,
messages=[
{"role": "system", "content": "你是一个知识丰富的AI助手，用简洁的口语化方式回答。"},
{"role": "user", "content": prompt}
],
max_tokens=300
)
reply_text = response.choices[0].message.content
# 5. 语音合成
audio_result = self.tts.synthesize(reply_text)
if isinstance(audio_result, tuple):
reply_audio, sr = audio_result
else:
reply_audio, sr = audio_result, 22050
return query_text, reply_text, reply_audio
11.2 语音文档索引
class VoiceDocumentIndexer:
"""语音文档索引器：将音频/视频内容索引为可检索的知识"""
def __init__(self):
self.asr = ParaformerASR()
self.embedder = None # 文本嵌入模型
def index_audio_file(self, audio_path: str, metadata: dict = None) -> List[dict]:
"""索引音频文件"""
# 1. 语音识别
result = self.asr.transcribe(audio_path)
# 2. 分段
segments = result.get("sentences", [])
# 3. 为每个片段生成嵌入并索引
indexed = []
for i, seg in enumerate(segments):
entry = {
"content": seg["text"],
"start_time": seg.get("start", 0),
"end_time": seg.get("end", 0),
"source": audio_path,
"type": "audio",
"metadata": metadata or {},
"segment_index": i
}
indexed.append(entry)
return indexed
def index_meeting(self, audio_path: str, meeting_info: dict = None) -> List[dict]:
"""索引会议录音"""
# 带说话人分离的识别
segments = self.asr.transcribe_with_speaker(audio_path)
indexed = []
for seg in segments:
entry = {
"content": f"[{seg['speaker']}] {seg['text']}",
"speaker": seg["speaker"],
"start_time": seg["start"],
"end_time": seg["end"],
"source": audio_path,
"type": "meeting",
"metadata": meeting_info or {}
}
indexed.append(entry)
return indexed
第十二章：场景落地实战
12.1 智能客服语音系统
class VoiceCustomerService:
"""AI语音客服系统"""
def __init__(self):
self.agent = VoiceAgent(VoiceAgentConfig(
system_prompt="""你是一个专业的客服助手。
规则：
1. 先确认用户问题类型（咨询/投诉/技术支持/其他）
2. 用简洁明了的语言回答
3. 如果无法解决，引导用户转人工
4. 注意识别用户情绪，投诉时先表示歉意
5. 每次回复控制在3句话以内"""
))
self.faq_knowledge = VoiceRAGSystem(VoiceRAGConfig())
async def handle_call(self, audio_stream):
"""处理来电"""
# 初始化对话
greeting = "您好，我是AI客服小助手，请问有什么可以帮您？"
yield await self._speak(greeting)
# 处理用户输入
async for audio_chunk in self.agent.process_audio_stream(audio_stream):
yield audio_chunk
async def _speak(self, text: str) -> bytes:
"""合成语音"""
audio = self.agent.tts.synthesize(text)
if isinstance(audio, tuple):
return audio[0]
return audio
12.2 语音笔记系统
class VoiceNoteSystem:
"""AI语音笔记：将语音转为结构化笔记"""
def __init__(self):
self.asr = ParaformerASR()
self.llm = None # LLM客户端
async def create_note_from_audio(self, audio_path: str) -> dict:
"""从音频创建结构化笔记"""
# 1. 语音识别
transcription = self.asr.transcribe(audio_path)
raw_text = transcription["text"]
# 2. LLM结构化处理
prompt = f"""请将以下语音转录内容整理为结构化笔记：
原始内容：
{raw_text}
请输出JSON格式：
{{
"title": "笔记标题",
"summary": "内容摘要（50字以内）",
"key_points": ["要点1", "要点2", ...],
"action_items": ["待办事项1", ...],
"categories": ["分类标签"],
"full_text": "整理后的完整文本"
}}"""
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
note = json.loads(response.choices[0].message.content)
note["raw_transcription"] = raw_text
note["segments"] = transcription.get("segments", [])
note["source_file"] = audio_path
return note
async def summarize_meeting(self, audio_path: str) -> dict:
"""会议纪要生成"""
# 带说话人分离
segments = self.asr.transcribe_with_speaker(audio_path)
# 整理会议内容
meeting_text = "\n".join([
f"[{s['speaker']}] ({s['start']}ms-{s['end']}ms): {s['text']}"
for s in segments
])
prompt = f"""请根据以下会议记录生成会议纪要：
{meeting_text}
输出格式：
1. 会议主题
2. 参与者
3. 讨论要点（分点列出）
4. 决议事项
5. 后续行动"""
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return {
"summary": response.choices[0].message.content,
"participants": list(set(s["speaker"] for s in segments)),
"duration_ms": segments[-1]["end"] if segments else 0,
"segments": segments
}
12.3 智能音箱应用架构
class SmartSpeakerApp:
"""智能音箱应用架构"""
def __init__(self):
# 核心组件
self.wake_word_detector = WakeWordDetector(keyword="小智小智")
self.vad = SileroVAD()
self.asr = WhisperASR(model_size="base") # 音箱用base足够
self.tts = CosyVoiceTTS()
self.llm = None
# 技能系统
self.skills = {
"weather": WeatherSkill(),
"music": MusicSkill(),
"alarm": AlarmSkill(),
"smart_home": SmartHomeSkill(),
"chat": ChatSkill(),
}
async def run(self):
"""主运行循环"""
print("智能音箱启动，等待唤醒词...")
while True:
# 1. 唤醒词检测
if await self._detect_wake_word():
print("唤醒词检测到！")
# 2. 提示音
await self._play_prompt_sound()
# 3. 录音
audio = await self._record_user_speech()
# 4. ASR
text = self.asr.transcribe(audio)["text"]
print(f"用户说: {text}")
# 5. 意图识别和技能路由
skill_name, params = await self._route_intent(text)
# 6. 执行技能
response = await self.skills[skill_name].execute(params)
# 7. TTS播放
await self._speak(response)
async def _detect_wake_word(self) -> bool:
"""唤醒词检测"""
# 实际实现使用Porcupine或Snowboy
return True
async def _record_user_speech(self) -> np.ndarray:
"""录制用户语音（VAD控制）"""
import pyaudio
p = pyaudio.PyAudio()
stream = p.open(
format=pyaudio.paInt16,
channels=1,
rate=16000,
input=True,
frames_per_buffer=512
)
audio_chunks = []
silence_count = 0
max_silence = 15 # 15个静音块后停止
while True:
data = stream.read(512, exception_on_overflow=False)
audio = np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0
if self.vad.is_speech(audio):
audio_chunks.append(audio)
silence_count = 0
elif audio_chunks:
silence_count += 1
if silence_count >= max_silence:
break
stream.stop_stream()
stream.close()
p.terminate()
if audio_chunks:
return np.concatenate(audio_chunks)
return np.array([], dtype=np.float32)
async def _route_intent(self, text: str) -> tuple:
"""意图识别和路由"""
prompt = f"""判断以下用户意图，返回JSON格式：
用户说：{text}
可选意图：
- weather: 查询天气
- music: 播放音乐
- alarm: 设置闹钟
- smart_home: 控制智能家居
- chat: 闲聊
返回：{{"intent": "xxx", "params": {{}}}}"""
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
intent = result.get("intent", "chat")
params = result.get("params", {})
params["original_text"] = text
return intent, params
最佳实践总结
性能优化清单
ASR优化
- 使用Paraformer或Whisper Turbo获得最佳速度/质量平衡
- 启用流式识别，边录边识别
- GPU加速是必须的
LLM优化
- 使用流式API，首token延迟 < 500ms
- 控制max_tokens，语音场景100-300字足够
- 保持上下文窗口简洁
TTS优化
- 使用流式合成，按句子分批
- CosyVoice/F5-TTS适合中文实时场景
- 预热模型，避免首次加载延迟
系统优化
- 异步并行：录音、ASR、LLM、TTS在不同协程中执行
- 音频缓冲：预缓冲2-3个chunk再开始播放
- 网络优化：使用WebSocket长连接，避免频繁建连
架构选择指南
| 场景 | 推荐架构 | 延迟目标 |
|---|---|---|
| 智能客服 | 级联式 | < 2s |
| 实时翻译 | 级联式 | < 3s |
| 语音助手 | 端到端 | < 1s |
| 智能音箱 | 级联式 | < 1.5s |
| 语音游戏 | 端到端 | < 500ms |
安全与隐私
- 音频数据安全：传输使用TLS加密，存储加密
- 用户隐私：明确告知录音，提供关闭选项
- 内容安全：ASR和LLM输出需经过内容审核
- 合规要求：遵循当地隐私法规（如GDPR、个人信息保护法）
总结
本教程系统性地介绍了AI实时语音交互的完整技术栈：
- 架构层面：从级联式到端到端，理解不同架构的权衡
- 模型层面：GPT-4o Realtime、Gemini Live、Qwen2.5-Omni等主流方案对比
- 核心组件：ASR（Whisper/Paraformer）、TTS（VITS/XTTS/CosyVoice/F5-TTS）
- 工程实现：WebSocket/WebRTC实时通信、流式处理管线
- 高级话题：情感识别、声音克隆、多语言、语音RAG
- 场景落地：智能客服、语音笔记、智能音箱
AI语音交互正处于快速发展期，端到端模型的能力在不断增强。建议开发者：
- 先用级联式架构快速验证产品
- 在用户体验要求高的场景尝试端到端方案
- 关注开源模型的进展，降低部署成本
- 重视延迟优化，这是语音交互体验的核心
本教程内容为技术分享，所有代码示例均为教学目的编写。实际部署时请参考各项目的官方文档。
