LLM应用开发——语音交互系统实战

LLM 应用开发系列

• Embedding 嵌入向量
• 向量数据库
• RAG 检索增强生成
• Prompt Engineering
• LangChain 框架
• Agent 智能体
• Function Calling
• 部署与优化
• 安全防护
• 对话记忆管理
• Streaming 流式输出
• LlamaIndex 框架
• 结构化输出
• 多模态 LLM 应用
• LLM 评估与测试
• 多模型路由
• 缓存策略
• 成本优化
• 监控与可观测性
• 高级 RAG 技术
• GraphRAG 知识图谱
• 长上下文处理
• LLM 微调实战
• 本地模型部署
• MCP 协议开发
• AI 搜索引擎
• 代码生成与执行
• 文档智能处理
• 语音交互系统
• 多租户架构
• Prompt 版本管理
• 混合检索策略
• 对话系统设计
• 内容审核与过滤
• 个性化推荐
• 知识库构建
• AI 写作助手
• 实时协作
• 边缘部署

┌─────────────────────────────────────────────────────────────────┐
│                      语音对话流水线                             │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. [VAD] → 检测用户是否开始/停止说话                           │
│                                                                 │
│  2. [ASR] → 语音转文字 (Whisper / FunASR)                       │
│                                                                 │
│  3. [LLM] → 理解意图并生成文本回复                              │
│                                                                 │
│  4. [TTS] → 文字转语音 (OpenAI TTS / Fish Speech)               │
│                                                                 │
│  5. [播放] → 音频流式输出                                       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

from faster_whisper import WhisperModel

model_size = "base"
model = WhisperModel(model_size, device="cuda", compute_type="float16")

def transcribe(audio_path: str):
    segments, info = model.transcribe(audio_path, beam_size=5)
    text = "".join([segment.text for segment in segments])
    return text

from openai import OpenAI

client = OpenAI()

def text_to_speech_stream(text: str):
    response = client.audio.speech.create(
        model="tts-1",
        voice="alloy",
        input=text,
    )
    # 流式写入文件或播放器
    response.stream_to_file("output.mp3")

---

#### 核心挑战：低延迟与打断处理

在语音对话中，**延迟 (Latency)** 是用户体验的杀手。如果回复延迟超过 1 秒，对话就会变得尴尬。

#### 1. VAD (静音检测) 与打断逻辑
系统必须知道用户什么时候开始说话，什么时候说完。
- **Silero VAD**：目前最流行的轻量级 VAD 模型，可以在 CPU 上实时运行。
- **打断逻辑**：当用户在 AI 说话时突然插话，系统必须立即停止 TTS 播放，并清空当前的生成队列。

```python
# 打断逻辑伪代码
def on_user_speech_start():
    if tts_player.is_playing():
        tts_player.stop()
        llm_chain.cancel_current_generation()
        print("用户插话，已停止播放")

# Realtime API (WebSocket) 概念示例
import websockets

async def voice_chat():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    async with websockets.connect(url, extra_headers=headers) as ws:
        # 发送音频流
        await ws.send(audio_chunk)
        # 接收音频流并直接播放
        async for message in ws:
            if message.type == "response.audio.delta":
                play_audio(message.delta)

---

#### 实时性优化：低延迟架构设计

在语音对话中，延迟超过 1 秒就会让用户感到不自然。我们需要在每个环节进行极致优化。

#### 1. 语音活动检测 (VAD) 与打断处理

使用 `Silero VAD` 可以精准识别用户是否说话结束，并支持“打断”逻辑：如果用户在 AI 播放音频时开始说话，立即停止播放。

```python
import torch
import numpy as np

# 加载 Silero VAD 模型
model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad', model='silero_vad')
(get_speech_timestamps, save_audio, read_audio, VADIterator, collect_chunks) = utils

def process_audio_stream(audio_chunk, is_ai_speaking):
    """
    处理实时音频流
    audio_chunk: 采样率为 16000 的音频片段
    is_ai_speaking: 当前 AI 是否正在播放声音
    """
    # 检测当前片段是否包含人声
    speech_prob = model(torch.from_numpy(audio_chunk), 16000).item()
    
    if speech_prob > 0.5:
        if is_ai_speaking:
            # 触发打断逻辑
            stop_ai_playback()
            clear_audio_buffer()
            print("检测到用户插话，停止 AI 播放")
        return True # 正在说话
    return False

async def streaming_pipeline(user_audio):
    # 1. ASR (使用 Faster-Whisper 流式)
    text_stream = await asr.transcribe_stream(user_audio)
    
    # 2. LLM (流式生成)
    full_response = ""
    async for chunk in llm.astream(text_stream):
        full_response += chunk
        # 遇到句号、问号等结束符，立即合成语音
        if chunk in ["。", "！", "？", ".", "!", "?"]:
            audio_segment = await tts.generate(full_response)
            await audio_player.play(audio_segment)
            full_response = "" # 清空缓冲区，准备下一句

from fish_speech import FishSpeechClient

client = FishSpeechClient(api_key="your_key")

async def clone_voice(text: str, reference_audio_path: str):
    """
    使用参考音频克隆声音并合成新文本
    """
    audio_content = await client.tts(
        text=text,
        ref_audio=reference_audio_path,
        format="mp3"
    )
    with open("cloned_voice.mp3", "wb") as f:
        f.write(audio_content)

import asyncio
import websockets
import json

async def realtime_chat():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
    headers = {
        "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(url, extra_headers=headers) as ws:
        # 发送配置
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy"
            }
        }))

        # 发送音频数据 (Base64)
        # await ws.send(json.dumps({"type": "input_audio_buffer.append", "audio": "..."}))