基于FastAPI和Vosk的实时语音转文字系统开发实践

阿里云教程2个月前发布黄

23 2 0

在人工智能和语音识别技术日益普及的今天，构建一个高效、实时的语音转文字系统变得越来越重大。本文将详细介绍如何使用FastAPI和Vosk模型构建一个功能完整的语音识别系统，支持实时麦克风录音识别和音频文件上传识别。

功能设计

我们的语音识别系统包含以下核心功能：

实时语音识别：通过浏览器麦克风实时录音，并将语音实时转换为文字显示在网页上
音频文件识别：支持上传多种格式的音频文件（WAV、MP3、M4A、FLAC等）进行批量识别
Web界面交互：提供直观友善的用户界面，方便用户操作和查看识别结果

技术选型

后端框架：FastAPI

我们选择FastAPI作为后端框架，主要缘由包括：

高性能：基于Starlette和Pydantic，具有异步处理能力
易用性：自动生成交互式API文档
类型提示：完整的类型检查，减少运行时错误
WebSocket支持：原生支持WebSocket，适合实时通信

语音识别引擎：Vosk

Vosk是一个开源的离线语音识别工具包，具有以下优势：

离线识别：无需网络连接，保护用户隐私
多语言支持：支持包括中文在内的多种语言
高准确率：基于Kaldi语音识别工具包
轻量级模型：提供小型模型，适合部署

项目依赖管理：uv

使用uv作为Python包管理工具，具有以下优势：

速度快：比pip快10-100倍
兼容性好：与pip完全兼容
功能丰富：集成了虚拟环境管理、依赖解析等功能
现代化：支持pyproject.toml等现代Python标准

前端技术

HTML/CSS/JavaScript：标准Web技术栈
WebSocket：实现实时语音数据传输
MediaRecorder API：浏览器原生录音API

音频处理

FFmpeg：用于音频格式转换，支持多种音频格式
PyDub：Python音频处理库（备选方案）

系统架构

基于FastAPI和Vosk的实时语音转文字系统开发实践

语音转文字系统组成

项目结构

SpeechRecognizer/
├── main.py              # 主应用文件
├── requirements.txt     # 依赖列表
├── pyproject.toml       # 项目配置文件
├── README.md            # 说明文档
├── templates/           # HTML模板
│   └── index.html       # 主页面
└── static/              # 静态文件
    ├── css/             # 样式文件
    │   └── style.css    # 主样式文件
    └── js/              # JavaScript文件
        └── main.js      # 主脚本文件

核心代码实现

1. FastAPI应用主文件

from fastapi import FastAPI, File, UploadFile, WebSocket, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import uvicorn
import json
import wave
import os
from vosk import Model, KaldiRecognizer
import logging
import subprocess
import tempfile

app = FastAPI()

# 挂载静态文件目录
app.mount("/static", StaticFiles(directory="static"), name="static")

# 设置模板目录
templates = Jinja2Templates(directory="templates")

# 初始化Vosk模型
model = Model(lang="zh")

