OpenHarmony WS63 设备把一段语音送到服务端时，上传的不是 WAV 文件，也不是可以直接交给识别模型的浮点数组，而是一串有先后关系的 Opus packet。服务端必须先保留同一个解码器状态，把 packet 还原为 PCM16，再做数值归一化，最后把样本交给 sherpa-onnx 的中文 Zipformer CTC 模型。任何一层把采样率、声道数、帧边界或生命周期理解错，最终都可能只得到空文本或解码错误。

小鸿 AI 当前服务端把这条链路集中在 asr_service.py，由 server.py 在收到 listen/stop 后调用。它是 OpenHarmony mini/LiteOS-M 设备语音链路的服务端一半：WS63 与 CI1302 负责采集、VAD 和 Opus 上行，Python 服务负责解码与中文识别。本篇只讨论当前真实源码已经存在的实现和边界，不把源码阅读或替身测试写成一次新的实机识别结论。

ASR 入口不是每来一帧就识别一次

服务端的 Session 会先累积本轮 audio_packets。设备发送的 binary frame 经过协议头剥离后进入这个列表，直到 VAD 段结束对应的 listen/stop 到达，才在线程中执行 recognize_session_audio()。因此当前实现的基本单位是一整段 utterance，而不是持续返回 partial result 的实时字幕。

入口先检查 hello 中声明的格式是否为 opus，然后把 packet 列表、采样率和声道数传给 ASR 封装：

def recognize_session_audio(session: Session) -> str:
    if session.audio_format.lower() != "opus":
        raise asr_service.AudioDecodeError(
            f"unsupported uplink audio format: {session.audio_format}"
        )
    return asr_service.transcribe_opus_packets(
        session.audio_packets,
        sample_rate=session.sample_rate,
        channels=session.channels,
    )

这段代码也说明了一条边界：名字叫 StreamingChineseASR，底层也使用 sherpa-onnx 的 OnlineRecognizer，但服务端没有边收包边输出识别文字。当前流程仍然是先收齐一轮 packet，再一次性送入 recognizer。把“使用在线识别器 API”直接写成“已经实现流式增量字幕”，会夸大源码能力。

libopus 必须沿着一段语音复用解码状态

Opus packet 不是彼此完全无关的文件片段。当前 OpusPacketDecoder 在构造时只创建一次 decoder，随后顺序解码整个 packet iterable，结束后再销毁。这样可以保留一段语音内部的编解码状态，也避免为每个 40 ms packet 反复创建 native 对象。

库加载没有把系统文件名写死为唯一值，而是先问 ctypes.util.find_library("opus")，再尝试 Linux 常见 soname：

@staticmethod
def _load_library() -> ctypes.CDLL:
    candidates = [ctypes.util.find_library("opus"), "libopus.so.0", "libopus.so"]
    for candidate in candidates:
        if not candidate:
            continue
        try:
            return ctypes.CDLL(candidate)
        except OSError:
            continue
    raise ASRUnavailableError("libopus is not installed")

requirements-asr.txt 只固定了 numpy 和 sherpa-onnx，没有把 libopus 当成 Python wheel 安装。项目 README 对应要求系统侧安装 libopus0。因此部署检查要同时覆盖 Python 虚拟环境与操作系统动态库；只看到 pip install 成功，不能证明 Opus 解码路径可用。

采样率和声道校验只是第一道门

构造器接受 Opus 规范支持的 8000、12000、16000、24000、48000 Hz，以及 1 或 2 声道；不在集合内会抛出 AudioDecodeError。但“构造器允许”不等于“项目已验证”。当前 WS63 hello、服务端 hello、模型配置与冒烟样本共同指向 16000 Hz、单声道、40 ms，这才是项目已经形成一致契约的组合。

服务端 hello 解析会采用设备传来的 sample_rate 和 channels，而 sherpa recognizer 创建时明确配置为 16000 Hz。双声道 PCM 也没有额外的下混处理。因此对当前项目最准确的表述是：libopus 封装具有更宽的参数检查范围，端到端已设计并使用的是 16 kHz 单声道；其他组合不能仅凭这两个集合就宣称已经支持。

每次调用 opus_decode() 时，缓冲区按 Opus 允许的最长 120 ms 准备，而实际设备 packet 是 40 ms。返回值是“每声道 sample 数”，所以复制字节数还要乘声道数和每个 int16 的 2 字节：

