14b - 语音克隆与 TTS-STT

# Python
pip install elevenlabs
 
# TypeScript/Node.js
npm install elevenlabs

from elevenlabs import ElevenLabs
import os
 
client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
 
# 步骤 1：上传音频样本创建即时克隆语音
voice = client.voices.add(
    name="我的克隆语音",
    description="用于中文客服场景的克隆语音",
    files=["my-voice-sample.mp3"],  # 本地音频文件路径
    # labels={"language": "zh", "use_case": "customer_service"}
)
 
print(f"语音克隆成功！Voice ID: {voice.voice_id}")
 
# 步骤 2：使用克隆语音生成 TTS
audio = client.text_to_speech.convert(
    voice_id=voice.voice_id,
    text="你好，欢迎致电智能客服中心，请问有什么可以帮助您的？",
    model_id="eleven_flash_v2_5",  # 低延迟模型
    output_format="mp3_44100_128"
)
 
# 步骤 3：保存音频文件
with open("output.mp3", "wb") as f:
    for chunk in audio:
        f.write(chunk)
 
print("音频生成完成：output.mp3")

import { ElevenLabsClient } from "elevenlabs";
import * as fs from "fs";
 
const client = new ElevenLabsClient({
  apiKey: process.env.ELEVENLABS_API_KEY,
});
 
async function cloneAndGenerate() {
  // 步骤 1：上传音频样本创建即时克隆
  const voice = await client.voices.add({
    name: "我的克隆语音",
    description: "用于中文客服场景的克隆语音",
    files: [fs.createReadStream("my-voice-sample.mp3")],
  });
 
  console.log(`语音克隆成功！Voice ID: ${voice.voice_id}`);
 
  // 步骤 2：使用克隆语音生成 TTS
  const audioStream = await client.textToSpeech.convertAsStream(
    voice.voice_id,
    {
      text: "你好，欢迎致电智能客服中心，请问有什么可以帮助您的？",
      model_id: "eleven_flash_v2_5",
      output_format: "mp3_44100_128",
    }
  );
 
  // 步骤 3：保存音频
  const chunks: Buffer[] = [];
  for await (const chunk of audioStream) {
    chunks.push(Buffer.from(chunk));
  }
  fs.writeFileSync("output.mp3", Buffer.concat(chunks));
  console.log("音频生成完成：output.mp3");
}
 
cloneAndGenerate();

# ElevenLabs 专业语音克隆（PVC）
from elevenlabs import ElevenLabs
import os
 
client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
 
# 步骤 1：创建 PVC 语音
voice = client.voices.add(
    name="品牌专属语音",
    description="公司品牌语音，用于所有对外语音交互",
    files=[
        "recording_part1.wav",
        "recording_part2.wav",
        "recording_part3.wav",
    ],
)
 
# 步骤 2：等待训练完成（PVC 需要几小时）
# 可通过 Dashboard 或 API 查询状态
voice_info = client.voices.get(voice_id=voice.voice_id)
print(f"语音状态: {voice_info.name}")

pip install resemble

import resemble
import os
 
resemble.api_key = os.getenv("RESEMBLE_API_KEY")
 
# 步骤 1：创建项目
project = resemble.v2.projects.create(
    name="客服语音项目",
    description="智能客服系统使用的克隆语音",
    language="zh"
)
project_uuid = project["item"]["uuid"]
 
# 步骤 2：上传音频样本创建语音
# 快速克隆：仅需 10 秒音频
voice = resemble.v2.voices.create(
    name="客服克隆语音",
    dataset_url="https://your-storage.com/voice-sample.wav",
    consent="https://your-storage.com/consent-recording.wav",  # 授权录音
    clone_type="rapid"  # rapid（快速）或 professional（专业）
)
voice_uuid = voice["item"]["uuid"]
print(f"语音创建成功：{voice_uuid}")
 
# 步骤 3：使用克隆语音生成音频
clip = resemble.v2.clips.create_sync(
    project_uuid=project_uuid,
    voice_uuid=voice_uuid,
    body="欢迎使用我们的智能客服系统，请问有什么可以帮助您的？",
    output_format="wav",
    sample_rate=44100
)
 
# 下载生成的音频
audio_url = clip["item"]["audio_src"]
print(f"音频生成完成：{audio_url}")

# 基础安装
pip install chatterbox-tts
 
# 或从源码安装（获取最新功能）
git clone https://github.com/resemble-ai/chatterbox.git
cd chatterbox
pip install -e .

import torch
import torchaudio
from chatterbox.tts import ChatterboxTTS
 
# 加载模型（首次运行会自动从 HuggingFace 下载）
device = "cuda" if torch.cuda.is_available() else "cpu"
model = ChatterboxTTS.from_pretrained(device=device)
 
# 零样本语音克隆：用参考音频的声音生成新文本
text = "你好，我是你的 AI 助手，今天有什么可以帮你的吗？"
reference_audio = "reference_voice.wav"  # 几秒钟的参考音频
 
# 生成语音
wav = model.generate(
    text=text,
    audio_prompt_path=reference_audio,
    exaggeration=0.5,  # 情感夸张度：0.0（平淡）到 1.0（夸张）
    cfg_weight=0.5     # 引导权重：控制与参考音频的相似度
)
 
# 保存输出
torchaudio.save("cloned_output.wav", wav, model.sr)
print("克隆语音生成完成：cloned_output.wav")

from chatterbox.tts import ChatterboxTTS
 
# 加载 Turbo 模型
model = ChatterboxTTS.from_pretrained(
    device="cuda",
    model_id="ResembleAI/chatterbox-turbo"  # Turbo 变体
)
 
# Turbo 支持副语言标签（paralinguistic tags）
text = "哈哈 [laugh]，这个问题很有趣 [chuckle]。让我想想..."
wav = model.generate(
    text=text,
    audio_prompt_path="reference_voice.wav",
    exaggeration=0.7  # 更高的情感表达
)
 
