适用场景
这套方案适合已经有基础移动底盘或机械臂,希望把“听懂一句话、看着画面干活”接入控制流程的团队:输入端是一帧第一视角画面加一句任务指令,输出端是一段可执行的动作序列,中间由安全围栏拦截危险动作,由失败回退兜住异常。它解决的核心问题不是“模型能不能答对”,而是“模型输出的动作能不能被放心地执行”。
环境与前置条件
硬件:Lion-ER-35B-A3B 属于稀疏激活的 MoE 结构,35B 是总参数量,A3B 指推理时每 token 约激活 3B 参数。这个设计主要降低算力消耗,不降低显存占用——权重仍然要整体驻留显存。按经验估算:
| 精度 | 每参数字节 | 35B 权重大致占用 |
|---|---|---|
| BF16 / FP16 | 2 | 约 70 GB |
| FP8 / INT8 | 1 | 约 35 GB |
| INT4 量化 | 0.5 | 约 18 GB |
再叠加 AI 词典:KV Cache">KV Cache(随 --max-model-len 线性增长)和激活开销。因此单卡 24 GB 需要量化,单卡 80 GB 可跑 BF16 但上下文不宜开太大,双卡 80 GB 是较稳妥的起点。具体权重文件大小以下载后 du -sh 的实际结果为准。
软件:Linux(Ubuntu 系为主流)、可用的 NVIDIA 驱动、Python 3.10 及以上、足够的 CUDA 运行时(版本匹配关系以所用推理框架的官方文档为准)。推理框架推荐 vLLM 或 SGLang,两者都提供 OpenAI 兼容接口,便于复用现有客户端代码。
磁盘:模型权重目录预留 100 GB 以上,另外给日志和录制的画面数据留出空间。
其他:如果直接控制真机,务必先准备好急停按钮和人工接管开关,这是硬件层面的事,软件围栏只是第二道保险。
分步骤部署
步骤 1:确认 GPU 与驱动可用
```bash
nvidia-smi
```
预期能看到 GPU 型号、单卡显存、驱动版本,右上角有 CUDA Version 字段。若提示命令不存在,说明驱动未安装,先解决驱动再往下走。
步骤 2:建立隔离的 Python 环境
```bash
python3 -m venv ~/lion-er-env
source ~/lion-er-env/bin/activate
python -V
pip install -U pip setuptools wheel
```
这一步是为了避免和系统自带的 Python 包冲突。python -V 输出 3.10 及以上即可。
步骤 3:安装推理框架
```bash
pip install vllm
```
不锁定版本时默认装最新稳定版。若需固定版本,版本号以官方文档当前版本为准。安装完成后执行 python -c "import vllm; print(vllm.__version__)",能打印出版本号说明装好了。
步骤 4:获取模型权重
权重来源以官方发布页面为准,常见形态是 Hugging Face 或 ModelScope 上的模型仓库。以 Hugging Face 为例:
```bash
pip install -U huggingface_hub
huggingface-cli download <模型仓库名> --local-dir ~/models/lion-er-35b-a3b
```
把 <模型仓库名> 换成官方页面给出的仓库标识。下载完成后确认目录里至少包含配置文件、分词器文件和权重分片:
```bash
ls ~/models/lion-er-35b-a3b
du -sh ~/models/lion-er-35b-a3b
```
步骤 5:做一次离线自检
先用离线推理确认权重能加载、能出结果,避免直接调服务时把问题混在一起:
```python
from vllm import LLM, SamplingParams
llm = LLM(model="/home/<用户名>/models/lion-er-35b-a3b",
tensor_parallel_size=2, # 等于 GPU 卡数
max_model_len=8192,
trust_remote_code=True) # 仅在信任权重来源时开启
out = llm.generate(["用一句话说明什么是抓取动作。"],
SamplingParams(temperature=0.2, max_tokens=128))
print(out[0].outputs[0].text)
```
能打印出一段通顺文本,说明权重、分词器、并行切分都正常。
步骤 6:启动 OpenAI 兼容服务
```bash
vllm serve ~/models/lion-er-35b-a3b \
--served-model-name lion-er-35b-a3b \
--tensor-parallel-size 2 \
--max-model-len 8192 \
--gpu-memory-utilization 0.90 \
--trust-remote-code \
--port 8000
```
参数含义:--served-model-name 是客户端调用时填的模型名,随便起但要和请求里一致;--tensor-parallel-size 是卡数;--max-model-len 决定上下文上限,直接决定 KV Cache 占用;--gpu-memory-utilization 控制单卡显存使用比例,卡上还跑别的进程时要调低。端口 8000 只是示例,可自行更换。
成功标志是日志末尾出现类似 Uvicorn running on http://0.0.0.0:8000 的提示,并且不再有加载进度输出。
步骤 7:约定动作协议
模型输出必须是机器可校验的结构,不要让解析器直接执行自然语言。建议用 JSON 数组,一步一个动作:
```json
{
"task": "把桌上的蓝色杯子推到我右手边",
"steps": [
{"action": "look", "params": {"pan": 0.2, "tilt": -0.3}, "expect": "杯子出现在画面中央", "on_fail": "rescan"},
{"action": "move", "params": {"lin_x": 0.15, "ang_z": 0.0, "duration_s": 1.0}, "expect": "距离缩短到一臂范围", "on_fail": "stop"},
{"action": "move", "params": {"lin_x": 0.0, "ang_z": -0.4, "duration_s": 0.8}, "expect": "杯口位于夹爪正前方", "on_fail": "stop"}
]
}
```
把允许的动作写进提示词,并在服务端用结构化输出(如 guided_json、response_format 一类能力,具体参数名以框架文档为准)约束返回格式,能显著降低解析失败率。
步骤 8:实现安全围栏
围栏做三件事:白名单过滤、参数裁剪、任务级拒绝。
```python
guard.py
ACTION_SPEC = {
"move": {"lin_x": (-0.3, 0.3), "ang_z": (-0.6, 0.6), "duration_s": (0.1, 2.0)},
"look": {"pan": (-1.2, 1.2), "tilt": (-0.5, 0.9)},
"grasp": {"width_m": (0.0, 0.10)},
"stop": {},
}
MAX_STEPS = 12
MAX_TOTAL_SECONDS = 20.0
FORBIDDEN = ["丢到楼下", "砸", "剪断", "靠近人脸", "全速"]
def guard_command(text: str):
for kw in FORBIDDEN:
if kw in text:
return False, f"指令命中禁用词:{kw}"
return True, ""
def guard_plan(plan: dict):
steps = plan.get("steps", [])
if not steps:
return None, "空计划"
if len(steps) > MAX_STEPS:
return None, f"步骤数超限 {len(steps)}"
safe, total = [], 0.0
for i, s in enumerate(steps):
act = s.get("action")
if act not in ACTION_SPEC:
return None, f"第{i}步出现未知动作:{act}"
params = {}
for k, v in (s.get("params") or {}).items():
if k not in ACTION_SPEC[act]:
return None, f"第{i}步 {act} 含未定义参数:{k}"
lo, hi = ACTION_SPEC[act][k]
params[k] = max(lo, min(hi, float(v))) # 越界即裁剪,不直接执行原值
d = params.get("duration_s", 0.0)
total += d
if total > MAX_TOTAL_SECONDS:
return None, "累计时长超限"
safe.append({"action": act, "params": params,
"expect": s.get("expect", ""),
"on_fail": s.get("on_fail", "stop")})
return safe, ""
```
注意裁剪和拒绝要分开:参数微超限裁剪掉更稳,未知动作、未知参数、禁用词则直接拒绝整份计划,宁可不出动作也不出危险动作。
步骤 9:闭环执行与失败回退
```python
loop.py
import json, time, requests
from guard import guard_command, guard_plan
BASE = "http://127.0.0.1:8000/v1"
HEADERS = {"Authorization": "Bearer EMPTY", "Content-Type": "application/json"}
def ask(prompt, image_b64=None, retries=2):
content = [{"type": "text", "text": prompt}]
if image_b64:
content.append({"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}})
body = {"model": "lion-er-35b-a3b",
"messages": [{"role": "user", "content": content}],
"temperature": 0.1, "max_tokens": 1024,
"response_format": {"type": "json_object"}}
last = None
for _ in range(retries + 1):
try:
r = requests.post(f"{BASE}/chat/completions", headers=HEADERS,
json=body, timeout=60)
r.raise_for_status()
return json.loads(r.json()["choices"][0]["message"]["content"])
except Exception as e:
last = e
time.sleep(1.0)
raise RuntimeError(f"模型调用失败:{last}")
def execute_step(step):
"""替换成你的真机 SDK 调用;这里返回 True/False 表示是否达成 expect"""
print(f"[执行] {step['action']} {step['params']}")
return True
def run_task(instruction, frame_b64, max_replan=2):
ok, msg = guard_command(instruction)
if not ok:
return {"status": "rejected", "reason": msg}
for attempt in range(max_replan + 1):
prompt = (f"任务:{instruction}\n"
f"这是第 {attempt + 1} 次尝试,请根据第一视角画面给出后续动作。\n"
f"只允许动作:move/look/grasp/stop。按指定 JSON 格式输出。")
try:
plan = ask(prompt, frame_b64)
except RuntimeError as e:
print(f"[回退L2] 模型不可用:{e}")
return {"status": "fallback_stop", "reason": str(e)}
steps, err = guard_plan(plan)
if steps is None:
print(f"[回退L1] 计划被围栏拒绝:{err}")
continue
for i, step in enumerate(steps):
if not execute_step(step):
print(f"[回退L1] 第{i}步未达成预期:{step['expect']}")
if step["on_fail"] == "stop":
execute_step({"action": "stop", "params": {}})
return {"status": "aborted", "at_step": i}
break # rescan 之类,跳出内层去重新规划
else:
return {"status": "ok", "steps": len(steps)}
execute_step({"action": "stop", "params": {}})
return {"status": "need_human"}
```
三级回退逻辑:L1 计划不合规或执行未达预期,重规划(有次数上限);L2 模型服务不可用,立即执行 stop 并返回;L3 重试耗尽,停车等待人工接管。不要在回退分支里再调用模型,否则故障时会陷入循环。
步骤 10:接进 ROS 2(可选)
把 execute_step 换成发布速度指令即可,以下为骨架,ROS 2 发行版差异以官方文档为准:
```python
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
class Executor(Node):
def __init__(self):
super().__init__("lion_er_executor")
self.pub = self.create_publisher(Twist, "/cmd_vel", 10)
def move(self, lin_x, ang_z, duration_s):
msg = Twist()
msg.linear.x = lin_x
msg.angular.z = ang_z
end = self.get_clock().now().nanoseconds / 1e9 + duration_s
while self.get_clock().now().nanoseconds / 1e9 < end:
self.pub.publish(msg)
rclpy.spin_once(self, timeout_sec=0.05)
self.pub.publish(Twist()) # 收敛到零速,别省这一步
```
验证部署是否成功
1. 服务存活与模型列表
```bash
curl -s http://127.0.0.1:8000/v1/models | python -m json.tool
```
预期返回 data 数组,其中 id 字段等于启动时的 --served-model-name。
2. 纯文本推理
```bash
curl -s http://127.0.0.1:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{"model":"lion-er-35b-a3b",
"messages":[{"role":"user","content":"用一句话说明什么是抓取动作。"}],
"max_tokens":128}'
```
预期返回含 choices[0].message.content 的 JSON,内容是与抓取相关的自然语言描述。
3. 多模态输入(若服务端已支持图像输入)
把一张第一视角画面转成 base64 后按步骤 9 的格式提交,预期模型在回答中能描述画面里的物体与相对方位。若返回内容明显与图像无关,说明图像没被正确解码,回到步骤 9 检查 data URL 的 MIME 前缀。
4. 围栏与回退联调(dry-run)
```bash
python -c "
from guard import guard_plan
plan = {'steps':[{'action':'move','params':{'lin_x':9.9,'ang_z':0,'duration_s':1}}]}
print(guard_plan(plan))
"
```
预期看到 lin_x 被裁剪到 0.3,而非原值 9.9。再把 action 改成 explode 重跑,预期返回 None 并给出“未知动作”提示。最后断开模型服务再跑一次 run_task,预期在数秒内返回 fallback_stop 且机器人保持静止——这一步必须过,否则不要上真机。
常见报错与解决
报错 1:torch.OutOfMemoryError: CUDA out of memory
原因:权重加 KV Cache 超出单卡显存,--max-model-len 开太大是最常见诱因。
解决:先降显存水位和上下文,再考虑加卡。
```bash
vllm serve ~/models/lion-er-35b-a3b \
--served-model-name lion-er-35b-a3b \
--tensor-parallel-size 2 \
--max-model-len 4096 \
--gpu-memory-utilization 0.85 \
--trust-remote-code --port 8000
```
报错 2:ValueError: The model requires trust_remote_code=True
原因:模型仓库带自定义建模代码,框架默认不执行外部代码。
解决:确认权重来源可信后加 --trust-remote-code;不放心的做法是通读仓库里的建模文件再启用。
```bash
vllm serve ~/models/lion-er-35b-a3b --trust-remote-code \
--served-model-name lion-er-35b-a3b --port 8000
```
报错 3:OSError: [Errno 98] Address already in use
原因:端口被上一个没退干净的进程占用。
解决:
```bash
lsof -i :8000
kill -9 <PID>
```
报错 4:客户端报 Connection refused 或 404 model not found
原因:服务端还没加载完就发请求;或请求里的 model 字段和 --served-model-name 不一致。
解决:等日志出现 Uvicorn running 后再调用;用 curl /v1/models 核对 id 字段,逐字对齐请求体里的 model 值。
报错 5:json.decoder.JSONDecodeError 频繁出现
原因:模型在 JSON 前后夹带解释性文字,或输出被 max_tokens 截断。
解决:开启结构化输出约束(guided_json、response_format 等,参数名以框架文档为准),把 max_tokens 提到 1024 以上,并在客户端保留解析失败重试,重试仍失败则走 L1 回退重规划。
报错 6:日志显示动作已下发,但机器人不动
原因:参数单位不一致(弧度与角度混用),或速度指令发完没有归零导致看门狗触发。
解决:在围栏里写死单位约定(角度统一用弧度),并确保每条 move 结束后补发一次零速指令。
后续维护
备份:需要备份四样东西——模型权重目录(或至少记录来源与版本标识)、系统提示词与动作 schema、围栏参数表、以及每次任务的输入画面与输出动作日志。前两样决定行为,后两样决定出问题后能不能复盘。权重目录体积大,可只记录来源信息,重下即可。
升级:换模型权重或换推理框架版本都属于高风险变更。做法是并行起一个新端口跑新版本,用同一批历史任务回放对比动作质量与围栏拦截率,确认无回退再切流量;旧版本至少保留一个可回滚周期。
日志与监控:至少盯四个指标——请求延迟(P50/P95)、结构化解析失败率、围栏拦截率、回退触发率。解析失败率突然上升通常意味着提示词被改动或模型换了;围栏拦截率上升可能是模型开始输出越界参数,也可能是场景变了;回退触发率上升则要优先查服务端健康度。建议给 fallback_stop 和 need_human 单独设告警,这两类事件直接对应现场停机。
日常还可以固定跑一套小规模回放集:十几个典型指令配对应画面,每次升级前后各跑一遍,看动作序列的一致性。这套东西不复杂,但能挡住大部分“升级完才发现不对劲”的情况。
