MHS 协议快速开始
动手练习请访问 7 天学会 MHS 协议 子站中的设备模拟器与闭环演示。
完成入门后请阅读 开发指南;需要开源对照见 GitHub 相关项目。
目标
本指南帮助你在无需真实硬件的前提下,建立 MHS 的心智模型:理解 read/write 原语、解析 Device Description、完成第一次闭环验证,并对比 MCP 开环与 MHS 闭环的执行轨迹差异。重点是建立 Physical AI 思维,而非一次部署生产集群。
前置知识
- 了解 MCP 的 tool/resource 基本概念(若不了解,可先阅读本站 MCP 协议文档)。
- 会使用 Python 或 TypeScript 调用 HTTP/CLI(本指南示例以 Python 伪代码为主,概念与语言无关)。
- 具备基础物理量概念:功率(W)、温度(°C)、 位置(mm)、转速(RPM)等。
环境准备
MHS 参考实现与工具链随生态演进;学习阶段推荐:
| 组件 | 用途 | 说明 |
|---|---|---|
| MHS CLI | 对设备 read/write | 类似 curl,面向物理设备 |
| 设备模拟器 | 无硬件练习 | 子站 LAB.01 提供虚拟激光器等 |
| Device Description 编辑器 | 编写/校验描述文件 | 子站 LAB.02 |
| MCP Python SDK | 桥接数字 Agent | 可选,用于 MCP↔MHS 联调 |
本地 Python 环境(3.11+):
python -m venv .venv && source .venv/bin/activate
pip install mcp # MCP SDK,用于理解桥接层
# MHS CLI 安装方式以官方发布为准;学习阶段可用子站在线模拟器
第一步:理解 Device Description
在连接任何设备前,先读它的「说明书」——机器可读的 Device Description。示例(虚拟激光器):
device:
vendor: VirtuLab
model: VL-800
firmware: "2.1"
serial: SIM-LASER-001
capabilities:
power_w:
access: read_write
range: [0.0, 8.0]
unit: W
resolution: 0.01
verify: read_after_write
temperature_c:
access: read_only
range: [15.0, 45.0]
unit: "°C"
safety:
hard_limits:
power_w: 5.0
soft_limits:
power_w: [0.0, 4.5]
interlocks:
- condition: "power_w > 0"
requires:
cooling_lpm: { min: 2.0 }
metadata:
calibration_date: "2026-01-15"
environment: { max_humidity_pct: 80 }
关注四类信息:
- capabilities:Agent 能 read/write 什么;
verify: read_after_write表示 write 后必须 read 验收。 - safety:硬边界 5.0 W;软边界建议不超过 4.5 W;互锁要求冷却流量。
- access 模式:read_only / read_write / write_only。
- 物理 metadata:校准日期影响置信度权重。
第二步:第一次 read
state = mhs.read("laser-001")
print(state.power_w)
# StateValue(value=0.0, unit='W', confidence=0.99, ts='2026-03-05T10:00:00Z')
print(state.temperature_c)
# StateValue(value=22.3, unit='°C', confidence=0.95, ts='...')
read 返回的是状态向量,不是字符串 "temperature is fine"。编排器应检查:
ts是否足够新(例如 < 500 ms)confidence是否高于任务阈值(例如 > 0.9)- 单位是否与 Description 一致
第三步:第一次 write + 闭环验证
TARGET = 2.5
TOL = 0.05
MAX_RETRY = 3
for attempt in range(MAX_RETRY):
mhs.write("laser-001", {"power_w": TARGET})
state = mhs.read("laser-001")
actual = state.power_w.value
if abs(actual - TARGET) <= TOL:
print(f"闭环成功: {actual} W")
break
correction = TARGET + (TARGET - actual) # 简单 PI 修正示例
print(f"尝试 {attempt+1}: 实测 {actual} W,修正为 {correction}")
else:
mhs.safe_state("laser-001") # 进入安全态:功率归零等
raise RuntimeError("无法在容差内达到目标功率")
对比 MCP 开环:
await mcp.call("laser.set_power", {"watts": 2.5}) # 收到 OK 即继续 — 危险
第四步:边界与互锁体验
故意触发安全机制以理解分层防御:
# 应被 Driver 拒绝:超过硬边界
try:
mhs.write("laser-001", {"power_w": 6.0})
except SafetyViolation as e:
print(e) # hard_limit power_w <= 5.0
# 互锁:未开冷却时禁止出光
try:
mhs.write("laser-001", {"power_w": 1.0, "cooling_lpm": 0.5})
except InterlockError as e:
print(e) # cooling_lpm >= 2.0 required
所有越界尝试应出现在审计日志中,供事后复盘与合规审查。
第五步:设备发现
设备接入 MHS 总线后,Agent 通过发现 API 列出设备并拉取 Description,无需硬编码型号:
devices = mhs.discover()
for d in devices:
desc = mhs.get_description(d.id)
print(d.id, desc.device.model, list(desc.capabilities.keys()))
这是「AI 的 USB-C」体验的核心:换设备不改 Agent 代码,只改 Description 与 Driver。
第六步:MCP 桥接一瞥
MHS Driver 可向 MCP 暴露 tools:
{
"name": "laser.set_power",
"description": "Set laser power with MHS closed-loop verify",
"inputSchema": {
"type": "object",
"properties": {
"power_w": { "type": "number", "minimum": 0, "maximum": 5.0 }
}
},
"x-mhs": {
"device_id": "laser-001",
"verify": "read_after_write",
"tolerance": { "power_w": 0.05 }
}
}
MCP Client 调用 tool 时,桥接层自动执行 MHS write+read 验证,把物理闭环包装成 Agent 熟悉的 tool 接口。
第七步:多设备最小编排
任务:位移台移动到 (10, 10) mm,相机拍照,激光器设 2.0 W,再拍照。
plan = [
("stage-001", {"x_mm": 10, "y_mm": 10}, lambda s: s.x_mm.value == 10),
("camera-001", {"capture": True}, lambda s: s.last_frame_id.value > 0),
("laser-001", {"power_w": 2.0}, lambda s: abs(s.power_w.value - 2.0) < 0.05),
("camera-001", {"capture": True}, lambda s: s.last_frame_id.value > prev_frame),
]
for device_id, cmd, accept in plan:
mhs.write(device_id, cmd)
state = mhs.read(device_id)
if not accept(state):
action = policy.decide(device_id, state) # retry | skip | abort | human
handle(action)
每一步必须验证再进入下一步——这是 Physical AI 与数字自动化的分水岭。
CLI 快速参考
学习阶段可用 MHS CLI 直接操作模拟设备(命令名为概念示例):
# 列出设备
mhs discover
# 读取状态
mhs read laser-001 --format json
# 写入并自动验证
mhs write laser-001 --set power_w=2.5 --verify --tolerance 0.05
# 查看设备描述
mhs describe laser-001
# 查看审计日志
mhs audit --device laser-001 --since 1h
CLI 与代码 API 语义一致,便于脚本化调试与 CI 中的 Driver 冒烟测试。
虚拟设备实验:VirtuLab VL-800
子站 LAB.01 提供 VirtuLab VL-800 虚拟激光器,支持:
| 实验 | 操作 | 观察点 |
|---|---|---|
| 基础闭环 | 设 2.5 W | write-read 曲线、修正次数 |
| 硬边界 | 设 6.0 W | Driver 拒绝与审计条目 |
| 互锁 | 低冷却 + 出光 | InterlockError 消息 |
| 漂移注入 | 开启传感器噪声 | confidence 下降与编排暂停 |
| 通信故障 | 模拟 Modbus 超时 | 重试与 safe_state |
建议按顺序完成,并在笔记中记录每次失败时的状态向量快照,而非仅记录错误码。
参数模式详解
Device Description 中的 access 与 verify 组合决定编排行为:
| access | verify | 编排语义 |
|---|---|---|
| read_only | — | 只读监控,禁止 write |
| read_write | read_after_write | 标准闭环 write |
| read_write | none | 仅用于无反馈执行器(需额外说明) |
| write_only | read_after_write | 写后通过关联传感器 read 验证 |
枚举约束示例:mode: { enum: [idle, run, maintenance] } — write 仅在 maintenance 下允许调某些参数。
状态新鲜度策略
def is_fresh(state_value, max_age_ms=500):
age = now_ms() - parse_ts(state_value.timestamp)
return age <= max_age_ms and state_value.confidence >= 0.9
state = mhs.read("stage-001")
if not is_fresh(state.x_mm):
mhs.write("stage-001", {"refresh": True}) # 或触发重新 poll
state = mhs.read("stage-001")
不同设备动力学不同:激光功率可能 10–100 Hz poll;位移台到位可 poll 每 100 ms 直到 stable=True 通道出现。
与实验室 SOP 的对齐
将现有标准操作程序(SOP)映射为编排 plan 节点时,每个 SOP 步骤应对应:
- write 命令与参数
- accept 谓词(何谓「本步成功」)
- 超时与重试上限
- 失败策略(retry / skip / abort / human)
- 是否需要人工 sign-off
这样 Agent 自动化不是「黑盒替代实验员」,而是可审计的 SOP 执行引擎。
错误分类入门
| 类型 | 示例 | 建议动作 |
|---|---|---|
| 通信 | Modbus 超时 | 重试 → 降级 → 人工 |
| 传感器 | 置信度骤降 | 暂停 write,请求校准 |
| 执行器 | 到位超时 | 重试有限次数 → safe state |
| 安全 | 硬边界 | 拒绝 + 审计,不自动重试 |
| 逻辑 | 互锁不满足 | 先满足依赖参数 |
CMU 药物实验案例中 6 种故障自动拦截,正是建立在这种分类与策略表之上。
常见问题(快速开始阶段)
Q:没有真实设备怎么办?
使用子站 MHS 设备模拟器(LAB.01),支持故障注入(通信中断、传感器漂移)。
Q:read 频率设多少?
取决于设备动力学:激光功率可能 10–100 Hz;位移台到位可能 poll 每 100 ms 直到 stable。
Q:MHS 与 PyVISA 的关系?
PyVISA 是仪器控制库;MHS Driver 可以内部调用 PyVISA,对外只暴露 read/write。
Q:能否跳过 verify 加速?
仅当 Description 允许且后果可逆;涉及样品安全或不可逆操作时禁止跳过。
动手检查清单
- 读懂一份完整 Device Description
- 完成 read 并解释 confidence 与 ts
- 完成 write + read 闭环达到容差
- 触发一次硬边界拒绝
- 理解互锁报错信息
- 对比 MCP 开环与 MHS 闭环伪代码
- 在模拟器观察故障注入后的编排行为
- 用 CLI 或伪代码完成多设备三步编排
下一步
附录:Nyquist 采样直觉
对于缓慢热惯性系统(分钟级),1 Hz read 足够;对于振动抑制(kHz),需要 Driver 事件驱动或硬件触发。Description 应声明 recommended_poll_hz 与 settling_time_ms,编排器据此调度,避免无意义的高频 read 占满总线。
闭环策略模板库
以下模板可直接嵌入编排器,覆盖常见 Physical AI 场景:
模板 A:设定值跟踪(Setpoint Tracking)
适用于激光功率、温度设定等连续量:
async def setpoint_loop(device_id, channel, target, tol, max_iter=5):
for i in range(max_iter):
await mhs.write(device_id, {channel: target})
await asyncio.sleep(settling_time(device_id, channel))
s = await mhs.read(device_id)
v = getattr(s, channel).value
if abs(v - target) <= tol:
return s
target = target + (target - v) * 0.5 # 阻尼修正
await mhs.safe_state(device_id)
raise SetpointError(device_id, channel, target)
模板 B:到位等待(Move and Wait)
适用于位移台、机械臂关节:
async def move_until_stable(device_id, cmd, stable_key="stable", timeout_s=60):
await mhs.write(device_id, cmd)
deadline = time.time() + timeout_s
while time.time() < deadline:
s = await mhs.read(device_id)
if getattr(s, stable_key).value is True:
return s
await asyncio.sleep(0.1)
raise MoveTimeout(device_id)
模板 C:互锁预检(Interlock Pre-check)
在 write 前显式 read 依赖量,便于调试:
async def write_with_interlock_check(device_id, cmd, desc):
state = await mhs.read(device_id)
violations = check_interlocks(desc.safety.interlocks, state, cmd)
if violations:
audit.log("interlock_would_fail", violations)
raise InterlockError(violations)
return await mhs.write(device_id, cmd)
Modbus 温控器最小示例(概念)
Day 2 挑战要求为 Modbus RTU 温控器写最小 Driver。寄存器映射示例:
| 寄存器 | 含义 | MHS 通道 |
|---|---|---|
| 0x0001 | 当前温度 × 0.1 °C | temperature_c (read) |
| 0x0002 | 设定温度 × 0.1 °C | setpoint_c (write) |
Driver 伪代码:
class ModbusTempDriver:
def read(self):
raw = self.modbus.read_register(0x0001)
return StateVector(temperature_c=StateValue(raw/10, "°C", 0.98, now()))
def write(self, cmd):
if "setpoint_c" in cmd:
if cmd["setpoint_c"] > self.desc.safety.hard_limits["setpoint_c"]:
raise SafetyViolation("setpoint exceeds hard limit")
self.modbus.write_register(0x0002, int(cmd["setpoint_c"] * 10))
此示例展示:协议细节留在 Driver,Agent 只见 temperature_c 与 setpoint_c。
资源与进度追踪
建议在学习快速开始的同时,在 7 天 MHS 子站 勾选 Day 1–2 晚间实战项。子站进度与本文档检查清单互补:文档重理解,子站重交互。