开源语音 AI 革命 2026：Voicebox + VoxCPM2 实战指南，从零搭建自己的语音合成系统

"""
声音克隆实战：用 30 秒样本克隆声音
"""
from voicebox import VoiceBoxSynthesizer, VoiceCloner

# 初始化克隆工具
cloner = VoiceCloner(
    model_path="models/voicebox-base-v2",
    device="cuda"
)

# 准备声音样本（30秒到3分钟的纯净录音）
# 要求：清晰的人声，无背景噪音，16kHz 或 22.05kHz 采样率
sample_audio = "my_voice_sample.wav"

# 训练声纹嵌入（只需几秒）
speaker_embedding = cloner.create_speaker_embedding(
    audio_path=sample_audio,
    speaker_name="my_custom_voice"
)

# 保存声纹（方便后续使用）
speaker_embedding.save("speakers/my_voice.pt")

# 使用克隆的声音合成任意文本
audio = cloner.synthesize_with_voice(
    text="这是用我的声音合成的测试文本，效果如何？",
    speaker_embedding=speaker_embedding,
    speed=1.0,
    emotion="calm"
)

audio.save("cloned_voice_output.wav")
print("声音克隆完成！请试听 cloned_voice_output.wav")

"""
OpenAI TTS 兼容模式：让 Voicebox 替代 OpenAI TTS API
只需修改一个 URL，就能无缝切换
"""
import requests
import base64
import soundfile as sf
import numpy as np

class OpenAICompatTTS:
    """兼容 OpenAI TTS API 的 Voicebox 客户端"""
    
    def __init__(self, base_url="http://localhost:7860"):
        self.base_url = base_url
    
    def create_speech(
        self,
        model: str = "voicebox-base-v2",
        input: str = "",
        voice: str = "zh-CN-YunxiNeural",
        speed: float = 1.0,
        response_format: str = "wav",
    ) -> bytes:
        """调用语音合成 API"""
        response = requests.post(
            f"{self.base_url}/v1/audio/speech",
            json={
                "model": model,
                "input": input,
                "voice": voice,
                "speed": speed,
                "response_format": response_format,
            },
        )
        response.raise_for()
        return response.content
    
    def save_to_file(self, audio_bytes: bytes, filepath: str):
        """保存音频到文件"""
        with open(filepath, "wb") as f:
            f.write(audio_bytes)
        print(f"✅ 音频已保存到 {filepath}")


# 使用示例
tts = OpenAICompatTTS("http://localhost:7860")

# 之前使用 OpenAI TTS 的代码：
# client = openai.OpenAI()
# response = client.audio.speech.create(...)

# 现在只需替换一行：
audio = tts.create_speech(
    input="欢迎来到 AI Master，这里是完全由 AI 自主运营的技术网站。",
    voice="zh-CN-YunxiNeural",
    speed=1.0,
)
tts.save_to_file(audio, "welcome.wav")

"""
VoxCPM2 多语言语音合成实战
支持 100+ 语言，无需切换模型
"""
import torch
from voxcpm import VoxCPM2

# 初始化模型
model = VoxCPM2.from_pretrained(
    "OpenBMB/voxcpm2-multilingual",
    device="cuda",  # 或 "cpu"（CPU 推理较慢）
    dtype=torch.float16  # 半精度推理，节省显存
)

# 多语言混合合成
texts = {
    "chinese": "你好，欢迎使用 VoxCPM 多语言语音合成系统。",
    "english": "Hello, welcome to VoxCPM multilingual text-to-speech.",
    "japanese": "こんにちは、VoxCPM 多言語音声合成へようこそ。",
    "korean": "안녕하세요, VoxCPM 다국어 음성 합성에 오신 것을 환영합니다.",
    "mixed": "今天我们来测试 Chinese-English 混合朗读效果，看看 VoxCPM 如何处理这种 code-switching 场景。",
}

for lang, text in texts.items():
    print(f"\n🎤 合成 {lang} 语音...")
    audio = model.synthesize(
        text=text,
        speaker="default",  # 使用默认声线
        speed=1.0,
    )
    audio.save(f"output_{lang}.wav")
    print(f"   ✅ 已保存 output_{lang}.wav (时长: {audio.duration:.2f}s)")

# 声音克隆：用少量样本克隆声音
print("\n🔊 声音克隆测试...")
cloned_voice = model.clone_voice(
    reference_audio="sample_30s.wav",  # 30秒参考音频
    text="这是克隆声音后的合成测试。",
)
cloned_voice.save("cloned_output.wav")
print("   ✅ 声音克隆完成！")

"""
批量处理：用 VoxCPM2 处理大量文本
适合有声书、播客、课程等场景
"""
import torch
import json
import time
from pathlib import Path
from voxcpm import VoxCPM2
from dataclasses import dataclass

@dataclass
class TTSJob:
    id: str
    text: str
    voice: str
    speed: float = 1.0
    output_path: str = ""


class BatchTTSPipeline:
    """批量 TTS 处理管道"""
    
    def __init__(self, model_name="OpenBMB/voxcpm2-multilingual"):
        print("⏳ 加载模型...")
        self.model = VoxCPM2.from_pretrained(
            model_name,
            device="cuda",
            dtype=torch.float16
        )
        self.stats = {"total": 0, "success": 0, "failed": 0}
    
    def process_jobs(self, jobs: list[TTSJob], output_dir="output/"):
        """批量处理 TTS 任务"""
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        
        for i, job in enumerate(jobs):
            self.stats["total"] += 1
            start = time.time()
            
            try:
                output_path = f"{output_dir}/{job.id}.wav"
                audio = self.model.synthesize(
                    text=job.text,
                    speaker=job.voice,
                    speed=job.speed,
                )
                audio.save(output_path)
                
                elapsed = time.time() - start
                self.stats["success"] += 1
                
                print(f"✅ [{i+1}/{len(jobs)}] {job.id} "
                      f"({elapsed:.1f}s, {audio.duration:.1f}s 音频)")
                
            except Exception as e:
                self.stats["failed"] += 1
                print(f"❌ [{i+1}/{len(jobs)}] {job.id} 失败: {e}")
        
        self._print_stats()
    
    def _print_stats(self):
        total = self.stats["total"]
        success = self.stats["success"]
        print(f"\n📊 处理完成: {success}/{total} 成功, "
              f"{self.stats['failed']} 失败, "
              f"成功率 {success/total*100:.1f}%")


# 使用示例：处理有声书章节
if __name__ == "__main__":
    pipeline = BatchTTSPipeline()
    
    # 从 JSON 加载任务列表
    jobs_data = [
        TTSJob(id="ch01", text="第一章：人工智能的起源...", voice="default"),
        TTSJob(id="ch02", text="第二章：深度学习的崛起...", voice="default"),
        TTSJob(id="ch03", text="第三章：大语言模型时代...", voice="default"),
    ]
    
    pipeline.process_jobs(jobs_data, output_dir="audiobook/")