torchaudio.save("turbo_output.wav", wav, model.sr)

语音克隆合规检查清单：
□ 获取语音所有者的书面同意（含使用范围和期限）
□ 录制语音所有者的口头授权确认
□ 明确标注 AI 生成的语音内容
□ 实施 AI 水印技术（如 Resemble.ai 的内置水印）
□ 建立语音使用审计日志
□ 制定语音数据的保留和删除策略
□ 禁止用于冒充、欺诈或未经授权的用途
□ 定期审查合规状态

你是一个语音克隆技术顾问。请根据以下需求帮我设计语音克隆方案：

项目类型：[品牌语音 / 有声书 / 客服 / 游戏角色 / ...]
语音来源：[自己的声音 / 配音演员 / 已有录音]
质量要求：[高保真 / 可接受 / 快速原型]
部署方式：[云端 API / 本地部署 / 混合]
预算范围：[$X/月]
合规要求：[GDPR / 中国个保法 / 无特殊要求]
使用场景：[实时对话 / 离线生成 / 批量生产]

请推荐：
1. 最适合的克隆平台和方案
2. 音频样本准备建议
3. 预估成本
4. 合规注意事项

# ElevenLabs HTTP 流式 TTS — Python
from elevenlabs import ElevenLabs
import os
 
client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
 
def stream_tts(text: str, voice_id: str = "JBFqnCBsd6RMkjVDRZzb"):
    """流式生成 TTS 音频"""
    audio_stream = client.text_to_speech.convert_as_stream(
        voice_id=voice_id,
        text=text,
        model_id="eleven_flash_v2_5",  # 低延迟模型
        output_format="mp3_44100_128",
        voice_settings={
            "stability": 0.5,        # 稳定性：0.0-1.0
            "similarity_boost": 0.75, # 相似度增强：0.0-1.0
            "style": 0.0,            # 风格：0.0-1.0（仅 v3 支持）
            "use_speaker_boost": True # 说话人增强
        }
    )
    return audio_stream
 
# 使用示例
stream = stream_tts("你好，我是你的 AI 助手。今天天气不错，有什么可以帮你的吗？")
 
# 方式 1：保存到文件
with open("tts_output.mp3", "wb") as f:
    for chunk in stream:
        f.write(chunk)
 
# 方式 2：实时播放（需要 pyaudio 或类似库）
# import pyaudio
# p = pyaudio.PyAudio()
# stream_out = p.open(format=pyaudio.paInt16, channels=1, rate=44100, output=True)
# for chunk in stream:
#     stream_out.write(chunk)

// ElevenLabs HTTP 流式 TTS — TypeScript
import { ElevenLabsClient } from "elevenlabs";
import * as fs from "fs";
 
const client = new ElevenLabsClient({
  apiKey: process.env.ELEVENLABS_API_KEY,
});
 
async function streamTTS(text: string, voiceId: string = "JBFqnCBsd6RMkjVDRZzb") {
  const audioStream = await client.textToSpeech.convertAsStream(voiceId, {
    text,
    model_id: "eleven_flash_v2_5",
    output_format: "mp3_44100_128",
    voice_settings: {
      stability: 0.5,
      similarity_boost: 0.75,
      style: 0.0,
      use_speaker_boost: true,
    },
  });
 
  // 保存到文件
  const writeStream = fs.createWriteStream("tts_output.mp3");
  for await (const chunk of audioStream) {
    writeStream.write(chunk);
  }
  writeStream.end();
  console.log("TTS 音频生成完成");
}
 
streamTTS("你好，我是你的 AI 助手。今天天气不错，有什么可以帮你的吗？");

# ElevenLabs WebSocket 实时 TTS — Python
import asyncio
import websockets
import json
import os
 
ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
VOICE_ID = "JBFqnCBsd6RMkjVDRZzb"
MODEL_ID = "eleven_flash_v2_5"
 
async def realtime_tts(text_chunks: list[str]):
    """
    WebSocket 实时 TTS：逐块发送文本，逐块接收音频。
    适合与 LLM 流式输出配合使用。
    """
    uri = (
        f"wss://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}"
        f"/stream-input?model_id={MODEL_ID}"
        f"&output_format=mp3_44100_128"
    )
 
    async with websockets.connect(
        uri,
        additional_headers={"xi-api-key": ELEVENLABS_API_KEY}
    ) as ws:
        # 发送初始配置（BOS — Beginning of Stream）
        await ws.send(json.dumps({
            "text": " ",  # 空格触发连接初始化
            "voice_settings": {
                "stability": 0.5,
                "similarity_boost": 0.75
            },
            "generation_config": {
                "chunk_length_schedule": [120, 160, 250, 290]
            }
        }))
 
        audio_chunks = []
 
        # 异步接收音频
        async def receive_audio():
            async for message in ws:
                data = json.loads(message)
                if data.get("audio"):
                    import base64
                    audio_chunks.append(base64.b64decode(data["audio"]))
                if data.get("isFinal"):
                    break
 
        receiver = asyncio.create_task(receive_audio())
 
        # 逐块发送文本（模拟 LLM 流式输出）
        for chunk in text_chunks:
            await ws.send(json.dumps({"text": chunk}))
            await asyncio.sleep(0.05)  # 模拟 LLM 生成间隔
 
        # 发送结束信号（EOS — End of Stream）
        await ws.send(json.dumps({"text": ""}))
 
        await receiver
 
    # 合并音频
    with open("realtime_output.mp3", "wb") as f:
        for chunk in audio_chunks:
            f.write(chunk)
    print(f"实时 TTS 完成，共 {len(audio_chunks)} 个音频块")
 
# 模拟 LLM 流式输出
text_chunks = [
    "你好，",
    "我是你的 AI 助手。",
    "今天有什么",
    "可以帮你的吗？"
]
 
