跑 MiniCPM-Robot 先算显存占用、再对推理脚本的输入输出

文章导读
跑 MiniCPM-Robot 之前,值得先做两件事:用显存监控命令实测一次前向的峰值占用,再把推理脚本能收哪些参数、返回什么字段列清楚,和自己下游程序对齐。这两步都不用改模型代码——前者靠一条采样命令加一段分段打点,后者靠 `--help` 和一次带打印的试跑。做完之后,“本地设备撑不撑得住”和“输出怎么接进程序”就有可复现的答案,而不是靠参数量估算。
📋 目录
  1. Ⅰ 用显存监控命令记录一次前向的峰值
  2. Ⅱ 把加载、前向、释放三段拆开测
  3. Ⅲ 读推理脚本的参数表与返回值结构
  4. Ⅳ 写最小调用示例把输出接到自己的程序
  5. Ⅴ 显存不够时的降级顺序
A A

跑 MiniCPM-Robot 之前,值得先做两件事:用显存监控命令实测一次前向的峰值占用,再把推理脚本能收哪些参数、返回什么字段列清楚,和自己下游程序对齐。这两步都不用改模型代码——前者靠一条采样命令加一段分段打点,后者靠 `--help` 和一次带打印的试跑。做完之后,“本地设备撑不撑得住”和“输出怎么接进程序”就有可复现的答案,而不是靠参数量估算。

建议把判断顺序固定成:先测峰值显存,再定输入输出契约。显存侧用 nvidia-smi 定时采样,配合 torch.cuda.max_memory_allocated 分段打点,拿到加载、前向、释放三段的实测值;契约侧用 `--help` 列出可配参数,跑一次最小样例确认返回字段名。设备够不够用取决于峰值还剩多少余量,不够时按分辨率、精度、批大小、生成长度的顺序逐项下调,每改一项都重测一次峰值。参数名以仓库实际配置为准。

用显存监控命令记录一次前向的峰值

凭参数量和精度估算显存容易偏,因为峰值常常出现在中间激活和临时张量上,那部分和输入尺寸强相关。更可靠的做法是让脚本真跑起来,同时用外部命令按固定间隔采一次显存。

开两个终端,一个采样一个推理:

# 终端 A:每 200ms 采一次,写入文件
nvidia-smi `--query-gpu`=timestamp,memory.used,memory.total \
  `--format`=csv,noheader,nounits -lms 200 > gpu_mem.csv

# 终端 B:跑一次推理
python infer.py `--image` ./demo.jpg `--prompt` "描述画面"

只想看一个数的话,也可以用一个 Python 采样线程,边跑边记:

import subprocess, time, threading

stop = False
def sample(path="mem.log", interval=0.2):
    with open(path, "w") as f:
        while not stop:
            out = subprocess.check_output([
                "nvidia-smi", "`--query-gpu`=memory.used",
                "`--format`=csv,noheader,nounits"
            ]).decode().strip().splitlines()[0]
            f.write(f"{time.time():.2f},{out}\n")
            time.sleep(interval)

t = threading.Thread(target=sample); t.start()
run_once()                 # 替换成你自己的一次前向
stop = True; t.join()

读峰值的方法:先在空闲状态跑一次 nvidia-smi 记下基线,再取采样文件里推理期间的 memory.used 最大值,两者之差就是这次前向大致吃掉的显存。200ms 间隔可能漏掉瞬时尖峰,想保守一点把间隔调到 50ms;但采样本身有开销,读到的值偏高一点是可以接受的。

把加载、前向、释放三段拆开测

整体峰值只回答“够不够”,说不清“被谁吃掉”。用 PyTorch 自带的显存统计在三个位置打点,就能分清加载、前向、释放各自占多少。

import gc, torch

def mark(tag):
    torch.cuda.synchronize()
    alloc = torch.cuda.memory_allocated() / 1024**2
    peak  = torch.cuda.max_memory_allocated() / 1024**2
    print(f"[{tag}] allocated={alloc:.0f}MB peak={peak:.0f}MB")
    torch.cuda.reset_peak_memory_stats()   # 每段重新计峰

mark("start")                          # 空载基线
model, processor = load_model()        # 换成仓库的加载函数
mark("after_load")                     # 加载段

inputs = build_inputs(...)
outs = model.generate(**inputs)
mark("after_forward")                  # 前向段

del outs, inputs, model
gc.collect(); torch.cuda.empty_cache()
mark("after_release")                  # 释放段

每段结束时的记录方式关键在 reset_peak_memory_stats():重置之后再打点,peak 就只反映这一段;allocated 是当前仍占用的量,用来判断释放是否真的生效。after_release 的 allocated 如果还接近 after_load,通常说明有引用没断(比如把输出张量挂到了全局变量上),这时重启进程比在代码里翻引用更省时间。

三段数出来后常见的现象是:加载占一块固定值,前向峰值随输入分辨率变化,能否压回基线取决于代码。三次测量之间建议重启进程,避免缓存分配器把上一轮的空闲块算进来,读数虚高。

