FunASR 流式语音识别怎么接？按音频分块输出中文转写

接实时语音转写时，不能把一个普通离线识别模型不断重复调用，就当作流式 ASR。FunASR 的 paraformer-zh-streaming 接受连续音频块，并通过会话缓存保存上下文。这里先用本地 WAV 模拟逐块输入，把采样率、缓存和最后一块处理正确，再接麦克风或 WebSocket。

本文面向会使用 Python 的开发者，采用 Python 3.11、FunASR 1.4.16、16 kHz 单声道音频与 CPU 路径。代码依据 2026 年 10 月 1 日核对的官方示例改写，未下载模型运行，也未测实时延迟；文件逐块输入是接口演示，不是完整的麦克风采集服务。

一、准备模型环境与正确的音频

建立独立虚拟环境，安装 PyTorch、FunASR 1.4.16 和读音频所需的 SoundFile。以下是 Windows PowerShell 命令：

py -3.11 -m venv .venv-funasr
.\.venv-funasr\Scripts\python.exe -m pip install --upgrade pip
.\.venv-funasr\Scripts\python.exe -m pip install torch torchaudio
.\.venv-funasr\Scripts\python.exe -m pip install funasr==1.4.16 soundfile
.\.venv-funasr\Scripts\python.exe -m pip check

准备一段有权处理的普通话录音，转换成 16 kHz 单声道 WAV，命名为 speech-16k.wav。已有 FFmpeg 时可以使用：

ffmpeg -i source.wav -ac 1 -ar 16000 -c:a pcm_s16le speech-16k.wav

实际数据必须重采样；不能只把文件头的采样率数字改为 16000。检查转换文件能否播放，声音有没有被截断。官方流式例子采用 16 kHz 单声道，模型与离线 Paraformer 是不同检查点，依据见 FunASR 官方 README。

二、用同一个 cache 连续送入音频块

保存下面代码为 stream_demo.py。它读取真实录音，不包含固定“识别成功”文本；每行输出的内容由模型决定。

import soundfile as sf
from funasr import AutoModel

audio, sr = sf.read("speech-16k.wav", dtype="float32")
if sr != 16000 or audio.ndim != 1:
    raise ValueError("需要 16 kHz 单声道音频")
if len(audio) == 0:
    raise ValueError("音频为空")

model = AutoModel(model="paraformer-zh-streaming", device="cpu")
chunk_size = [0, 10, 5]
stride = chunk_size[1] * 960
cache = {}
n_chunks = (len(audio) - 1) // stride + 1

for i in range(n_chunks):
    block = audio[i * stride:(i + 1) * stride]
    result = model.generate(
        input=block,
        cache=cache,
        is_final=(i == n_chunks - 1),
        chunk_size=chunk_size,
        encoder_chunk_look_back=4,
        decoder_chunk_look_back=1,
    )
    text = result[0].get("text", "") if result else ""
    print(f"chunk={i} final={i == n_chunks - 1} text={text}")

运行：

.\.venv-funasr\Scripts\python.exe stream_demo.py

本例 stride=9600，在 16 kHz 输入下等于每块 0.6 秒音频。这个数字描述输入块的长度，不是端到端延迟保证；推理时间、模型前瞻、设备采集和网络都可能增加等待。首次运行还要下载模型，不能把下载阶段计入日常推理延迟。

三、缓存、尾块和空文本各代表什么

• 同一段连续音频复用 cache：不要在每个循环里重新写 cache={}，否则会丢掉流式上下文。
• 新会话使用新 cache：两个用户或两段独立录音不能共用同一个字典；模型实例能否并发使用还需要按实际运行方式核验。
• 最后一块设置 is_final：即使它不足 stride，也要送入并标记结束，避免结尾内容没有被提交。
• 空 text 不等于失败：短块、静音或模型等待更多上下文时可能暂时没有文字。用整段实际发言和最终输出判断，不要每块都要求必须出一句话。

先给这段录音人工写一份参考文本。完成后回听开头、停顿之后和最后一句，检查字有没有丢、重复或跨会话串入；再换第二段录音，确认新 cache 的结果不混入第一段内容。本文不预先给出你的录音会产生哪句文字。

四、从文件演示接到麦克风，还缺哪些工作

演示循环会尽快处理文件块，没有按真实采集时钟等待。接麦克风时，采集回调要把单声道 16 kHz 样本依顺序放入缓冲区，凑够 stride 再提交；不要在音频回调内做长时间阻塞推理。停止录音时送出剩余样本并标记 final，同时保留原始音频供校对。

对多客户端服务，应为每条连接建立独立队列与缓存，并明确连接中断、尾块未到、模型异常的处理。若准备使用现成网络服务，官方实时转写服务文档区分了 online、offline 和 2pass 模式：2pass 可在流式初稿后做句末修正，前端应区分临时结果与最终结果，避免把改写后的同一句追加两遍。该服务不是上面 Python 演示自动启动出来的。

五、最容易出问题的地方

流式识别输出仍是初稿。这个代码没有接入标点模型、说话人区分或会议摘要，不应把它的能力扩大成“自动生成完整会议纪要”。先把正确输入、连续上下文和结束语义跑通，才有可靠基础。