asyncio.run(realtime_tts(text_chunks))

// ElevenLabs WebSocket 实时 TTS — TypeScript
import WebSocket from "ws";
 
const ELEVENLABS_API_KEY = process.env.ELEVENLABS_API_KEY!;
const VOICE_ID = "JBFqnCBsd6RMkjVDRZzb";
const MODEL_ID = "eleven_flash_v2_5";
 
async function realtimeTTS(textChunks: string[]): Promise<Buffer> {
  const uri =
    `wss://api.elevenlabs.io/v1/text-to-speech/${VOICE_ID}` +
    `/stream-input?model_id=${MODEL_ID}&output_format=mp3_44100_128`;
 
  return new Promise((resolve, reject) => {
    const ws = new WebSocket(uri, {
      headers: { "xi-api-key": ELEVENLABS_API_KEY },
    });
 
    const audioChunks: Buffer[] = [];
 
    ws.on("open", async () => {
      // 发送初始配置
      ws.send(
        JSON.stringify({
          text: " ",
          voice_settings: { stability: 0.5, similarity_boost: 0.75 },
          generation_config: { chunk_length_schedule: [120, 160, 250, 290] },
        })
      );
 
      // 逐块发送文本
      for (const chunk of textChunks) {
        ws.send(JSON.stringify({ text: chunk }));
        await new Promise((r) => setTimeout(r, 50));
      }
 
      // 发送结束信号
      ws.send(JSON.stringify({ text: "" }));
    });
 
    ws.on("message", (data: string) => {
      const parsed = JSON.parse(data);
      if (parsed.audio) {
        audioChunks.push(Buffer.from(parsed.audio, "base64"));
      }
      if (parsed.isFinal) {
        ws.close();
      }
    });
 
    ws.on("close", () => resolve(Buffer.concat(audioChunks)));
    ws.on("error", reject);
  });
}
 
// 使用示例
const chunks = ["你好，", "我是你的 AI 助手。", "今天有什么", "可以帮你的吗？"];
realtimeTTS(chunks).then((audio) => {
  require("fs").writeFileSync("realtime_output.mp3", audio);
  console.log("实时 TTS 完成");
});

# Deepgram Aura-2 TTS — Python REST API
from deepgram import DeepgramClient, SpeakOptions
import os
 
deepgram = DeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
def generate_tts(text: str, output_file: str = "deepgram_output.mp3"):
    """使用 Deepgram Aura-2 生成 TTS 音频"""
    options = SpeakOptions(
        model="aura-2-en-us",     # Aura-2 英文模型
        encoding="mp3",            # 输出格式
        sample_rate=24000,         # 采样率
    )
 
    # 生成音频并保存到文件
    response = deepgram.speak.v("1").save(
        output_file,
        {"text": text},
        options
    )
 
    print(f"音频生成完成：{output_file}")
    print(f"Content-Type: {response.content_type}")
    return response
 
generate_tts("Hello, I am your AI assistant. How can I help you today?")

// Deepgram Aura-2 TTS — TypeScript REST API
import { createClient } from "@deepgram/sdk";
import * as fs from "fs";
 
const deepgram = createClient(process.env.DEEPGRAM_API_KEY!);
 
async function generateTTS(text: string, outputFile: string = "deepgram_output.mp3") {
  const response = await deepgram.speak.request(
    { text },
    {
      model: "aura-2-en-us",
      encoding: "mp3",
      sample_rate: 24000,
    }
  );
 
  const stream = await response.getStream();
  if (stream) {
    const chunks: Uint8Array[] = [];
    const reader = stream.getReader();
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      chunks.push(value);
    }
    const buffer = Buffer.concat(chunks);
    fs.writeFileSync(outputFile, buffer);
    console.log(`音频生成完成：${outputFile}`);
  }
}
 
generateTTS("Hello, I am your AI assistant. How can I help you today?");

# Deepgram Aura-2 WebSocket 流式 TTS — Python
from deepgram import DeepgramClient, SpeakWebSocketEvents, SpeakWSOptions
import os
import asyncio
 
async def stream_tts_websocket(text: str):
    """使用 Deepgram WebSocket 流式生成 TTS"""
    deepgram = DeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
    # 创建 WebSocket 连接
    dg_connection = deepgram.speak.websocket.v("1")
 
    audio_chunks = []
 
    # 注册事件处理器
    def on_audio_data(self, result, **kwargs):
        """接收音频数据块"""
        audio_chunks.append(result.data)
 
    def on_flushed(self, result, **kwargs):
        """文本块处理完成"""
        print("音频块刷新完成")
 
    dg_connection.on(SpeakWebSocketEvents.AudioData, on_audio_data)
    dg_connection.on(SpeakWebSocketEvents.Flushed, on_flushed)
 
    # 配置选项
    options = SpeakWSOptions(
        model="aura-2-en-us",
        encoding="linear16",
        sample_rate=24000
    )
 
    # 启动连接
    if dg_connection.start(options):
        # 发送文本（可以分块发送）
        dg_connection.send_text(text)
        dg_connection.flush()  # 触发音频生成
 
        await asyncio.sleep(2)  # 等待音频生成完成
        dg_connection.finish()
 
    print(f"流式 TTS 完成，共接收 {len(audio_chunks)} 个音频块")
    return audio_chunks
 
asyncio.run(stream_tts_websocket("Hello, how can I help you today?"))

# TTS 参数调优示例 — 不同场景的配置
TTS_CONFIGS = {
    "customer_service": {
        "model_id": "eleven_flash_v2_5",
        "voice_settings": {
            "stability": 0.5,
            "similarity_boost": 0.75,
            "style": 0.0,
            "use_speaker_boost": True
        }
    },
    "audiobook": {
        "model_id": "eleven_multilingual_v2",
        "voice_settings": {
            "stability": 0.3,
            "similarity_boost": 0.85,
            "style": 0.4,
            "use_speaker_boost": True
        }
    },
    "ivr_menu": {
        "model_id": "eleven_flash_v2_5",
        "voice_settings": {
            "stability": 0.8,
            "similarity_boost": 0.5,
            "style": 0.0,
            "use_speaker_boost": False
        }
    }
}

