多模态语音合成与识别：从 Whisper + TTS 到实时语音交互的 LLM 实践

import librosa
import numpy as np

# 加载音频
audio, sr = librosa.load("speech.wav", sr=16000)  # 16kHz

# 计算梅尔频谱
mel_spec = librosa.feature.melspectrogram(
    y=audio,
    sr=sr,
    n_mels=80,           # 梅尔滤波器数量
    n_fft=400,           # FFT 窗口大小（25ms @ 16kHz）
    hop_length=160,      # 帧移（10ms）
    f_min=0, f_max=8000  # 频率范围
)

# 转换为对数尺度（更符合人耳感知）
log_mel_spec = librosa.power_to_db(mel_spec, ref=np.max)

# 输出 shape: (80, T)
# 80 个梅尔频段，T 个时间帧

import webrtcvad

# WebRTC VAD（Google 开源，轻量 CPU 实现）
vad = webrtcvad.Vad(3)  # 灵敏度 0-3（3 最严格）

# 将音频分帧检测
frame_duration = 30  # 支持 10/20/30ms
is_speech = vad.is_speech(frame_bytes, sample_rate=16000)

# 更现代的 Silero VAD（基于 PyTorch，精度更高）
import torch
model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad',
                              model='silero_vad',
                              force_reload=True)
(get_speech_timestamps,
 save_audio,
 read_audio,
 VADIterator,
 collect_chunks) = utils

wav = read_audio('long_audio.wav', sampling_rate=16000)
speech_timestamps = get_speech_timestamps(wav, model, sampling_rate=16000)
# 输出: [{'start': 0.52, 'end': 3.41}, {'start': 5.12, 'end': 8.90}, ...]

from openai import OpenAI
client = OpenAI()

audio_file = open("speech.mp3", "rb")
transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=audio_file,
    language="zh",  # 指定语言提升准确率
    response_format="verbose_json",  # 含时间戳
    timestamp_granularities=["word"],  # 词级时间戳
)
print(transcript.text)

pip install faster-whisper

from faster_whisper import WhisperModel

model = WhisperModel("large-v3", device="cuda", compute_type="float16")

segments, info = model.transcribe("audio.mp3", beam_size=5, language="zh")

print(f"检测到语言: {info.language}, 概率: {info.language_probability:.2f}")

for segment in segments:
    print(f"[{segment.start:.2f} → {segment.end:.2f}] {segment.text}")

response = client.audio.speech.create(
    model="tts-1",
    voice="alloy",  # alloy, echo, fable, onyx, nova, shimmer
    input="欢迎来到 Birdor，您的 AI 助手已准备就绪。",
    speed=1.0,
)

response.stream_to_file("output.mp3")

from elevenlabs import generate, play, set_api_key

set_api_key("YOUR_KEY")

audio = generate(
    text="这是一段高质量的语音合成演示。",
    voice="Bella",  # 或自定义克隆声音
    model="eleven_multilingual_v2",
)

play(audio)

pip install TTS

from TTS.api import TTS

# 加载多语言模型
tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2")

# 克隆声音（用 6 秒参考音频）
tts.tts_to_file(
    text="你好，这是用你的声音合成的中文语音。",
    speaker_wav="reference_voice.wav",
    language="zh",
    file_path="cloned_output.wav",
)

import asyncio
import numpy as np
import torch

class RealtimeVoiceAgent:
    def __init__(self):
        self.vad = VADModel()          # 语音活动检测
        self.asr = WhisperModel("base")  # 语音识别
        self.llm = OpenAIClient()      # 大语言模型
        self.tts = TTSClient()         # 语音合成
        self.is_speaking = False       # Agent 是否在说话
        self.user_interrupt = False    # 用户是否打断

    async def audio_stream_loop(self):
        """持续监听音频输入"""
        audio_buffer = []

        async for chunk in microphone_stream():
            # VAD 检测
            speech_prob = self.vad(chunk)

            if speech_prob > 0.5:
                audio_buffer.append(chunk)
            elif audio_buffer:
                # 用户说完一句，开始处理
                await self.process_utterance(audio_buffer)
                audio_buffer = []

    async def process_utterance(self, audio_chunks):
        # 1. ASR
        text = self.asr.transcribe(np.concatenate(audio_chunks))

        # 2. 如果 Agent 正在说话，先中断
        if self.is_speaking:
            self.user_interrupt = True
            await self.tts.stop()

        # 3. LLM 推理
        response_text = await self.llm.chat(text)

        # 4. TTS 合成（流式）
        self.is_speaking = True
        await self.tts.speak_streaming(response_text)
        self.is_speaking = False

# 使用 aiortc 实现浏览器 ↔ 服务器的实时音频通信
from aiortc import RTCPeerConnection, MediaStreamTrack
from aiortc.contrib.media import MediaPlayer, MediaRecorder

class AudioTrack(MediaStreamTrack):
    kind = "audio"

    async def recv(self):
        frame = await self.track.recv()
        # 将音频帧送入 VAD/ASR 管道
        audio_data = frame.to_ndarray()
        await process_audio_chunk(audio_data)
        return frame

# 前端使用 Web Audio API + getUserMedia
# 音频通过 WebRTC 传输到 Python 服务端

from TTS.api import TTS

tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2")

# 只需 6 秒参考音频即可克隆
tts.tts_to_file(
    text="这是用克隆声音说出的中文句子。",
    speaker_wav="my_voice_6s.wav",
    language="zh",
    file_path="cloned.wav",
)

# 使用 Hugging Face 上的 UTMOS 模型
from transformers import pipeline

utmos = pipeline("audio-classification", model="speechbrain/UTMOS-strong-563M")
score = utmos("synthetic_audio.wav")
print(f"Predicted MOS: {score[0]['score']}")  # 0-5 分

# 简单示例：使用模型库嵌入水印
from wavmark import WatermarkEncoder, WatermarkDecoder

encoder = WatermarkEncoder()
audio, sr = librosa.load("synthetic.wav", sr=16000)
watermarked = encoder.encode(audio, message="copyright_birdor_2025")

# 解码验证
decoder = WatermarkDecoder()
msg = decoder.decode(watermarked)
print(f"Watermark: {msg}")