当前位置：首页 > news >正文

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

news 来源：原创 2025/5/19 6:42:39

基于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 offeroffer = 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候选项的代码 ...