基于WebRTC的实时语音对话系统：从语音识别到AI回复

news2025/5/19 20:29:20

基于WebRTC的实时语音对话系统：从语音识别到AI回复

在当今数字化时代，实时语音交互已成为人机界面的重要组成部分。本文将深入探讨一个基于WebRTC技术的实时语音对话系统，该系统集成了语音识别(ASR)、大语言模型(LLM)和语音合成(TTS)技术，实现了完整的语音到语音的交互体验。
源码

系统架构概览

我们的系统采用了前后端分离的架构，主要包含以下核心组件：

前端WebRTC客户端：负责音频采集和播放
后端FastAPI服务：处理WebRTC连接和信令
语音识别模块：基于Sherpa-ONNX的实时语音转文字
大语言模型接口：连接讯飞星火API进行自然语言处理
语音合成模块：将AI回复转换为自然语音

整个系统的工作流程如下图所示：

用户语音输入 → WebRTC传输 → 语音识别(ASR) → 大语言模型处理 → 语音合成(TTS) → WebRTC传输 → 用户听到回复

技术亮点

1. 基于WebRTC的实时音频传输

WebRTC（Web Real-Time Communication）是一项革命性技术，它允许网页浏览器之间进行实时的音视频通信，无需安装任何插件。在我们的系统中，WebRTC负责处理以下关键任务：

建立浏览器与服务器之间的P2P连接
实时采集用户麦克风的音频数据
将服务器生成的语音回复实时传回浏览器

以下是前端建立WebRTC连接的核心代码片段：

// 前端WebRTC连接建立代码
const pc = new RTCPeerConnection(configuration);
const audioTrack = await navigator.mediaDevices.getUserMedia({audio: true})
    .then(stream => stream.getAudioTracks()[0]);
pc.addTrack(audioTrack);

2. 实时语音识别技术

我们使用了Sherpa-ONNX作为语音识别引擎，它具有以下优势：

支持流式识别，可实时处理音频
使用ONNX格式模型，推理速度快
支持中英双语识别

语音识别的核心处理流程如下：

async def run_asr_on_track(track: MediaStreamTrack, websocket: WebSocket):
    """接收WebRTC音频流，执行ASR识别"""
    asr_stream = stt_recognizer.create_stream()
    
    try:
        while True:
            # 接收音频帧
            frame = await track.recv()
            
            # 转换为ASR所需格式
            _audio = AudioSegment(
                data=bytes(frame.planes[0]),
                sample_width=frame.format.bytes,
                frame_rate=frame.sample_rate,
                channels=len(frame.layout.channels)
            )
            _audio = _audio.set_frame_rate(ASR_SAMPLE_RATE).set_channels(1)
            
            # 转换为float32并归一化
            samples = np.array(_audio.get_array_of_samples()).astype(np.float32) / 32768.0
            
            # 送入识别器处理
            asr_stream.accept_waveform(ASR_SAMPLE_RATE, samples)
            
            # 解码当前可用的音频
            while stt_recognizer.is_ready(asr_stream):
                stt_recognizer.decode_stream(asr_stream)
            
            # 获取中间结果并发送
            current_result = stt_recognizer.get_result(asr_stream)
            if current_result:
                await websocket.send_text(json.dumps({
                    "type": "asr_partial_result",
                    "data": current_result
                }))
                
    finally:
        # 处理最终结果
        full_transcript = stt_recognizer.get_result(asr_stream)
        await websocket.send_text(json.dumps({
            "type": "asr_result", 
            "data": full_transcript
        }))

3. 大语言模型集成

系统集成了讯飞星火大语言模型API，用于理解用户语音内容并生成智能回复。这部分实现了：

异步API调用，不阻塞主线程
错误处理和重试机制
结构化的请求和响应处理

以下是调用讯飞星火API的核心代码：

async def call_xunfei_api(text):
    """调用讯飞星火大模型API"""
    headers = {
        "Authorization": f"Bearer {XUNFEI_API_KEY}",
        "Content-Type": "application/json"
    }
    
    data = {
        "model": "generalv3.5",
        "messages": [
            {
                "role": "user",
                "content": text
            }
        ],
        "stream": False
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(XUNFEI_API_URL, headers=headers, json=data) as response:
            if response.status == 200:
                result = await response.json()
                if result.get("code") == 0:
                    content = result["choices"][0]["message"]["content"]
                    return content

4. 语音合成与回传

系统使用Sherpa-ONNX的TTS模块将文本转换为自然语音，并通过WebRTC回传给用户：

async def run_tts_and_send(text: str, tts_track: TTSAudioTrack):
    """生成TTS音频并通过WebRTC发送"""
    # 生成语音
    generated_audio = tts_model.generate(text, sid=0, speed=1.0)
    
    # 转换为int16格式
    samples_int16 = (generated_audio.samples * 32767).astype(np.int16)
    
    # 添加到WebRTC音频轨道
    await tts_track.add_audio_bytes_pcm(
        samples_int16, 
        generated_audio.sample_rate, 
        1  # 单声道
    )

技术挑战与解决方案

1. 音频格式转换与重采样

WebRTC音频通常使用48kHz采样率的Opus编码，而ASR和TTS模型可能需要不同的采样率（如16kHz）。我们使用pydub库解决了这个问题：

# 将WebRTC音频转换为ASR所需格式
_audio = AudioSegment(
    data=bytes(frame.planes[0]),
    sample_width=frame.format.bytes,
    frame_rate=frame.sample_rate,
    channels=len(frame.layout.channels)
)
_audio = _audio.set_frame_rate(ASR_SAMPLE_RATE).set_channels(1)

2. 语音活动检测(VAD)

为了提高识别准确率，我们实现了语音活动检测，只在检测到语音时进行处理：

# 使用VAD检测是否有语音活动
is_speech = self.vad.is_speech(pcm_bytes, frame.sample_rate)
if is_speech:
    self.last_audio_time = current_time
    # 将当前帧的音频数据添加到缓冲区
    self.audio_buffer.append({
        'data': pcm_bytes,
        'format': frame.format,
        'sample_rate': frame.sample_rate,
        'channels': len(frame.layout.channels)
    })

3. WebRTC信令处理

WebRTC需要复杂的信令交换来建立连接。我们使用WebSocket实现了自定义信令服务器：

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    pc = RTCPeerConnection()
    
    # 处理客户端发来的信令消息
    while True:
        message_str = await websocket.receive_text()
        message = json.loads(message_str)
        
        if message["type"] == "offer":
            # 处理SDP offer
            offer = RTCSessionDescription(sdp=message["sdp"], type=message["type"])
            await pc.setRemoteDescription(offer)
            answer = await pc.createAnswer()
            await pc.setLocalDescription(answer)
            await websocket.send_text(
                json.dumps({"type": "answer", "sdp": pc.localDescription.sdp})
            )
        elif message["type"] == "candidate":
            # 处理ICE candidate
            # ... 处理ICE候选项的代码 ...