用 FunASR 实现实时流式语音识别(边说边出字)
实时字幕、语音助手、会议转写——这些场景要的不是"录完再转",而是边说边出字:用户话音未落,文字已经在屏幕上滚动。这就是流式语音识别(Streaming ASR)。
FunASR 的流式 Paraformer 用分块(chunk)+ 缓存(cache)解码,可以做到 600ms 级延迟的实时出字,而且 CPU 上就能跑。
安装
pip install -U funasr modelscope
完整代码(可直接运行)
from funasr import AutoModel
import soundfile as sf
chunk_size = [0, 10, 5] # 600 ms chunks (10 * 60 ms center)
encoder_chunk_look_back = 4 # encoder look-back chunks
decoder_chunk_look_back = 1 # decoder look-back chunks
model = AutoModel(model="paraformer-zh-streaming")
audio, sr = sf.read("speech.wav", dtype="float32") # 16 kHz mono
chunk_stride = chunk_size[1] * 960 # 600 ms @ 16 kHz
cache = {}
n_chunks = (len(audio) - 1) // chunk_stride + 1
for i in range(n_chunks):
chunk = audio[i * chunk_stride : (i + 1) * chunk_stride]
is_final = i == n_chunks - 1
res = model.generate(
input=chunk, cache=cache, is_final=is_final,
chunk_size=chunk_size,
encoder_chunk_look_back=encoder_chunk_look_back,
decoder_chunk_look_back=decoder_chunk_look_back,
)
if res[0]["text"]:
print(res[0]["text"], end="", flush=True) # emit partial text
输出是逐步增长的(示例)
今天 # ~600 ms after speech starts
今天天气
今天天气真
今天天气真不错 # is_final=True
每喂入一个 600ms 的音频块,模型就吐出这一块新识别到的文字,屏幕上的句子逐字增长,直到 is_final=True 收尾。实测一段 12 秒音频被切成 20 个块,逐块返回增量结果。
关键参数
| 参数 | 含义 |
|---|---|
chunk_size=[0,10,5] | [左看, 中心, 右看] 块大小,单位 60ms。中心 10 → 每块 600ms(延迟与精度的平衡点) |
encoder_chunk_look_back=4 | 编码器回看的历史块数(越大上下文越足、延迟略增) |
decoder_chunk_look_back=1 | 解码器回看的历史块数 |
cache={} | 跨块传递的状态,必须在多次调用间复用同一个 dict |
is_final | 最后一块置 True,触发收尾解码 |
最佳实践:2-pass(流式 + 离线)
流式模型为了低延迟,精度会略低于离线模型。生产里常用 2-pass 方案:
- 第一遍(流式):用流式 Paraformer 实时出字,给用户即时反馈;
- 第二遍(离线):一句话结束后(用 VAD 判定停顿),把整段音频再用 SenseVoice / 离线 Paraformer 重转一遍,得到更准的最终结果替换屏幕上的临时文本。
FunASR 官方的 WebSocket 服务(funasr-runtime-sdk-online-cpu)就内置了这套 2-pass 流式协议,可直接部署。
典型场景
- 实时字幕:直播、会议、课堂的即时上字幕。
- 语音助手:边听边理解,降低端到端响应延迟。
- 客服/质检:通话实时转写 + 实时关键词告警。
FunASR 是通义实验室开源的工业级语音识别工具包,离线 + 流式全覆盖,中文又快又准。
在 GitHub 上 Star FunASR ★