延迟排名（从低到高）：
1. ElevenLabs Flash v2.5  — ~75ms TTFB
2. Deepgram Aura-2        — ~90ms TTFB
3. ElevenLabs Turbo v2.5  — ~135ms TTFB
4. OpenAI TTS             — ~200ms TTFB
5. ElevenLabs v3          — ~300ms TTFB

# 非流式（等待全部生成完再返回）— 延迟高
audio = client.text_to_speech.convert(
    voice_id=voice_id,
    text=long_text,
    model_id="eleven_flash_v2_5"
)
 
# 流式（边生成边返回）— 延迟低
audio_stream = client.text_to_speech.convert_as_stream(
    voice_id=voice_id,
    text=long_text,
    model_id="eleven_flash_v2_5"
)
# 第一个音频块在 ~75ms 后就到达
for chunk in audio_stream:
    play_audio(chunk)  # 立即播放

import asyncio
from elevenlabs import AsyncElevenLabs
 
async_client = AsyncElevenLabs(api_key="your-api-key")
 
async def parallel_tts(sentences: list[str], voice_id: str):
    """将长文本拆分为句子，并行生成 TTS"""
    tasks = [
        async_client.text_to_speech.convert_as_stream(
            voice_id=voice_id,
            text=sentence,
            model_id="eleven_flash_v2_5"
        )
        for sentence in sentences
    ]
    # 并行生成，按顺序播放
    results = await asyncio.gather(*tasks)
    return results
 
# 使用示例
sentences = [
    "你好，欢迎致电客服中心。",
    "我是你的 AI 助手小明。",
    "请问有什么可以帮助您的？"
]
# asyncio.run(parallel_tts(sentences, "your-voice-id"))

# WebSocket 连接预热 — 避免首次请求的连接建立延迟
import time
 
class TTSConnectionPool:
    """TTS WebSocket 连接池，保持连接预热"""
 
    def __init__(self, api_key: str, voice_id: str, pool_size: int = 3):
        self.api_key = api_key
        self.voice_id = voice_id
        self.pool_size = pool_size
        self.connections = []
 
    async def warmup(self):
        """预热连接池"""
        for _ in range(self.pool_size):
            conn = await self._create_connection()
            self.connections.append(conn)
        print(f"连接池预热完成：{self.pool_size} 个连接就绪")
 
    async def get_connection(self):
        """获取一个可用连接"""
        if self.connections:
            conn = self.connections.pop(0)
            # 异步补充连接
            asyncio.create_task(self._replenish())
            return conn
        return await self._create_connection()
 
    async def _create_connection(self):
        """创建新的 WebSocket 连接"""
        # 实际实现中连接到 ElevenLabs WebSocket API
        pass
 
    async def _replenish(self):
        """补充连接池"""
        conn = await self._create_connection()
        self.connections.append(conn)

你是一个 TTS 配置优化专家。请根据以下需求帮我优化 TTS 配置：

应用场景：[实时对话 / 有声书 / IVR / 游戏 / 播客]
目标延迟：[<100ms / <200ms / <500ms / 无严格要求]
语音质量要求：[最高 / 高 / 可接受]
目标语言：[中文 / 英文 / 多语言]
并发量：[X 路同时通话]
月度预算：[$X]

请推荐：
1. 最佳 TTS 模型和提供商
2. 推荐的参数配置（stability, similarity_boost 等）
3. 延迟优化策略
4. 预估月度成本

# Deepgram Nova-3 实时流式 STT — Python
from deepgram import (
    DeepgramClient,
    LiveTranscriptionEvents,
    LiveOptions,
)
import os
import asyncio
 
async def realtime_stt():
    """Deepgram 实时流式语音转文本"""
    deepgram = DeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
    # 创建实时连接
    dg_connection = deepgram.listen.asyncwebsocket.v("1")
 
    # 存储转录结果
    transcript_parts = []
 
    # 事件处理器：接收中间结果（interim results）
    async def on_message(self, result, **kwargs):
        sentence = result.channel.alternatives[0].transcript
        if sentence:
            if result.is_final:
                # 最终结果 — 高置信度
                transcript_parts.append(sentence)
                print(f"[最终] {sentence}")
            else:
                # 中间结果 — 实时反馈，可能会修正
                print(f"[中间] {sentence}")
 
    async def on_utterance_end(self, result, **kwargs):
        """说话人停顿检测 — 用于判断用户说完了"""
        print("[轮次结束] 用户可能说完了")
 
    async def on_error(self, error, **kwargs):
        print(f"[错误] {error}")
 
    # 注册事件
    dg_connection.on(LiveTranscriptionEvents.Transcript, on_message)
    dg_connection.on(LiveTranscriptionEvents.UtteranceEnd, on_utterance_end)
    dg_connection.on(LiveTranscriptionEvents.Error, on_error)
 
    # 配置选项
    options = LiveOptions(
        model="nova-3",           # 最新最准确的模型
        language="zh",            # 中文
        smart_format=True,        # 智能格式化（标点、数字等）
        diarize=True,             # 说话人分离
        interim_results=True,     # 启用中间结果（降低感知延迟）
        utterance_end_ms=1000,    # 停顿 1 秒判定为说完
        vad_events=True,          # 语音活动检测事件
        encoding="linear16",      # 音频编码
        sample_rate=16000,        # 采样率
        channels=1                # 单声道
    )
 
    # 启动连接
    if await dg_connection.start(options):
        print("Deepgram 实时 STT 连接已建立")
 
        # 模拟发送音频数据（实际场景中从麦克风或电话流获取）
        # await dg_connection.send(audio_data)
 
        # 保持连接
        await asyncio.sleep(30)
 
        # 关闭连接
        await dg_connection.finish()
 
    return " ".join(transcript_parts)
 
