4.2 Python 接口
hbm_runtime 是基于 pybind11 的 Python 绑定接口,用于访问和操作底层 libhbucp / libdnn C++ 库,提供高性能的神经网络模型加载与推理能力。
该接口封装了底层模型运行时细节,使 Python 用户能够方便地加载单个或多个神经网络模型,查询并管理模型的输入/输出元信息,并灵活执行推理操作。接口支持多种输入数据格式,并在必要时自动将输入转换为 C 连续(C-contiguous)存储,以保证底层访问正确性与效率。
此外,新版接口在推理阶段会在 C++ 侧释放 Python GIL,使 Python 多线程可并发调用 run();对多模型推理场景,底层会自动使用多线程并行调度各模型推理任务,以提升吞吐。
适用场景
- 在 Python 环境中快速集成和调用 hbm_runtime 运行时能力。
- 机器人视觉、智能边缘计算等对推理效率和调度灵活性有较高要求的应用。
- 需要同时加载和管理多个模型,并在不同推理调用中按需配置任务调度参数(优先级/核心绑定/设备 ID 等)的场景。
- 需要查询模型编译期 BPU 相关信息(例如编译期 BPU core 数)以辅助运行时资源配置与一致性检查的场景。
关键特性
- 多模型支持
- 支持加载单个模型或多个模型组成的模型组;每个模型均可独立获取输入/输出元信息并进行推理。
- run() 支持对多模型输入进行一次性推理,并按模型名返回结果(即使只跑单个模型也返回
{model_name: {...}}的嵌套结构)。
- 灵活输入格式
- 支持单输入(numpy.ndarray);
- 支持单模型多输入字典(Dict[str, np.ndarray],key 为 input tensor name);
- 支持多模型多输入结构(Dict[str, Dict[str, np.ndarray]],外层 key 为 model name,内层 key 为 input tensor name)。
- 所有输入均自动检查是否为 C 连续内存格式,必要时进行拷贝,以保证底层高效正确访问(非连续输入可能带来额外拷贝开销)。
- 调度参数配置:默认参数 + 单次调用覆盖(run-local)
- 支持通过 set_scheduling_params(...) 设置模型级默认调度参数(持久化在 runtime 内部,可多次复用)。
- 同时支持在每次 run() 调用时,通过可选参数对调度进行单次覆盖(run-time overrides),覆盖规则为:run() 参数优先于默认参数,且该覆盖仅作用于本次调用,不影响其他线程/其他 run() 调用。
- 多线程推理能力
- Python 多线程并发调用 run():推理阶段在 C++ 内部释放 GIL,使多个 Python 线程可同时发起推理调用。
- 多模型并行推理:当输入为多模型结构时,底层会为每个模型启动线程并行执行推理任务(multi-threaded launch),在多核 BPU 系统上可提升吞吐;单模型场景则仅对应一个推理线程。
安装说明(Installation)
本模块 hbm_runtime 是基于 C++ 实现的高性能推理运行时 Python 接口,依赖 pybind11 和地平线提供的底层推理库(如 libdnn, libhbucp 等)。支持通过系统 DEB 包(.deb) 的方式进行安装,适用于 Python 3.10 及以上版本。
系统依赖
| 依赖项 | 最低版本 | 说明 |
|---|---|---|
| Python | ≥ 3.10 | 推荐使用 Python 3.10 |
| pip | ≥ 22.0 | 安装 wheel 包所需 |
| pybind11 | 任意 | 构建时使用,安装包时不需要依赖 |
| scikit-build-core | ≥ 0.7 | 构建 wheel 包时使用(仅源码构建) |
| 地平线基础库 | 根据平台 | 如 libdnn.so、libucp.so 等,通常由 BSP 提供 |
构建wheel包
构建wheel包的方式有三种下面分别介绍。
安装deb时构建
在hobot-dnn包的安装过程中添加了hbm_runtime的wheel构建,deb包安装完成后就会生成hbm-runtime的whl包。
#通过源安装
sudo apt-get install hobot-dnn
#通过本地deb包安装(注意不同时间编译的包名称,以实际情况为准)
dpkg -i hobot-dnn_4.0.4-20250909195426_arm64.deb
#安装完成后可在板端的/tmp目录下找到wheel包
ls /tmp
#注意不同版本号whl包名称不同,xxx代表版本
#hbm_runtime-x.x.x-cp310-cp310-manylinux_2_34_aarch64.whl
编译系统软件时构建
在系统软件的镜像编译时会安装hobot-dnn的deb,安装的过程中会构建hbm-runtime的whl包,并转存到out/product/deb_packages目录
sudo ./pack_image.sh
ls out/product/deb_packages
#注意不同版本号whl包名称不同,xxx代表版本
#hbm_runtime-x.x.x-cp310-cp310-manylinux_2_34_aarch64.whl
在端侧构建
#进入hbm_runtime的源码库
cd /usr/hobot/lib/hbm_runtime
#运行构建命令
./build.sh
#查看构建好的wheel包
ls dist/
#注意不同版本号whl包名称不同,xxx代表版本
#hbm_runtime-x.x.x-cp310-cp310-manylinux_2_34_aarch64.whl
安装方式
使用 wheel 包
使用wheel 安装的方式有两种,选其一即可
-
通过本地 wheel 包安装
- 找到通过“构建wheel包”小节中构建的whl文件。
#示例:使用 pip 安装本地whl包(注意不同版本号whl包名称不同,xxx代表版本)
pip install hbm_runtime-x.x.x-cp310-cp310-manylinux_2_34_aarch64.whl -
从pypi源安装
pip install hbm_runtime
使用 .deb 包安装
使用deb安装的方式有两种,选其一即可
-
通过本地 DEB 包安装
# 示例:安装 DEB 包(注意不同时间编译的包名称,以实际情况为准)
sudo dpkg -i hobot-dnn_4.0.2-20250714201215_arm64.deb -
通过apt源安装
sudo apt-get install hobot-dnn -
常见问题
- 如果 .deb 安装后文件未生效,检查是否有其他依赖阻止其覆盖(如已有老版本 hobot-spdev)。
- 可使用 dpkg -L hobot-dnn 查看文件是否成功部署。
卸载说明
-
卸载 pip 安装的包:
pip uninstall hbmruntime -
卸载 .deb 安装的包:
sudo apt remove hobot-dnn
快速开始(Quick Start)
本节介绍如何使用hbm_runtime进行模型加载和推理。只需几行代码,即可运行模型并获取输出结果。
环境准备
请确保已正确安装 HBMRuntime(详见安装说明),并已具备模型文件 hbm 模型。
示例
单线程推理
单线程单模型单输入推理
适用于模型只有一个输入张量的情况。
import numpy as np
from hbm_runtime import HB_HBMRuntime
# 加载模型
model = HB_HBMRuntime("/opt/hobot/model/s600/basic/lanenet256x512.hbm")
# 获取模型名与输入名
model_name = model.model_names[0]
input_name = model.input_names[model_name][0] # 假设模型只有一个输入
# 获取该输入对应的 shape
input_shape = model.input_shapes[model_name][input_name]
# 构造 numpy 输入
input_tensor = np.ones(input_shape, dtype=np.float32)
# 执行推理
outputs = model.run(input_tensor)
# 获取输出结果
output_array = outputs[model_name]
print("Output:", output_array)
单线程单模型多输入推理
适用于模型有多个输入张量的情况。
import numpy as np
from hbm_runtime import HB_HBMRuntime
hb_dtype_map = {
"U8": np.uint8,
"S8": np.int8,
"F32": np.float32,
"F16": np.float16,
"U16": np.uint16,
"S16": np.int16,
"S32": np.int32,
"U32": np.uint32,
"BOOL8": np.bool_,
}
# 加载模型
model = HB_HBMRuntime("/opt/hobot/model/s600/basic/yolov5x_672x672_nv12.hbm")
# 获取模型名(假设只加载了一个模型)
model_name = model.model_names[0]
# 准备输入名和 shape
input_names = model.input_names[model_name]
input_shapes = model.input_shapes[model_name]
input_dtypes = model.input_dtypes[model_name]
# 构造输入字典
input_tensors = {}
for name in input_names:
shape = input_shapes[name]
np_dtype = hb_dtype_map.get(input_dtypes[name].name, np.float32) # fallback
input_tensors[name] = np.ones(shape, dtype=np_dtype)
# 可选:指定推理优先级和 BPU 设备
priority = {model_name: 5}
bpu_cores = {model_name: [0]}
model.set_scheduling_params(
priority=priority,
bpu_cores=bpu_cores
)
# 执行推理,可选指定优先级和 BPU 核
results = model.run(input_tensors)
# 输出结果
for output_name, output_data in results[model_name].items():
print(f"Output: {output_name}, shape={output_data.shape}")
单线程多模型多输入推理
适用于多模型有多个输入张量的情况,注意这里的多模型可以是多个 HBM 文件,也可以是单个 HBM 文件里面包含多个模型。
"""Multi-model inference quick start."""
import numpy as np
from hbm_runtime import HB_HBMRuntime
MODEL_PATHS = [
"/opt/hobot/model/s600/basic/yolov5x_672x672_nv12.hbm",
"/opt/hobot/model/s600/basic/resnet18_224x224_nv12.hbm",
]
DTYPE_MAP = {
"U8": np.uint8, "S8": np.int8,
"F16": np.float16, "F32": np.float32,
}
# Load models
rt = HB_HBMRuntime(MODEL_PATHS)
# Build inputs from model metadata
inputs = {
m: {
inp: np.random.rand(*rt.input_shapes[m][inp]).astype(
DTYPE_MAP.get(rt.input_dtypes[m][inp].name, np.float32)
)
for inp in rt.input_names[m]
}
for m in rt.model_names
}
# Optional: default scheduling params
rt.set_scheduling_params(
priority={m: 5 for m in rt.model_names},
bpu_cores={m: [0] for m in rt.model_names},
)
# Run inference (multi-model, parallel internally)
outputs = rt.run(inputs)
# Print results
for m, outs in outputs.items():
print(f"[{m}]")
for name, arr in outs.items():
print(f" {name}: {arr.shape}, {arr.dtype}")
多线程推理
多线程单模型单输入推理
适用于模型只有一个输入张量的情况。
import threading
import numpy as np
from hbm_runtime import HB_HBMRuntime
# Load model
model = HB_HBMRuntime("/opt/hobot/model/s600/basic/asr.hbm")
model_name = model.model_names[0]
input_name = model.input_names[model_name][0]
input_shape = model.input_shapes[model_name][input_name]
# Shared input (read-only)
input_tensor = np.ones(input_shape, dtype=np.float32)
def worker(core_id: int):
outputs = model.run(
input_tensor,
model_name=model_name,
priority={model_name: 5},
bpu_cores={model_name: [core_id]},
custom_id={model_name: core_id}, # optional
)
# Print minimal info
outs = outputs[model_name]
first_name, first_arr = next(iter(outs.items()))
print(f"[T{core_id}] {first_name}: shape={first_arr.shape}, dtype={first_arr.dtype}")
threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
for t in threads: t.start()
for t in threads: t.join()
多线程单模型多输入推理
适用于模型有多个输入张量的情况。
import threading
import numpy as np
from hbm_runtime import HB_HBMRuntime
hb_dtype_map = {
"U8": np.uint8, "S8": np.int8,
"F16": np.float16, "F32": np.float32,
"U16": np.uint16, "S16": np.int16,
"U32": np.uint32, "S32": np.int32,
"BOOL8": np.bool_,
}
# Load single model
model = HB_HBMRuntime("/opt/hobot/model/s600/basic/yolov5x_672x672_nv12.hbm")
model_name = model.model_names[0]
# Build input tensors (shared, read-only)
input_tensors = {
name: np.ones(
model.input_shapes[model_name][name],
dtype=hb_dtype_map.get(model.input_dtypes[model_name][name].name, np.float32)
)
for name in model.input_names[model_name]
}
def worker(core_id: int):
results = model.run(
input_tensors,
model_name=model_name,
priority={model_name: 5},
bpu_cores={model_name: [core_id]},
custom_id={model_name: core_id}, # optional, for tracing
)
out_name, out_arr = next(iter(results[model_name].items()))
print(f"[T{core_id}] {out_name}: {out_arr.shape}, {out_arr.dtype}")
# Launch 4 threads, bind to BPU cores 0~3
threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
for t in threads: t.start()
for t in threads: t.join()