快速上手Fun-ASR语音识别：WebSocket流式接口开发与前端集成
·
快速上手Fun-ASR语音识别：WebSocket流式接口开发与前端集成
1. 项目概述与核心价值
Fun-ASR-MLT-Nano-2512是阿里通义实验室推出的轻量级多语言语音识别模型，支持31种语言的高精度识别。这个800M参数的模型在消费级GPU上即可流畅运行，特别适合需要实时语音转写的应用场景。
传统语音识别方案通常需要上传完整音频文件，等待处理完成后才能获取结果。但在实际业务中，我们经常需要：
- 实时会议记录：边说边转文字
- 在线教育互动：即时反馈学生发言
- 智能客服对话：实时理解用户需求
这些场景都要求流式识别能力——语音数据持续输入，识别结果实时输出。本文将带你从零构建一个基于WebSocket的流式语音识别接口，并完成前端集成。
2. 环境准备与模型部署
2.1 基础环境要求
确保你的开发环境满足以下条件：
- 操作系统：Ubuntu 20.04或更高版本
- Python：3.8及以上
- GPU：NVIDIA显卡（可选但推荐）
- 内存：至少8GB
- 磁盘空间：5GB以上
2.2 快速部署模型
通过以下命令一键启动模型服务：
# 安装依赖
pip install -r requirements.txt
apt-get install -y ffmpeg
# 启动服务
cd /root/Fun-ASR-MLT-Nano-2512
nohup python app.py > /tmp/funasr_web.log 2>&1 &
echo $! > /tmp/funasr_web.pid
服务启动后，可以通过http://localhost:7860访问基础的Gradio界面。但我们的目标是构建更强大的流式接口。
3. WebSocket流式接口开发
3.1 项目结构调整
首先创建专门处理流式识别的目录结构：
Fun-ASR-MLT-Nano-2512/
├── model_loader.py # 模型单例加载器
├── ws_server/
│ ├── __init__.py
│ ├── server.py # WebSocket主服务
│ └── stream_processor.py # 流式处理核心
└── app.py # 原Gradio服务（保留）
3.2 核心流式处理器实现
stream_processor.py负责音频数据的实时处理：
import numpy as np
import torch
from model_loader import get_model
class ASRStreamProcessor:
def __init__(self, device="cuda:0"):
self.model = get_model(device)
self.cache = {} # 识别状态缓存
self.audio_buffer = np.array([], dtype=np.float32)
self.sampling_rate = 16000
def append_audio_chunk(self, chunk_bytes: bytes) -> None:
"""追加音频数据块"""
audio_np = np.frombuffer(chunk_bytes, dtype=np.int16).astype(np.float32)
audio_np /= 32768.0 # 归一化
self.audio_buffer = np.concatenate([self.audio_buffer, audio_np])
def process_stream(self, language: str = "中文") -> Dict:
"""执行流式识别"""
if len(self.audio_buffer) < self.sampling_rate * 0.2: # 至少200ms音频
return {"text": "", "is_final": False}
waveform = torch.from_numpy(self.audio_buffer).unsqueeze(0)
try:
res = self.model.generate(
input=waveform,
cache=self.cache,
batch_size=1,
language=language,
itn=True
)
if res and len(res) > 0:
self.cache = res[0].get("cache", {})
return {
"text": res[0]["text"].strip() if res else "",
"is_final": False,
"confidence": res[0].get("confidence", 0.8) if res else 0.0
}
except Exception as e:
print(f"处理错误: {e}")
return {"text": "", "is_final": False}
def reset(self) -> None:
"""重置会话状态"""
self.audio_buffer = np.array([], dtype=np.float32)
self.cache = {}
3.3 WebSocket服务实现
server.py提供WebSocket接口：
import asyncio
import json
import websockets
from stream_processor import ASRStreamProcessor
async def handle_client(websocket, path):
processor = ASRStreamProcessor()
try:
async for message in websocket:
data = json.loads(message)
cmd = data.get("cmd")
if cmd == "start":
processor.reset()
await websocket.send(json.dumps({
"status": "started",
"language": data.get("language", "中文")
}))
elif cmd == "audio":
import base64
audio_bytes = base64.b64decode(data["data"])
processor.append_audio_chunk(audio_bytes)
result = processor.process_stream(language=data.get("language", "中文"))
await websocket.send(json.dumps(result))
elif cmd == "stop":
result = processor.process_stream()
await websocket.send(json.dumps({
"text": result["text"],
"is_final": True,
"status": "finished"
}))
finally:
print("客户端断开连接")
async def main():
async with websockets.serve(handle_client, "0.0.0.0", 8765):
print("WebSocket服务已启动: ws://localhost:8765")
await asyncio.Future()
if __name__ == "__main__":
asyncio.run(main())
启动服务：
cd ws_server && python server.py
4. 前端集成实战
4.1 HTML页面结构
创建index.html实现录音和实时展示：
<!DOCTYPE html>
<html>
<head>
<title>Fun-ASR 流式识别</title>
<style>
body { font-family: sans-serif; max-width: 800px; margin: 40px auto; }
#output { background: #f5f5f5; padding: 15px; min-height: 100px; }
button { padding: 10px 20px; margin: 5px; }
</style>
</head>
<body>
<h1>Fun-ASR流式识别演示</h1>
<p>状态：<span id="status">未连接</span></p>
<button onclick="startMic()">开始录音</button>
<button onclick="stopMic()" disabled>停止</button>
<div>
<label>语言：</label>
<select id="langSelect">
<option value="中文">中文</option>
<option value="English">English</option>
</select>
</div>
<div id="output"></div>
<script>
let socket;
let mediaRecorder;
async function startMic() {
try {
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
socket = new WebSocket('ws://localhost:8765');
socket.onopen = () => {
document.getElementById('status').textContent = '已连接';
document.querySelector('button[onclick="stopMic()"]').disabled = false;
document.querySelector('button[onclick="startMic()"]').disabled = true;
socket.send(JSON.stringify({
cmd: "start",
language: document.getElementById('langSelect').value
}));
};
socket.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.text) {
document.getElementById('output').innerHTML +=
`<div>${data.text}</div>`;
}
};
mediaRecorder = new MediaRecorder(stream);
mediaRecorder.ondataavailable = (event) => {
if (socket.readyState === WebSocket.OPEN) {
const reader = new FileReader();
reader.onload = () => {
const arrayBuffer = reader.result;
const base64 = btoa(String.fromCharCode(...new Uint8Array(arrayBuffer)));
socket.send(JSON.stringify({
cmd: "audio",
data: base64,
language: document.getElementById('langSelect').value
}));
};
reader.readAsArrayBuffer(event.data);
}
};
mediaRecorder.start(200); // 每200ms发送一次数据
} catch (err) {
console.error('录音失败:', err);
document.getElementById('status').textContent = '错误: ' + err.message;
}
}
function stopMic() {
if (mediaRecorder && mediaRecorder.state === 'recording') {
mediaRecorder.stop();
socket.send(JSON.stringify({ cmd: "stop" }));
document.querySelector('button[onclick="startMic()"]').disabled = false;
document.querySelector('button[onclick="stopMic()"]').disabled = true;
}
}
</script>
</body>
</html>
4.2 关键实现细节
- 音频采集：使用浏览器MediaRecorder API获取麦克风输入
- 数据传输：将PCM音频数据转为base64编码通过WebSocket发送
- 实时展示：服务端返回的识别结果即时显示在页面上
- 语言切换：支持动态切换识别语言
5. 生产环境优化建议
5.1 服务稳定性增强
# 在stream_processor.py中添加资源监控
import gc
class ASRStreamProcessor:
# ...原有代码...
def process_stream(self, language: str = "中文") -> Dict:
# 限制缓存大小
if len(self.cache) > 50:
keys_to_keep = list(self.cache.keys())[-10:]
self.cache = {k: self.cache[k] for k in keys_to_keep}
gc.collect()
# ...原有处理逻辑...
5.2 前端自动重连机制
// 在前端代码中添加
let reconnectAttempts = 0;
socket.onclose = () => {
if (reconnectAttempts < 3) {
setTimeout(() => {
startMic();
reconnectAttempts++;
}, 3000);
}
};
socket.onopen = () => {
reconnectAttempts = 0;
// ...原有代码...
};
5.3 Docker容器化部署
创建Dockerfile.ws：
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y \
ffmpeg \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt && \
pip install websockets
COPY . .
EXPOSE 7860 8765
CMD ["sh", "-c", "python app.py & python ws_server/server.py"]
构建并运行：
docker build -f Dockerfile.ws -t funasr-ws .
docker run -d -p 7860:7860 -p 8765:8765 --gpus all funasr-ws
6. 总结与下一步
通过本文，你已经完成了：
- Fun-ASR模型的WebSocket流式接口开发
- 前端实时语音采集与识别展示集成
- 生产环境部署优化方案
这套方案可以轻松集成到各种Web应用中，为你的产品添加实时语音识别能力。下一步可以考虑：
- 实现自动语言检测
- 添加标点恢复功能
- 支持更多音频格式输入
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)