- Introducción y Beneficios del Sistema
Este artículo aborda la implementación de un servicio de reconocimiento de voz basado en el modelo SenseVoice-small-onnx, optimizado mediante ONNX, con un enfoque principal en la actualización de modelos en tiempo real sin interrupciones del servicio. El modelo, al estar cuantizado, posee un tamaño reducido de 230 MB y ofrece una velocidad de inferencia rápida, procesando 10 segundos de audio en aproximadamente 70 milisegundos. Soporta más de 50 idiomas, incluyendo mandarín, cantonés, inglés, japonés y coreano.
En entornos de producción, es crítico mantener la disponibilidad del servicio. Los métodos tradicionales de actualización requieren detener el servicio, lo que genera tiempos de inactividad. La solución presentada utiliza una arquitectura que permite cambiar entre versiones del modelo de forma dinámica, garantizando continuidad operativa. Los desafíos clave resueltos incluyen: actualización sin detener el servicio, enrutamiento de solicitudes durante la transición, gestión de múltiples versiones del modelo y mecanismos de monitoreo y reversión.
- Preparación del Entorno e Implementación Inicial
2.1 Requisitos e Instalación de Dependencias
El entorno debe cumplir con: Python 3.8 o superior, al menos 2 GB de memoria RAM disponible, y soporte para ONNX Runtime en CPU o GPU.
Instale los paquetes necesarios con los siguientes comandos:
pip install funasr-onnx gradio fastapi uvicorn soundfile jieba
pip install prometheus-client psutil # Opcional, para monitoreo
2.2 Despliegue del Modelo Base
Descargue el modelo cuantizado en el directorio designado. La siguiente verificación asegura su correcta carga:
from funasr_onnx import SenseVoiceSmall
gestor_modelos = SenseVoiceSmall(
"/ruta/a/modelos/sensevoice-small-onnx-quant",
batch_size=10,
quantize=True
)
resultado_prueba = gestor_modelos(["audio_prueba.wav"], language="auto", use_itn=True)
print("Carga exitosa:", resultado_prueba[0] is not None)
- Diseño de la Arquitectura de Actualización en Caliente
3.1 Principios Fundamentales
La arquitectura se basa en un gestor de modelos que carga versiones múltiples simultáneamente y enruta las solicitudes hacia la versión activa. Los componentes principales son: cargador de modelos, administrador de versiones, controlador de enrutamiento y verificador de salud.
3.2 Implementación del Gestor de Modelos
La clase GestorModelosAdmin permite la carga, conmutación y des-carga de versiones del modelo. Se utiliza un bloqueo de hilos para garantizar la seguridad en entornos concurrentes.
import threading
import time
from typing import Dict
from funasr_onnx import SenseVoiceSmall
class GestorModelosAdmin:
def __init__(self, ruta_base: str):
self.ruta_base = ruta_base
self.modelos_cargados: Dict[str, SenseVoiceSmall] = {}
self.versiones_activas: Dict[str, dict] = {}
self.version_actual = "inicial"
self.bloqueo = threading.RLock()
def cargar_version(self, etiqueta: str, ruta_modelo: str, tamano_lote: int = 10):
with self.bloqueo:
if etiqueta in self.modelos_cargados:
return False
try:
modelo = SenseVoiceSmall(ruta_modelo, batch_size=tamano_lote, quantize=True)
if modelo([], language="auto") is not None:
self.modelos_cargados[etiqueta] = modelo
self.versiones_activas[etiqueta] = {"ruta": ruta_modelo, "timestamp": time.time()}
return True
except Exception as error:
print(f"Error al cargar la versión {etiqueta}: {error}")
return False
def conmutar_version(self, nueva_etiqueta: str):
with self.bloqueo:
if nueva_etiqueta in self.modelos_cargados:
self.version_actual = nueva_etiqueta
return True
return False
def obtener_modelo_activo(self):
with self.bloqueo:
return self.modelos_cargados.get(self.version_actual)
def descargar_version(self, etiqueta: str):
with self.bloqueo:
if etiqueta != self.version_actual and etiqueta in self.modelos_cargados:
del self.modelos_cargados[etiqueta]
del self.versiones_activas[etiqueta]
return True
return False
- Estrategia de Actualización sin Tiempo de Inactividad
4.1 Flujo de Actualización
El proceso incluye: preparación del nuevo modelo en un directorio temporal, carga en segundo plano para validación, conmutación del tráfico a la nueva versión y limpieza diferida de versiones antiguas tras completar solicitudes pendientes.
4.2 Controlador de Actualización
La clase ControladorActualizaciones orquesta el ciclo de vida de las versiones, incluyendo preparación, despliegue y reversión.
import os
import shutil
import threading
import time
class ControladorActualizaciones:
def __init__(self, gestor: GestorModelosAdmin, directorio_temp: str):
self.gestor = gestor
self.directorio_temp = directorio_temp
os.makedirs(directorio_temp, exist_ok=True)
def preparar_nueva_version(self, origen: str, etiqueta: str = None):
etiqueta = etiqueta or f"v{int(time.time())}"
destino = os.path.join(self.directorio_temp, etiqueta)
os.makedirs(destino, exist_ok=True)
for archivo in os.listdir(origen):
if archivo.endswith('.onnx') or archivo.endswith('.json'):
shutil.copy2(os.path.join(origen, archivo), os.path.join(destino, archivo))
return etiqueta, destino
def ejecutar_actualizacion_cero_parada(self, nueva_ruta: str, etiqueta: str = None):
etiqueta, ruta_temp = self.preparar_nueva_version(nueva_ruta, etiqueta)
if not self.gestor.cargar_version(etiqueta, ruta_temp):
return False
if not self.gestor.conmutar_version(etiqueta):
return False
versiones_antiguas = [v for v in self.gestor.modelos_cargados if v != etiqueta]
def limpiar_antiguas():
time.sleep(300) # Espera para solicitudes pendientes
for v in versiones_antiguas:
self.gestor.descargar_version(v)
shutil.rmtree(os.path.join(self.directorio_temp, v), ignore_errors=True)
threading.Thread(target=limpiar_antiguas, daemon=True).start()
return True
def revertir_a_version(self, version_objetivo: str = None):
versiones_disponibles = list(self.gestor.modelos_cargados.keys())
objetivo = version_objetivo if version_objetivo in versiones_disponibles else (versiones_disponibles[0] if versiones_disponibles else None)
return self.gestor.conmutar_version(objetivo) if objetivo else False
4.3 Integración con FastAPI
El servicio web expone endpoints para la transcripción de audio y la administración de actualizaciones.
from fastapi import FastAPI, UploadFile, File, HTTPException
import tempfile
app = FastAPI()
gestor = GestorModelosAdmin("/ruta/modelos")
controlador = ControladorActualizaciones(gestor, "/ruta/temp")
@app.on_event("startup")
async def inicio():
gestor.cargar_version("inicial", "/ruta/modelos")
@app.post("/transcribir")
async def transcribir(audio: UploadFile = File(...)):
modelo_activo = gestor.obtener_modelo_activo()
if not modelo_activo:
raise HTTPException(status_code=503, detail="Modelo no disponible")
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
tmp.write(await audio.read())
resultado = modelo_activo([tmp.name], language="auto")
os.unlink(tmp.name)
return {"texto": resultado[0] if resultado else ""}
@app.post("/admin/actualizar_modelo")
async def actualizar(ruta_modelo: str, etiqueta: str = None):
exito = controlador.ejecutar_actualizacion_cero_parada(ruta_modelo, etiqueta)
return {"exito": exito, "version_actual": gestor.version_actual}
@app.post("/admin/revertir")
async def revertir(version_objetivo: str = None):
exito = controlador.revertir_a_version(version_objetivo)
return {"exito": exito}
- Monitoreo y Mantenimiento
5.1 Métricas de Rendimiento
Se implementan métricas para rastrear el uso de memoria, tiempo de procesamiento y conteo de solicitudes. Un middleware en FastAPI captura estas métricas por versión del modelo.
5.2 Verificación de Salud y Recuperación Automática
La clase VerificadorSalud ejecuta pruebas periódicas del modelo activo. Si se detectan fallos consecutivos, intenta cambiar a una versión estable alternativa.
class VerificadorSalud:
def __init__(self, gestor: GestorModelosAdmin):
self.gestor = gestor
self.fallos_consecutivos = 0
async def verificar(self):
modelo = self.gestor.obtener_modelo_activo()
try:
if modelo is not None and modelo([], language="auto") is not None:
self.fallos_consecutivos = 0
return True
except:
self.fallos_consecutivos += 1
return False
async def recuperar_automaticamente(self):
if self.fallos_consecutivos >= 3:
for version in self.gestor.modelos_cargados:
if self.gestor.conmutar_version(version):
if await self.verificar():
return True
return False
- Despliegue y Validación Práctica
Un script de shell automatiza la instalación y configuración del servicio como un daemon del sistema. Las pruebas de actualización se realizan mediante solicitueds continuas al endpoint de transcripción durante el proceso de cambio de modelo, verificando la continuidad del servicio.
- Recomendaciones para Producción
Para garantizar robustez: utilice convenciones de versionado semántico, mantenga siempre al menos una versión estable para reversión, configure alertas para degradaciones del rendimiento, y realice pruebas exhaustivas en entornos pre-productivos. Optimice el parámetro batch\_size según la carga, y considere el uso de GPU para acelerar la inferencia.