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

SenseVoice-small-onnx语音识别部署：模型版本管理与灰度发布

news 2026/3/27 8:32:31

SenseVoice-small-onnx语音识别部署：模型版本管理与灰度发布

部署前提：本文假设您已基于sensevoice-small-onnx-quant模型完成基础服务部署，熟悉基本的 REST API 调用和 Python 集成方式。

1. 为什么需要模型版本管理

在实际生产环境中，语音识别服务很少会"一套模型用到老"。随着业务发展，您可能会遇到这些场景：

模型升级：从 small 版本升级到 medium 或 large 版本提升识别准确率
量化优化：尝试不同的量化策略平衡精度和性能
多版本并行：A/B 测试不同模型版本的效果
热修复：紧急修复某个语言识别问题而不影响整体服务

如果没有规范的版本管理，这些操作很容易导致服务中断、识别结果不一致、难以回滚等问题。

2. 模型版本管理基础方案

2.1 目录结构规范化

首先建立清晰的模型存储结构，避免文件混乱：

/root/ai-models/ └── sensevoice/ ├── small-onnx-quant/ # 当前生产版本 │ ├── model_quant.onnx │ ├── config.json │ └── version.txt # 记录版本号：v1.0.0 ├── small-onnx-quant-v1.0.1/ # 新版本准备目录 │ ├── model_quant.onnx │ ├── config.json │ └── version.txt # v1.0.1 └── model-registry.json # 模型注册表

2.2 版本注册表管理

创建model-registry.json来集中管理所有版本：

{ "models": { "sensevoice-small-onnx-quant": { "current": "v1.0.0", "versions": { "v1.0.0": { "path": "/root/ai-models/sensevoice/small-onnx-quant", "status": "production", "quantize": true, "languages": ["zh", "en", "yue", "ja", "ko", "auto"], "deploy_time": "2024-01-15T10:00:00Z" }, "v1.0.1": { "path": "/root/ai-models/sensevoice/small-onnx-quant-v1.0.1", "status": "staging", "quantize": true, "languages": ["zh", "en", "yue", "ja", "ko", "auto"], "deploy_time": null } } } } }

2.3 自动化版本切换脚本

编写简单的版本管理脚本model-manager.py：

import json import os import shutil from pathlib import Path class ModelVersionManager: def __init__(self, registry_path): self.registry_path = registry_path self.registry = self._load_registry() def _load_registry(self): if os.path.exists(self.registry_path): with open(self.registry_path, 'r') as f: return json.load(f) return {"models": {}} def switch_version(self, model_name, version): """切换模型版本""" if model_name not in self.registry["models"]: raise ValueError(f"Model {model_name} not found") if version not in self.registry["models"][model_name]["versions"]: raise ValueError(f"Version {version} not found") # 更新当前版本 self.registry["models"][model_name]["current"] = version # 保存注册表 with open(self.registry_path, 'w') as f: json.dump(self.registry, f, indent=2) print(f"Switched {model_name} to version {version}") def get_current_model_path(self, model_name): """获取当前版本模型路径""" current_version = self.registry["models"][model_name]["current"] return self.registry["models"][model_name]["versions"][current_version]["path"] # 使用示例 manager = ModelVersionManager("/root/ai-models/model-registry.json") current_path = manager.get_current_model_path("sensevoice-small-onnx-quant")

3. 灰度发布实施方案

3.1 基于权重的流量分发

修改您的服务代码，支持多版本并行和流量分发：

