MOSS-VL-Realtime 想跑“实时”,第一件事不是改帧率参数,而是把一次问答拆成抽帧、预处理、推理、解码四段,分别量出毫秒数。单帧总耗时决定了理论帧率上限,四段的分布决定了卡顿应该往哪里改。建议先在本机跑 20 帧取中位数,拿到一张分段耗时表,再用 1 除以单帧总耗时反推可承受的抽帧频率,最后留出余量定帧率并连续复测十分钟。
适用场景:用 MOSS-VL-Realtime 做摄像头或屏幕流上的实时画面问答,机器型号和输入分辨率尚未确定。操作动作:先用 time.perf_counter 给四段打点,连跑 20 帧取中位数;固定 prompt,只梯度改变输入尺寸找耗时拐点;按“1÷单帧总耗时”算出理论帧率并打折作为实际配置。验证方式:看运行日志里各段耗时与丢弃计数,连续复测十分钟观察抖动与内存。风险边界:单帧耗时随输入尺寸、量化方式、显存带宽变化明显,本机测出的数值不能直接搬到别的机器上。
把一次问答拆成抽帧、预处理、推理、解码四段分别打点计时
打点的意义在于把“卡”这个模糊感受换成可比的数字。抽帧是从视频源或摄像头缓冲区取一帧,预处理是缩放、归一化、转张量,推理是模型前向,解码是把输出 token 还原成文本。下面这段骨架与具体 SDK 无关,把 grab_frame、preprocess、model.generate、decode 替换成本地实际调用即可,放在单独脚本里执行,不放进业务主循环。
import time, statistics
def timed_frame(source, prompt, model):
t0 = time.perf_counter()
img = grab_frame(source) # 抽帧:同步模式下会阻塞,异步取最新帧时接近 0
t1 = time.perf_counter()
inputs = preprocess(img) # 预处理:缩放 / 归一化 / 转张量
t2 = time.perf_counter()
out = model.generate(inputs, prompt) # 推理
t3 = time.perf_counter()
text = decode(out) # 解码成文本
t4 = time.perf_counter()
return {
"grab_ms": (t1 - t0) * 1000,
"pre_ms": (t2 - t1) * 1000,
"infer_ms": (t3 - t2) * 1000,
"decode_ms": (t4 - t3) * 1000,
"total_ms": (t4 - t0) * 1000,
}
rows = [timed_frame(src, PROMPT, model) for _ in range(20)]
for k in rows[0]:
print(k, round(statistics.median(r[k] for r in rows), 2))
字段说明:grab_ms 反映取帧是否阻塞,用异步缓冲时这一项很小,不能据此认为整条链路很快;pre_ms 与输入尺寸强相关;infer_ms 通常是最大项;decode_ms 取决于输出长度上限,把 max_new_tokens 调小会明显改变这一项,测试时要固定住。取中位数而不是平均值,是因为首帧会带模型加载、显存分配、算子编译等一次性开销,平均值会被首帧拉偏;建议把第 1 帧单独打印,剩余 19 帧不参与判断也可以,或者跑满 20 帧后中位数与最大值一起看。
固定问题文本长度,只改输入尺寸观察单帧耗时变化
输入尺寸是单帧耗时最敏感的旋钮。测试时把 prompt 写成同一句话,长度和内容都不改,只改送进模型的画面分辨率,从低到高逐档跑,每档仍按 20 帧取中位数。档位设置以本地模型的图像配置支持为准,可以先试短边 336、448、560、672 这类间距,遇到模型或显存不接受的值就跳过,不要硬改内部张量形状。
记录表建议至少包含这些字段,方便横向比较:
- 输入尺寸(宽×高或短边)、缩放方式
- pre_ms 中位数 / 最大值
- infer_ms 中位数 / 最大值
- 单帧 total_ms 中位数
- 显存占用峰值(nvidia-smi 或框架自带统计)
- 同一问题的回答是否仍然正确或明显退化
判读方式:如果 pre_ms 随尺寸近似线性增长而 infer_ms 增长更快,说明推理是瓶颈,继续加大分辨率收益很低;如果某一档之后 total_ms 出现跳变(例如跨过某个分块或显存阈值),那一档之前就是可用区间。分辨率降下来后回答质量下降属于业务取舍,需要人工看几帧典型画面再决定,不能只看耗时。
用实测耗时反推可承受的抽帧频率与队列长度
有了单帧总耗时的中位数 T(毫秒),理论可承受帧率约为 1000 ÷ T。这个数值是上限,实际配置要打折:建议留出约三分之一到一半的余量给突发负载、垃圾回收和其他进程争用,即配置帧率取 1000 ÷ T × 0.5 到 0.7 之间的值。如果 T 是 800ms,理论约 1.25 帧/秒,实际按 0.6 到 0.9 帧/秒设置更稳妥。
只定帧率不够,还要观察队列是否在堆积。需要盯的字段有三个:queue_len(当前待处理帧数)、frame_age_ms(被处理帧从采集到送入模型的间隔)、drop_count(累计丢弃帧数)。frame_age_ms 持续上升,说明处理速度跟不上采集速度,画面问答的回答会对应到已经过时的画面。
def on_new_frame(frame, now_ms):
if queue_len() >= MAX_QUEUE: # MAX_QUEUE 建议取 1 或 2
dropped = pop_oldest() # 丢弃最旧的帧,保留最新
drop_count += 1
push(frame, enqueue_ms=now_ms)
# 消费侧始终取最新一帧,而不是队首
frame = pop_latest()
age_ms = now_ms - frame.enqueue_ms
if age_ms > MAX_AGE_MS:
skip(frame) # 过旧帧直接不回答案,避免答非所问
丢弃旧帧是工程上的取舍,不是性能优化:它换来的只是“回答对应当前画面”,不会让模型变快。业务上如果不能接受漏帧,就得降分辨率或降问题复杂度,而不是把队列调长。
在运行日志里确认卡顿落在哪一段
线上或联调时无法随手打断点,日志字段要提前埋好。建议每处理一帧打一行,字段包括 frame_id、grab_ms、pre_ms、infer_ms、decode_ms、total_ms、queue_len、frame_age_ms、drop_count。
frame=1042 grab=3.1 pre=41.6 infer=612.4 decode=88.2 total=745.3 queue=0 age=12 drop=0
frame=1043 grab=2.8 pre=42.0 infer=744.9 decode=90.1 total=879.8 queue=2 age=930 drop=1
frame=1044 grab=2.9 pre=41.2 infer=690.3 decode=87.5 total=821.9 queue=1 age=410 drop=2
判读规则可以写得很直接:total_ms 变大而 infer_ms 基本不变,问题在预处理或解码;infer_ms 单独抖动而其他段平稳,通常是显存压力或同机其他进程抢占,可以对照显存占用与系统负载看;grab_ms 变大,说明采集侧或读取方式成了瓶颈,比如同步读摄像头;queue 与 age 一起上涨而 total_ms 没变,是配置的帧率本身超过了机器能力,需要下调抽帧频率,而不是继续加缓冲。
留出余量后定帧率并连续复测
按前面的换算把帧率定下来后,不要只看两分钟的平均值就收工。复测脚本至少保留这些可调参数:目标帧率、输入尺寸、prompt 文本、采样时长、日志输出路径、最大队列长度。把目标帧率设为最终配置值,输入尺寸和 prompt 与线上一致,连续跑十分钟,中间不要手动干预。
十分钟里重点看两件事:一是各段耗时的抖动,把 infer_ms 的最大值和中位数放在一起比,最大值明显偏离中位数说明存在突发阻塞;二是内存与显存是否单调增长,如果随时间持续爬升而任务量不变,可能存在缓存未释放或对象累积的问题,需要先排查再上线。
判定标准建议写成可比对的句子而不是感觉:在目标帧率下连续运行十分钟,drop_count 增长缓慢且 frame_age_ms 不持续上升;infer_ms 最大值不超过中位数的若干倍(倍数按本机历史数据定);进程内存与显存回落到稳定区间而非持续增长。三条同时满足,再把帧率写进配置;不满足就回到第二节降输入尺寸,或回到第三节下调帧率。