-
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/ # 与策略、ROS、硬件无关的具名关节类型
platform/ # 机器人接入边界和关节 I/O 编码器
policies/ # ELF3 内置的具体策略
normal.py
amp.py
beyondmimic.py
depth.py
joints.py # ELF3 关节布局
framework/ 不导入任何 ELF3 策略或关节表；policies/ 可以依赖通用框架和项目的
joints.py。迁移到另一台机器人时，可以整体复用 framework/，再提供目标机器人的关节
布局、平台适配器、策略和 Mod。
机器人状态布局 State Layout
-> 策略观测布局 Policy Observation Layout
-> 策略动作布局 Policy Action Layout
-> 运行时控制布局 Runtime Control Layout
-> 可选的硬件固定布局 Hardware Fixed Layout
布局由不可变的 JointLayout 表示，关节名称是跨组件的语义主键。数字索引只在首次绑定
或消息顺序改变时编译，不能作为组件间的隐式协议。
- State Layout 可以比策略更多。例如 31 DoF 机器人状态可以包含 29 个策略关节和两个夹爪。
- Observation Layout 是策略真正读取的关节及顺序。
- Action Layout 是策略真正输出的关节及顺序，可以与 observation 不同。
- Control Layout 是状态机和 Transition 发送完整命令时使用的布局。
- Hardware Layout 只用于不携带关节名称、只能接收固定数组顺序的接口。
策略类中的 PolicyJointContract 是唯一权威：
from bxi_example_py_elf3.framework.inference import PolicyJointContract
from bxi_example_py_elf3.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.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 关节完整硬件命令。
ELF3 为此分别声明 ELF3_POLICY_JOINTS 和 ELF3_CONTROL_JOINTS。目前控制器只负责与
策略相同的 29 个关节，所以二者指向同一个不可变布局，保持零复制 fast path；将来控制器
扩展到 31 个关节时，只扩展 ELF3_CONTROL_JOINTS，模型类仍引用固定 29 维的
ELF3_POLICY_JOINTS。
框架提供两条明确路径：
from bxi_example_py_elf3.framework.joints import (
ExactJointTargetAssembler,
PartialJointTargetAssembler,
)
# 只允许同一组关节，顺序可以不同。
complete = ExactJointTargetAssembler(control_layout).compose(policy_output.joints)
# 允许子集，但剩余关节必须来自调用者明确提供的完整 fallback。
complete = PartialJointTargetAssembler(control_layout).compose(
policy_output.joints,
fallback=hold_or_other_controller_target,
)禁止默认补零、默认沿用上帧或忽略额外关节，因为这些选择代表不同的控制所有权和安全 语义。运行时状态机目前只接受完整 Control Layout。
携带名称的消息无需强制变成硬件固定顺序：
encoder = NamedJointCommandEncoder()
target = encoder.encode(output.joints, required_layout=control_layout)
publish(names=target.layout.names, position=target.position)默认要求完整关节集合。只有下游协议确实支持局部关节命令时才显式开启：
encoder = NamedJointCommandEncoder(supports_partial=True)不携带名称的硬件数组必须声明固定布局并编码：
encoder = FixedOrderJointCommandEncoder(hardware_layout)
hardware_target = encoder.encode(output.joints)encoder 在 source layout 首次出现时编译索引，之后复用输出数组。
ModelSpec 保存按优先级排列的开放 ModelArtifact，没有固定的“onnx/openvino/rknn 字段”。
增加后端只需实现新的 artifact 和 factory，不需要扩大策略接口。
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,
"config": {"optimization_level": 3}
}' python3 your_program.py转换缓存通过源模型和设置指纹判断是否有效。x86 是否能转换取决于 Rockchip 提供的
rknn-toolkit2 wheel；真正执行 .rknn 通常使用目标板上的 rknn-toolkit-lite2。
-
JointLayout和映射只在启动、首次绑定或消息名称顺序变化时创建。 - 相同 observation layout 使用零复制 fast path；不同布局使用
np.take(..., out=...)。 -
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
跨平台对比时保持模型、warmup、iterations、电源模式和 CPU 负载一致。报告目录
tools/benchmark/results/ 被 Git 忽略，benchmark 程序本身纳入版本管理。