想把 Matrix-Game3.5 这类流式交互模型接进自己的 demo，卡点通常不在“模型能不能跑”，而在喂帧和取帧的节奏对不上：键盘或鼠标动了，画面要等好几拍才跟上，或者缓冲区越积越深，最后变成一段延迟很大的录播。建议先把输入帧率、输出步长、缓冲长度这三个参数固定下来，再接界面；否则界面上的每一种卡顿都分不清是手感问题还是排队问题。

接入顺序建议是：输入侧按固定间隔抽帧，并给每帧打相对时间戳；输出侧用有界队列，长度先设 2，满了丢最旧帧；同一段输入连跑两次，确认延迟统计量可复现；再用“入队—出队—上屏”三个打点，把端到端延迟拆成生成段和消费段。参数最终以日志里能观测到的延迟和丢帧数为准。接口字段名、步长取值范围要以仓库实际暴露的为准，不确定处先用占位常量顶上。

先把输入侧抽帧与时间戳对齐

交互 demo 里最容易出问题的是“帧率依赖读盘速度或摄像头交付速度”，同一个输入文件在不同机器上喂进去的帧数不一样，后面就没法对比。做法是先抽出帧源，再按固定间隔往下游推。下面这段骨架可以同时用于视频文件和摄像头，替换点在 fps 和 cap.read() 之后的预处理。

import time, cv2

def frame_source(path=None, cam_index=0, fps=8):
    src = cam_index if path is None else path
    cap = cv2.VideoCapture(src)
    if not cap.isOpened():
        raise RuntimeError(f"open failed: {src}")
    interval = 1.0 / fps
    next_t = time.monotonic()
    idx = 0
    t0 = None
    while True:
        ok, frame = cap.read()
        if not ok:
            break
        now = time.monotonic()
        if now < next_t:
            time.sleep(next_t - now)
        t_read = time.monotonic()
        if t0 is None:
            t0 = t_read
        yield {
            "i": idx,
            "frame": frame,
            "t_src": cap.get(cv2.CAP_PROP_POS_MSEC) / 1000.0,
            "t_rel": round(t_read - t0, 4),
        }
        idx += 1
        next_t += interval
    cap.release()

时间戳字段建议至少保留三个：i 是帧序号，t_src 是源侧时间（视频文件有，摄像头可能为 0），t_rel 是相对首帧的读取时刻。对齐时以下游统一用 t_rel 为准，不要直接信任 t_src：部分容器的时间戳会跳变或有空洞，接进交互链路后会把延迟统计带歪。验证方式很直接，打印前十帧和后十帧的 t_rel 差值，应接近 1/fps；如果出现大片重复或跳跃，说明抽帧阻塞在预处理上，要先把预处理挪出抽帧循环。

给流式输出加一个有界缓冲队列

生成侧的出帧间隔通常不是等距的，显示侧又要求稳定刷新，中间必须有队列，但这个队列要有上限。交互场景通常用“丢旧留新”：画面新鲜度比完整性重要。队列长度先设 2，长度越大越能吸收抖动，但累积延迟也越大，这个关系可以直接在日志里看出来。

from collections import deque
import threading

class FrameQueue:
    def __init__(self, maxlen=2, policy="drop_oldest"):
        self.q = deque()
        self.maxlen = maxlen
        self.policy = policy
        self.lock = threading.Lock()
        self.dropped = 0
        self.peak = 0

    def put(self, item):
        with self.lock:
            if len(self.q) >= self.maxlen:
                self.dropped += 1
                if self.policy == "drop_oldest":
                    self.q.popleft()
                else:
                    return False
            self.q.append(item)
            self.peak = max(self.peak, len(self.q))
            return True

    def get(self):
        with self.lock:
            return self.q.popleft() if self.q else None

观察项建议固定三个：dropped 累计丢弃数、peak 峰值长度、相邻两次出队的间隔。如果 peak 长时间顶着 maxlen 且 dropped 快速增长，说明生成速度跟不上输入，此时调大队列只会把延迟藏起来，正确的动作是降输入帧率或改输出步长。反过来，如果 dropped 一直是 0 但画面仍不连续，问题多半在消费侧的上屏节奏，不在队列。

用同一段输入复跑两次确认可复现

调参之前先确认这份 demo 自己跟自己对得上。输入固定方式：用一段本地视频代替摄像头，代码里禁用随机拖拽和鼠标干预，喂帧路径走同一份 frame_source(path=..., fps=8)。复跑 2 次即可，有条件跑 3 次更稳。

对比方法是比统计量，不是比像素：取每帧的端到端延迟，看中位数、最大值和丢帧总数，再比两次运行的丢帧序号是否集中在前若干帧（通常是冷启动预热）。判定标准可以这样定：两次的丢帧总数接近、延迟中位数差距落在队列长度带来的抖动范围内，就认为可复现；如果某一次中位数明显抬高或丢帧大量出现，先查机器是否有其他任务抢占，别急着改模型参数。生成结果本身如果带采样随机性，逐帧画面不一致是正常的，这里只要求时间序列可比。

记录每帧的入队与出队时间戳

打点位置建议三处：生成结果进队列前、从队列取出后、实际送达显示回调时。用 JSON 行式日志，字段名固定下来，方便后面直接喂给脚本统计。

{"frame": 17, "enq": 1720000000.123, "deq": 1720000000.418,
 "shown": 1720000000.459, "gen_ms": 295, "disp_ms": 41, "q_len": 2, "fps_in": 8, "step": 2}

两段耗时的算法：生成段取 deq - enq，即这一帧从生成完成到被消费方拿走的等待时间；消费段取 shown - deq，即上屏本身的开销。端到端是 shown - 该帧的 t_rel 对应时刻。如果 gen_ms 一直很小而端到端很大，时间花在上屏和渲染；如果 gen_ms 随队列长度线性增长，就是典型的堆积。q_len 字段别省，它是判断上面两种情况的分水岭。

按实测结果回调输入帧率与输出步长

每次只改一个参数，复跑同一段输入，记录延迟中位数、丢帧数和画面连续性（是否出现明显跳帧或停顿）。可以参照下面这张对照表，档位数值按自己仓库支持的取值范围替换。

参数最终落点建议选在“丢帧数为 0 或极少、延迟中位数稳定、帧间间隔肉眼看不跳”的那一档，而不是延迟最低的一档。矩阵式交互里延迟和连续性经常互相牺牲，把对照组留在日志里，比事后凭手感回忆可靠。具体步长与帧率的合法取值、以及模型输入是否要求固定尺寸，需要结合仓库实际暴露的接口确认后再写死到配置里。