低延迟小智AI服务端搭建-ASR篇（续）：CPU可跑 - 链载Ai

如果你：

对高成本及其厌恶；
个人玩家，不关心用户体验。

可以试试接下来的CPU推理方案。

本篇将首先介绍sherpa-onnx，一款高性能的语音处理开源项目。

然后，选择其中的模型，实现小智AI服务端ASR的实时CPU推理。

1. sherpa-onnx 简介

https://github.com/k2-fsa/sherpa-onnx

sherpa-onnx是一款跨平台、多语言的语音处理工具包，使用 onnxruntime 实现。支持的功能非常丰富，包括：

语音活动检测
语音转文本
文本转语音
说话人分割

老规矩，简单介绍下亮点：

跨平台兼容：包括 Windows、macOS、Linux以及Android和iOS等操作系统。它还支持各种嵌入式系统。
多语言API：提供11种主流编程语言接口。
高性能：基于ONNX运行时，适合在各种计算能力的设备上部署。

2. 模型选择

模型列表：https://k2-fsa.github.io/sherpa/onnx/pretrained_models/index.html

根据上篇的实测，VAD 模型参数量较少，延时可忽略。

因此，VAD 模型选择和上篇保持一致。

2.1 ASR 模型

要想 ASR 识别效果不拉跨，模型参数量少不了。

同时还得尽可能满足低延迟。

综上，ASR模型，我们可以考虑选择支持流式推理的paraformer，这也是阿里云上的流式推理模型。

2.2 声纹向量模型

考虑到模型参数量和推理延时正相关，为了兼顾性能和延时，可以选择3dspeaker_speech_campplus_sv_zh_en_16k-common_advanced

3. CPU 推理方案

3.1 模型加载

VAD 模型加载和上篇保持一致，ASR 和声纹识别模型，采用sherpa-onnx的方式加载。

class ModelManager:
  def __init__(self):
    self.vad_model = None
    self.asr_model = None
    self.sv_model = None

  def load_models(self):
    self.vad_model = AutoModel(model="ckpts/speech_fsmn_vad_zh-cn-16k-common-pytorch")
   
    model_dir ="ckpts/sherpa-onnx-streaming-paraformer-bilingual-zh-en"
    self.asr_model = sherpa_onnx.OnlineRecognizer.from_paraformer(
      encoder=f'{model_dir}/encoder.int8.onnx',
      decoder=f'{model_dir}/decoder.int8.onnx',
      tokens=f'{model_dir}/tokens.txt')

    model_dir ="ckpts/3dspeaker_speech_campplus_sv_zh_en_16k-common_advanced.onnx"
    config = sherpa_onnx.SpeakerEmbeddingExtractorConfig(model=model_dir)
    self.sv_model = sherpa_onnx.SpeakerEmbeddingExtractor(config)

3.2 模型推理

服务的核心逻辑是：

当 VAD 检测到有活动音频时，asr_model 创建音频流self.stream：

ifself.stream is None:
  self.stream = self.model_manager.asr_model.create_stream()
  self.input_stream(self.audio_buffer)

然后，每来一个音频片段，都压入音频流：

def on_audio_frame(self, frame):
  frame_fp32 = np.frombuffer(frame, dtype=np.int16).astype(np.float32) / 32768
 ifself.stream:
    self.input_stream(frame_fp32)

实测发现，streaming-paraformer所谓的流式推理是：每 10 个 60ms 音频会触发一次解码。

而单次解码耗时 < 0.3s。

因此，rtf 确实可以做到 < 1。

为此，把音频数据压入一个消费队列，进行异步处理，理论上完全可以做到实时推理。

改造过程如下：

首先，音频数据不再直接解码，而是压入队列：

# self.input_stream(frame_fp32)
self.asr_queue.put(frame_fp32)

然后，创建一个 worker 异步消费队列中的音频：

def _decode_worker(self):
 whileTrue:
    try:
     # 使用阻塞方式获取音频数据
      chunk = self.asr_queue.get(timeout=0.01)
     ifself.stream: # 确保stream存在
        self.running = True
        self.input_stream(chunk)
    except Empty:
      self.running = False # 清除解码事件
     continue# 队列为空时继续等待
    except Exception as e:
      logging.error(f"Error in decode worker: {e}", exc_info=True)

最后，声纹向量模型获取音频特征：

def generate_embedding(self):
 ifself.audio_buffer.shape[0] == 0:
   return[]
 # last 3 seconds for speaker embedding
  stream = self.model_manager.sv_model.create_stream()
  stream.accept_waveform(SAMPLE_RATE, self.audio_buffer[-SAMPLE_RATE*3:])
  stream.input_finished()
  embedding = self.model_manager.sv_model.compute(stream)
 returnembedding

3.3 延时测试

streaming-paraformer流式推理：每 10 个 60ms 音频触发一次解码：

模型	CPU 推理 (秒)
paraformer	0.09-0.25
paraformer-int8	0.08-0.09

声纹向量模型，音频时长最长3s，单次推理：

模型	CPU 推理 (秒)
3dspeaker	0.09

3.4 服务器配置和成本

上述三个模型加载进来，占用内存 1.0 G+。

所以，如果不考虑高并发的情况，虚拟机配置需至少确保 2c2g。

如果考虑主流云厂商，包年包月的价格如下：

阿里云：https://www.aliyun.com/minisite/goods?userCode=ggqtukm3

腾讯云：https://curl.qcloud.com/BLm2fgkN

当然，如果只是想玩玩，压根没想用上一年啊。

推荐试试sealos的产品，根据使用资源，按需付费。

注册体验：https://cloud.sealos.run/?uid=QDGJoX2_Qp

新人注册送 10 元余额，足够玩一阵了。

我们以2c2g为例，并挂载一块20G的磁盘，每天的费用大概在 1.5 元左右，不用时关掉就好。

写在最后

本文分享了小智AI服务端ASR的CPU推理方案，并对配置和成本进行了估算。