-
Notifications
You must be signed in to change notification settings - Fork 12
Inference Architecture
推理框架的目标是让策略类只保留“如何组织观测、如何解码动作”这两类代码，同时统一 后端选择、关节映射、历史帧、预热、性能监控和资源释放。配置使用 Python 类，不使用 YAML 描述模型或关节契约。
bxi_example_py_elf3/
framework/ # 可移植的运控框架
inference/
api.py # InferenceFrame / PolicyOutput
contract.py # PolicyJointContract / JointInputBinding
policy.py # JointPolicy 和 Builder/Decoder 组合方式
history.py # 无逐帧申请的历史环形缓冲
model.py # 可扩展 ModelArtifact / ModelSpec
runtime.py # 后端注册、选择和回退
monitor.py # 可选耗时统计
backends/ # ONNX Runtime / OpenVINO / RKNN
joints/ # Layout、显式默认值和命令解析器
platform/ # 机器人接入边界和关节 I/O 编码器
policies/ # ELF3 内置的具体策略
joints.py # 模型关节契约与策略内部顺序
normal.py
amp.py
beyondmimic.py
depth.py
framework/ 不导入任何 ELF3 策略或关节表；policies/ 可以依赖通用框架和项目的
policies/joints.py。完整机器人布局由平台首个合法的具名状态快照建立，不由策略关节表
决定。迁移到另一台机器人时，可以整体复用 framework/，再提供目标机器人的平台适配器、
策略、缺失关节默认目标和 Mod。
机器人状态布局 State Layout
-> 策略观测布局 Policy Observation Layout
-> 策略动作布局 Policy Action Layout
-> 完整机器人布局 Robot Layout
-> 可选的硬件固定布局 Hardware Fixed Layout
布局由不可变的 JointLayout 表示，关节名称是跨组件的语义主键。数字索引只在首次绑定
或消息顺序改变时编译，不能作为组件间的隐式协议。
- State Layout 可以比策略更多。例如 31 DoF 机器人状态可以包含 29 个策略关节和两个夹爪。
- Observation Layout 是策略真正读取的关节及顺序。
- Action Layout 是策略真正输出的关节及顺序，可以与 observation 不同。
- Robot Layout 是平台当前实际具有的完整关节集合，也是最终
MotorFrame的布局。 - Hardware Layout 只用于不携带关节名称、只能接收固定数组顺序的接口。
每周期的实际路径是：
完整 RobotObservation（N）
-> PolicyJointContract 按名称选择 observation（M）
-> 策略输出自己的 action MotorFrame（K）
-> JointCommandResolver：K + 平台显式默认目标 -> 完整 Robot Frame（N）
-> Transition：两端都先解析成 N，再插值
-> 具名消息直接发布，或映射到 Hardware Layout
M、K 和 N 可以不同，只要 observation 名称存在于 Robot Layout、action 名称是 Robot
Layout 的子集，且 N-K 的每个关节都有平台默认目标。
策略类中的 PolicyJointContract 是唯一权威：
from bxi_example_py_elf3.framework.inference import PolicyJointContract
from bxi_example_py_elf3.policies.joints import ELF3_POLICY_JOINTS
class WalkPolicy:
joint_contract = PolicyJointContract(
observation=ELF3_POLICY_JOINTS,
action=ELF3_POLICY_JOINTS,
)模型 metadata 只能校验该声明，不能在运行时反过来修改契约。声明和 metadata 不一致应在 加载时失败，不能猜测顺序。
平台每周期更新一个长期复用的 InferenceFrame：
InferenceFrame(
joints=joint_state_view, # JointStateView(layout, position, velocity)
quat_wxyz=quat_wxyz,
angular_velocity=omega,
command=command,
base_linear_velocity=None,
world_position=None,
depth=None,
depth_frame_id=None,
)所有内置策略使用同一调用形式：
policy.reset(frame)
output = policy.step(frame, dt=0.02, advance=True)
output.joints.layout
output.joints.position
output.joints.kp
output.joints.kd
output.estimated_velocity
output.completed调用者不再把裸 q, dq, quat, omega 按约定顺序传入。JointInputBinding 根据策略类声明
自动取得关节；来源布局与策略布局完全相同时直接使用原 view，不发生复制。布局不同或
来源包含额外关节时，映射写入策略长期持有的 float32 缓冲。
advance=False 用于预热和 Transition 双侧采样。策略不得在这种调用中推进时间、动作帧
或历史索引；HistoryBuffer.preview_append_into() 可以预览“如果追加当前帧”的连续输入，
但不修改环形缓冲状态。
下面是当前公共 API 的精简示例。关节契约、后端、稳定输入、历史和稳定输出均完整保留； 实际项目可继续加入相位、VAE 速度或不同 observation 字段。
import numpy as np
from bxi_example_py_elf3.framework.inference import (
HistoryBuffer,
InferenceFrame,
InferenceRuntime,
JointPolicy,
ModelSpec,
PolicyJointContract,
PolicyOutput,
default_runtime,
)
from bxi_example_py_elf3.policies.joints import ELF3_POLICY_JOINTS
class AmpPolicy(JointPolicy):
joint_contract = PolicyJointContract(
observation=ELF3_POLICY_JOINTS,
action=ELF3_POLICY_JOINTS,
)
def __init__(
self,
model_path: str,
default_position: np.ndarray,
action_scale: np.ndarray,
kp: np.ndarray,
kd: np.ndarray,
*,
runtime: InferenceRuntime | None = None,
backend: str = "auto",
) -> None:
super().__init__() # 创建具名输入绑定和稳定 PolicyOutput
self._runtime = runtime or default_runtime()
model = ModelSpec.portable_onnx(
model_path,
input_names=("obs",),
output_names=("actions",),
)
self._backend = self._runtime.open_backend(model, backend=backend)
count = self.joint_contract.action.dof_num
self._default = np.asarray(default_position, dtype=np.float32).reshape(count)
self._scale = np.asarray(action_scale, dtype=np.float32).reshape(count)
self._kp = np.asarray(kp, dtype=np.float32).reshape(count)
self._kd = np.asarray(kd, dtype=np.float32).reshape(count)
self._single = np.empty(96, dtype=np.float32)
self._obs = np.empty((1, 960), dtype=np.float32)
self._inputs = {"obs": self._obs} # 字典和数组身份始终稳定
self._history = HistoryBuffer(10, 96, dtype=np.float32)
self._action = np.zeros(count, dtype=np.float32)
self._scaled = np.empty(count, dtype=np.float32)
self._joint_delta = np.empty(count, dtype=np.float32)
self._gravity = np.empty(3, dtype=np.float32)
# 直接在 JointPolicy 的输出缓冲中生成位置，不增加一次复制。
self._target = self._target_buffer.position
np.copyto(self._target, self._default)
self.publish_output(self._target, self._kp, self._kd)
def reset(self, frame: InferenceFrame) -> None:
joints = self.bind_joints(frame) # 自动按类契约选取和排列
self._action.fill(0.0)
self._build_single(frame, joints.position, joints.velocity)
self._history.fill(self._single)
self._history.write_into(self._obs[0])
np.copyto(self._target, self._default)
def step(
self,
frame: InferenceFrame,
dt: float,
*,
advance: bool = True,
) -> PolicyOutput:
del dt
joints = self.bind_joints(frame)
self._build_single(frame, joints.position, joints.velocity)
if advance:
self._history.append(self._single)
self._history.write_into(self._obs[0])
else:
self._history.preview_append_into(self._single, self._obs[0])
raw = self._backend.run(self._inputs)["actions"]
np.copyto(self._action, np.asarray(raw).reshape(-1))
np.multiply(self._action, self._scale, out=self._scaled)
np.add(self._default, self._scaled, out=self._target)
return self.publish_output(self._target)
def _build_single(self, frame, q, dq) -> None:
if frame.command is None:
raise ValueError("AMP policy requires frame.command")
obs = self._single
obs[0:3] = frame.angular_velocity
self._project_gravity(frame.quat_wxyz, self._gravity)
obs[3:6] = self._gravity
obs[6:9] = frame.command
np.subtract(q, self._default, out=self._joint_delta)
obs[9:38] = self._joint_delta
obs[38:67] = dq
obs[67:96] = self._action
@staticmethod
def _project_gravity(quat, out) -> None:
w, x, y, z = quat
out[0] = 2.0 * (w * y - x * z)
out[1] = -2.0 * (w * x + y * z)
out[2] = 2.0 * (x * x + y * y) - 1.0
def close(self) -> None:
self._backend.close()这是“手写策略”的方式。希望进一步拆分单元测试时，也可以使用
InputBuilder + OutputDecoder + Policy 组合模板；两种方式共享完全相同的
InferenceFrame、PolicyOutput、关节契约和后端运行时。
31 关节状态进入 29 关节策略是安全的：策略 observation 是 source 的子集，绑定时按名称 抽取 29 个关节。反方向不能默认成立：29 个动作不足以形成 31 关节完整硬件命令。
policies/joints.py 中的 ELF3_POLICY_JOINTS 只固定旧模型的 29 维契约；完整 Robot
Layout 直接来自首个合法的 ActuatorStates.name。因此机器人增加两个关节不会改变旧模型
的输入宽度。新 31 关节模型只需声明自己的 31 关节 observation/action layout。
旧模型没有输出的新关节必须在平台类配置中给出固定安全目标：
from bxi_example_py_elf3.framework.joints import (
JointCommandDefaults,
JointDefault,
)
ROBOT_COMMAND_DEFAULTS = JointCommandDefaults(
{
"left_gripper_joint": JointDefault(position=0.0, kp=20.0, kd=0.5),
"right_gripper_joint": JointDefault(position=0.0, kp=20.0, kd=0.5),
}
)把它传给 RobotControlRuntime(command_defaults=ROBOT_COMMAND_DEFAULTS)。状态和策略产生的
MotorFrame 保留自己的自然布局；JointCommandResolver 在该布局第一次出现时完成以下
工作：
- 检查模型是否输出了机器人不存在的关节。
- 为 Robot Layout 中缺失的关节查找显式默认目标。
- 编译 source→robot 数字索引并缓存。
- 控制周期只向长期复用的完整输出缓冲写入，不查名称、不申请数组。
若缺少任一默认项，首次绑定立即报错。不能静默补零、沿用上帧或忽略额外关节，因为这些 行为代表不同的控制所有权和安全语义。
Transition 也先把源状态和目标状态各自解析到完整 Robot Layout，再执行插值。因此从 31 关节状态切到旧 29 关节状态时，新增关节会从源状态平滑过渡到显式默认目标，而不是 在切换开始时突跳。
携带名称的消息无需强制变成硬件固定顺序：
publish(names=frame.layout.names, position=frame.qpos)框架最终返回的 MotorFrame 已经是完整 Robot Layout。只有绕过状态机、直接处理
PolicyOutput 且下游协议确实支持局部命令时，才使用：
encoder = NamedJointCommandEncoder(supports_partial=True)不携带名称的硬件数组必须声明固定布局并编码：
encoder = FixedOrderJointCommandEncoder(hardware_layout)
hardware_target = encoder.encode(policy_output.joints)encoder 在 source layout 首次出现时编译索引，之后复用输出数组。
ModelSpec 保存按优先级排列的开放 ModelArtifact，没有固定的“onnx/openvino/rknn 字段”。
增加后端只需实现新的 artifact 和 factory，不需要扩大策略接口。
策略类传入 ONNX 路径时统一使用 ModelSpec.portable_onnx(...)，默认顺序是
rknn -> openvino -> onnxruntime：先找同名 .rknn，需要时在显式环境变量授权下
由 ONNX 转换，再回退到 OpenVINO 和 ONNX Runtime。
from bxi_example_py_elf3.framework.inference import (
ModelSpec, OnnxArtifact, OpenVinoArtifact, RknnArtifact,
)
model = ModelSpec(
artifacts=(
RknnArtifact(
"walk.rknn",
source_onnx="walk.onnx",
target="rk3588",
input_shapes=(("obs", (1, 960)),),
output_shapes=(("actions", (1, 29)),),
),
OpenVinoArtifact("walk.onnx", device="CPU"),
OnnxArtifact("walk.onnx"),
),
input_names=("obs",),
output_names=("actions",),
)运行时按 artifacts 顺序选择第一个可用后端。缺少库时会给出安装命令并继续回退；显式 指定不可用后端则抛错。平台上真正的优先顺序应以 benchmark 为准，不能只根据“已安装” 推断速度。
OpenVINO 的 Intel GPU 插件可能枚举到 NVIDIA GPU。框架会拒绝这种设备，避免
CL_INVALID_COMMAND_QUEUE 导致进程崩溃；NVIDIA 应使用 ONNX Runtime CUDA/TensorRT。
ONNX 到 RKNN 是构建步骤，不是每帧推理步骤。默认绝不转换；只有显式设置环境变量才在 加载时检查缓存并按需转换：
BXI_RKNN_CONVERT_ON_LOAD=rk3588 python3 your_program.py
BXI_RKNN_CONVERT_ON_LOAD='{
"target": "rk3588",
"do_quantization": true,
"dataset": "/data/calibration.txt",
"force_rebuild": false,
"outputs": ["actions"],
"config": {"optimization_level": 3}
}' python3 your_program.py转换缓存通过源模型和设置指纹判断是否有效。x86 是否能转换取决于 Rockchip 提供的
rknn-toolkit2 wheel；真正执行 .rknn 通常使用目标板上的 rknn-toolkit-lite2。
outputs 用于裁剪 RKNN 中实际需要的输出，也会进入缓存指纹。benchmark 默认只转换
actions；这既减少无用图节点，也避开 RKNN Toolkit 2.3.2 在合并不同 shape 的并行
Gather 输出时可能触发的优化器错误。需要多个运行时输出的策略必须显式列出全部名称。
框架层面支持 N 个输入和 N 个输出，但 RKNN Lite 的运行时接口只返回按顺序排列的 list，不携带输出名。因此策略类必须把真实需要的输入、输出名称声明清楚：
model = ModelSpec.portable_onnx(
"policy.onnx",
input_names=("obs", "depth", "command"),
output_names=("actions", "value", "latent"),
)portable_onnx(...) 会从 ONNX 读取模型描述，生成同名 .rknn artifact，并把
output_names 作为 RKNN 转换时的 outputs。运行时再按相同顺序把 RKNN 返回的 list
映射回字典。只要转换时的 outputs 顺序和策略声明一致，N 输入 / N 输出不会因为框架
结构本身出错。
部署时仍需注意这些边界：
- 只声明策略实际使用的输出。只需要
actions就不要保留调试输出，RKNN 会裁掉无用图， 转换更稳，推理也更轻。 - 多输出模型如果触发 RKNN Toolkit 优化器 bug，框架会回退 OpenVINO / ONNX Runtime； 这属于具体 Toolkit 和 ONNX 图结构兼容性问题，不是控制周期逻辑错误。
- RKNN 转换更适合静态 shape。动态 batch 或动态空间尺寸需要先导出成固定 shape，或通过 artifact / 环境变量明确转换参数。
- RKNN artifact 本身不保存 ONNX metadata。框架会从
source_onnx读取关节名、增益、 shape 等信息；部署.rknn时建议保留旁边的.onnx，至少保留同源 metadata 文件的能力。 - 如果某个输出只用于训练或离线分析，不要放进
output_names。运行时输出越少，后端选择 和跨平台 benchmark 越可控。
-
JointLayout和映射只在启动、首次绑定或消息名称顺序变化时创建。 - 相同 observation layout 使用零复制 fast path；不同布局使用
np.take(..., out=...)。 - 状态输出布局第一次出现时编译命令映射；后续只写预分配的完整 Robot Frame。
-
InferenceFrame、PolicyOutput、输入字典、输入数组、历史存储和目标数组长期复用。 - 不在控制周期创建 Session、
InferRequest、深拷贝或按名称执行list.index()。 - ONNX Runtime 使用 I/O Binding；OpenVINO 使用持久 InferRequest 和共享输入 Tensor。
- 监控默认关闭，关闭时不读取性能时钟。
# 所有发现到的模型 × 本机可用后端，子进程隔离 native driver 崩溃
python3 tools/benchmark/backend_benchmark.py --quick
# 策略级 Git 基线、分位延迟和 Python allocation
python3 tools/benchmark/inference_benchmark.py --baseline-ref HEAD
# 31→29 观测映射、29→31 默认补齐、31 关节重排和完整布局 fast path
python3 tools/benchmark/joint_mapping_benchmark.py跨平台对比时保持模型、warmup、iterations、电源模式和 CPU 负载一致。报告目录
tools/benchmark/results/ 被 Git 忽略，benchmark 程序本身纳入版本管理。