# asyncio.run(realtime_stt())

// Deepgram Nova-3 实时流式 STT — TypeScript
import { createClient, LiveTranscriptionEvents } from "@deepgram/sdk";
 
const deepgram = createClient(process.env.DEEPGRAM_API_KEY!);
 
async function realtimeSTT() {
  // 创建实时连接
  const connection = deepgram.listen.live({
    model: "nova-3",
    language: "zh",
    smart_format: true,
    diarize: true,
    interim_results: true,
    utterance_end_ms: 1000,
    vad_events: true,
    encoding: "linear16",
    sample_rate: 16000,
    channels: 1,
  });
 
  const transcriptParts: string[] = [];
 
  // 监听转录结果
  connection.on(LiveTranscriptionEvents.Transcript, (data) => {
    const sentence = data.channel.alternatives[0].transcript;
    if (sentence) {
      if (data.is_final) {
        transcriptParts.push(sentence);
        console.log(`[最终] ${sentence}`);
      } else {
        console.log(`[中间] ${sentence}`);
      }
    }
  });
 
  // 监听说话人停顿
  connection.on(LiveTranscriptionEvents.UtteranceEnd, () => {
    console.log("[轮次结束] 用户可能说完了");
    // 在这里触发 LLM 处理
  });
 
  connection.on(LiveTranscriptionEvents.Error, (err) => {
    console.error("[错误]", err);
  });
 
  connection.on(LiveTranscriptionEvents.Open, () => {
    console.log("Deepgram 实时 STT 连接已建立");
 
    // 发送音频数据（实际场景中从麦克风或 WebRTC 获取）
    // connection.send(audioBuffer);
  });
 
  // 30 秒后关闭
  setTimeout(() => {
    connection.requestClose();
    console.log("完整转录:", transcriptParts.join(" "));
  }, 30000);
}
 
realtimeSTT();

# Deepgram 批量转录 — Python
from deepgram import DeepgramClient, PrerecordedOptions
import os
 
deepgram = DeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
def transcribe_file(audio_path: str):
    """转录本地音频文件"""
    with open(audio_path, "rb") as audio:
        source = {"buffer": audio.read()}
 
    options = PrerecordedOptions(
        model="nova-3",
        language="zh",
        smart_format=True,
        diarize=True,           # 说话人分离
        paragraphs=True,        # 自动分段
        summarize="v2",         # 自动摘要
        detect_language=True,   # 自动语言检测
        punctuate=True,         # 自动标点
    )
 
    response = deepgram.listen.rest.v("1").transcribe_file(source, options)
 
    # 获取转录结果
    transcript = response.results.channels[0].alternatives[0]
    print(f"转录文本：{transcript.transcript}")
    print(f"置信度：{transcript.confidence}")
 
    # 说话人分离结果
    if transcript.words:
        current_speaker = None
        for word in transcript.words:
            if word.speaker != current_speaker:
                current_speaker = word.speaker
                print(f"\n[说话人 {current_speaker}]", end=" ")
            print(word.punctuated_word, end=" ")
 
    return transcript
 
transcribe_file("meeting_recording.mp3")

// Deepgram 批量转录 — TypeScript
import { createClient } from "@deepgram/sdk";
import * as fs from "fs";
 
const deepgram = createClient(process.env.DEEPGRAM_API_KEY!);
 
async function transcribeFile(audioPath: string) {
  const audioBuffer = fs.readFileSync(audioPath);
 
  const { result } = await deepgram.listen.prerecorded.transcribeFile(
    audioBuffer,
    {
      model: "nova-3",
      language: "zh",
      smart_format: true,
      diarize: true,
      paragraphs: true,
      summarize: "v2",
      detect_language: true,
      punctuate: true,
    }
  );
 
  const transcript = result.results.channels[0].alternatives[0];
  console.log(`转录文本：${transcript.transcript}`);
  console.log(`置信度：${transcript.confidence}`);
 
  return transcript;
}
 
transcribeFile("meeting_recording.mp3");

# ElevenLabs Scribe v2 STT — Python
from elevenlabs import ElevenLabs
import os
 
client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
 
def transcribe_with_scribe(audio_path: str):
    """使用 ElevenLabs Scribe v2 转录音频"""
    with open(audio_path, "rb") as audio_file:
        result = client.speech_to_text.convert(
            file=audio_file,
            model_id="scribe_v2",          # Scribe v2 模型
            language_code="zh",             # 中文
            diarize=True,                   # 说话人分离
            tag_audio_events=True,          # 标记音频事件（笑声、音乐等）
            # keyterm_prompts=["AI Agent", "语音克隆"]  # 关键词提示
        )
 
    print(f"转录文本：{result.text}")
    print(f"语言：{result.language_code}")
 
    # 逐词结果（含时间戳）
    if result.words:
        for word in result.words:
            print(f"  [{word.start:.2f}s - {word.end:.2f}s] {word.text}")
 
    return result
 
transcribe_with_scribe("customer_call.mp3")

# ElevenLabs Scribe v2 实时 STT — Python
import asyncio
import websockets
import json
import os
 