读推理脚本的参数表与返回值结构

显存有数了,接下来固定输入输出契约。先从脚本自带的帮助信息入手,比自己猜参数名可靠:

跑 MiniCPM-Robot 先算显存占用、再对推理脚本的输入输出
python infer.py `--help`
python infer.py `--help` | head -60

下面这张表是这类推理脚本常见的参数形态,具体名称和默认值以 `--help` 输出和仓库配置文件为准,不要照抄:

  • 模型路径(常见形如 `--model-path`):指向本地权重目录。确认方式:路径写错时通常报找不到 config 或权重文件。
  • 精度(`--dtype` bfloat16 / float16 / int8):直接影响显存和速度。确认方式:改完重跑一次前向,看加载日志里的 dtype 是否真的变了,再对照峰值。
  • 设备(`--device` cuda:0 / cpu):决定是否走 GPU。确认方式:日志里参数所在 device 是否如预期。
  • 输入(`--image` / `--video` / `--prompt`):决定一次调用喂什么。确认方式:给一个不存在的路径,看报错是否落在读取输入这一步。
  • 尺寸与批(`--max-size` / `--batch-size`):最影响峰值的一对参数。确认方式:调小后按上一节重测前向峰值。
  • 生成长度(`--max-new-tokens`):影响耗时和后半段显存增长。确认方式:输出是否被截断在半句。
  • 采样参数(`--temperature` / `--top-p`):影响输出稳定性,调试阶段可以先调低温度减少随机性。
  • 输出(`--output-json` 之类):决定结果落到 stdout 还是文件,后面接程序时用得上。

返回值要先看清是“只返回一个字符串”,还是“返回一个 dict”。常见字段大致分三类:文本结果(可能叫 text / response / output)、结构化结果(面向控制或定位时,会有动作序列、坐标、位姿一类的字段)、附加信息(耗时、输入尺寸等 meta)。最直接的确认方式是把脚本输出重定向看一眼:python infer.py ... > out.json,再 head -c 800 out.json。字段名字照抄实际输出,不要按习惯写法假设。

写最小调用示例把输出接到自己的程序

最小调用有两条路:能 import 就直接调函数;跨进程或环境隔离就用 subprocess 调命令行再读 JSON。后者对排查更友好,因为脚本的报错会原样出现在 stderr 里,不会被自己的异常处理吞掉。

import json, subprocess

def run_infer(image_path, prompt, timeout=120):
    cmd = [
        "python", "infer.py",
        "`--image`", image_path,
        "`--prompt`", prompt,
        "`--output-json`", "-",      # 脚本不支持 stdout 就改成临时文件再读
    ]
    p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if p.returncode != 0:
        raise RuntimeError(f"infer 失败: {p.stderr[-800:]}")
    return json.loads(p.stdout)

res = run_infer("./demo.jpg", "描述画面并给出下一步动作")
print("keys:", list(res.keys()))
print("text:", res.get("text"))
print("action:", res.get("action"))

校验方式:先打印 list(res.keys()) 确认字段名和预期一致,再逐字段取值。字段缺失时的表现要提前想好——直接 res["action"] 会抛 KeyError,很快暴露;用 .get() 得到 None,容易被业务层当成“没有动作”静默放过。所以在适配层做一次显式断言更稳妥:

required = ["text"]
missing = [k for k in required if res.get(k) in (None, "")]
if missing:
    raise ValueError(f"返回缺少字段: {missing}, 实际字段: {list(res.keys())}")

如果脚本只吐纯文本、没有 JSON,就在适配层自己包一层 dict,把原始输出放进 raw 字段,下游只依赖这层包装。脚本以后改格式,需要改的只是这一处。

显存不够时的降级顺序

峰值超过可用显存时,按下面顺序逐项下调,每改一项都回到第一节重测峰值,确认这一项确实起了作用再动下一项。所有项都以仓库已有配置为准,不要自己加仓库没提供的开关。

  1. 输入分辨率或帧数:降到还能看清目标的下限。验证方式:用同一段输入重测峰值,同时看输出文本是否还能描述关键物体。
  2. 批大小:设为 1,把并发交给进程外排队。验证方式:峰值下降,且没有变成逐条循环导致整体耗时不可接受。
  3. 精度:仓库若支持 bfloat16 / float16,优先用它;int8 或量化权重要看仓库是否提供了对应的加载分支。验证方式:加载日志确认 dtype 真的变了,同时对比输出是否明显退化。
  4. 生成长度与采样:调低 max-new-tokens。验证方式:单次耗时变短,且输出没有被截断在半句。
  5. 释放与复用:单次推理跑完显式 empty_cache();批量场景改成一个进程循环处理,而不是并发起多个进程。验证方式:连续跑若干次后 allocated 是否回到接近基线。

把模型放 CPU 或做 offload、依赖 swap,都属于能让进程不崩的兜底,不是提速方案,延迟会明显变差。设备到底够不够,最终看的是峰值之外还剩多少余量,而这取决于同一张卡上还跑不跑别的进程,需要结合自己的环境确认。