@app.get("/", response_class=HTMLResponse)
async def get(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

@app.post("/upload/")
async def upload_audio(file: UploadFile = File(...)):
    # 保存并处理上传的音频文件
    file_location = f"temp_{file.filename}"
    with open(file_location, "wb+") as file_object:
        file_object.write(file.file.read())
    
    result = process_audio_file(file_location)
    os.remove(file_location)
    
    return {"text": result}

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    rec = KaldiRecognizer(model, 16000)
    rec.SetWords(True)
    
    try:
        while True:
            data = await websocket.receive_bytes()
            if rec.AcceptWaveform(data):
                result = rec.Result()
                await websocket.send_text(result)
            else:
                result = rec.PartialResult()
                await websocket.send_text(result)
    except Exception as e:
        print(f"WebSocket连接错误: {e}")
    finally:
        await websocket.close()

2.音频文件处理函数

def convert_to_wav(input_file):
    """将音频文件转换为WAV格式"""
    temp_wav = tempfile.NamedTemporaryFile(suffix='.wav', delete=False)
    temp_wav.close()
    
    # 使用ffmpeg转换音频格式
    cmd = [
        'ffmpeg', '-i', input_file, 
        '-ar', '16000',  # 设置采样率为16kHz
        '-ac', '1',      # 设置单声道
        '-acodec', 'pcm_s16le',  # 设置编码为16位PCM
        '-y', temp_wav.name
    ]
    
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        raise Exception(f"音频转换失败: {result.stderr}")
    
    return temp_wav.name

def process_audio_file(file_path):
    """处理音频文件并返回识别结果"""
    temp_wav_path = None
    try:
        # 检查文件扩展名并转换格式
        _, ext = os.path.splitext(file_path.lower())
        if ext != '.wav':
            temp_wav_path = convert_to_wav(file_path)
            actual_file_path = temp_wav_path
        else:
            actual_file_path = file_path
            
        # 打开WAV文件进行识别
        wf = wave.open(actual_file_path, "rb")
        rec = KaldiRecognizer(model, wf.getframerate())
        result = ""
        
        # 读取音频数据并识别
        while True:
            data = wf.readframes(4000)
            if len(data) == 0:
                break
            if rec.AcceptWaveform(data):
                res = json.loads(rec.Result())
                result += res.get("text", "") + " "
        
        res = json.loads(rec.FinalResult())
        result += res.get("text", "")
        
        return result
    finally:
        # 清理临时文件
        if temp_wav_path and os.path.exists(temp_wav_path):
            os.remove(temp_wav_path)

3. 前端JavaScript实现

// WebSocket连接和实时录音处理
let socket = null;
let audioContext = null;
let scriptProcessor = null;
let microphoneStream = null;

// 开始录音
startBtn.addEventListener('click', async () => {
    // 连接WebSocket
    socket = new WebSocket(`ws://${window.location.host}/ws`);
    
    // 配置音频处理
    const stream = await navigator.mediaDevices.getUserMedia({ 
        audio: {
            sampleRate: 16000,
            channelCount: 1,
            echoCancellation: true,
            noiseSuppression: true
        } 
    });
    
    // 创建AudioContext处理音频
    audioContext = new AudioContext({ sampleRate: 16000 });
    microphoneStream = audioContext.createMediaStreamSource(stream);
    scriptProcessor = audioContext.createScriptProcessor(4096, 1, 1);
    
    // 连接音频处理节点
    microphoneStream.connect(scriptProcessor);
    scriptProcessor.connect(audioContext.destination);
    
    // 处理音频数据并发送到服务器
    scriptProcessor.onaudioprocess = function(e) {
        if (socket && socket.readyState === WebSocket.OPEN) {
            const inputData = e.inputBuffer.getChannelData(0);
            const pcmData = convertFloat32ToInt16(inputData);
            socket.send(pcmData.buffer);
        }
    };
});

// 音频格式转换
function convertFloat32ToInt16(buffer) {
    let l = buffer.length;
    const int16Buffer = new Int16Array(l);
    while (l--) {
        int16Buffer[l] = Math.min(1, buffer[l]) * 0x7FFF;
    }
    return int16Buffer;
}

项目依赖管理

使用uv管理项目依赖

创建pyproject.toml文件：

[project]
name = "speech-recognizer"
version = "0.1.0"
description = "基于FastAPI和Vosk的语音转文字系统"
dependencies = [
    "fastapi>=0.104.0",
    "uvicorn>=0.24.0",
    "vosk>=0.3.45",
    "python-multipart>=0.0.6",
    "websockets>=12.0",
    "numpy>=1.24.0",
    "jinja2>=3.1.2"
]
requires-python = ">=3.8"

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

安装依赖：

# 创建虚拟环境
uv venv

# 激活虚拟环境
source .venv/bin/activate

# 安装依赖
uv pip install -r requirements.txt

部署和运行

环境准备

安装Python 3.7+
安装依赖包：
pip install fastapi uvicorn vosk python-multipart websockets
安装FFmpeg：
# macOS
brew install ffmpeg
# Ubuntu/Debian
sudo apt install ffmpeg

运行应用

python main.py

访问 http://localhost:8000 即可使用系统。

性能优化提议

模型选择：根据应用场景选择合适的Vosk模型（大模型精度高但资源消耗大）
音频预处理：适当降噪和增强语音信号
并发处理：使用异步处理提高并发能力
缓存机制：对识别结果进行缓存，避免重复处理

总结

基于FastAPI和Vosk构建语音识别系统具有实时性强、支持格式多、部署简单等优点，可广泛应用于会议记录、语音助手、教学辅助等场景。

未来可以进一步优化的方向包括：

集成更多语言模型
添加语音活动检测(VAD)
实现说话人分离功能
优化移动端适配

通过不断迭代和优化，这个语音识别系统可以为用户提供更好的使用体验。

阿里云教程

文章版权归作者所有，未经允许请勿转载。

一个小时玩转阿里云 | 免费的自助实验平台上线了

阿里云教程

2个月前

0180

英伟达发布581.94热修复驱动，Win11游戏性能最高回升50%

阿里云教程

2个月前

0160

2025年还在所有项目堆TS？我踩坑后换回JSDoc爽翻了

阿里云教程

2个月前

9170

在STM32上移植CmBacktrace：实现HardFault异常的精准追踪与调试

阿里云教程

2个月前

0240

2 条评论

教你做睿智人投稿者

大神💪

2个月前

回复
不吃面包的女孩投稿者

收藏了，感谢分享

2个月前

回复

基于FastAPI和Vosk的实时语音转文字系统开发实践

功能设计

技术选型

后端框架：FastAPI

语音识别引擎：Vosk

项目依赖管理：uv

前端技术

音频处理

系统架构

项目结构

核心代码实现

1. FastAPI应用主文件

2.音频文件处理函数

3. 前端JavaScript实现

项目依赖管理

使用uv管理项目依赖

部署和运行

环境准备

运行应用

性能优化提议

总结

OpenAI大举从苹果硬件工程团队挖人；中国6G技术试验第一阶段完成

hugo创建静态网页并部署到github

相关文章

一个小时玩转阿里云 | 免费的自助实验平台上线了

英伟达发布581.94热修复驱动，Win11游戏性能最高回升50%

2025年还在所有项目堆TS？我踩坑后换回JSDoc爽翻了

在STM32上移植CmBacktrace：实现HardFault异常的精准追踪与调试

2 条评论