async def scribe_realtime_stt():
    """ElevenLabs Scribe v2 实时流式 STT"""
    uri = "wss://api.elevenlabs.io/v1/speech-to-text/realtime"
 
    async with websockets.connect(
        uri,
        additional_headers={"xi-api-key": os.getenv("ELEVENLABS_API_KEY")}
    ) as ws:
        # 发送初始配置
        await ws.send(json.dumps({
            "type": "config",
            "config": {
                "model_id": "scribe_v2_realtime",
                "language_code": "zh",
                "encoding": "pcm_16000",
                "sample_rate": 16000
            }
        }))
 
        # 接收转录结果
        async def receive_transcripts():
            async for message in ws:
                data = json.loads(message)
                if data.get("type") == "transcript":
                    text = data.get("text", "")
                    is_final = data.get("is_final", False)
                    if text:
                        prefix = "[最终]" if is_final else "[中间]"
                        print(f"{prefix} {text}")
 
        receiver = asyncio.create_task(receive_transcripts())
 
        # 发送音频数据（实际场景中从麦克风获取）
        # await ws.send(audio_bytes)
 
        await asyncio.sleep(30)
        await ws.close()
 
# asyncio.run(scribe_realtime_stt())

# Deepgram 关键词提示 — 提升专业术语识别准确率
options = LiveOptions(
    model="nova-3",
    language="zh",
    keywords=[
        "AI Agent:2",      # 权重 2（最高）
        "语音克隆:1.5",
        "TTS:1.5",
        "STT:1.5",
        "ElevenLabs:1",
        "Deepgram:1"
    ]
)

# 音频预处理 — 提升 STT 准确率
import numpy as np
 
def preprocess_audio(audio_data: np.ndarray, sample_rate: int) -> np.ndarray:
    """
    音频预处理管线：
    1. 降噪
    2. 音量归一化
    3. 重采样到目标采样率
    """
    # 1. 简单降噪（高通滤波，去除低频噪音）
    from scipy.signal import butter, filtfilt
    b, a = butter(4, 80 / (sample_rate / 2), btype='high')
    audio_filtered = filtfilt(b, a, audio_data)
 
    # 2. 音量归一化
    max_val = np.max(np.abs(audio_filtered))
    if max_val > 0:
        audio_normalized = audio_filtered / max_val * 0.9
 
    # 3. 确保 16kHz 采样率（STT 推荐）
    if sample_rate != 16000:
        from scipy.signal import resample
        num_samples = int(len(audio_normalized) * 16000 / sample_rate)
        audio_resampled = resample(audio_normalized, num_samples)
        return audio_resampled
 
    return audio_normalized

# 多 STT 模型对比 — 选择最适合你场景的模型
import asyncio
import time
 
async def benchmark_stt(audio_path: str):
    """对比不同 STT 模型的准确率和延迟"""
    results = {}
 
    # Deepgram Nova-3
    start = time.perf_counter()
    dg_result = await transcribe_with_deepgram(audio_path, model="nova-3")
    results["Deepgram Nova-3"] = {
        "text": dg_result.transcript,
        "confidence": dg_result.confidence,
        "latency_ms": (time.perf_counter() - start) * 1000
    }
 
    # ElevenLabs Scribe v2
    start = time.perf_counter()
    el_result = await transcribe_with_elevenlabs(audio_path)
    results["ElevenLabs Scribe v2"] = {
        "text": el_result.text,
        "latency_ms": (time.perf_counter() - start) * 1000
    }
 
    # 打印对比结果
    for name, result in results.items():
        print(f"\n{name}:")
        print(f"  文本: {result['text'][:100]}...")
        print(f"  延迟: {result['latency_ms']:.0f}ms")
        if "confidence" in result:
            print(f"  置信度: {result['confidence']:.3f}")
 
    return results

你是一个 STT 集成专家。请根据以下需求帮我设计 STT 集成方案：

应用场景：[实时对话 / 会议转录 / 呼叫中心 / 语音搜索]
目标语言：[中文 / 英文 / 多语言 / 方言]
音频来源：[麦克风 / 电话 / 视频会议 / 录音文件]
准确率要求：[最高 / 高 / 可接受]
延迟要求：[<200ms / <500ms / 无严格要求]
特殊需求：[说话人分离 / 情感分析 / 关键词检测 / PII 脱敏]
月度用量：[X 小时/月]
预算：[$X/月]

请推荐：
1. 最佳 STT 模型和提供商
2. 推荐的参数配置
3. 音频预处理建议
4. 准确率优化策略
5. 预估月度成本

用户语音 → [STT: Deepgram Nova-3] → 文本
                                       ↓
                              [LLM: GPT-4o-mini]
                                       ↓
                                    回复文本
                                       ↓
              音频输出 ← [TTS: ElevenLabs Flash] ← 回复文本

# 完整的 STT + LLM + TTS 语音对话管线 — Python
import asyncio
import os
import time
from openai import AsyncOpenAI
from elevenlabs import AsyncElevenLabs
from deepgram import DeepgramClient, LiveTranscriptionEvents, LiveOptions
 