for packet in packets:
    if not packet:
        continue
    encoded = (ctypes.c_ubyte * len(packet)).from_buffer_copy(packet)
    sample_count = self._lib.opus_decode(
        self._decoder,
        encoded,
        len(packet),
        pcm,
        max_samples,
        0,
    )
    if sample_count < 0:
        raise AudioDecodeError(self._error_text(sample_count))
    decoded.extend(ctypes.string_at(pcm, sample_count * self.channels * 2))

空 packet 会被跳过；如果整轮没有产生 PCM，则抛出 no decodable Opus audio。当前实现没有用空 packet 主动触发 packet-loss concealment，也没有在服务端重排 packet。WebSocket 本身提供有序传输，但网络断开或客户端漏发的处理仍需要通过会话日志和端到端测试确认。

PCM16 进入模型前要转换到浮点幅值

libopus 通过 ctypes.c_int16 输出主机原生字节序的 PCM16；当前部署平台通常是小端，但源码没有显式把字节序固定为 little-endian。transcribe_pcm16() 用 NumPy 的原生 int16 从同一 buffer 建立视图，转换为 float32 后除以 32768，使样本进入大致 [-1, 1) 的范围。这里没有另外做降噪、自动增益、重采样或声道混合；输入质量仍取决于设备侧采集与 CI1302 输出。

samples = np.frombuffer(pcm16, dtype=np.int16).astype(np.float32)
samples /= 32768.0
recognizer = self._load()
with self._decode_lock:
    stream = recognizer.create_stream()
    stream.accept_waveform(sample_rate, samples)

np.frombuffer() 不会凭空修复奇数长度或错误字节序，幸运的是 PCM 来自 libopus 自己填充的 c_int16 数组，字节数由 sample_count * channels * 2 计算。真正值得监控的是 PCM 时长、峰值、静音比例与识别耗时；只有最终文本长度的日志，仍不足以区分“用户没说话”“音频幅值太低”和“模型解码失败”。

sherpa-onnx 模型采用懒加载而不是启动即加载

StreamingChineseASR 初始化时不立即导入 sherpa-onnx，也不立即创建 recognizer。第一次真实识别才进入 _load()：先检查 ASR 开关和模型文件，再导入模块并构造 OnlineRecognizer。这样服务进程可以先启动 HTTP 与 WebSocket，但也意味着健康检查看到模型文件存在，不等于第一次模型加载一定成功。

当前模型配置来自实际代码：

self._recognizer = sherpa_onnx.OnlineRecognizer.from_zipformer2_ctc(
    model=str(ASR_MODEL_FILE),
    tokens=str(ASR_TOKENS_FILE),
    num_threads=ASR_NUM_THREADS,
    sample_rate=16000,
    feature_dim=80,
    enable_endpoint_detection=False,
    decoding_method="greedy_search",
    provider="cpu",
)

模型目录默认指向 sherpa-onnx-streaming-zipformer-small-ctc-zh-int8-2025-04-01，必须至少包含 model.int8.onnx 与 tokens.txt。ASR_NUM_THREADS 默认 2，且被限制为不小于 1。当前 provider 是 CPU，解码方法是 greedy search，endpoint detection 被关闭；分段终点由设备 VAD 与 listen/stop 决定，而不是识别器在服务端自行切句。

补半秒静音是当前 utterance 收尾策略

把整段波形送入 stream 后，代码还追加 0.5 秒零值样本，然后调用 input_finished()，循环 decode_stream() 直到 recognizer 不再 ready，最后读取结果。这个处理帮助 CTC 解码器消化尾部，但它不是设备真实录到的 0.5 秒环境声。

stream.accept_waveform(
    sample_rate,
    np.zeros(int(sample_rate * 0.5), dtype=np.float32),
)
stream.input_finished()
while recognizer.is_ready(stream):
    recognizer.decode_stream(stream)
return recognizer.get_result(stream).strip()

因此时延分析要区分两部分：用户真实说话的音频时长，以及识别收尾时额外喂入的零样本。零样本是计算输入，不会让服务端真的等待 500 ms，但会增加一定特征与解码工作。若未来改成真正的在线 partial result，endpoint、尾静音和 VAD 的职责需要重新划分，不能直接在现有循环旁边多加一个回调就算完成。

两把锁分别保护加载和解码

