Fun-ASR-Realtime 接进实时字幕，先把音频分片和采样率对齐

把 Fun-ASR-Realtime 这类实时识别接进字幕链路，最容易卡住的不是识别端本身，而是你推上去的音频和它期望的输入没对齐：要么一直收不到返回，要么首字很晚、字幕整句才蹦出来。建议把顺序固定成三步——先确认采集端真实输出的采样率和声道数，再选一个固定分片时长，最后用最小推流脚本打印每一帧返回。三处对齐之后，延迟高低才是可测量、可调整的问题。

判断：如果接上后没有返回或延迟偏高，优先怀疑音频格式而不是网络。操作上先在采集端读出实际采样率、声道和编码，按识别端要求做转换或直接配置采集参数；再固定一个分片时长，常见可以先从 100ms 量级起步。验证方式是打印每帧返回文本和到达时刻，看首字返回是否稳定、整句结束是否及时。边界：识别端支持的采样率、位深和编码范围要以服务端返回和日志为准，不要只凭采集端配置推断。

确认采集端输出的采样率与声道数

录音软件界面上显示的采样率，不一定等于设备实际交给程序的值。先用命令把真实值读出来，再决定是否要转换。Linux 下可以用 arecord 打印硬件参数，或者录一小段再用 ffprobe 读文件：

arecord -D plughw:0,0 `--dump-hw-params` -d 1 /tmp/probe.wav
ffprobe -v error -show_entries stream=sample_rate,channels,codec_name -of default=nw=1 /tmp/probe.wav

设备名换成自己机器上的；macOS 可以在音频 MIDI 设置里看输入格式，或者用 sox `--i` 读一段已有录音。读完之后，在采集代码里把输出格式写死，不要依赖系统默认值：

SAMPLE_RATE = 16000   # 按识别端要求填，常见 16000
CHANNELS = 1          # 实时字幕一般推单声道
SAMPLE_FORMAT = 's16le'
BLOCK_MS = 100

改完配置，录 10 秒再跑一次 ffprobe，确认文件里的 sample_rate 和 channels 与采集配置一致。如果文件值和配置对不上，问题在采集库或驱动层，先别去改推流参数。

按固定时长切音频分片并写清选取依据

分片时长要固定，不要随回调抖动。取值依据通常有三条：识别端期望的单帧长度、字幕上屏的节奏、一次网络往返的耗时。可以先从 100ms 起步，再按 20ms 或 50ms 的步进调整，观察返回变化。

分片过大的可观察现象：首条识别文本回来得晚，字幕往往整句一起出现，实时感下降。分片过小的可观察现象：返回条数变多但每条只有一两个字，尾字容易被截断，同一个词可能重复出现；推得过于密集时，服务端可能合并帧或限流。

采样率：采集端输出与识别端要求一致，常见 16000，以服务端要求为准
声道：多声道先下混成单声道，或按服务端要求保留
编码与位深：s16le、f32le、pcm、opus 等按服务端要求选一种并固定
分片时长：一个固定值，按毫秒写在配置里，不在回调里动态改
帧字节数：按采样率 × 声道 × 位深 ÷ 8 × 时长(秒) 算出来，保证是整数

写最小推流脚本，打印每一帧返回

先跑通链路再谈优化。下面是一个通用 WebSocket 推流骨架，地址和参数帧字段名都写成占位符，按服务端实际要求替换；它只做一件事：逐帧发送，并把每条返回文本和到达时刻打印出来。

import asyncio, json, time, websockets

WS_URL = 'wss://<host>/<path>?<query>'
SAMPLE_RATE = 16000
CHANNELS = 1
CHUNK_MS = 100
FRAME_BYTES = int(SAMPLE_RATE * CHANNELS * 2 * CHUNK_MS / 1000)

async def main():
    async with websockets.connect(WS_URL) as ws:
        await ws.send(json.dumps({
            'type': 'start',
            'sample_rate': SAMPLE_RATE,
            'channels': CHANNELS,
            'format': 's16le',
        }))
        t0 = time.monotonic()
        with open('input.pcm', 'rb') as f:
            while True:
                frame = f.read(FRAME_BYTES)
                if not frame:
                    break
                await ws.send(frame)
                try:
                    msg = await asyncio.wait_for(ws.recv(), timeout=0.5)
                    print(round(time.monotonic() - t0, 3), msg)
                except asyncio.TimeoutError:
                    pass
        await ws.send(json.dumps({'type': 'stop'}))

asyncio.run(main())

运行方式：把 input.pcm 换成前面配置录好的单声道 PCM 文件，先跑完整段音频。判断链路是否通的依据是最少收到一条非空返回；如果一条都没有，先看首帧参数是否被服务端接受，再核对每帧字节数和采样率是否匹配。

用同一段音频对比不同分片时长的返回

把分片时长从猜测变成测量，关键是只改一个变量。准备同一段 10 到 20 秒、内容清晰的单声道音频，其余参数全部固定，只改 CHUNK_MS，每次重跑脚本。

首字返回时刻：记录从发送第一帧到第一条非空识别文本的时间差
整句结束时刻：收到服务端 final 标记或 stop 之后的最终结果时记时间差
返回条数：统计整段音频期间收到的返回条数，用来判断返回是否过碎

chunk_ms=80   first_text=1.24s  final=10.31s  returns=63
chunk_ms=100  first_text=1.10s  final=10.28s  returns=51
chunk_ms=150  first_text=1.35s  final=10.33s  returns=34
chunk_ms=200  first_text=1.62s  final=10.41s  returns=26

上面只是记录格式示意，数值要用自己的音频跑出来。对比时重点看首字返回时刻与整句结束时刻的差值，以及返回条数是否随分片变小而明显增多；选一个首字不太晚、返回又不过碎的固定值。

把采样率、声道、编码写进配置文件

把这些值从代码里挪到配置文件，改错时才能一眼定位。建议的配置项清单：

endpoint：WebSocket 地址与路径，含鉴权参数
sample_rate：采集与推流共用的采样率
channels：声道数
audio_format：s16le / f32le / pcm / opus 之一
chunk_ms：分片时长
frame_bytes：由前三项算出的帧长度，或在启动时校验
auth_token / timeout：鉴权与超时

写错时的典型特征：采样率写成 8000 而音频是 16000，服务端通常不报错，但返回为空或识别成乱码；声道写成 2 而实际是单声道，可能返回空，也可能出现串字和语速异常；编码类型填错，日志里会出现解码失败或 invalid audio format 一类错误；frame_bytes 不是整数帧，服务端可能报帧长不匹配，或者一直等更多数据不返回；漏发首帧参数，服务端按默认值解析，同样表现为没有结果。改完配置只跑一遍最小脚本，确认首帧参数应答和第一条非空返回都正常即可。