class VoiceAgent:
    """语音 AI Agent：STT → LLM → TTS 完整管线"""
 
    def __init__(self):
        self.deepgram = DeepgramClient(
            api_key=os.getenv("DEEPGRAM_API_KEY")
        )
        self.openai = AsyncOpenAI(
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.elevenlabs = AsyncElevenLabs(
            api_key=os.getenv("ELEVENLABS_API_KEY")
        )
        self.voice_id = "JBFqnCBsd6RMkjVDRZzb"
        self.conversation_history = []
        self.system_prompt = """你是一个友好的中文 AI 客服助手。
        - 用简洁、自然的口语回答问题
        - 每次回复控制在 2-3 句话以内（适合语音输出）
        - 如果无法解决问题，建议转接人工客服"""
 
    async def process_speech(self, user_text: str) -> bytes:
        """
        处理用户语音输入，返回 AI 语音回复。
        完整流程：用户文本 → LLM 推理 → TTS 语音
        """
        start_time = time.perf_counter()
 
        # 步骤 1：LLM 推理
        self.conversation_history.append({
            "role": "user",
            "content": user_text
        })
 
        llm_response = await self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": self.system_prompt},
                *self.conversation_history[-10:]  # 保留最近 10 轮对话
            ],
            max_tokens=150,  # 限制回复长度（语音场景不宜过长）
            temperature=0.7
        )
 
        assistant_text = llm_response.choices[0].message.content
        self.conversation_history.append({
            "role": "assistant",
            "content": assistant_text
        })
 
        llm_time = time.perf_counter() - start_time
        print(f"[LLM] {llm_time*1000:.0f}ms — {assistant_text}")
 
        # 步骤 2：TTS 语音合成
        tts_start = time.perf_counter()
        audio_stream = await self.elevenlabs.text_to_speech.convert_as_stream(
            voice_id=self.voice_id,
            text=assistant_text,
            model_id="eleven_flash_v2_5",
            output_format="mp3_44100_128"
        )
 
        audio_chunks = []
        async for chunk in audio_stream:
            audio_chunks.append(chunk)
 
        tts_time = time.perf_counter() - tts_start
        total_time = time.perf_counter() - start_time
        print(f"[TTS] {tts_time*1000:.0f}ms")
        print(f"[总延迟] {total_time*1000:.0f}ms")
 
        return b"".join(audio_chunks)
 
    async def start_listening(self):
        """启动实时 STT 监听"""
        dg_connection = self.deepgram.listen.asyncwebsocket.v("1")
 
        async def on_transcript(self_dg, result, **kwargs):
            sentence = result.channel.alternatives[0].transcript
            if sentence and result.is_final:
                print(f"\n[用户] {sentence}")
                # 触发 LLM + TTS 处理
                audio = await self.process_speech(sentence)
                # 播放音频（实际场景中发送到用户端）
                print(f"[音频] 生成 {len(audio)} 字节")
 
        dg_connection.on(
            LiveTranscriptionEvents.Transcript, on_transcript
        )
 
        options = LiveOptions(
            model="nova-3",
            language="zh",
            smart_format=True,
            interim_results=True,
            utterance_end_ms=1000,
            encoding="linear16",
            sample_rate=16000,
            channels=1
        )
 
        if await dg_connection.start(options):
            print("语音 Agent 已启动，等待用户输入...")
            # 实际场景中持续发送音频数据
            # await dg_connection.send(audio_data)
 
# 使用示例
# agent = VoiceAgent()
# asyncio.run(agent.start_listening())

// 完整的 STT + LLM + TTS 语音对话管线 — TypeScript
import { createClient, LiveTranscriptionEvents } from "@deepgram/sdk";
import { ElevenLabsClient } from "elevenlabs";
import OpenAI from "openai";
 
interface ConversationMessage {
  role: "system" | "user" | "assistant";
  content: string;
}
 
class VoiceAgent {
  private deepgram = createClient(process.env.DEEPGRAM_API_KEY!);
  private openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
  private elevenlabs = new ElevenLabsClient({
    apiKey: process.env.ELEVENLABS_API_KEY,
  });
  private voiceId = "JBFqnCBsd6RMkjVDRZzb";
  private conversationHistory: ConversationMessage[] = [];
  private systemPrompt = `你是一个友好的中文 AI 客服助手。
    - 用简洁、自然的口语回答问题
    - 每次回复控制在 2-3 句话以内
    - 如果无法解决问题，建议转接人工客服`;
 
  async processSpeech(userText: string): Promise<Buffer> {
    const startTime = performance.now();
 
    // 步骤 1：LLM 推理
    this.conversationHistory.push({ role: "user", content: userText });
 
    const llmResponse = await this.openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [
        { role: "system", content: this.systemPrompt },
        ...this.conversationHistory.slice(-10),
      ],
      max_tokens: 150,
      temperature: 0.7,
    });
 
    const assistantText = llmResponse.choices[0].message.content!;
    this.conversationHistory.push({
      role: "assistant",
      content: assistantText,
    });
 
    const llmTime = performance.now() - startTime;
    console.log(`[LLM] ${llmTime.toFixed(0)}ms — ${assistantText}`);
 
    // 步骤 2：TTS 语音合成
    const ttsStart = performance.now();
    const audioStream = await this.elevenlabs.textToSpeech.convertAsStream(
      this.voiceId,
      {
        text: assistantText,
        model_id: "eleven_flash_v2_5",
        output_format: "mp3_44100_128",
      }
    );
 
    const chunks: Buffer[] = [];
    for await (const chunk of audioStream) {
      chunks.push(Buffer.from(chunk));
    }
 
    const ttsTime = performance.now() - ttsStart;
    const totalTime = performance.now() - startTime;
    console.log(`[TTS] ${ttsTime.toFixed(0)}ms`);
    console.log(`[总延迟] ${totalTime.toFixed(0)}ms`);
 
    return Buffer.concat(chunks);
  }
 
  async startListening() {
    const connection = this.deepgram.listen.live({
      model: "nova-3",
      language: "zh",
      smart_format: true,
      interim_results: true,
      utterance_end_ms: 1000,
      encoding: "linear16",
      sample_rate: 16000,
      channels: 1,
    });
 
    connection.on(LiveTranscriptionEvents.Transcript, async (data) => {
      const sentence = data.channel.alternatives[0].transcript;
      if (sentence && data.is_final) {
        console.log(`\n[用户] ${sentence}`);
        const audio = await this.processSpeech(sentence);
        console.log(`[音频] 生成 ${audio.length} 字节`);
      }
    });
 
    connection.on(LiveTranscriptionEvents.Open, () => {
      console.log("语音 Agent 已启动，等待用户输入...");
    });
  }
}
 
// 使用示例
// const agent = new VoiceAgent();
// agent.startListening();

# 流式管线：LLM 流式输出 → TTS WebSocket 流式输入
import asyncio
import json
import websockets
from openai import AsyncOpenAI
 