服务端使用全局 _RECOGNIZER 实例。_load_lock 防止两个并发请求同时创建模型；_decode_lock 则让同一个 recognizer 的 create/accept/decode/get_result 区域串行执行。server.py 虽然用 asyncio.to_thread() 避免 ASR 阻塞事件循环，但多个设备同时结束说话时，最终仍会在 decode lock 前排队。

这不是代码错误，而是明确的容量选择：单模型实例减少内存与重复加载成本，串行区段换取线程安全。要评估是否需要 worker pool 或多实例，必须记录队列等待时间、纯解码时间、峰值并发和模型内存，而不能只看单次本地识别有多快。

错误分类决定设备最终听到什么

ASR 封装区分 ASRUnavailableError 与 AudioDecodeError。前者覆盖 ASR 被关闭、模型文件缺失、sherpa-onnx 或 libopus 不可用；后者覆盖非法参数、decoder 创建失败、packet 解码失败与无有效 PCM。server.py 对这两类错误返回不同中文提示，并为其他异常保留第三条兜底。

识别成功但结果为空也不是异常，它会返回“没有听清问题”。识别成功且有文本，才调用 answer_question()。无 binary frame 的 stop 则走唤醒文本或预置问题分支。这些分支解释了为什么设备能正常播放一句回答，并不必然证明那句话来自真实语音识别。

封装的最后两步非常短，却明确了组件边界：

packet_list = list(packets)
with OpusPacketDecoder(sample_rate=sample_rate, channels=channels) as decoder:
    pcm16 = decoder.decode(packet_list)
return _RECOGNIZER.transcribe_pcm16(pcm16, sample_rate=sample_rate)

先物化 packet iterable 可以保证同一轮只消费一次，context manager 保证正常和异常路径都销毁 decoder。若 native decoder 在构造中途失败，_decoder 尚不存在时，close() 也通过 getattr() 安全处理。

项目里有三种测试，证明范围并不相同

protocol_smoke.py 用 fake recognizer、fake LLM 和 fake TTS 验证 hello、packet 收集、stop 顺序与 ASR→LLM→TTS 编排。本轮重新运行通过，输出包含 asr_to_llm=ok 和 4 个下行 Opus 包。但 recognizer 被替换，所以这不是 libopus 或中文模型的真实测试。

asr_model_smoke.py 会读取模型目录中的 WAV，用系统 libopus 编成 40 ms packet，再调用 transcribe_opus_packets()。它覆盖“WAV→Opus→PCM→模型→文本”，需要 Linux 动态库、NumPy、sherpa-onnx 和模型文件齐全。本轮没有在当前工作机执行该模型冒烟。

live_asr_roundtrip.py 则会连接真实 WebSocket，发送项目保存的 packet 文件，等待 TTS start、sentence_start、binary audio 和 stop。它还会经过 LLM 与 TTS 外部依赖。本轮没有连接生产服务，也没有据此生成新的线上通过记录。

health 的 model_ready 只检查文件存在

当前 status() 返回 enabled、固定 backend 名称，以及两个文件是否存在：

def status() -> dict:
    return {
        "enabled": ASR_ENABLED,
        "backend": "sherpa-onnx-zipformer-small-ctc",
        "model_ready": ASR_MODEL_FILE.is_file() and ASR_TOKENS_FILE.is_file(),
    }

它没有主动加载 ONNX、跑一段音频或验证 libopus。因此 enabled=true 且 model_ready=true 只适合当作配置与文件探针，不能替代一次真实识别。更严格的就绪探针可以在部署阶段运行模型 smoke，把结果写进发布证据；常规 health 则保持轻量，避免每次探活都消耗模型推理资源。

本轮能够确认什么，仍然不能确认什么

本轮逐行核对了 asr_service.py、server.py、asr_model_smoke.py、live_asr_roundtrip.py、protocol_smoke.py、依赖文件和 README，并用逐文件 SHA-256 固定证据版本。确定性协议冒烟重新通过，能确认 stop 后调用链、错误分支接口和下行帧编排没有在当前源码中断裂。

本轮没有在 Linux 服务器加载实际 Zipformer INT8 模型，没有用真实普通话录音测字错率，也没有重新连接 WS63 实机验证噪声、远场、口音、丢包和并发。文章中的六张图应是依据这些源码绘制的数据流与边界图，不是实机截图。后续若要把“模型可用”提升为“设备体验可用”，至少还需要固定语料集、识别准确率、首字/整句时延、并发排队时间，以及一条从 CI1302 到设备播报的可回放端到端记录。