from funasr_onnx import SenseVoiceSmall import random class MultiVersionASRService: def __init__(self, registry_path): self.manager = ModelVersionManager(registry_path) self.models = self._load_models() def _load_models(self): """加载所有活跃版本的模型""" models = {} registry = self.manager.registry for model_name, model_info in registry["models"].items(): for version, version_info in model_info["versions"].items(): if version_info["status"] in ["production", "staging"]: model_key = f"{model_name}-{version}" models[model_key] = { "instance": SenseVoiceSmall( version_info["path"], batch_size=10, quantize=version_info["quantize"] ), "weight": 100 if version_info["status"] == "production" else 10, "version": version } return models def transcribe(self, audio_path, language="auto"): """根据权重选择模型进行转录""" # 选择模型 model_key = self._select_model() model = self.models[model_key]["instance"] # 执行转录 result = model([audio_path], language=language, use_itn=True) return { "text": result[0], "model_version": self.models[model_key]["version"], "model_key": model_key } def _select_model(self): """根据权重随机选择模型""" total_weight = sum(model["weight"] for model in self.models.values()) selected = random.uniform(0, total_weight) current = 0 for key, model in self.models.items(): current += model["weight"] if selected <= current: return key

3.2 基于用户标识的灰度发布

对于更精细的控制，可以基于用户ID或设备ID进行分流：

def get_model_by_user(user_id, audio_path, language="auto"): """根据用户ID选择模型版本""" # 简单的哈希算法决定用户使用哪个版本 user_hash = hash(user_id) % 100 if user_hash < 10: # 10% 流量到新版本 model_path = "/root/ai-models/sensevoice/small-onnx-quant-v1.0.1" version = "v1.0.1" else: model_path = "/root/ai-models/sensevoice/small-onnx-quant" version = "v1.0.0" model = SenseVoiceSmall(model_path, batch_size=10, quantize=True) result = model([audio_path], language=language, use_itn=True) return { "text": result[0], "model_version": version, "user_id": user_id }

3.3 API 扩展支持版本选择

扩展您的 REST API 支持版本参数：

from fastapi import FastAPI, File, UploadFile, HTTPException from pydantic import BaseModel import uuid app = FastAPI() class TranscribeRequest(BaseModel): language: str = "auto" use_itn: bool = True version: str = "auto" # 新增版本参数 @app.post("/api/v2/transcribe") async def transcribe_audio( file: UploadFile = File(...), request: TranscribeRequest = None ): """支持版本选择的转录接口""" if request is None: request = TranscribeRequest() # 保存上传的音频文件 file_path = f"/tmp/{uuid.uuid4()}_{file.filename}" with open(file_path, "wb") as buffer: content = await file.read() buffer.write(content) # 选择模型版本 if request.version == "auto": # 自动选择逻辑，可以基于用户ID或其他策略 user_id = "default" # 实际应从请求头或token获取 result = get_model_by_user(user_id, file_path, request.language) else: # 指定版本 model_path = f"/root/ai-models/sensevoice/small-onnx-quant-{request.version}" model = SenseVoiceSmall(model_path, batch_size=10, quantize=True) transcription = model([file_path], language=request.language, use_itn=request.use_itn) result = { "text": transcription[0], "model_version": request.version } # 清理临时文件 os.remove(file_path) return result

4. 监控与回滚机制

4.1 版本性能监控

实施监控来评估不同版本的表现：

import time import logging from datetime import datetime def monitored_transcribe(audio_path, language="auto", version="v1.0.0"): """带监控的转录函数""" start_time = time.time() try: model_path = f"/root/ai-models/sensevoice/small-onnx-quant-{version}" model = SenseVoiceSmall(model_path, batch_size=10, quantize=True) result = model([audio_path], language=language, use_itn=True) end_time = time.time() duration = end_time - start_time # 记录性能指标 log_performance({ "version": version, "duration": duration, "audio_length": get_audio_length(audio_path), "success": True, "timestamp": datetime.now().isoformat() }) return result[0] except Exception as e: end_time = time.time() duration = end_time - start_time log_performance({ "version": version, "duration": duration, "success": False, "error": str(e), "timestamp": datetime.now().isoformat() }) raise e def log_performance(metrics): """记录性能日志""" logging.info(f"ASR Performance: {metrics}") # 这里可以扩展到推送指标到监控系统 # 如 Prometheus, Datadog, 或自定义监控平台

4.2 自动化回滚机制

当检测到新版本有问题时自动回滚：

