ONNX模型动态批处理SenseVoice-Small ONNX服务吞吐量优化教程1. 引言为什么需要优化语音识别服务的吞吐量想象一下你搭建了一个语音识别服务单个用户上传一段10秒的音频模型能在70毫秒内快速返回结果体验非常流畅。但突然有10个、100个用户同时上传音频进行识别你的服务响应开始变慢甚至排队等待用户体验直线下降。这就是典型的吞吐量瓶颈问题。对于像SenseVoice-Small这样优秀的语音识别模型其单次推理速度已经非常快10秒音频仅需70毫秒。但在真实的生产环境中我们面对的是高并发请求。如果每次只处理一个用户的请求再快的模型也无法满足大量用户同时使用的需求。这就好比只有一个收银台的超市收银员手速再快也架不住排起长队。动态批处理Dynamic Batching正是解决这个问题的关键技术。它允许服务端将短时间内收到的多个请求“打包”成一个批次一次性送给模型进行推理然后再将结果分别返回给每个用户。这能极大地提升GPU等计算资源的利用率从而显著提高服务的整体吞吐量单位时间内能处理的请求数。本文将手把手带你为基于ONNX格式的SenseVoice-Small语音识别模型实现动态批处理功能。我们将从原理讲起然后改造现有的Gradio WebUI服务代码最后对比优化前后的性能差异。目标是让你不只能部署一个“能跑”的服务更能部署一个“高效、能抗住压力”的生产级服务。2. 理解动态批处理的核心原理在开始动手之前我们先花点时间搞清楚动态批处理到底是怎么工作的以及为什么它能提升性能。2.1 什么是批处理你可以把模型的推理过程想象成工厂的流水线。单个音频推理就像流水线只为一件产品开机大部分时间机器都在空转等待。批处理则是把多件相似的产品多个音频同时放上流水线一次性加工完成。对于深度学习模型尤其是运行在GPU上的模型批处理能带来两大好处提升计算并行度GPU拥有成千上万个核心擅长同时处理大量相似的计算。批量处理数据能让这些核心“吃饱”利用率接近100%而不是处理单个样本时大部分核心在闲置。分摊固定开销每次模型推理都有一些固定的开销比如将数据从内存传输到GPUPCIe传输、启动GPU计算任务等。处理一个样本要付一次这个“启动费”处理100个样本也只需要付一次或略微增加平均到每个样本的成本就大大降低了。2.2 “动态”又是什么意思传统的批处理通常是静态的。比如在训练模型时我们固定每个批次batch的大小是32或64。但在在线服务中请求是随机、实时到达的我们无法预先凑齐一个固定数量的批次。动态批处理聪明在这里等待与收集服务不会一收到请求就立刻推理而是设置一个很短的等待窗口例如50毫秒。动态组批在这个窗口期内到达的所有请求会被收集起来。窗口结束时无论收集到几个请求比如3个、5个或10个都把它们组成一个批次送给模型。处理与分发模型一次性推理完这个批次服务再将结果拆分分别返回给对应的用户。这样既享受了批处理的计算效率又保证了服务的实时性等待窗口很短。对于SenseVoice-Small这种单次推理很快的模型动态批处理能将吞吐量提升数倍甚至数十倍。2.3 ONNX Runtime与动态批处理ONNX Runtime (ORT) 是运行ONNX模型的高性能推理引擎。它原生支持动态批处理但需要模型和输入输出做一些特定的配置才能生效。关键点在于输入维度模型的输入张量需要包含一个批次维度batch dimension并且这个维度通常被标记为动态的dynamic用“-1”或“N”表示。这意味着模型可以接受任意批次大小的输入。会话配置在创建ORT推理会话InferenceSession时我们可以配置执行提供器如CUDAExecutionProvider的优化选项来启用批处理。SenseVoice-Small的ONNX模型通常已经支持动态批次维度这为我们实施优化提供了基础。3. 环境准备与代码结构分析在动手改造之前我们先确保环境一致并理解现有代码的结构。3.1 环境确认假设你已经通过ModelScope的镜像成功部署了SenseVoice-Small ONNX服务并能通过Gradio WebUI界面正常使用。我们的优化工作将在该环境基础上进行。核心的Python库依赖通常包括onnxruntime-gpu(或onnxruntime)用于推理ONNX模型。modelscope用于下载和管理ModelScope的模型。gradio用于构建Web界面。numpy,soundfile,librosa等用于音频处理。你的环境应该已经具备了这些。3.2 现有代码流程分析根据提供的路径/usr/local/bin/webui.py我们推断现有的服务流程大致如下加载模型使用ModelScope从仓库下载或从本地加载SenseVoice-Small的ONNX模型。创建推理会话使用ONNX Runtime创建一个InferenceSession。定义处理函数接收Gradio上传的音频文件或字节数据。对音频进行预处理重采样、转换为特征等。调用session.run()进行推理。对推理结果进行后处理得到文本、情感等信息。启动Gradio界面将处理函数与UI组件上传按钮、录音按钮、文本框绑定并启动Web服务。当前模式是每个用户请求触发一次完整的“预处理-推理-后处理”流程且推理是单样本的。我们的目标是将其改造为多个请求先被收集预处理后批量推理再分别后处理返回。4. 实现动态批处理服务我们将创建一个新的服务文件比如webui_batch.py。以下是核心实现步骤。4.1 创建批处理推理引擎这是最核心的部分我们将封装一个类来管理模型和批处理逻辑。import onnxruntime as ort import numpy as np import threading import time from queue import Queue from typing import List, Dict, Any, Optional import copy class SenseVoiceBatchInferenceEngine: def __init__(self, model_path: str, batch_timeout_ms: int 50, max_batch_size: int 16): 初始化批处理推理引擎。 Args: model_path: ONNX模型文件路径。 batch_timeout_ms: 批处理超时时间毫秒。收集请求等待的最大时间。 max_batch_size: 最大批次大小。防止单个批次过大导致内存溢出。 self.batch_timeout_ms batch_timeout_ms / 1000.0 # 转换为秒 self.max_batch_size max_batch_size # 配置ONNX Runtime以优化批处理性能 # 对于GPU使用CUDA执行提供器并开启优化 providers [CUDAExecutionProvider] if ort.get_device() GPU else [CPUExecutionProvider] sess_options ort.SessionOptions() # 关键配置启用ORTOptimizer它对动态形状和批处理有更好支持 # sess_options.graph_optimization_level ort.GraphOptimizationLevel.ORT_ENABLE_ALL # 对于动态批处理更推荐在会话创建后由ORT自动管理以下配置供参考 # sess_options.add_session_config_entry(session.dynamic_blocking, 1) # sess_options.add_session_config_entry(session.enable_sequential_execution, 0) print(f正在加载ONNX模型: {model_path}) self.session ort.InferenceSession(model_path, sess_optionssess_options, providersproviders) print(模型加载完毕。) # 获取模型输入输出信息确认支持动态批次 self.input_name self.session.get_inputs()[0].name input_shape self.session.get_inputs()[0].shape print(f模型输入名称: {self.input_name}, 形状: {input_shape}) # 通常第一个维度是批次维度应为-1或正整数 if input_shape[0] 0: print(f警告模型输入形状为固定批次 {input_shape[0]}可能不支持动态批处理。尝试强制推理。) # 请求队列和批处理线程 self.request_queue Queue() self.result_dict {} # 用于存储请求ID到结果的映射 self.lock threading.Lock() self.batch_thread threading.Thread(targetself._batch_processing_loop, daemonTrue) self.batch_thread.start() print(f批处理推理引擎已启动超时{batch_timeout_ms}ms最大批次{max_batch_size}) def _preprocess_audio(self, audio_data: np.ndarray, sample_rate: int) - np.ndarray: 音频预处理。这里需要根据SenseVoice模型的具体要求实现。 例如重采样到16kHz提取Fbank特征归一化等。 这是一个简化示例你需要替换成真实的预处理逻辑。 # 假设模型输入需要 [batch, time, feature] 形状的log-mel特征 # 此处仅为示例实际预处理请参考SenseVoice官方代码 target_sr 16000 if sample_rate ! target_sr: # 使用librosa等库进行重采样 import librosa audio_data librosa.resample(audio_data, orig_srsample_rate, target_srtarget_sr) # 提取Fbank特征 (示例参数需调整) # features extract_fbank(audio_data, target_sr) # 这里我们模拟一个特征 time_length len(audio_data) // (target_sr // 100) # 模拟每10ms一帧 feature_dim 80 # 假设特征维度80 features np.random.randn(time_length, feature_dim).astype(np.float32) # 添加批次维度 features np.expand_dims(features, axis0) # 形状变为 [1, time, feature] return features def async_recognize(self, audio_data: np.ndarray, sample_rate: int, request_id: str) - None: 异步识别接口。将请求放入队列立即返回。 Args: audio_data: 音频波形数据。 sample_rate: 音频采样率。 request_id: 唯一请求ID用于后续获取结果。 # 预处理音频注意预处理在放入队列前完成避免在批处理线程中重复计算 # 但更高效的做法是将原始音频和参数入队在批处理线程中统一预处理避免主线程阻塞。 # 这里为了简化我们传入预处理后的特征。 features self._preprocess_audio(audio_data, sample_rate) with self.lock: self.result_dict[request_id] {status: processing, result: None} # 将请求放入队列 self.request_queue.put({ request_id: request_id, features: features, audio_length: features.shape[1] # 记录时间长度可用于填充 }) def get_result(self, request_id: str) - Optional[Dict[str, Any]]: 根据request_id获取识别结果。 with self.lock: return self.result_dict.get(request_id) def _batch_processing_loop(self): 批处理线程的主循环。 while True: batch_requests [] batch_data [] audio_lengths [] # 步骤1收集第一个请求 try: first_request self.request_queue.get(timeoutself.batch_timeout_ms) batch_requests.append(first_request) batch_data.append(first_request[features]) audio_lengths.append(first_request[audio_length]) start_time time.time() except Queue.Empty: # 超时没有请求继续等待 continue # 步骤2在超时时间内尽可能收集更多请求 while len(batch_requests) self.max_batch_size: try: next_request self.request_queue.get(timeoutmax(0, self.batch_timeout_ms - (time.time() - start_time))) batch_requests.append(next_request) batch_data.append(next_request[features]) audio_lengths.append(next_request[audio_length]) except Queue.Empty: # 超时停止收集开始处理当前批次 break # 步骤3动态填充组成一个批次 # 由于音频长度可能不同需要填充到同一长度 max_len max(audio_lengths) padded_batch [] for feat, length in zip(batch_data, audio_lengths): if length max_len: # 填充假设在时间轴axis1上填充 pad_width [(0,0), (0, max_len - length), (0,0)] padded_feat np.pad(feat, pad_width, modeconstant, constant_values0) else: padded_feat feat padded_batch.append(padded_feat) batch_input np.concatenate(padded_batch, axis0) # 形状 [batch_size, max_time, feature] # 步骤4批量推理 try: ort_inputs {self.input_name: batch_input} outputs self.session.run(None, ort_inputs) # 假设第一个输出是识别结果文本logits或序列 batch_results outputs[0] except Exception as e: print(f批量推理失败: {e}) batch_results [None] * len(batch_requests) # 步骤5处理结果并分发给每个请求 for i, req in enumerate(batch_requests): request_id req[request_id] original_length audio_lengths[i] result_for_this batch_results[i] if batch_results[i] is not None else None # 这里需要对result_for_this进行后处理例如解码成文本 # 假设有一个 _postprocess 函数 final_text self._postprocess(result_for_this, original_length) if result_for_this is not None else 识别失败 with self.lock: self.result_dict[request_id] {status: done, result: final_text} print(f请求 {request_id} 处理完成。) # 标记队列任务完成 self.request_queue.task_done() def _postprocess(self, model_output: np.ndarray, audio_length: int) - str: 后处理函数将模型输出解码为文本。 此处需要根据SenseVoice模型的实际输出和解码方式实现如CTC解码。 这里返回模拟结果。 # 模拟解码过程 # 实际应调用SenseVoice的解码器 return f模拟识别文本 (音频长度: {audio_length} 帧)4.2 集成到Gradio Web服务接下来我们修改Gradio应用使用上面的批处理引擎。import gradio as gr import numpy as np import soundfile as sf import uuid import time from pathlib import Path # 假设批处理引擎类定义在同一个文件或已导入 # from sensevoice_batch_engine import SenseVoiceBatchInferenceEngine # 初始化引擎 MODEL_PATH path/to/your/sensevoice-small.onnx # 替换为实际模型路径 engine SenseVoiceBatchInferenceEngine(model_pathMODEL_PATH, batch_timeout_ms50, max_batch_size8) def recognize_audio(audio_file_path: str) - str: Gradio处理函数。 # 生成唯一请求ID request_id str(uuid.uuid4()) # 读取音频文件 try: audio_data, sample_rate sf.read(audio_file_path) # 如果音频是双声道转换为单声道 if len(audio_data.shape) 1: audio_data np.mean(audio_data, axis1) except Exception as e: return f读取音频文件失败: {e} # 提交异步识别请求 engine.async_recognize(audio_data, sample_rate, request_id) # 轮询获取结果简单实现生产环境建议用WebSocket或长轮询 max_wait_time 30 # 最大等待30秒 start_time time.time() while time.time() - start_time max_wait_time: result_info engine.get_result(request_id) if result_info and result_info[status] done: return result_info[result] time.sleep(0.01) # 短暂休眠避免CPU空转 return 识别超时 # 构建Gradio界面 demo gr.Interface( fnrecognize_audio, inputsgr.Audio(typefilepath, label上传或录制音频), outputsgr.Textbox(label识别结果), titleSenseVoice-Small 语音识别 (动态批处理优化版), description上传音频文件或使用麦克风录制体验高并发下的高效识别服务。, examples[[example_audio1.wav], [example_audio2.wav]] # 可提供示例音频路径 ) if __name__ __main__: # 设置共享允许多用户并发访问 demo.queue(concurrency_count10) # 设置Gradio队列的并发数 demo.launch(server_name0.0.0.0, server_port7860, shareFalse)4.3 关键改造点解析异步化recognize_audio函数不再同步调用模型而是将请求提交给engine.async_recognize后立即返回并通过轮询实际生产应用建议用更高效的方式如WebSocket等待结果。这避免了Gradio工作线程被长时间阻塞。请求队列SenseVoiceBatchInferenceEngine内部维护一个请求队列 (request_queue) 和一个结果字典 (result_dict)。批处理线程一个独立的守护线程 (_batch_processing_loop) 持续运行负责从队列收集请求、组批、推理和分发结果。动态组批策略线程采用“先取一个再等一段时间收集更多”的策略在延迟和吞吐量之间取得平衡。输入填充由于音频长度不一在组批时需要进行填充padding以确保输入张量形状一致。这里在时间轴上进行零填充。5. 性能测试与优化效果对比理论再好也需要数据验证。我们来设计一个简单的测试对比优化前后的性能。5.1 测试方法我们可以写一个简单的客户端脚本模拟多个用户同时发送请求。# test_client.py import requests import time import threading import json BASE_URL http://localhost:7860 # 假设Gradio服务运行在本机 def send_request(audio_path): 模拟单个用户请求 start time.time() try: # 注意Gradio接口通常通过API调用这里简化处理。 # 实际测试可能需要根据Gradio的API格式构造请求。 files {audio: open(audio_path, rb)} response requests.post(f{BASE_URL}/api/predict, filesfiles) result response.json() latency time.time() - start print(f请求完成耗时: {latency:.3f}s, 结果: {result}) return latency except Exception as e: print(f请求失败: {e}) return None def concurrent_test(num_clients10, audio_pathtest.wav): 并发测试 threads [] latencies [] def worker(): lat send_request(audio_path) if lat: latencies.append(lat) start_total time.time() for i in range(num_clients): t threading.Thread(targetworker) threads.append(t) t.start() time.sleep(0.05) # 稍微错开启动时间模拟真实并发 for t in threads: t.join() total_time time.time() - start_total print(f\n 并发测试报告 (客户端数{num_clients}) ) print(f总耗时: {total_time:.3f}s) if latencies: print(f平均延迟: {sum(latencies)/len(latencies):.3f}s) print(f最大延迟: {max(latencies):.3f}s) print(f最小延迟: {min(latencies):.3f}s) print(f吞吐量 (请求/秒): {num_clients / total_time:.2f}) if __name__ __main__: # 测试优化前的服务假设运行在7861端口 # BASE_URL http://localhost:7861 # concurrent_test(10) # 测试优化后的服务运行在7860端口 BASE_URL http://localhost:7860 concurrent_test(20) # 尝试更多并发5.2 预期优化效果指标优化前 (无批处理)优化后 (动态批处理)提升说明单请求延迟~70ms (纯模型推理) 网络/队列开销可能略增(增加~50ms等待窗口)为换取吞吐量个体延迟稍有牺牲但仍在可接受范围200ms。并发吞吐量低 (约 1 / (推理时间开销))高 (数倍至数十倍提升)GPU利用率从个位数提升到80%以上单位时间处理请求数大幅增加。GPU利用率低 (峰值利用率低)高且平稳批量计算使GPU核心充分工作利用率曲线平滑在高位。资源效率低 (为每个请求单独启动计算)高固定开销数据传输、内核启动被大量请求分摊。简单估算假设单次推理70ms动态批处理等待窗口50ms。在并发请求充足的情况下理想状态是每120ms处理一个批次。如果批次大小为8则平均每个请求的处理时间为120ms/8 15ms吞吐量理论上是原来的约4.7倍 (70ms / 15ms)。实际提升取决于请求的密集程度和批次大小。6. 总结与进阶建议通过本文的实践我们成功为SenseVoice-Small ONNX语音识别服务加上了动态批处理能力。从简单的单请求处理升级为能够高效应对高并发的生产级服务。关键点在于理解批处理原理并利用队列和独立线程实现请求的收集与批量推理。6.1 核心要点回顾动态批处理通过收集短时间内多个请求并批量推理极大提升了GPU利用率和系统吞吐量。实现核心是生产者-消费者模式Gradio接口作为生产者提交请求到队列独立的批处理线程作为消费者从队列取请求、组批、推理、返回结果。ONNX Runtime支持动态形状输入是实施该优化的基础。需要在延迟等待组批的时间和吞吐量批次大小之间根据业务需求进行权衡。6.2 进阶优化方向更智能的批处理策略当前的超时等待策略比较简单。可以引入基于批次大小和最长等待时间的双重触发机制或者根据请求的预估计算量如音频长度进行优先级排序和组批。使用专门的服务框架对于更复杂的生产环境建议使用专门的模型服务框架如Triton Inference Server或TensorFlow Serving。它们内置了更成熟、更高效的动态批处理、模型版本管理、监控等功能。预处理/后处理卸载可以将音频解码、特征提取等预处理工作以及解码、格式化等后处理工作放到CPU上异步执行甚至使用专用线程池进一步释放GPU压力提升整体流水线效率。性能监控与调优持续监控服务的延迟P50, P99、吞吐量、GPU利用率、队列长度等指标根据实际负载动态调整批处理超时时间和最大批次大小。结合量化与图优化确保使用的ONNX模型是经过量化如FP16/INT8和ONNX Runtime图优化过的这能进一步降低延迟、提高吞吐量。希望这篇教程能帮助你构建出性能更强大的语音识别服务。动态批处理是优化在线推理服务性价比的利器掌握它让你在应对真实业务流量时更加从容。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
ONNX模型动态批处理：SenseVoice-Small ONNX服务吞吐量优化教程
发布时间：2026/6/9 4:36:19
ONNX模型动态批处理SenseVoice-Small ONNX服务吞吐量优化教程1. 引言为什么需要优化语音识别服务的吞吐量想象一下你搭建了一个语音识别服务单个用户上传一段10秒的音频模型能在70毫秒内快速返回结果体验非常流畅。但突然有10个、100个用户同时上传音频进行识别你的服务响应开始变慢甚至排队等待用户体验直线下降。这就是典型的吞吐量瓶颈问题。对于像SenseVoice-Small这样优秀的语音识别模型其单次推理速度已经非常快10秒音频仅需70毫秒。但在真实的生产环境中我们面对的是高并发请求。如果每次只处理一个用户的请求再快的模型也无法满足大量用户同时使用的需求。这就好比只有一个收银台的超市收银员手速再快也架不住排起长队。动态批处理Dynamic Batching正是解决这个问题的关键技术。它允许服务端将短时间内收到的多个请求“打包”成一个批次一次性送给模型进行推理然后再将结果分别返回给每个用户。这能极大地提升GPU等计算资源的利用率从而显著提高服务的整体吞吐量单位时间内能处理的请求数。本文将手把手带你为基于ONNX格式的SenseVoice-Small语音识别模型实现动态批处理功能。我们将从原理讲起然后改造现有的Gradio WebUI服务代码最后对比优化前后的性能差异。目标是让你不只能部署一个“能跑”的服务更能部署一个“高效、能抗住压力”的生产级服务。2. 理解动态批处理的核心原理在开始动手之前我们先花点时间搞清楚动态批处理到底是怎么工作的以及为什么它能提升性能。2.1 什么是批处理你可以把模型的推理过程想象成工厂的流水线。单个音频推理就像流水线只为一件产品开机大部分时间机器都在空转等待。批处理则是把多件相似的产品多个音频同时放上流水线一次性加工完成。对于深度学习模型尤其是运行在GPU上的模型批处理能带来两大好处提升计算并行度GPU拥有成千上万个核心擅长同时处理大量相似的计算。批量处理数据能让这些核心“吃饱”利用率接近100%而不是处理单个样本时大部分核心在闲置。分摊固定开销每次模型推理都有一些固定的开销比如将数据从内存传输到GPUPCIe传输、启动GPU计算任务等。处理一个样本要付一次这个“启动费”处理100个样本也只需要付一次或略微增加平均到每个样本的成本就大大降低了。2.2 “动态”又是什么意思传统的批处理通常是静态的。比如在训练模型时我们固定每个批次batch的大小是32或64。但在在线服务中请求是随机、实时到达的我们无法预先凑齐一个固定数量的批次。动态批处理聪明在这里等待与收集服务不会一收到请求就立刻推理而是设置一个很短的等待窗口例如50毫秒。动态组批在这个窗口期内到达的所有请求会被收集起来。窗口结束时无论收集到几个请求比如3个、5个或10个都把它们组成一个批次送给模型。处理与分发模型一次性推理完这个批次服务再将结果拆分分别返回给对应的用户。这样既享受了批处理的计算效率又保证了服务的实时性等待窗口很短。对于SenseVoice-Small这种单次推理很快的模型动态批处理能将吞吐量提升数倍甚至数十倍。2.3 ONNX Runtime与动态批处理ONNX Runtime (ORT) 是运行ONNX模型的高性能推理引擎。它原生支持动态批处理但需要模型和输入输出做一些特定的配置才能生效。关键点在于输入维度模型的输入张量需要包含一个批次维度batch dimension并且这个维度通常被标记为动态的dynamic用“-1”或“N”表示。这意味着模型可以接受任意批次大小的输入。会话配置在创建ORT推理会话InferenceSession时我们可以配置执行提供器如CUDAExecutionProvider的优化选项来启用批处理。SenseVoice-Small的ONNX模型通常已经支持动态批次维度这为我们实施优化提供了基础。3. 环境准备与代码结构分析在动手改造之前我们先确保环境一致并理解现有代码的结构。3.1 环境确认假设你已经通过ModelScope的镜像成功部署了SenseVoice-Small ONNX服务并能通过Gradio WebUI界面正常使用。我们的优化工作将在该环境基础上进行。核心的Python库依赖通常包括onnxruntime-gpu(或onnxruntime)用于推理ONNX模型。modelscope用于下载和管理ModelScope的模型。gradio用于构建Web界面。numpy,soundfile,librosa等用于音频处理。你的环境应该已经具备了这些。3.2 现有代码流程分析根据提供的路径/usr/local/bin/webui.py我们推断现有的服务流程大致如下加载模型使用ModelScope从仓库下载或从本地加载SenseVoice-Small的ONNX模型。创建推理会话使用ONNX Runtime创建一个InferenceSession。定义处理函数接收Gradio上传的音频文件或字节数据。对音频进行预处理重采样、转换为特征等。调用session.run()进行推理。对推理结果进行后处理得到文本、情感等信息。启动Gradio界面将处理函数与UI组件上传按钮、录音按钮、文本框绑定并启动Web服务。当前模式是每个用户请求触发一次完整的“预处理-推理-后处理”流程且推理是单样本的。我们的目标是将其改造为多个请求先被收集预处理后批量推理再分别后处理返回。4. 实现动态批处理服务我们将创建一个新的服务文件比如webui_batch.py。以下是核心实现步骤。4.1 创建批处理推理引擎这是最核心的部分我们将封装一个类来管理模型和批处理逻辑。import onnxruntime as ort import numpy as np import threading import time from queue import Queue from typing import List, Dict, Any, Optional import copy class SenseVoiceBatchInferenceEngine: def __init__(self, model_path: str, batch_timeout_ms: int 50, max_batch_size: int 16): 初始化批处理推理引擎。 Args: model_path: ONNX模型文件路径。 batch_timeout_ms: 批处理超时时间毫秒。收集请求等待的最大时间。 max_batch_size: 最大批次大小。防止单个批次过大导致内存溢出。 self.batch_timeout_ms batch_timeout_ms / 1000.0 # 转换为秒 self.max_batch_size max_batch_size # 配置ONNX Runtime以优化批处理性能 # 对于GPU使用CUDA执行提供器并开启优化 providers [CUDAExecutionProvider] if ort.get_device() GPU else [CPUExecutionProvider] sess_options ort.SessionOptions() # 关键配置启用ORTOptimizer它对动态形状和批处理有更好支持 # sess_options.graph_optimization_level ort.GraphOptimizationLevel.ORT_ENABLE_ALL # 对于动态批处理更推荐在会话创建后由ORT自动管理以下配置供参考 # sess_options.add_session_config_entry(session.dynamic_blocking, 1) # sess_options.add_session_config_entry(session.enable_sequential_execution, 0) print(f正在加载ONNX模型: {model_path}) self.session ort.InferenceSession(model_path, sess_optionssess_options, providersproviders) print(模型加载完毕。) # 获取模型输入输出信息确认支持动态批次 self.input_name self.session.get_inputs()[0].name input_shape self.session.get_inputs()[0].shape print(f模型输入名称: {self.input_name}, 形状: {input_shape}) # 通常第一个维度是批次维度应为-1或正整数 if input_shape[0] 0: print(f警告模型输入形状为固定批次 {input_shape[0]}可能不支持动态批处理。尝试强制推理。) # 请求队列和批处理线程 self.request_queue Queue() self.result_dict {} # 用于存储请求ID到结果的映射 self.lock threading.Lock() self.batch_thread threading.Thread(targetself._batch_processing_loop, daemonTrue) self.batch_thread.start() print(f批处理推理引擎已启动超时{batch_timeout_ms}ms最大批次{max_batch_size}) def _preprocess_audio(self, audio_data: np.ndarray, sample_rate: int) - np.ndarray: 音频预处理。这里需要根据SenseVoice模型的具体要求实现。 例如重采样到16kHz提取Fbank特征归一化等。 这是一个简化示例你需要替换成真实的预处理逻辑。 # 假设模型输入需要 [batch, time, feature] 形状的log-mel特征 # 此处仅为示例实际预处理请参考SenseVoice官方代码 target_sr 16000 if sample_rate ! target_sr: # 使用librosa等库进行重采样 import librosa audio_data librosa.resample(audio_data, orig_srsample_rate, target_srtarget_sr) # 提取Fbank特征 (示例参数需调整) # features extract_fbank(audio_data, target_sr) # 这里我们模拟一个特征 time_length len(audio_data) // (target_sr // 100) # 模拟每10ms一帧 feature_dim 80 # 假设特征维度80 features np.random.randn(time_length, feature_dim).astype(np.float32) # 添加批次维度 features np.expand_dims(features, axis0) # 形状变为 [1, time, feature] return features def async_recognize(self, audio_data: np.ndarray, sample_rate: int, request_id: str) - None: 异步识别接口。将请求放入队列立即返回。 Args: audio_data: 音频波形数据。 sample_rate: 音频采样率。 request_id: 唯一请求ID用于后续获取结果。 # 预处理音频注意预处理在放入队列前完成避免在批处理线程中重复计算 # 但更高效的做法是将原始音频和参数入队在批处理线程中统一预处理避免主线程阻塞。 # 这里为了简化我们传入预处理后的特征。 features self._preprocess_audio(audio_data, sample_rate) with self.lock: self.result_dict[request_id] {status: processing, result: None} # 将请求放入队列 self.request_queue.put({ request_id: request_id, features: features, audio_length: features.shape[1] # 记录时间长度可用于填充 }) def get_result(self, request_id: str) - Optional[Dict[str, Any]]: 根据request_id获取识别结果。 with self.lock: return self.result_dict.get(request_id) def _batch_processing_loop(self): 批处理线程的主循环。 while True: batch_requests [] batch_data [] audio_lengths [] # 步骤1收集第一个请求 try: first_request self.request_queue.get(timeoutself.batch_timeout_ms) batch_requests.append(first_request) batch_data.append(first_request[features]) audio_lengths.append(first_request[audio_length]) start_time time.time() except Queue.Empty: # 超时没有请求继续等待 continue # 步骤2在超时时间内尽可能收集更多请求 while len(batch_requests) self.max_batch_size: try: next_request self.request_queue.get(timeoutmax(0, self.batch_timeout_ms - (time.time() - start_time))) batch_requests.append(next_request) batch_data.append(next_request[features]) audio_lengths.append(next_request[audio_length]) except Queue.Empty: # 超时停止收集开始处理当前批次 break # 步骤3动态填充组成一个批次 # 由于音频长度可能不同需要填充到同一长度 max_len max(audio_lengths) padded_batch [] for feat, length in zip(batch_data, audio_lengths): if length max_len: # 填充假设在时间轴axis1上填充 pad_width [(0,0), (0, max_len - length), (0,0)] padded_feat np.pad(feat, pad_width, modeconstant, constant_values0) else: padded_feat feat padded_batch.append(padded_feat) batch_input np.concatenate(padded_batch, axis0) # 形状 [batch_size, max_time, feature] # 步骤4批量推理 try: ort_inputs {self.input_name: batch_input} outputs self.session.run(None, ort_inputs) # 假设第一个输出是识别结果文本logits或序列 batch_results outputs[0] except Exception as e: print(f批量推理失败: {e}) batch_results [None] * len(batch_requests) # 步骤5处理结果并分发给每个请求 for i, req in enumerate(batch_requests): request_id req[request_id] original_length audio_lengths[i] result_for_this batch_results[i] if batch_results[i] is not None else None # 这里需要对result_for_this进行后处理例如解码成文本 # 假设有一个 _postprocess 函数 final_text self._postprocess(result_for_this, original_length) if result_for_this is not None else 识别失败 with self.lock: self.result_dict[request_id] {status: done, result: final_text} print(f请求 {request_id} 处理完成。) # 标记队列任务完成 self.request_queue.task_done() def _postprocess(self, model_output: np.ndarray, audio_length: int) - str: 后处理函数将模型输出解码为文本。 此处需要根据SenseVoice模型的实际输出和解码方式实现如CTC解码。 这里返回模拟结果。 # 模拟解码过程 # 实际应调用SenseVoice的解码器 return f模拟识别文本 (音频长度: {audio_length} 帧)4.2 集成到Gradio Web服务接下来我们修改Gradio应用使用上面的批处理引擎。import gradio as gr import numpy as np import soundfile as sf import uuid import time from pathlib import Path # 假设批处理引擎类定义在同一个文件或已导入 # from sensevoice_batch_engine import SenseVoiceBatchInferenceEngine # 初始化引擎 MODEL_PATH path/to/your/sensevoice-small.onnx # 替换为实际模型路径 engine SenseVoiceBatchInferenceEngine(model_pathMODEL_PATH, batch_timeout_ms50, max_batch_size8) def recognize_audio(audio_file_path: str) - str: Gradio处理函数。 # 生成唯一请求ID request_id str(uuid.uuid4()) # 读取音频文件 try: audio_data, sample_rate sf.read(audio_file_path) # 如果音频是双声道转换为单声道 if len(audio_data.shape) 1: audio_data np.mean(audio_data, axis1) except Exception as e: return f读取音频文件失败: {e} # 提交异步识别请求 engine.async_recognize(audio_data, sample_rate, request_id) # 轮询获取结果简单实现生产环境建议用WebSocket或长轮询 max_wait_time 30 # 最大等待30秒 start_time time.time() while time.time() - start_time max_wait_time: result_info engine.get_result(request_id) if result_info and result_info[status] done: return result_info[result] time.sleep(0.01) # 短暂休眠避免CPU空转 return 识别超时 # 构建Gradio界面 demo gr.Interface( fnrecognize_audio, inputsgr.Audio(typefilepath, label上传或录制音频), outputsgr.Textbox(label识别结果), titleSenseVoice-Small 语音识别 (动态批处理优化版), description上传音频文件或使用麦克风录制体验高并发下的高效识别服务。, examples[[example_audio1.wav], [example_audio2.wav]] # 可提供示例音频路径 ) if __name__ __main__: # 设置共享允许多用户并发访问 demo.queue(concurrency_count10) # 设置Gradio队列的并发数 demo.launch(server_name0.0.0.0, server_port7860, shareFalse)4.3 关键改造点解析异步化recognize_audio函数不再同步调用模型而是将请求提交给engine.async_recognize后立即返回并通过轮询实际生产应用建议用更高效的方式如WebSocket等待结果。这避免了Gradio工作线程被长时间阻塞。请求队列SenseVoiceBatchInferenceEngine内部维护一个请求队列 (request_queue) 和一个结果字典 (result_dict)。批处理线程一个独立的守护线程 (_batch_processing_loop) 持续运行负责从队列收集请求、组批、推理和分发结果。动态组批策略线程采用“先取一个再等一段时间收集更多”的策略在延迟和吞吐量之间取得平衡。输入填充由于音频长度不一在组批时需要进行填充padding以确保输入张量形状一致。这里在时间轴上进行零填充。5. 性能测试与优化效果对比理论再好也需要数据验证。我们来设计一个简单的测试对比优化前后的性能。5.1 测试方法我们可以写一个简单的客户端脚本模拟多个用户同时发送请求。# test_client.py import requests import time import threading import json BASE_URL http://localhost:7860 # 假设Gradio服务运行在本机 def send_request(audio_path): 模拟单个用户请求 start time.time() try: # 注意Gradio接口通常通过API调用这里简化处理。 # 实际测试可能需要根据Gradio的API格式构造请求。 files {audio: open(audio_path, rb)} response requests.post(f{BASE_URL}/api/predict, filesfiles) result response.json() latency time.time() - start print(f请求完成耗时: {latency:.3f}s, 结果: {result}) return latency except Exception as e: print(f请求失败: {e}) return None def concurrent_test(num_clients10, audio_pathtest.wav): 并发测试 threads [] latencies [] def worker(): lat send_request(audio_path) if lat: latencies.append(lat) start_total time.time() for i in range(num_clients): t threading.Thread(targetworker) threads.append(t) t.start() time.sleep(0.05) # 稍微错开启动时间模拟真实并发 for t in threads: t.join() total_time time.time() - start_total print(f\n 并发测试报告 (客户端数{num_clients}) ) print(f总耗时: {total_time:.3f}s) if latencies: print(f平均延迟: {sum(latencies)/len(latencies):.3f}s) print(f最大延迟: {max(latencies):.3f}s) print(f最小延迟: {min(latencies):.3f}s) print(f吞吐量 (请求/秒): {num_clients / total_time:.2f}) if __name__ __main__: # 测试优化前的服务假设运行在7861端口 # BASE_URL http://localhost:7861 # concurrent_test(10) # 测试优化后的服务运行在7860端口 BASE_URL http://localhost:7860 concurrent_test(20) # 尝试更多并发5.2 预期优化效果指标优化前 (无批处理)优化后 (动态批处理)提升说明单请求延迟~70ms (纯模型推理) 网络/队列开销可能略增(增加~50ms等待窗口)为换取吞吐量个体延迟稍有牺牲但仍在可接受范围200ms。并发吞吐量低 (约 1 / (推理时间开销))高 (数倍至数十倍提升)GPU利用率从个位数提升到80%以上单位时间处理请求数大幅增加。GPU利用率低 (峰值利用率低)高且平稳批量计算使GPU核心充分工作利用率曲线平滑在高位。资源效率低 (为每个请求单独启动计算)高固定开销数据传输、内核启动被大量请求分摊。简单估算假设单次推理70ms动态批处理等待窗口50ms。在并发请求充足的情况下理想状态是每120ms处理一个批次。如果批次大小为8则平均每个请求的处理时间为120ms/8 15ms吞吐量理论上是原来的约4.7倍 (70ms / 15ms)。实际提升取决于请求的密集程度和批次大小。6. 总结与进阶建议通过本文的实践我们成功为SenseVoice-Small ONNX语音识别服务加上了动态批处理能力。从简单的单请求处理升级为能够高效应对高并发的生产级服务。关键点在于理解批处理原理并利用队列和独立线程实现请求的收集与批量推理。6.1 核心要点回顾动态批处理通过收集短时间内多个请求并批量推理极大提升了GPU利用率和系统吞吐量。实现核心是生产者-消费者模式Gradio接口作为生产者提交请求到队列独立的批处理线程作为消费者从队列取请求、组批、推理、返回结果。ONNX Runtime支持动态形状输入是实施该优化的基础。需要在延迟等待组批的时间和吞吐量批次大小之间根据业务需求进行权衡。6.2 进阶优化方向更智能的批处理策略当前的超时等待策略比较简单。可以引入基于批次大小和最长等待时间的双重触发机制或者根据请求的预估计算量如音频长度进行优先级排序和组批。使用专门的服务框架对于更复杂的生产环境建议使用专门的模型服务框架如Triton Inference Server或TensorFlow Serving。它们内置了更成熟、更高效的动态批处理、模型版本管理、监控等功能。预处理/后处理卸载可以将音频解码、特征提取等预处理工作以及解码、格式化等后处理工作放到CPU上异步执行甚至使用专用线程池进一步释放GPU压力提升整体流水线效率。性能监控与调优持续监控服务的延迟P50, P99、吞吐量、GPU利用率、队列长度等指标根据实际负载动态调整批处理超时时间和最大批次大小。结合量化与图优化确保使用的ONNX模型是经过量化如FP16/INT8和ONNX Runtime图优化过的这能进一步降低延迟、提高吞吐量。希望这篇教程能帮助你构建出性能更强大的语音识别服务。动态批处理是优化在线推理服务性价比的利器掌握它让你在应对真实业务流量时更加从容。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
 介绍){kind=tracking}
