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

SenseVoice-small-onnx语音识别教程：API限流与并发处理配置

news 2026/5/12 1:10:16

SenseVoice-small-onnx语音识别教程：API限流与并发处理配置

1. 引言：为什么需要API限流和并发处理

当你开始使用SenseVoice-small-onnx语音识别服务时，可能会遇到这样的问题：同时有多个用户上传音频文件，服务器响应变慢，甚至出现超时错误。这就是我们需要API限流和并发处理的原因。

SenseVoice-small-onnx是一个基于ONNX量化的多语言语音识别模型，支持中文、粤语、英语、日语、韩语等50多种语言的自动检测。虽然单个音频文件的识别速度很快（10秒音频仅需70毫秒），但在实际应用中，我们需要处理多个并发请求。

本文将手把手教你如何配置API限流和并发处理，让你的语音识别服务既能高效运行，又不会因为请求过多而崩溃。无论你是个人开发者还是企业用户，这些配置都能让你的服务更加稳定可靠。

2. 环境准备与基础配置

在开始配置限流和并发之前，我们需要先确保基础环境正确安装。SenseVoice-small-onnx使用FastAPI作为Web框架，这为我们提供了很好的并发处理基础。

2.1 安装必要依赖

# 安装核心依赖 pip install funasr-onnx fastapi uvicorn python-multipart # 安装可选依赖（用于限流和监控） pip install slowapi python-jose[cryptography] passlib[bcrypt]

2.2 基础服务启动代码

先来看一个最简单的服务启动示例：

from fastapi import FastAPI, File, UploadFile from funasr_onnx import SenseVoiceSmall import os app = FastAPI(title="SenseVoice语音识别服务") # 初始化模型 model_path = "/root/ai-models/danieldong/sensevoice-small-onnx-quant" model = SenseVoiceSmall(model_path, batch_size=10, quantize=True) @app.post("/api/transcribe") async def transcribe_audio( file: UploadFile = File(...), language: str = "auto", use_itn: bool = True ): """语音转写接口""" # 保存上传的文件 file_path = f"/tmp/{file.filename}" with open(file_path, "wb") as f: f.write(await file.read()) # 进行语音识别 result = model([file_path], language=language, use_itn=use_itn) # 清理临时文件 os.remove(file_path) return {"text": result[0]["text"]} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)

这个基础版本还没有任何限流和并发控制，接下来我们会逐步添加这些功能。

3. API限流配置实战

限流是保护服务不被过多请求压垮的重要手段。我们将使用slowapi库来实现灵活的限流策略。

3.1 安装和配置限流中间件

from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from fastapi import Request from fastapi.responses import JSONResponse # 初始化限流器 limiter = Limiter(key_func=get_remote_address) app = FastAPI(title="SenseVoice语音识别服务") # 将限流器挂载到app app.state.limiter = limiter # 添加限流异常处理 @app.exception_handler(RateLimitExceeded) async def rate_limit_handler(request: Request, exc: RateLimitExceeded): return JSONResponse( status_code=429, content={"detail": "请求过于频繁，请稍后再试"} )

3.2 配置不同的限流策略

根据不同的API端点，我们可以设置不同的限流策略：

@app.post("/api/transcribe") @limiter.limit("5/minute") # 每分钟最多5次请求 async def transcribe_audio( request: Request, # 必须添加request参数 file: UploadFile = File(...), language: str = "auto", use_itn: bool = True ): """语音转写接口 - 有限流保护""" # 处理逻辑同上 pass @app.get("/api/health") @limiter.limit("30/minute") # 健康检查接口限制宽松一些 async def health_check(request: Request): """健康检查接口""" return {"status": "healthy", "model_loaded": True}

3.3 基于用户身份的限流

对于需要用户认证的场景，我们可以实现更精细的限流：

from slowapi import Limiter from slowapi.util import get_remote_address def get_user_identifier(request: Request): """根据用户身份生成限流标识""" # 如果有用户认证，使用用户ID if hasattr(request.state, "user_id"): return str(request.state.user_id) # 否则使用IP地址 return get_remote_address(request) # 使用自定义标识符 limiter = Limiter(key_func=get_user_identifier)

4. 并发处理优化配置

SenseVoice-small-onnx模型本身支持批量处理，我们可以利用这个特性来提高并发处理能力。

4.1 模型批量处理配置

from funasr_onnx import SenseVoiceSmall from concurrent.futures import ThreadPoolExecutor import asyncio # 优化模型配置 model = SenseVoiceSmall( model_path="/root/ai-models/danieldong/sensevoice-small-onnx-quant", batch_size=20, # 增加批量大小 quantize=True, device="cpu" # 或者 "cuda" 如果有GPU ) # 创建线程池处理IO密集型操作 thread_pool = ThreadPoolExecutor(max_workers=10)

4.2 异步处理优化

