Speech-to-Speech语音大模型完全教程
前言
2024-2025年，AI语音交互经历了革命性的突破。从GPT-4o Realtime API的惊艳亮相，到Gemini Live的自然对话，再到Qwen2.5-Omni的开源多模态能力——语音AI已经从"能用"进化到了"好用"，甚至"惊艳"。
本教程将系统讲解Speech-to-Speech（S2S）语音大模型的核心技术、主流方案对比、以及如何从零构建一个实时语音对话系统。
第一章：Speech-to-Speech 技术概览
1.1 传统语音交互 vs 端到端语音交互
传统的语音AI系统采用级联架构（Cascaded Pipeline），而新一代系统趋向端到端架构：
传统级联架构：
麦克风 → ASR(语音转文本) → LLM(文本推理) → TTS(文本转语音) → 扬声器
延迟：ASR ~300ms + LLM ~500ms + TTS ~200ms = ~1000ms+
端到端架构：
麦克风 → 语音大模型(直接理解+生成语音) → 扬声器
延迟：~200-500ms
混合架构（当前主流）：
麦克风 → 语音编码器 → LLM(共享表示) → 语音解码器 → 扬声器
延迟：~300-600ms
1.2 主流语音大模型对比
| 模型 | 厂商 | 架构类型 | 开源 | 多语言 | 实时对话 | 延迟 | 情感表达 |
|---|---|---|---|---|---|---|---|
| GPT-4o Realtime | OpenAI | 端到端 | ❌ | ✅ | ✅ | ~300ms | ⭐⭐⭐⭐⭐ |
| Gemini 2.5 Live | 端到端 | ❌ | ✅ | ✅ | ~400ms | ⭐⭐⭐⭐ | |
| Qwen2.5-Omni | 阿里 | 端到端 | ✅ | ✅ | ✅ | ~500ms | ⭐⭐⭐⭐ |
| Mini-Omni2 | 开源社区 | 混合 | ✅ | 部分 | ✅ | ~600ms | ⭐⭐⭐ |
| VITA | 开源社区 | 混合 | ✅ | 部分 | ✅ | ~500ms | ⭐⭐⭐ |
| LLaMA-Omni | 开源社区 | 混合 | ✅ | 英文 | ✅ | ~400ms | ⭐⭐⭐ |
1.3 级联方案技术栈
对于大多数开发者，采用级联方案（ASR + LLM + TTS）是最实用的选择：
"""
级联S2S系统核心架构
"""
import asyncio
import numpy as np
from dataclasses import dataclass
from typing import AsyncGenerator, Optional
from enum import Enum
class EmotionType(Enum):
NEUTRAL = "neutral"
HAPPY = "happy"
SAD = "sad"
ANGRY = "angry"
SURPRISED = "surprised"
@dataclass
class VoiceConfig:
"""语音配置"""
speaker_id: str = "default"
language: str = "zh"
speed: float = 1.0
pitch: float = 0.0
emotion: EmotionType = EmotionType.NEUTRAL
sample_rate: int = 24000
@dataclass
class ConversationTurn:
"""对话轮次"""
role: str # "user" or "assistant"
text: str
audio: Optional[np.ndarray] = None
emotion: Optional[EmotionType] = None
latency_ms: float = 0
class SpeechToSpeechSystem:
"""Speech-to-Speech 系统基类"""
def __init__(self):
self.conversation_history = []
self.system_prompt = "你是一个友好的AI语音助手。请用简洁自然的口语风格回答。"
async def process_audio(self, audio_input: np.ndarray) -> AsyncGenerator[ConversationTurn, None]:
"""处理音频输入，流式返回音频输出"""
raise NotImplementedError
async def asr(self, audio: np.ndarray) -> str:
"""语音识别"""
raise NotImplementedError
async def llm_generate(self, text: str) -> AsyncGenerator[str, None]:
"""文本生成（流式）"""
raise NotImplementedError
async def tts(self, text: str, voice_config: VoiceConfig) -> np.ndarray:
"""语音合成"""
raise NotImplementedError
第二章：语音编码器（ASR）深度解析
2.1 Whisper 模型详解
Whisper 是 OpenAI 开源的通用语音识别模型，也是目前最广泛使用的 ASR 模型之一：
"""
Whisper 语音识别实战
"""
# pip install openai-whisper torch
import whisper
import numpy as np
import time
def transcribe_with_whisper(
audio_path: str,
model_size: str = "large-v3",
language: str = "zh",
task: str = "transcribe",
) -> dict:
"""
使用 Whisper 进行语音识别
Args:
audio_path: 音频文件路径
model_size: 模型大小 (tiny/base/small/medium/large-v3)
language: 语言代码
task: "transcribe" (同语言转录) 或 "translate" (翻译成英文)
Returns:
包含文本、时间戳等信息的字典
"""
# 加载模型（首次运行会下载）
model = whisper.load_model(model_size)
start = time.time()
# 转录
result = model.transcribe(
audio_path,
language=language,
task=task,
beam_size=5, # beam search 宽度
best_of=5, # 采样5次取最优
temperature=(0.0, 0.2, 0.4, 0.6, 0.8, 1.0), # 温度回退
compression_ratio_threshold=2.4,
logprob_threshold=-1.0,
no_speech_threshold=0.6,
condition_on_previous_text=True, # 利用上文提高准确率
initial_prompt="以下是普通话的句子。", # 中文提示
word_timestamps=True, # 词级别时间戳
)
elapsed = time.time() - start
print(f"模型: {model_size}")
print(f"耗时: {elapsed:.2f}s")
print(f"识别结果: {result['text']}")
print(f"语言: {result['language']}")
print(f"段落数: {len(result['segments'])}")
# 输出带时间戳的段落
for seg in result['segments']:
start_t = seg['start']
end_t = seg['end']
text = seg['text'].strip()
print(f" [{start_t:.1f}s - {end_t:.1f}s] {text}")
return result
# 使用示例
# result = transcribe_with_whisper("recording.wav", model_size="large-v3")
2.2 流式语音识别
实时语音对话需要流式ASR，即边听边识别：
"""
流式语音识别 - 实时转录
"""
import numpy as np
import torch
import whisper
from collections import deque
import threading
import time
class StreamingASR:
"""流式语音识别器（基于 Whisper 的 VAD + 分段识别）"""
def __init__(
self,
model_size: str = "base", # 流式场景用小模型保证速度
language: str = "zh",
sample_rate: int = 16000,
chunk_duration_ms: int = 30, # 每个音频块30ms
silence_threshold: float = 0.01, # 静音阈值
silence_duration_ms: int = 800, # 静音超过800ms认为说完一句
min_speech_ms: int = 300, # 最短语音段300ms
):
self.model = whisper.load_model(model_size)
self.language = language
self.sample_rate = sample_rate
self.chunk_size = int(sample_rate * chunk_duration_ms / 1000)
self.silence_threshold = silence_threshold
self.silence_chunks = int(silence_duration_ms / chunk_duration_ms)
self.min_speech_chunks = int(min_speech_ms / chunk_duration_ms)
self.audio_buffer = []
self.silence_count = 0
self.is_speaking = False
self.transcript_callback = None
def on_transcript(self, callback):
"""注册转录回调函数"""
self.transcript_callback = callback
def _compute_energy(self, audio_chunk: np.ndarray) -> float:
"""计算音频能量"""
return np.sqrt(np.mean(audio_chunk.astype(np.float32) ** 2)) / 32768.0
def feed_audio(self, audio_chunk: np.ndarray):
"""输入音频数据（PCM 16bit）"""
energy = self._compute_energy(audio_chunk)
if energy > self.silence_threshold:
# 检测到语音
self.audio_buffer.append(audio_chunk)
self.silence_count = 0
self.is_speaking = True
elif self.is_speaking:
# 语音中的静音
self.audio_buffer.append(audio_chunk)
self.silence_count += 1
# 静音足够长，认为一句话结束
if self.silence_count >= self.silence_chunks:
if len(self.audio_buffer) >= self.min_speech_chunks:
self._process_segment()
self.audio_buffer = []
self.silence_count = 0
self.is_speaking = False
def _process_segment(self):
"""处理一个完整的语音段"""
if not self.audio_buffer:
return
# 合并音频块
audio = np.concatenate(self.audio_buffer)
audio_float = audio.astype(np.float32) / 32768.0
# Whisper 需要30秒的输入，短音频需要padding
if len(audio_float) < whisper.audio.SAMPLE_RATE * 30:
# Pad to 30 seconds
padded = np.zeros(whisper.audio.SAMPLE_RATE * 30, dtype=np.float32)
padded[:len(audio_float)] = audio_float
else:
padded = audio_float[:whisper.audio.SAMPLE_RATE * 30]
# 转录
audio_tensor = torch.from_numpy(padded)
# 使用 mel spectrogram
mel = whisper.log_mel_spectrogram(audio_tensor).to(self.model.device)
# 解码
options = whisper.DecodingOptions(
language=self.language,
beam_size=3,
without_timestamps=True,
)
result = whisper.decode(self.model, mel, options)
text = result.text.strip()
if text and self.transcript_callback:
self.transcript_callback(text)
# 使用示例
def on_text(text):
print(f"识别到: {text}")
# asr = StreamingASR(model_size="base", language="zh")
# asr.on_transcript(on_text)
#
# # 模拟实时音频输入
# import pyaudio
# p = pyaudio.PyAudio()
# stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000,
# input=True, frames_per_buffer=480)
# while True:
# data = np.frombuffer(stream.read(480), dtype=np.int16)
# asr.feed_audio(data)
2.3 FunASR 阿里语音识别
"""
FunASR - 阿里开源的语音识别工具包
优势：中文识别效果好，支持标点恢复和时间戳
"""
# pip install funasr modelscope torch torchaudio
from funasr import AutoModel
def transcribe_with_funasr(audio_path: str) -> dict:
"""使用 FunASR 进行中文语音识别"""
# 语音识别模型
model = AutoModel(
model="paraformer-zh", # 中文语音识别
vad_model="fsmn-vad", # 语音活动检测
punc_model="ct-punc", # 标点恢复
spk_model="cam++", # 说话人识别（可选）
)
# 识别
result = model.generate(
input=audio_path,
batch_size_s=300, # 动态batch，每批最大300秒
hotword="", # 热词（提升特定词汇识别率）
)
if result:
for item in result:
text = item.get("text", "")
timestamp = item.get("timestamp", [])
print(f"识别结果: {text}")
if timestamp:
for ts in timestamp:
print(f" [{ts[0]/1000:.1f}s - {ts[1]/1000:.1f}s]")
return result
# 流式识别
def streaming_funasr():
"""FunASR 流式识别"""
model = AutoModel(model="paraformer-zh-streaming")
# 模拟流式输入
import numpy as np
chunk_size = [0, 10, 5] # 编码器chunk大小配置（秒）
# 分块处理音频
# audio = load_audio("test.wav")
# for i in range(0, len(audio), 16000): # 每秒一块
# chunk = audio[i:i+16000]
# result = model.generate(input=chunk, chunk_size=chunk_size)
# if result:
# print(f"部分结果: {result[0]['text']}")
# transcribe_with_funasr("test.wav")
第三章：语音合成（TTS）深度解析
3.1 TTS 技术演进
现代TTS已经从"机器人朗读"进化到了"真人级语音合成"：
"""
主流TTS技术对比和使用
"""
# === 1. Edge-TTS（微软免费TTS，入门首选）===
# pip install edge-tts
import edge_tts
import asyncio
async def edge_tts_demo(text: str, voice: str = "zh-CN-XiaoxiaoNeural"):
"""Edge TTS - 免费、高质量的在线TTS"""
communicate = edge_tts.Communicate(
text=text,
voice=voice, # 语音角色
rate="+0%", # 语速调节
volume="+0%", # 音量调节
pitch="+0Hz", # 音调调节
)
# 生成音频文件
await communicate.save("output_edge.mp3")
# 流式生成（适合实时场景）
audio_chunks = []
async for chunk in communicate.stream():
if chunk["type"] == "audio":
audio_chunks.append(chunk["data"])
print(f"生成完成，音频大小: {sum(len(c) for c in audio_chunks) / 1024:.1f} KB")
return audio_chunks
# 可用的中文语音
# zh-CN-XiaoxiaoNeural - 女声，活泼
# zh-CN-YunxiNeural - 男声，沉稳
# zh-CN-XiaoyiNeural - 女声，温柔
# zh-CN-YunjianNeural - 男声，播音
# asyncio.run(edge_tts_demo("你好，我是你的AI语音助手，很高兴认识你！"))
# === 2. CosyVoice（阿里开源，支持声音克隆）===
# pip install cosyvoice
def cosyvoice_demo():
"""CosyVoice - 高质量中文TTS，支持声音克隆"""
# 需要从 https://github.com/FunAudioLLM/CosyVoice 下载
# from cosyvoice.cli.cosyvoice import CosyVoice
# from cosyvoice.utils.file_utils import load_wav
# import torchaudio
#
# cosyvoice = CosyVoice('pretrained_models/CosyVoice-300M')
#
# # 1. 基础语音合成
# for i, j in cosyvoice.inference_sft(
# '你好，我是通义语音助手。',
# '中文女',
# ):
# torchaudio.save(f'output_sft_{i}.wav', j['tts_speech'], 22050)
#
# # 2. 声音克隆（需要3-10秒参考音频）
# prompt_speech = load_wav('reference.wav', 16000)
# for i, j in cosyvoice.inference_cross_lingual(
# '这是一段克隆声音的测试。',
# prompt_speech=prompt_speech,
# ):
# torchaudio.save(f'output_clone_{i}.wav', j['tts_speech'], 22050)
pass
# === 3. ChatTTS（开源对话式TTS）===
def chattts_demo():
"""ChatTTS - 对话场景优化的TTS"""
# pip install chattts
# import ChatTTS
# import torch
# import torchaudio
#
# chat = ChatTTS.Chat()
# chat.load(compile=False)
#
# # 生成随机说话人
# rand_speaker = chat.sample_random_speaker()
#
# # 推理
# params_infer = ChatTTS.Chat.InferCodeParams(
# spk_emb=rand_speaker,
# temperature=0.3, # 越低越稳定
# top_P=0.7,
# top_K=20,
# )
#
# params_refine = ChatTTS.Chat.RefineTextParams(
# prompt='[oral_2][laugh_0][break_6]', # 口语化、笑声、停顿标记
# )
#
# texts = ['你好啊，今天天气真不错！', '嗯...让我想想怎么回答你。']
# wavs = chat.infer(
# texts,
# params_infer_code=params_infer,
# params_refine_text=params_refine,
# )
#
# for i, wav in enumerate(wavs):
# torchaudio.save(f'chattts_{i}.wav', torch.from_numpy(wav), 24000)
pass
# === 4. GPT-SoVITS（声音克隆专用）===
def gpt_sovits_demo():
"""GPT-SoVITS - 少样本声音克隆"""
# 需要从 https://github.com/RVC-Boss/GPT-SoVITS 安装
#
# 核心流程：
# 1. 准备参考音频（3-10秒清晰语音）
# 2. 提取说话人特征
# 3. 用目标文本合成语音
#
# 优势：
# - 仅需3秒参考音频即可克隆
# - 支持中英日粤多语言
# - 情感和语调可调
pass
3.2 流式TTS实现
"""
流式TTS - 边生成边播放，降低感知延迟
"""
import asyncio
import numpy as np
from typing import AsyncGenerator
class StreamingTTS:
"""流式语音合成器"""
def __init__(self, engine: str = "edge-tts"):
self.engine = engine
self.sample_rate = 24000
async def synthesize_stream(
self,
text_stream: AsyncGenerator[str, None],
voice: str = "zh-CN-XiaoxiaoNeural",
) -> AsyncGenerator[bytes, None]:
"""
流式合成：接收文本流，输出音频流
核心策略：按句子分割，每个句子独立合成并立即输出
"""
buffer = ""
sentence_endings = {"。", "！", "？", "；", ".", "!", "?", "\n"}
async for text_chunk in text_stream:
buffer += text_chunk
# 检测完整句子
while True:
end_idx = -1
for i, char in enumerate(buffer):
if char in sentence_endings:
end_idx = i
break
if end_idx == -1:
break
# 提取完整句子
sentence = buffer[:end_idx + 1].strip()
buffer = buffer[end_idx + 1:]
if sentence:
# 合成这个句子
async for audio_chunk in self._synthesize_sentence(sentence, voice):
yield audio_chunk
# 处理剩余文本
if buffer.strip():
async for audio_chunk in self._synthesize_sentence(buffer.strip(), voice):
yield audio_chunk
async def _synthesize_sentence(
self,
sentence: str,
voice: str,
) -> AsyncGenerator[bytes, None]:
"""合成单个句子"""
if self.engine == "edge-tts":
import edge_tts
communicate = edge_tts.Communicate(sentence, voice=voice)
async for chunk in communicate.stream():
if chunk["type"] == "audio":
yield chunk["data"]
async def synthesize_with_timestamps(
self,
text: str,
voice: str = "zh-CN-XiaoxiaoNeural",
) -> list:
"""带时间戳的合成（用于字幕同步）"""
import edge_tts
communicate = edge_tts.Communicate(text, voice=voice)
audio_data = b""
word_boundaries = []
async for chunk in communicate.stream():
if chunk["type"] == "audio":
audio_data += chunk["data"]
elif chunk["type"] == "WordBoundary":
word_boundaries.append({
"text": chunk["text"],
"offset": chunk["offset"], # 微秒
"duration": chunk["duration"], # 微秒
})
return {
"audio": audio_data,
"word_boundaries": word_boundaries,
}
# 模拟LLM流式输出 + 流式TTS
async def simulate_streaming_tts():
"""演示流式TTS效果"""
tts = StreamingTTS(engine="edge-tts")
# 模拟LLM的流式文本输出
async def mock_llm_stream():
tokens = ["你好", "，我是", "你的AI", "语音助手", "。",
"今天", "天气", "真不错", "，", "要不要", "出去", "走走", "？"]
for token in tokens:
await asyncio.sleep(0.05) # 模拟生成延迟
yield token
# 流式合成并输出
audio_chunks = []
async for audio_chunk in tts.synthesize_stream(mock_llm_stream()):
audio_chunks.append(audio_chunk)
# 在实际应用中，这里会直接发送到音频播放器
print(f"收到音频块: {len(audio_chunk)} bytes")
total_size = sum(len(c) for c in audio_chunks)
print(f"总音频大小: {total_size / 1024:.1f} KB")
# asyncio.run(simulate_streaming_tts())
第四章：实时语音对话系统
4.1 完整的语音对话系统
"""
完整的实时语音对话系统
集成 ASR + LLM + TTS，支持流式处理
"""
import asyncio
import numpy as np
import time
from dataclasses import dataclass, field
from typing import Optional, Callable, AsyncGenerator
from collections import deque
from enum import Enum
class SystemState(Enum):
IDLE = "idle" # 空闲，等待用户说话
LISTENING = "listening" # 正在监听用户语音
THINKING = "thinking" # LLM正在思考
SPEAKING = "speaking" # 正在播放AI回复
@dataclass
class ConversationConfig:
"""对话系统配置"""
asr_model: str = "whisper-base"
llm_model: str = "gpt-4o"
tts_voice: str = "zh-CN-XiaoxiaoNeural"
language: str = "zh"
sample_rate: int = 16000
vad_threshold: float = 0.01
silence_timeout_ms: int = 800
max_turn_duration_s: int = 30
interrupt_enabled: bool = True # 允许用户打断AI说话
class RealtimeVoiceAssistant:
"""实时语音助手"""
def __init__(self, config: ConversationConfig):
self.config = config
self.state = SystemState.IDLE
self.conversation_history = []
self.audio_queue = deque()
self.state_change_callbacks = []
# 统计
self.stats = {
"total_turns": 0,
"avg_asr_latency_ms": 0,
"avg_llm_latency_ms": 0,
"avg_tts_latency_ms": 0,
"avg_e2e_latency_ms": 0,
}
def on_state_change(self, callback: Callable):
"""注册状态变化回调"""
self.state_change_callbacks.append(callback)
def _set_state(self, new_state: SystemState):
if self.state != new_state:
self.state = new_state
for cb in self.state_change_callbacks:
cb(new_state)
async def process_voice_input(
self,
audio_data: np.ndarray,
) -> AsyncGenerator[bytes, None]:
"""
处理一轮语音交互
1. ASR: 语音转文本
2. LLM: 文本生成回复
3. TTS: 文本转语音
Yields: 音频数据块
"""
turn_start = time.time()
# === 阶段1：语音识别 ===
self._set_state(SystemState.THINKING)
asr_start = time.time()
user_text = await self._asr(audio_data)
asr_latency = (time.time() - asr_start) * 1000
if not user_text.strip():
self._set_state(SystemState.IDLE)
return
print(f"👤 用户: {user_text}")
print(f" ASR延迟: {asr_latency:.0f}ms")
# 添加到对话历史
self.conversation_history.append({
"role": "user",
"content": user_text,
})
# === 阶段2：LLM生成 ===
llm_start = time.time()
full_response = ""
# 流式生成文本
async for text_chunk in self._llm_stream(self.conversation_history):
full_response += text_chunk
llm_latency = (time.time() - llm_start) * 1000
print(f"🤖 助手: {full_response}")
print(f" LLM延迟: {llm_latency:.0f}ms")
# 添加AI回复到历史
self.conversation_history.append({
"role": "assistant",
"content": full_response,
})
# === 阶段3：语音合成 ===
self._set_state(SystemState.SPEAKING)
tts_start = time.time()
async for audio_chunk in self._tts_stream(full_response):
yield audio_chunk
tts_latency = (time.time() - tts_start) * 1000
print(f" TTS延迟: {tts_latency:.0f}ms")
# 更新统计
e2e_latency = (time.time() - turn_start) * 1000
self.stats["total_turns"] += 1
self.stats["avg_asr_latency_ms"] = self._update_avg(
self.stats["avg_asr_latency_ms"], asr_latency
)
self.stats["avg_llm_latency_ms"] = self._update_avg(
self.stats["avg_llm_latency_ms"], llm_latency
)
self.stats["avg_tts_latency_ms"] = self._update_avg(
self.stats["avg_tts_latency_ms"], tts_latency
)
self.stats["avg_e2e_latency_ms"] = self._update_avg(
self.stats["avg_e2e_latency_ms"], e2e_latency
)
print(f" 端到端延迟: {e2e_latency:.0f}ms")
self._set_state(SystemState.IDLE)
def _update_avg(self, old_avg: float, new_val: float) -> float:
n = self.stats["total_turns"]
return old_avg * (n - 1) / n + new_val / n
async def _asr(self, audio: np.ndarray) -> str:
"""语音识别（实现时接入实际ASR模型）"""
# 实际实现：
# import whisper
# model = whisper.load_model(self.config.asr_model)
# result = model.transcribe(audio, language=self.config.language)
# return result["text"]
await asyncio.sleep(0.1) # 模拟延迟
return "（模拟识别结果）你好，今天天气怎么样？"
async def _llm_stream(
self,
messages: list,
) -> AsyncGenerator[str, None]:
"""流式LLM生成（实现时接入实际LLM）"""
# 实际实现：
# from openai import AsyncOpenAI
# client = AsyncOpenAI()
# stream = await client.chat.completions.create(
# model=self.config.llm_model,
# messages=messages,
# stream=True,
# )
# async for chunk in stream:
# if chunk.choices[0].delta.content:
# yield chunk.choices[0].delta.content
# 模拟流式输出
response = "今天天气非常好！阳光明媚，温度适宜，很适合出去散步或者运动。"
for char in response:
await asyncio.sleep(0.02)
yield char
async def _tts_stream(self, text: str) -> AsyncGenerator[bytes, None]:
"""流式语音合成（实现时接入实际TTS）"""
# 实际实现：
# import edge_tts
# communicate = edge_tts.Communicate(text, voice=self.config.tts_voice)
# async for chunk in communicate.stream():
# if chunk["type"] == "audio":
# yield chunk["data"]
# 模拟音频输出
for i in range(0, len(text), 10):
await asyncio.sleep(0.05)
yield b'\x00' * 480 # 模拟音频块
# 使用示例
async def main():
config = ConversationConfig(
asr_model="whisper-base",
llm_model="gpt-4o",
tts_voice="zh-CN-XiaoxiaoNeural",
)
assistant = RealtimeVoiceAssistant(config)
# 注册状态回调
assistant.on_state_change(lambda s: print(f"状态变化: {s.value}"))
# 模拟一轮对话
mock_audio = np.random.randint(-32768, 32767, size=16000, dtype=np.int16)
async for audio_chunk in assistant.process_voice_input(mock_audio):
# 在实际应用中，这里会发送到音频播放器
pass
print(f"\n统计: {assistant.stats}")
# asyncio.run(main())
4.2 WebSocket 实时语音服务
"""
基于 WebSocket 的实时语音对话服务
"""
import asyncio
import json
import websockets
import numpy as np
import time
from typing import Optional
class VoiceWebSocketServer:
"""WebSocket 语音对话服务端"""
def __init__(self, host: str = "0.0.0.0", port: int = 8765):
self.host = host
self.port = port
self.clients = {}
async def handle_client(self, websocket, path):
"""处理单个客户端连接"""
client_id = str(id(websocket))
self.clients[client_id] = {
"websocket": websocket,
"conversation": [],
"audio_buffer": bytearray(),
}
print(f"客户端连接: {client_id}")
try:
async for message in websocket:
if isinstance(message, bytes):
# 音频数据
await self._handle_audio(client_id, message)
else:
# JSON 控制消息
data = json.loads(message)
await self._handle_control(client_id, data)
except websockets.exceptions.ConnectionClosed:
print(f"客户端断开: {client_id}")
finally:
del self.clients[client_id]
async def _handle_audio(self, client_id: str, audio_data: bytes):
"""处理接收到的音频数据"""
client = self.clients[client_id]
client["audio_buffer"].extend(audio_data)
# 检查是否检测到语音结束（简单实现）
audio_np = np.frombuffer(audio_data, dtype=np.int16)
energy = np.sqrt(np.mean(audio_np.astype(float) ** 2)) / 32768
if energy < 0.01: # 静音
client["silence_count"] = client.get("silence_count", 0) + 1
else:
client["silence_count"] = 0
# 静音超过阈值，处理语音
if client.get("silence_count", 0) > 20: # 约600ms静音
audio_buffer = bytes(client["audio_buffer"])
client["audio_buffer"] = bytearray()
client["silence_count"] = 0
if len(audio_buffer) > 3200: # 至少100ms音频
await self._process_speech(client_id, audio_buffer)
async def _process_speech(self, client_id: str, audio_data: bytes):
"""处理一段完整语音"""
client = self.clients[client_id]
ws = client["websocket"]
# 1. ASR
await ws.send(json.dumps({"type": "status", "state": "listening"}))
user_text = await self._asr(audio_data)
await ws.send(json.dumps({
"type": "transcript",
"text": user_text,
"role": "user",
}))
# 2. LLM 流式生成
await ws.send(json.dumps({"type": "status", "state": "thinking"}))
full_response = ""
async for chunk in self._llm_stream(client_id, user_text):
full_response += chunk
await ws.send(json.dumps({
"type": "text_chunk",
"text": chunk,
}))
await ws.send(json.dumps({
"type": "response_complete",
"text": full_response,
}))
# 3. TTS 流式合成并发送
await ws.send(json.dumps({"type": "status", "state": "speaking"}))
async for audio_chunk in self._tts_stream(full_response):
await ws.send(audio_chunk) # 二进制音频数据
await ws.send(json.dumps({"type": "audio_complete"}))
await ws.send(json.dumps({"type": "status", "state": "idle"}))
async def _asr(self, audio_data: bytes) -> str:
"""ASR处理"""
await asyncio.sleep(0.2)
return "（识别结果）你好"
async def _llm_stream(self, client_id: str, text: str):
"""LLM流式生成"""
response = "你好！很高兴和你聊天。有什么我可以帮你的吗？"
for char in response:
await asyncio.sleep(0.03)
yield char
async def _tts_stream(self, text: str):
"""TTS流式合成"""
# 实际使用 edge-tts 或其他TTS引擎
for i in range(10):
await asyncio.sleep(0.05)
yield b'\x00' * 960
async def _handle_control(self, client_id: str, data: dict):
"""处理控制消息"""
msg_type = data.get("type")
if msg_type == "config":
# 更新客户端配置
self.clients[client_id]["config"] = data.get("config", {})
print(f"客户端 {client_id} 更新配置: {data['config']}")
elif msg_type == "interrupt":
# 用户打断（停止当前TTS播放）
self.clients[client_id]["interrupted"] = True
print(f"客户端 {client_id} 请求打断")
async def start(self):
"""启动WebSocket服务"""
print(f"语音对话服务启动: ws://{self.host}:{self.port}")
async with websockets.serve(self.handle_client, self.host, self.port):
await asyncio.Future() # 永久运行
# 启动服务
# server = VoiceWebSocketServer()
# asyncio.run(server.start())
// === 浏览器端 JavaScript 客户端 ===
/*
class VoiceClient {
constructor(wsUrl) {
this.ws = new WebSocket(wsUrl);
this.ws.binaryType = 'arraybuffer';
this.mediaRecorder = null;
this.audioContext = null;
this.ws.onmessage = (event) => {
if (event.data instanceof ArrayBuffer) {
// 音频数据，直接播放
this.playAudio(event.data);
} else {
// JSON控制消息
const msg = JSON.parse(event.data);
this.handleMessage(msg);
}
};
}
async startListening() {
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
this.audioContext = new AudioContext({ sampleRate: 16000 });
const source = this.audioContext.createMediaStreamSource(stream);
const processor = this.audioContext.createScriptProcessor(4096, 1, 1);
processor.onaudioprocess = (e) => {
const audioData = e.inputBuffer.getChannelData(0);
// 转换为16bit PCM
const pcm = new Int16Array(audioData.length);
for (let i = 0; i < audioData.length; i++) {
pcm[i] = Math.max(-32768, Math.min(32767, audioData[i] * 32768));
}
this.ws.send(pcm.buffer);
};
source.connect(processor);
processor.connect(this.audioContext.destination);
}
playAudio(arrayBuffer) {
// 播放接收到的音频
this.audioContext.decodeAudioData(arrayBuffer, (buffer) => {
const source = this.audioContext.createBufferSource();
source.buffer = buffer;
source.connect(this.audioContext.destination);
source.start();
});
}
handleMessage(msg) {
switch(msg.type) {
case 'transcript':
console.log(`[${msg.role}]: ${msg.text}`);
break;
case 'text_chunk':
// 实时显示AI回复文本
break;
case 'status':
console.log(`状态: ${msg.state}`);
break;
}
}
}
*/
第五章：情感识别与表达
5.1 语音情感识别
"""
语音情感识别 (Speech Emotion Recognition, SER)
"""
# pip install transformers torch torchaudio
import torch
import torchaudio
from transformers import pipeline
def recognize_emotion(audio_path: str) -> dict:
"""
识别语音中的情感
使用预训练的情感识别模型
"""
# 方法1：使用 HuggingFace pipeline
classifier = pipeline(
"audio-classification",
model="ehcalabres/wav2vec2-lg-xlsr-en-speech-emotion-recognition",
)
results = classifier(audio_path)
print("情感识别结果:")
for r in results:
print(f" {r['label']}: {r['score']:.3f}")
return results
def extract_prosody_features(audio_path: str) -> dict:
"""
提取韵律特征（用于情感分析）
韵律特征包括：
- 语速（音节/秒）
- 音高（基频F0）
- 音量（能量）
- 音色（频谱特征）
"""
import numpy as np
# 加载音频
waveform, sample_rate = torchaudio.load(audio_path)
# 转为单声道
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
audio = waveform.squeeze().numpy()
# 1. 计算能量（音量）
frame_length = int(0.025 * sample_rate) # 25ms帧
hop_length = int(0.010 * sample_rate) # 10ms步长
energy = []
for i in range(0, len(audio) - frame_length, hop_length):
frame = audio[i:i + frame_length]
energy.append(np.sqrt(np.mean(frame ** 2)))
energy = np.array(energy)
# 2. 估计基频F0（简化版自相关方法）
def estimate_f0(frame, sr, min_freq=50, max_freq=500):
"""自相关法估计基频"""
correlation = np.correlate(frame, frame, mode='full')
correlation = correlation[len(correlation) // 2:]
min_lag = sr // max_freq
max_lag = sr // min_freq
if max_lag > len(correlation):
return 0
search = correlation[min_lag:max_lag]
if len(search) == 0:
return 0
lag = np.argmax(search) + min_lag
if lag == 0:
return 0
return sr / lag
f0_values = []
for i in range(0, len(audio) - frame_length, hop_length):
frame = audio[i:i + frame_length]
f0 = estimate_f0(frame, sample_rate)
if 50 < f0 < 500: # 有效F0范围
f0_values.append(f0)
f0_values = np.array(f0_values) if f0_values else np.array([0])
# 3. 语速估计（基于能量变化的过零率）
zero_crossings = np.sum(np.abs(np.diff(np.sign(audio))) > 0)
duration = len(audio) / sample_rate
features = {
"duration_s": duration,
"mean_energy": float(np.mean(energy)),
"max_energy": float(np.max(energy)),
"energy_std": float(np.std(energy)),
"mean_f0": float(np.mean(f0_values)),
"f0_std": float(np.std(f0_values)),
"f0_range": float(np.max(f0_values) - np.min(f0_values)) if len(f0_values) > 1 else 0,
"zero_crossing_rate": zero_crossings / len(audio),
}
# 情感推断（基于韵律特征的简单规则）
if features["mean_energy"] > 0.05 and features["f0_std"] > 50:
features["inferred_emotion"] = "excited/angry"
elif features["mean_f0"] > 200 and features["f0_std"] > 30:
features["inferred_emotion"] = "happy"
elif features["mean_energy"] < 0.02 and features["mean_f0"] < 150:
features["inferred_emotion"] = "sad"
else:
features["inferred_emotion"] = "neutral"
print("韵律特征:")
for k, v in features.items():
print(f" {k}: {v}")
return features
# 使用示例
# recognize_emotion("happy_speech.wav")
# extract_prosody_features("speech.wav")
5.2 情感化语音合成
"""
带情感控制的语音合成
"""
class EmotionTTS:
"""情感化语音合成器"""
# 情感到语音参数的映射
EMOTION_PARAMS = {
"neutral": {
"rate": "+0%",
"pitch": "+0Hz",
"volume": "+0%",
"voice": "zh-CN-XiaoxiaoNeural",
},
"happy": {
"rate": "+10%",
"pitch": "+10Hz",
"volume": "+5%",
"voice": "zh-CN-XiaoxiaoNeural",
},
"sad": {
"rate": "-15%",
"pitch": "-5Hz",
"volume": "-10%",
"voice": "zh-CN-XiaoyiNeural",
},
"angry": {
"rate": "+5%",
"pitch": "-5Hz",
"volume": "+15%",
"voice": "zh-CN-YunjianNeural",
},
"excited": {
"rate": "+15%",
"pitch": "+15Hz",
"volume": "+10%",
"voice": "zh-CN-XiaoxiaoNeural",
},
"calm": {
"rate": "-10%",
"pitch": "-3Hz",
"volume": "-5%",
"voice": "zh-CN-XiaoyiNeural",
},
}
def __init__(self):
self.current_emotion = "neutral"
def set_emotion(self, emotion: str):
"""设置当前情感"""
if emotion in self.EMOTION_PARAMS:
self.current_emotion = emotion
async def synthesize(self, text: str, emotion: str = None) -> bytes:
"""合成带情感的语音"""
import edge_tts
emotion = emotion or self.current_emotion
params = self.EMOTION_PARAMS.get(emotion, self.EMOTION_PARAMS["neutral"])
# 根据情感添加适当的文本修饰
modified_text = self._apply_emotion_text_modifiers(text, emotion)
communicate = edge_tts.Communicate(
text=modified_text,
voice=params["voice"],
rate=params["rate"],
pitch=params["pitch"],
volume=params["volume"],
)
audio_data = b""
async for chunk in communicate.stream():
if chunk["type"] == "audio":
audio_data += chunk["data"]
return audio_data
def _apply_emotion_text_modifiers(self, text: str, emotion: str) -> str:
"""
根据情感修改文本，增强表达效果
例如：
- 开心时添加语气词
- 悲伤时添加停顿
- 惊讶时添加感叹词
"""
if emotion == "happy":
# 添加开心的语气
if not text.endswith(("！", "~", "😊")):
text = text.rstrip("。") + "！"
elif emotion == "sad":
# 添加停顿，语速变慢
text = text.replace("，", "，... ")
text = text.replace("。", "... 。")
elif emotion == "excited":
# 加强感叹
text = text.replace("！", "！！")
if "！" not in text:
text = text.rstrip("。") + "！！"
elif emotion == "angry":
# 加强语气
if not text.endswith(("！", "!!!")):
text = text.rstrip("。") + "！"
return text
def detect_and_respond_emotion(self, user_text: str, user_audio_features: dict) -> tuple:
"""
根据用户情感调整AI回复的语调
策略：
- 用户开心 → AI也用开心的语调
- 用户悲伤 → AI用温柔安慰的语调
- 用户愤怒 → AI用冷静平和的语调
"""
user_emotion = user_audio_features.get("inferred_emotion", "neutral")
# 情感响应策略
response_emotion_map = {
"happy": "happy", # 积极回应
"sad": "calm", # 温柔回应
"angry": "calm", # 冷静回应
"excited": "happy", # 配合兴奋
"neutral": "neutral", # 正常回应
}
ai_emotion = response_emotion_map.get(user_emotion, "neutral")
# 生成回复文本时考虑情感
if user_emotion == "sad":
response_prefix = "我理解你的感受，"
elif user_emotion == "angry":
response_prefix = "我明白你的不满，"
elif user_emotion == "happy":
response_prefix = "很高兴听到这个！"
else:
response_prefix = ""
return ai_emotion, response_prefix
# 使用示例
async def emotion_demo():
tts = EmotionTTS()
# 不同情感的合成
test_cases = [
("今天天气真好！", "happy"),
("我很抱歉听到这个消息。", "sad"),
("你说什么？！", "angry"),
("太棒了，我们成功了！", "excited"),
("请告诉我更多信息。", "neutral"),
]
for text, emotion in test_cases:
print(f"文本: {text} | 情感: {emotion}")
# audio = await tts.synthesize(text, emotion)
# 保存或播放 audio
# asyncio.run(emotion_demo())
第六章：声音克隆与个性化
6.1 基于 Coqui XTTS 的声音克隆
"""
XTTS v2 声音克隆
只需6秒参考音频即可克隆任意声音
"""
# pip install TTS
from TTS.api import TTS
def clone_voice_demo():
"""XTTS v2 声音克隆演示"""
# 加载XTTS v2模型
tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2")
# 方法1：使用参考音频克隆声音
tts.tts_to_file(
text="你好，这是克隆声音的测试。我正在用你的声音说话。",
speaker_wav="reference_voice.wav", # 6-30秒的参考音频
language="zh",
file_path="cloned_output.wav",
)
print("声音克隆完成！已保存到 cloned_output.wav")
def multi_speaker_clone():
"""多说话人声音克隆"""
tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2")
# 定义多个说话人的参考音频
speakers = {
"张三": "zhangsan_voice.wav",
"李四": "lisi_voice.wav",
"王五": "wangwu_voice.wav",
}
# 为不同角色生成对话
dialogue = [
("张三", "大家好，今天会议开始。"),
("李四", "好的，我先汇报一下进度。"),
("王五", "我有几点补充意见。"),
]
for speaker, text in dialogue:
ref_audio = speakers[speaker]
tts.tts_to_file(
text=text,
speaker_wav=ref_audio,
language="zh",
file_path=f"dialogue_{speaker}.wav",
)
print(f"已生成 {speaker} 的语音: dialogue_{speaker}.wav")
# clone_voice_demo()
6.2 声音特征提取与管理
"""
声音特征管理系统
用于存储、检索和应用不同的声音特征
"""
import json
import hashlib
import numpy as np
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional, Dict
@dataclass
class VoiceProfile:
"""声音档案"""
id: str
name: str
language: str
gender: str # "male" / "female"
age_range: str # "child" / "young" / "middle" / "senior"
description: str
reference_audio_path: str
embedding: Optional[list] = None # 声音特征向量
tags: list = None
def __post_init__(self):
if self.tags is None:
self.tags = []
class VoiceProfileManager:
"""声音档案管理器"""
def __init__(self, storage_dir: str = "./voice_profiles"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(parents=True, exist_ok=True)
self.profiles: Dict[str, VoiceProfile] = {}
self._load_profiles()
def _load_profiles(self):
"""从磁盘加载所有声音档案"""
for profile_file in self.storage_dir.glob("*.json"):
with open(profile_file, "r", encoding="utf-8") as f:
data = json.load(f)
profile = VoiceProfile(**data)
self.profiles[profile.id] = profile
def create_profile(
self,
name: str,
reference_audio_path: str,
language: str = "zh",
gender: str = "unknown",
age_range: str = "young",
description: str = "",
tags: list = None,
) -> VoiceProfile:
"""创建新的声音档案"""
# 生成唯一ID
audio_hash = hashlib.md5(
f"{name}{reference_audio_path}".encode()
).hexdigest()[:8]
profile_id = f"voice_{audio_hash}"
profile = VoiceProfile(
id=profile_id,
name=name,
language=language,
gender=gender,
age_range=age_range,
description=description,
reference_audio_path=reference_audio_path,
tags=tags or [],
)
# 提取声音嵌入（实际实现需要声音编码模型）
# profile.embedding = self._extract_embedding(reference_audio_path)
# 保存
self.profiles[profile_id] = profile
self._save_profile(profile)
print(f"创建声音档案: {name} (ID: {profile_id})")
return profile
def _save_profile(self, profile: VoiceProfile):
"""保存声音档案到磁盘"""
file_path = self.storage_dir / f"{profile.id}.json"
with open(file_path, "w", encoding="utf-8") as f:
json.dump(asdict(profile), f, ensure_ascii=False, indent=2)
def get_profile(self, profile_id: str) -> Optional[VoiceProfile]:
"""获取声音档案"""
return self.profiles.get(profile_id)
def search_profiles(
self,
language: str = None,
gender: str = None,
tags: list = None,
) -> list:
"""搜索声音档案"""
results = []
for profile in self.profiles.values():
if language and profile.language != language:
continue
if gender and profile.gender != gender:
continue
if tags and not any(t in profile.tags for t in tags):
continue
results.append(profile)
return results
def list_profiles(self):
"""列出所有声音档案"""
print(f"\n已保存 {len(self.profiles)} 个声音档案:")
for pid, profile in self.profiles.items():
print(f" [{pid}] {profile.name} - {profile.language} "
f"({profile.gender}, {profile.age_range})")
if profile.description:
print(f" {profile.description}")
# 使用示例
# manager = VoiceProfileManager()
#
# # 创建声音档案
# manager.create_profile(
# name="小明",
# reference_audio_path="xiaoming_voice.wav",
# language="zh",
# gender="male",
# age_range="young",
# description="年轻男性，声音清亮",
# tags=["播音", "年轻", "男声"],
# )
#
# # 搜索
# male_voices = manager.search_profiles(gender="male")
# chinese_voices = manager.search_profiles(language="zh")
第七章：多语言语音交互
7.1 多语言ASR
"""
多语言语音识别
"""
import whisper
import torch
class MultilingualASR:
"""多语言语音识别器"""
# Whisper 支持的语言代码
SUPPORTED_LANGUAGES = {
"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: str = "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)
# 排序返回前5个最可能的语言
sorted_langs = sorted(probs.items(), key=lambda x: x[1], reverse=True)[:5]
print("语言检测结果:")
for lang, prob in sorted_langs:
lang_name = self.SUPPORTED_LANGUAGES.get(lang, lang)
print(f" {lang_name} ({lang}): {prob:.3f}")
return sorted_langs[0][0], probs
def transcribe_multilingual(self, audio_path: str) -> dict:
"""
多语言转录
自动检测语言并转录，支持混合语言
"""
result = self.model.transcribe(
audio_path,
task="transcribe", # 保持原语言
beam_size=5,
best_of=5,
condition_on_previous_text=True,
# 混合语言支持
initial_prompt="", # 空提示以支持多语言
)
detected_lang = result["language"]
lang_name = self.SUPPORTED_LANGUAGES.get(detected_lang, detected_lang)
print(f"检测语言: {lang_name} ({detected_lang})")
print(f"转录结果: {result['text']}")
return result
def translate_to_english(self, audio_path: str) -> str:
"""任意语言翻译成英文"""
result = self.model.transcribe(
audio_path,
task="translate", # Whisper 内置翻译功能
beam_size=5,
)
print(f"原文语言: {result['language']}")
print(f"英文翻译: {result['text']}")
return result["text"]
# 使用示例
# asr = MultilingualASR(model_size="large-v3")
#
# # 自动检测语言并转录
# asr.transcribe_multilingual("chinese_speech.wav")
# asr.transcribe_multilingual("english_speech.wav")
# asr.transcribe_multilingual("japanese_speech.wav")
#
# # 翻译成英文
# asr.translate_to_english("chinese_speech.wav")
7.2 多语言TTS
"""
多语言语音合成
"""
class MultilingualTTS:
"""多语言语音合成器"""
# 各语言推荐的 Edge-TTS 语音
VOICE_MAP = {
"zh": {
"female": "zh-CN-XiaoxiaoNeural",
"male": "zh-CN-YunxiNeural",
},
"en": {
"female": "en-US-JennyNeural",
"male": "en-US-GuyNeural",
},
"ja": {
"female": "ja-JP-NanamiNeural",
"male": "ja-JP-KeitaNeural",
},
"ko": {
"female": "ko-KR-SunHiNeural",
"male": "ko-KR-InJoonNeural",
},
"fr": {
"female": "fr-FR-DeniseNeural",
"male": "fr-FR-HenriNeural",
},
"de": {
"female": "de-DE-KatjaNeural",
"male": "de-DE-ConradNeural",
},
"es": {
"female": "es-ES-ElviraNeural",
"male": "es-ES-AlvaroNeural",
},
}
async def synthesize(
self,
text: str,
language: str = "zh",
gender: str = "female",
output_path: str = "output.wav",
) -> bytes:
"""多语言语音合成"""
import edge_tts
voice = self.VOICE_MAP.get(language, {}).get(gender)
if not voice:
raise ValueError(f"不支持的语言或性别: {language}/{gender}")
communicate = edge_tts.Communicate(text, voice=voice)
audio_data = b""
async for chunk in communicate.stream():
if chunk["type"] == "audio":
audio_data += chunk["data"]
with open(output_path, "wb") as f:
f.write(audio_data)
print(f"已合成 {language} 语音: {output_path} (voice: {voice})")
return audio_data
async def code_switch_synthesize(
self,
text_segments: list, # [(text, language), ...]
gender: str = "female",
output_path: str = "code_switch.wav",
):
"""
代码切换合成 - 同一段话中混合多种语言
例如："今天的topic是机器学习，Let's开始吧"
"""
import edge_tts
all_audio = b""
for text, language in text_segments:
voice = self.VOICE_MAP.get(language, {}).get(gender)
if not voice:
continue
communicate = edge_tts.Communicate(text, voice=voice)
async for chunk in communicate.stream():
if chunk["type"] == "audio":
all_audio += chunk["data"]
with open(output_path, "wb") as f:
f.write(all_audio)
print(f"代码切换语音已保存: {output_path}")
# 使用示例
async def multilingual_demo():
tts = MultilingualTTS()
# 单语言合成
await tts.synthesize("你好世界", language="zh", gender="female")
await tts.synthesize("Hello World", language="en", gender="female")
await tts.synthesize("こんにちは世界", language="ja", gender="female")
# 代码切换
await tts.code_switch_synthesize([
("今天我们来讨论", "zh"),
("machine learning", "en"),
("的基本概念", "zh"),
])
# asyncio.run(multilingual_demo())
第八章：语音Agent构建
8.1 多工具语音Agent
"""
语音Agent - 能够调用工具的语音助手
"""
import asyncio
import json
from typing import Dict, Any, Callable, Awaitable
class VoiceAgent:
"""语音Agent - 集成工具调用能力"""
def __init__(self):
self.tools: Dict[str, Callable] = {}
self.tool_descriptions: list = []
self.conversation_history = []
self.system_prompt = """你是一个智能语音助手。你可以帮助用户完成各种任务。
当需要执行操作时，请使用可用的工具。
回复要简洁自然，适合语音播报。"""
def register_tool(
self,
name: str,
description: str,
parameters: dict,
func: Callable[..., Awaitable],
):
"""注册工具"""
self.tools[name] = func
self.tool_descriptions.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters,
}
})
print(f"注册工具: {name} - {description}")
async def process_voice_command(self, user_text: str) -> str:
"""处理语音命令"""
self.conversation_history.append({
"role": "user",
"content": user_text,
})
# 调用LLM，带工具定义
response = await self._call_llm_with_tools()
# 检查是否需要调用工具
if response.get("tool_calls"):
# 执行工具调用
tool_results = []
for tool_call in response["tool_calls"]:
func_name = tool_call["function"]["name"]
func_args = json.loads(tool_call["function"]["arguments"])
print(f" 🔧 调用工具: {func_name}({func_args})")
if func_name in self.tools:
result = await self.tools[func_name](**func_args)
tool_results.append({
"tool_call_id": tool_call["id"],
"output": json.dumps(result, ensure_ascii=False),
})
# 将工具结果加入对话历史
self.conversation_history.append({
"role": "assistant",
"tool_calls": response["tool_calls"],
})
for tr in tool_results:
self.conversation_history.append({
"role": "tool",
"tool_call_id": tr["tool_call_id"],
"content": tr["output"],
})
# 再次调用LLM生成最终回复
final_response = await self._call_llm_with_tools()
assistant_text = final_response["content"]
else:
assistant_text = response["content"]
self.conversation_history.append({
"role": "assistant",
"content": assistant_text,
})
return assistant_text
async def _call_llm_with_tools(self) -> dict:
"""调用LLM（带工具定义）"""
# 实际实现使用 OpenAI API
# from openai import AsyncOpenAI
# client = AsyncOpenAI()
# response = await client.chat.completions.create(
# model="gpt-4o",
# messages=self.conversation_history,
# tools=self.tool_descriptions if self.tool_descriptions else None,
# )
# return response.choices[0].message
# 模拟返回
return {"content": "（模拟回复）好的，我来帮你处理。"}
# 创建语音Agent并注册工具
async def build_voice_agent():
agent = VoiceAgent()
# 注册工具：查询天气
async def get_weather(city: str) -> dict:
# 实际实现会调用天气API
return {
"city": city,
"temperature": "25°C",
"weather": "晴",
"humidity": "45%",
}
agent.register_tool(
name="get_weather",
description="查询指定城市的天气信息",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如：北京、上海",
}
},
"required": ["city"],
},
func=get_weather,
)
# 注册工具：设置闹钟
async def set_alarm(time: str, label: str) -> dict:
return {"status": "success", "alarm_time": time, "label": label}
agent.register_tool(
name="set_alarm",
description="设置闹钟提醒",
parameters={
"type": "object",
"properties": {
"time": {"type": "string", "description": "时间，如：08:30"},
"label": {"type": "string", "description": "闹钟标签"},
},
"required": ["time", "label"],
},
func=set_alarm,
)
# 注册工具：控制智能家居
async def control_device(device: str, action: str, value: Any = None) -> dict:
return {
"device": device,
"action": action,
"value": value,
"status": "success",
}
agent.register_tool(
name="control_device",
description="控制智能家居设备（灯、空调、窗帘等）",
parameters={
"type": "object",
"properties": {
"device": {"type": "string", "description": "设备名称"},
"action": {"type": "string", "description": "操作：turn_on/turn_off/set"},
"value": {"description": "设置值（可选）"},
},
"required": ["device", "action"],
},
func=control_device,
)
# 模拟语音命令
commands = [
"今天北京天气怎么样？",
"帮我设一个明早8点的闹钟",
"把客厅的灯关掉",
]
for cmd in commands:
print(f"\n👤 用户: {cmd}")
response = await agent.process_voice_command(cmd)
print(f"🤖 助手: {response}")
# asyncio.run(build_voice_agent())
第九章：低延迟优化策略
9.1 端到端延迟分析
"""
语音对话系统的延迟分析与优化
"""
import time
from dataclasses import dataclass
from typing import List
@dataclass
class LatencyBreakdown:
"""延迟分解"""
vad_ms: float = 0 # 语音活动检测
asr_ms: float = 0 # 语音识别
llm_first_token_ms: float = 0 # LLM首token延迟
llm_total_ms: float = 0 # LLM总延迟
tts_first_audio_ms: float = 0 # TTS首帧音频延迟
tts_total_ms: float = 0 # TTS总延迟
network_ms: float = 0 # 网络传输
audio_playback_ms: float = 0 # 音频播放缓冲
@property
def total_ms(self) -> float:
return (self.vad_ms + self.asr_ms + self.llm_first_token_ms +
self.tts_first_audio_ms + self.network_ms + self.audio_playback_ms)
@property
def user_perceived_ms(self) -> float:
"""用户感知延迟（从说完话到听到第一个音频）"""
return self.asr_ms + self.llm_first_token_ms + self.tts_first_audio_ms
def report(self) -> str:
"""生成延迟报告"""
lines = [
"=" * 50,
"延迟分析报告",
"=" * 50,
f"VAD检测: {self.vad_ms:>8.0f} ms",
f"ASR识别: {self.asr_ms:>8.0f} ms",
f"LLM首Token: {self.llm_first_token_ms:>8.0f} ms",
f"LLM总生成: {self.llm_total_ms:>8.0f} ms",
f"TTS首帧音频: {self.tts_first_audio_ms:>8.0f} ms",
f"TTS总合成: {self.tts_total_ms:>8.0f} ms",
f"网络传输: {self.network_ms:>8.0f} ms",
f"播放缓冲: {self.audio_playback_ms:>8.0f} ms",
"-" * 50,
f"用户感知延迟: {self.user_perceived_ms:>8.0f} ms",
f"总端到端延迟: {self.total_ms:>8.0f} ms",
"=" * 50,
]
# 优化建议
suggestions = []
if self.asr_ms > 300:
suggestions.append("ASR: 使用更小的模型（base/small）或流式识别")
if self.llm_first_token_ms > 500:
suggestions.append("LLM: 使用更快的模型或减少系统提示长度")
if self.tts_first_audio_ms > 300:
suggestions.append("TTS: 使用流式合成，按句子分段输出")
if self.network_ms > 100:
suggestions.append("网络: 使用WebSocket长连接，部署在用户附近")
if suggestions:
lines.append("\n优化建议:")
for s in suggestions:
lines.append(f" ⚡ {s}")
return "\n".join(lines)
class LatencyOptimizer:
"""延迟优化策略"""
@staticmethod
def get_optimization_strategies() -> dict:
return {
"ASR优化": {
"流式识别": "使用流式ASR，边听边识别，不等说完再处理",
"VAD优化": "使用Silero VAD，响应时间<10ms",
"模型选择": "实时场景用base/small，非实时用large",
"热词加速": "使用热词列表提高特定词汇识别速度",
"GPU加速": "确保ASR在GPU上运行",
},
"LLM优化": {
"流式生成": "使用streaming API，第一个token就返回",
"减少上下文": "只保留最近几轮对话历史",
"系统提示": "精简系统提示，减少处理时间",
"投机解码": "使用投机解码加速生成",
"本地部署": "使用vLLM本地部署，减少网络延迟",
},
"TTS优化": {
"流式合成": "按句子分段合成，不等全部文本生成完",
"预测合成": "在LLM生成过程中就开始合成已完成的句子",
"缓存常用回复": "高频回复预合成缓存",
"低延迟模型": "使用FastSpeech2等低延迟TTS模型",
"双缓冲播放": "合成和播放并行，减少等待",
},
"架构优化": {
"并行处理": "ASR/LLM/TTS流水线并行",
"预测响应": "根据上下文预测可能的回复并预生成",
"边缘部署": "将模型部署在用户附近",
"协议优化": "使用WebSocket或WebRTC替代HTTP",
},
}
@staticmethod
def print_optimization_guide():
"""打印优化指南"""
strategies = LatencyOptimizer.get_optimization_strategies()
print("\n" + "=" * 60)
print("语音对话系统延迟优化指南")
print("=" * 60)
for category, items in strategies.items():
print(f"\n📌 {category}:")
for name, desc in items.items():
print(f" • {name}: {desc}")
# 生成延迟报告示例
breakdown = LatencyBreakdown(
vad_ms=50,
asr_ms=250,
llm_first_token_ms=300,
llm_total_ms=1500,
tts_first_audio_ms=150,
tts_total_ms=800,
network_ms=30,
audio_playback_ms=50,
)
print(breakdown.report())
# 打印优化指南
LatencyOptimizer.print_optimization_guide()
总结
构建一个优秀的Speech-to-Speech语音AI系统需要整合多个技术栈：
- ASR选择：Whisper适合多语言场景，FunASR中文效果更好，流式场景需要VAD配合
- TTS选择：Edge-TTS免费且质量高，CosyVoice/GPT-SoVITS适合声音克隆场景
- 架构设计：级联方案最实用，端到端方案是未来趋势
- 延迟优化：流式处理是关键，ASR/LLM/TTS流水线并行可将感知延迟降到500ms以内
- 情感表达：韵律控制+情感TTS让AI语音更自然
- 个性化：声音克隆技术让每个用户都能拥有独特的AI助手声音
随着GPT-4o Realtime、Gemini Live等产品的推出，语音AI正在成为人机交互的新范式。掌握这些技术，你就能构建出真正自然流畅的语音交互体验。
本教程持续更新中，涵盖最新的语音AI技术和最佳实践。