async def streaming_pipeline(user_text: str):
    """
    流式管线：LLM 边生成文本，TTS 边合成语音。
    用户感知延迟 = LLM 首 token 延迟 + TTS TTFB ≈ 200-400ms
    """
    openai_client = AsyncOpenAI()
 
    # 步骤 1：建立 TTS WebSocket 连接
    voice_id = "JBFqnCBsd6RMkjVDRZzb"
    tts_uri = (
        f"wss://api.elevenlabs.io/v1/text-to-speech/{voice_id}"
        f"/stream-input?model_id=eleven_flash_v2_5"
        f"&output_format=mp3_44100_128"
    )
 
    async with websockets.connect(
        tts_uri,
        additional_headers={
            "xi-api-key": "your-elevenlabs-api-key"
        }
    ) as tts_ws:
        # 初始化 TTS 连接
        await tts_ws.send(json.dumps({
            "text": " ",
            "voice_settings": {
                "stability": 0.5,
                "similarity_boost": 0.75
            }
        }))
 
        audio_chunks = []
 
        # 异步接收 TTS 音频
        async def receive_audio():
            async for msg in tts_ws:
                data = json.loads(msg)
                if data.get("audio"):
                    import base64
                    chunk = base64.b64decode(data["audio"])
                    audio_chunks.append(chunk)
                    # 实时播放：play_audio(chunk)
                if data.get("isFinal"):
                    break
 
        receiver = asyncio.create_task(receive_audio())
 
        # 步骤 2：LLM 流式生成 → 直接发送到 TTS
        stream = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "用简洁的中文回答。"},
                {"role": "user", "content": user_text}
            ],
            stream=True,
            max_tokens=150
        )
 
        buffer = ""
        async for chunk in stream:
            delta = chunk.choices[0].delta.content or ""
            buffer += delta
 
            # 按句子边界发送到 TTS（逗号、句号、问号等）
            if any(p in buffer for p in ["，", "。", "！", "？", "；", ",", ".", "!", "?"]):
                await tts_ws.send(json.dumps({"text": buffer}))
                print(f"[发送到 TTS] {buffer}")
                buffer = ""
 
        # 发送剩余文本
        if buffer:
            await tts_ws.send(json.dumps({"text": buffer}))
 
        # 发送结束信号
        await tts_ws.send(json.dumps({"text": ""}))
        await receiver
 
    return b"".join(audio_chunks)
 
# asyncio.run(streaming_pipeline("请介绍一下语音克隆技术"))

音频采集（30 分钟专业录音）
    ↓
ElevenLabs PVC 专业克隆
    ↓
┌─────────────────────────────────────────┐
│           品牌语音（Voice ID）            │
├─────────────┬──────────────┬────────────┤
│  客服电话    │  App 语音助手  │  课程配音   │
│  Flash v2.5  │  Flash v2.5   │  v3 模型   │
│  低延迟优先  │  低延迟优先    │  质量优先   │
│  Vapi 编排   │  WebSocket    │  批量生成   │
└─────────────┴──────────────┴────────────┘

患者语音 → [Whisper Large v3: 本地 STT] → 文本
                                            ↓
                                   [本地 LLM: Llama 3]
                                            ↓
                                         回复文本
                                            ↓
         语音输出 ← [Chatterbox Turbo: 本地 TTS] ← 回复文本

# 全本地语音 Agent — 无需外部 API
import torch
import torchaudio
import whisper
from chatterbox.tts import ChatterboxTTS
from transformers import AutoModelForCausalLM, AutoTokenizer
 
class LocalVoiceAgent:
    """全本地部署的语音 Agent（HIPAA 合规）"""
 
    def __init__(self):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
 
        # 加载 STT 模型（Whisper）
        self.stt_model = whisper.load_model("large-v3", device=self.device)
 
        # 加载 TTS 模型（Chatterbox Turbo）
        self.tts_model = ChatterboxTTS.from_pretrained(
            device=self.device,
            model_id="ResembleAI/chatterbox-turbo"
        )
 
        # 加载 LLM（本地 Llama）
        self.tokenizer = AutoTokenizer.from_pretrained(
            "meta-llama/Llama-3.1-8B-Instruct"
        )
        self.llm = AutoModelForCausalLM.from_pretrained(
            "meta-llama/Llama-3.1-8B-Instruct",
            torch_dtype=torch.float16,
            device_map="auto"
        )
 
        self.reference_audio = "brand_voice.wav"
 
    def transcribe(self, audio_path: str) -> str:
        """STT：语音转文本"""
        result = self.stt_model.transcribe(
            audio_path,
            language="zh",
            task="transcribe"
        )
        return result["text"]
 
    def think(self, user_text: str) -> str:
        """LLM：生成回复"""
        messages = [
            {"role": "system", "content": "你是一个医疗咨询助手。"},
            {"role": "user", "content": user_text}
        ]
        inputs = self.tokenizer.apply_chat_template(
            messages, return_tensors="pt"
        ).to(self.device)
 
        outputs = self.llm.generate(
            inputs, max_new_tokens=150, temperature=0.7
        )
        return self.tokenizer.decode(
            outputs[0][inputs.shape[1]:], skip_special_tokens=True
        )
 
    def speak(self, text: str, output_path: str = "response.wav"):
        """TTS：文本转语音"""
        wav = self.tts_model.generate(
            text=text,
            audio_prompt_path=self.reference_audio,
            exaggeration=0.3,
            cfg_weight=0.5
        )
        torchaudio.save(output_path, wav, self.tts_model.sr)
        return output_path
 
    def process(self, audio_path: str) -> str:
        """完整处理流程"""
        # STT
        user_text = self.transcribe(audio_path)
        print(f"[用户] {user_text}")
 
        # LLM
        response = self.think(user_text)
        print(f"[助手] {response}")
 
        # TTS
        output = self.speak(response)
        print(f"[音频] {output}")
 
        return output
 
# agent = LocalVoiceAgent()
# agent.process("patient_question.wav")