跑 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,通常说明有引用没断(比如把输出张量挂到了全局变量上),这时重启进程比在代码里翻引用更省时间。
三段数出来后常见的现象是:加载占一块固定值,前向峰值随输入分辨率变化,能否压回基线取决于代码。三次测量之间建议重启进程,避免缓存分配器把上一轮的空闲块算进来,读数虚高。
读推理脚本的参数表与返回值结构
显存有数了,接下来固定输入输出契约。先从脚本自带的帮助信息入手,比自己猜参数名可靠:
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,把并发交给进程外排队。验证方式:峰值下降,且没有变成逐条循环导致整体耗时不可接受。
- 精度:仓库若支持 bfloat16 / float16,优先用它;int8 或量化权重要看仓库是否提供了对应的加载分支。验证方式:加载日志确认 dtype 真的变了,同时对比输出是否明显退化。
- 生成长度与采样:调低 max-new-tokens。验证方式:单次耗时变短,且输出没有被截断在半句。
- 释放与复用:单次推理跑完显式
empty_cache();批量场景改成一个进程循环处理,而不是并发起多个进程。验证方式:连续跑若干次后 allocated 是否回到接近基线。
把模型放 CPU 或做 offload、依赖 swap,都属于能让进程不崩的兜底,不是提速方案,延迟会明显变差。设备到底够不够,最终看的是峰值之外还剩多少余量,而这取决于同一张卡上还跑不跑别的进程,需要结合自己的环境确认。