@app.post("/api/transcribe/batch") @limiter.limit("10/minute") async def transcribe_batch( request: Request, files: List[UploadFile] = File(...), language: str = "auto", use_itn: bool = True ): """批量语音转写接口""" # 保存所有上传的文件 file_paths = [] for file in files: file_path = f"/tmp/{file.filename}" with open(file_path, "wb") as f: f.write(await file.read()) file_paths.append(file_path) # 使用线程池执行批量识别（避免阻塞事件循环） loop = asyncio.get_event_loop() try: result = await loop.run_in_executor( thread_pool, lambda: model(file_paths, language=language, use_itn=use_itn) ) # 清理临时文件 for file_path in file_paths: os.remove(file_path) return {"results": [r["text"] for r in result]} except Exception as e: # 发生错误时也要清理文件 for file_path in file_paths: if os.path.exists(file_path): os.remove(file_path) raise e

4.3 UVicorn服务器配置优化

通过调整UVicorn的启动参数，我们可以进一步提高并发处理能力：

if __name__ == "__main__": import uvicorn uvicorn.run( app, host="0.0.0.0", port=7860, # 优化并发配置 workers=4, # 根据CPU核心数调整 loop="uvloop", # 使用更高效的事件循环 http="httptools", # 使用更快的HTTP解析器 timeout_keep_alive=30, # 保持连接超时时间 limit_concurrency=100, # 最大并发连接数 limit_max_requests=1000 # 最大请求数后重启worker )

5. 实战：完整的生产环境配置

现在我们把所有配置整合起来，创建一个适合生产环境使用的完整示例：

from fastapi import FastAPI, Request, File, UploadFile, HTTPException from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from funasr_onnx import SenseVoiceSmall from concurrent.futures import ThreadPoolExecutor import asyncio import os import logging from typing import List # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 初始化应用和限流器 limiter = Limiter(key_func=get_remote_address) app = FastAPI(title="SenseVoice语音识别服务（生产环境）") app.state.limiter = limiter # 全局变量 model = None thread_pool = None @app.on_event("startup") async def startup_event(): """应用启动时初始化""" global model, thread_pool logger.info("正在加载语音识别模型...") model_path = "/root/ai-models/danieldong/sensevoice-small-onnx-quant" # 检查模型路径是否存在 if not os.path.exists(model_path): logger.error(f"模型路径不存在: {model_path}") raise RuntimeError("模型文件未找到") model = SenseVoiceSmall( model_path=model_path, batch_size=20, quantize=True ) # 创建线程池 thread_pool = ThreadPoolExecutor(max_workers=8) logger.info("模型加载完成，服务已就绪") @app.on_event("shutdown") async def shutdown_event(): """应用关闭时清理资源""" if thread_pool: thread_pool.shutdown() logger.info("服务已关闭") @app.exception_handler(RateLimitExceeded) async def rate_limit_handler(request: Request, exc: RateLimitExceeded): """限流异常处理""" return JSONResponse( status_code=429, content={"detail": "请求过于频繁，请稍后再试"} ) @app.post("/api/transcribe") @limiter.limit("6/minute") async def transcribe_audio( request: Request, file: UploadFile = File(...), language: str = "auto", use_itn: bool = True ): """单文件语音转写接口""" if not file.filename.lower().endswith(('.wav', '.mp3', '.m4a', '.flac')): raise HTTPException(400, "不支持的文件格式") # 保存临时文件 file_path = f"/tmp/{os.urandom(8).hex()}_{file.filename}" try: with open(file_path, "wb") as f: content = await file.read() if len(content) > 10 * 1024 * 1024: # 10MB限制 raise HTTPException(400, "文件大小超过限制") f.write(content) # 使用线程池执行识别 loop = asyncio.get_event_loop() result = await loop.run_in_executor( thread_pool, lambda: model([file_path], language=language, use_itn=use_itn) ) return { "text": result[0]["text"], "language": result[0].get("lang", "unknown") } finally: # 清理临时文件 if os.path.exists(file_path): os.remove(file_path) @app.get("/api/health") async def health_check(): """健康检查接口""" return { "status": "healthy", "model_loaded": model is not None, "concurrent_workers": thread_pool._max_workers if thread_pool else 0 } # 启动脚本 if __name__ == "__main__": import uvicorn uvicorn.run( app, host="0.0.0.0", port=7860, workers=2, loop="uvloop", http="httptools", timeout_keep_alive=30, limit_concurrency=50 )

6. 监控和性能调优建议

配置好限流和并发处理后，我们还需要监控服务运行状态，以便进一步优化。

6.1 添加监控指标

from prometheus_client import Counter, Histogram, generate_latest from fastapi import Response # 定义监控指标 REQUEST_COUNT = Counter('api_requests_total', 'Total API requests', ['method', 'endpoint']) REQUEST_DURATION = Histogram('api_request_duration_seconds', 'API request duration', ['endpoint']) @app.middleware("http") async def monitor_requests(request: Request, call_next): """监控中间件""" start_time = time.time() response = await call_next(request) duration = time.time() - start_time REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path).inc() REQUEST_DURATION.labels(endpoint=request.url.path).observe(duration) return response @app.get("/metrics") async def metrics(): """Prometheus指标端点""" return Response(generate_latest(), media_type="text/plain")