import json from typing import Dict, List class AutoRollbackManager: def __init__(self, registry_path, threshold=0.1): self.registry_path = registry_path self.error_threshold = threshold # 10% 错误率触发回滚 self.version_metrics: Dict[str, List[bool]] = {} def record_result(self, version: str, success: bool): """记录版本执行结果""" if version not in self.version_metrics: self.version_metrics[version] = [] self.version_metrics[version].append(success) # 只保留最近100条记录 if len(self.version_metrics[version]) > 100: self.version_metrics[version] = self.version_metrics[version][-100:] # 检查是否需要回滚 self._check_rollback(version) def _check_rollback(self, version: str): """检查错误率并触发回滚""" if version not in self.version_metrics or len(self.version_metrics[version]) < 20: return success_count = sum(1 for s in self.version_metrics[version] if s) total_count = len(self.version_metrics[version]) error_rate = 1 - (success_count / total_count) if error_rate > self.error_threshold: print(f"Error rate {error_rate:.2%} exceeds threshold, triggering rollback for {version}") self._perform_rollback(version) def _perform_rollback(self, failed_version: str): """执行回滚操作""" with open(self.registry_path, 'r') as f: registry = json.load(f) model_name = "sensevoice-small-onnx-quant" current_version = registry["models"][model_name]["current"] if current_version == failed_version: # 回滚到上一个稳定版本 stable_versions = [ v for v, info in registry["models"][model_name]["versions"].items() if info["status"] == "production" and v != failed_version ] if stable_versions: rollback_version = stable_versions[-1] # 选择最新的稳定版本 registry["models"][model_name]["current"] = rollback_version with open(self.registry_path, 'w') as f: json.dump(registry, f, indent=2) print(f"Rolled back from {failed_version} to {rollback_version}")

5. 完整部署示例

5.1 生产环境部署结构

/opt/voice-asr-service/ ├── app/ # 应用代码 │ ├── main.py # 主应用 │ ├── model_manager.py # 版本管理 │ ├── gray_release.py # 灰度发布逻辑 │ └── monitoring.py # 监控模块 ├── models/ # 模型存储 │ └── sensevoice/ │ ├── small-onnx-quant-v1.0.0/ │ ├── small-onnx-quant-v1.0.1/ │ └── model-registry.json ├── logs/ # 日志目录 ├── requirements.txt # 依赖文件 └── docker-compose.yml # Docker 配置

5.2 Docker 化部署

创建Dockerfile和docker-compose.yml支持多版本部署：

# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ ffmpeg \ libsndfile1 \ && rm -rf /var/lib/apt/lists/* # 复制代码和模型 COPY requirements.txt . COPY app/ ./app/ COPY models/ ./models/ # 安装Python依赖 RUN pip install -r requirements.txt EXPOSE 7860 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

# docker-compose.yml version: '3.8' services: voice-asr: build: . ports: - "7860:7860" volumes: - ./models:/app/models - ./logs:/app/logs environment: - MODEL_REGISTRY_PATH=/app/models/sensevoice/model-registry.json - LOG_LEVEL=INFO restart: unless-stopped

5.3 自动化部署脚本

创建部署脚本简化版本发布流程：

#!/bin/bash # deploy-new-version.sh VERSION=$1 MODEL_PATH="/root/ai-models/sensevoice/small-onnx-quant-${VERSION}" # 1. 检查新版本模型是否存在 if [ ! -d "$MODEL_PATH" ]; then echo "Error: Model version $VERSION not found at $MODEL_PATH" exit 1 fi # 2. 更新模型注册表 python3 -c " import json registry_path = '/root/ai-models/model-registry.json' with open(registry_path, 'r') as f: registry = json.load(f) # 更新版本状态 registry['models']['sensevoice-small-onnx-quant']['versions']['$VERSION']['status'] = 'staging' with open(registry_path, 'w') as f: json.dump(registry, f, indent=2) print('Version $VERSION set to staging') " # 3. 重新加载服务（根据实际部署方式调整） echo "Reloading service..." sudo systemctl reload voice-asr-service echo "Deployment completed for version $VERSION" echo "Use curl -X POST http://localhost:7860/management/switch-version -d '{\"version\": \"$VERSION\", \"weight\": 10}'"