AI Integration

HORUS's sub-microsecond IPC makes it well-suited for combining real-time control with AI inference. This guide covers patterns for integrating ML models into HORUS applications.

Overview

HORUS supports local AI inference through two main approaches (cloud APIs, covered further down, are a third path for higher-level reasoning):

Python ML Nodes (Recommended for Prototyping)

Use any Python ML library (PyTorch, TensorFlow, ONNX, etc.)
Hardware nodes handle camera/sensor capture
Pub/sub connects ML pipeline to control nodes
10-100ms typical inference latency

Rust Inference (For Production)

ONNX Runtime via ort crate
Tract (pure Rust inference engine)
1-50ms typical inference latency

Architecture Pattern

Loading diagram...

AI pipeline: Sensor → ML inference → Real-time control

The key insight: keep AI inference in dedicated nodes. HORUS topics decouple the fast control loop from slower ML processing, so a slow inference step doesn't block motor commands.

Python ML Integration

The fastest way to add AI to a HORUS application is through Python nodes. Python has the richest ML ecosystem, and HORUS's Python bindings give you full access to the pub/sub system.

Camera + ML Pipeline

from horus import Node, Scheduler
import numpy as np

# Simulated camera node (replace with your camera capture logic)
def camera_tick(node):
    # In a real robot, capture from camera hardware here. A topic declared as a
    # bare string carries at most 4KB per message, so this toy frame is
    # deliberately tiny — real frames go over a horus.Tensor topic (see below).
    frame = np.random.randint(0, 255, (24, 24, 3), dtype=np.uint8)
    node.send("cam.image_raw", frame.tolist())

cam = Node(name="camera", pubs=["cam.image_raw"], tick=camera_tick, rate=30, order=0)

# Stand-in for your model — replace with real inference. The control node
# below expects a dict back.
def process_frame(frame):
    return {"obstacle_detected": False}

# ML processing node
def ml_tick(node):
    if node.has_msg("cam.image_raw"):
        frame = node.recv("cam.image_raw")
        # Run your ML model here
        # e.g., model.predict(frame), torch inference, etc.
        result = process_frame(frame)
        node.send("detections", result)

ml_node = Node(
    name="ml_processor",
    subs=["cam.image_raw"],
    pubs=["detections"],
    tick=ml_tick,
    rate=10,  # Process at 10 FPS
    order=1
)

# Control node reacts to detections
def control_tick(node):
    if node.has_msg("detections"):
        detections = node.recv("detections")
        # React to ML output
        if detections.get("obstacle_detected"):
            node.send("cmd_vel", {"linear": 0.0, "angular": 0.5})
        else:
            node.send("cmd_vel", {"linear": 1.0, "angular": 0.0})

controller = Node(
    name="controller",
    subs=["detections"],
    pubs=["cmd_vel"],
    tick=control_tick,
    rate=30,
    order=2
)

scheduler = Scheduler()
scheduler.add(cam)
scheduler.add(ml_node)
scheduler.add(controller)
scheduler.run()

PyTorch Example

from horus import Node, run
import torch

# Load model once at startup
model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True)
model.eval()

def detect_tick(node):
    if node.has_msg("cam.image_raw"):
        frame = node.recv("cam.image_raw")
        results = model(frame)
        detections = results.pandas().xyxy[0].to_dict('records')
        node.send("detections", detections)

node = Node(
    name="yolo_detector",
    subs=["cam.image_raw"],
    pubs=["detections"],
    tick=detect_tick,
    rate=10
)

run(node)

ONNX Runtime (Python)

from horus import Node, run
import onnxruntime as ort
import numpy as np

session = ort.InferenceSession("model.onnx")
input_name = session.get_inputs()[0].name

def inference_tick(node):
    if node.has_msg("cam.image_raw"):
        frame = node.recv("cam.image_raw")
        # Preprocess
        input_data = np.array(frame).astype(np.float32)
        input_data = np.expand_dims(input_data, axis=0)
        # Run inference
        outputs = session.run(None, {input_name: input_data})
        node.send("ml.output", outputs[0].tolist())

node = Node(
    name="onnx_inference",
    subs=["cam.image_raw"],
    pubs=["ml.output"],
    tick=inference_tick,
    rate=15
)

run(node)

Pass Frames as Tensors, Not Lists

There is no HORUS-supplied inference base class — no horus.ml_utils, no horus.ai. The session, the preprocessing and the postprocessing are all your code, exactly as written above. What HORUS does supply is a zero-copy carrier for the arrays: horus.Tensor.

The camera example above sends frame.tolist(), which only works because its frame is a 24x24 stand-in. A topic declared as a bare string serializes to MessagePack and carries at most 4KB per message, so a real 640x480 RGB frame — ~920,000 Python ints, ~1.7MB packed — is rejected outright: ValueError: 'data' out of range: expected [0..4096], raised on every tick. Declare it as horus.Tensor instead: the pixels are written once into a shared-memory pool and only a 168-byte descriptor crosses the topic.

import numpy as np
import horus


def capture(node):
    frame = np.random.default_rng().integers(0, 255, (480, 640, 3), dtype=np.uint8)
    node.send("cam.image_raw", horus.Tensor.from_numpy(frame))


def infer(node):
    t = node.recv("cam.image_raw")
    if t is None:
        return
    frame = t.numpy()                    # zero-copy view of the same bytes
    node.send("detections", {"mean": float(frame.mean())})


horus.run(
    horus.Node(name="camera", tick=capture, rate=10, order=0,
               pubs=[horus.Pub("cam.image_raw", horus.Tensor)]),
    horus.Node(name="detector", tick=infer, rate=10, order=1,
               subs=[horus.Sub("cam.image_raw", horus.Tensor)],
               pubs=["detections"]),
    duration=1.0,
)

Give the subscriber the same rate as the publisher. Unlike the ordinary message topics used elsewhere on this page, a tensor topic hands over a reference rather than queueing copies, and a subscriber that ticks slower than the publisher receives nothing at all — recv() returns None every time. If inference only keeps up at 10 Hz, publish at 10 Hz.

See Python Tensors for ML for the full tensor and pool API, including the torch bridge and the current CPU-only device situation.

Rust Inference

For production deployments where you need maximum performance, integrate ML inference directly in Rust.

ONNX Runtime (ort crate)

The ort crate provides Rust bindings for ONNX Runtime:

# horus.toml
[dependencies]
horus = "0.2"
ort = "2.0"
ndarray = "0.15"

Or let the CLI write those two entries for you:

horus add ort --source crates.io
horus add ndarray --source crates.io

use horus::prelude::*;
use ort::{GraphOptimizationLevel, Session};
use ndarray::Array;

struct InferenceNode {
    session: Session,
    input_name: String,
}

impl InferenceNode {
    // horus::prelude re-exports a one-generic `Result<T>`, so the two-generic
    // std form has to be named explicitly here.
    fn new(model_path: &str) -> std::result::Result<Self, Box<dyn std::error::Error>> {
        let session = Session::builder()?
            .with_optimization_level(GraphOptimizationLevel::Level3)?
            .commit_from_file(model_path)?;

        let input_name = session.inputs[0].name.clone();

        Ok(Self { session, input_name })
    }

    fn infer(&self, input: &[f32]) -> Option<Vec<f32>> {
        let input_array = Array::from_shape_vec((1, input.len()), input.to_vec()).ok()?;
        let outputs = self.session.run(
            ort::inputs![&self.input_name => input_array.view()].ok()?
        ).ok()?;
        let output = outputs[0].try_extract_tensor::<f32>().ok()?;
        Some(output.view().iter().copied().collect())
    }
}

Tract (Pure Rust)

Tract runs ONNX models with zero external dependencies:

# horus.toml
[dependencies]
horus = "0.2"
tract-onnx = "0.21"

use tract_onnx::prelude::*;

fn load_model(path: &str) -> TractResult<SimplePlan<TypedFact, Box<dyn TypedOp>, Graph<TypedFact, Box<dyn TypedOp>>>> {
    tract_onnx::onnx()
        .model_for_path(path)?
        .into_optimized()?
        .into_runnable()
}

fn run_inference(model: &SimplePlan<TypedFact, Box<dyn TypedOp>, Graph<TypedFact, Box<dyn TypedOp>>>, input: &[f32]) -> Option<Vec<f32>> {
    let input_tensor = tract_ndarray::arr1(input).into_dyn();
    let result = model.run(tvec!(input_tensor.into())).ok()?;
    let output = result[0].to_array_view::<f32>().ok()?;
    Some(output.iter().copied().collect())
}

Model Format Comparison

Format	Crate	Use Case	External Deps
ONNX	`ort`	General (PyTorch, TF exports)	ONNX Runtime C lib
ONNX	`tract-onnx`	Pure Rust inference	None
TFLite	`tflite`	Edge/mobile models	TFLite C lib

Cloud API Integration

For complex reasoning tasks (task planning, scene understanding, natural language), call cloud APIs from HORUS nodes.

Python (Recommended)

from horus import Node, run
import requests
import os

API_KEY = os.environ["OPENAI_API_KEY"]

def planner_tick(node):
    if node.has_msg("user.goal"):
        goal = node.recv("user.goal")
        response = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "model": "gpt-4",
                "messages": [
                    {"role": "system", "content": "Generate robot action plans as JSON."},
                    {"role": "user", "content": goal}
                ],
                "max_tokens": 500
            }
        )
        plan = response.json()["choices"][0]["message"]["content"]
        node.send("robot.plan", plan)

node = Node(
    name="planner",
    subs=["user.goal"],
    pubs=["robot.plan"],
    tick=planner_tick,
    rate=1  # Check for goals once per second
)

run(node)

Rust (reqwest)

Add the HTTP client first:

horus add reqwest --source crates.io --features blocking,json

use reqwest::blocking::Client;
use serde::{Deserialize, Serialize};

#[derive(Serialize)]
struct ChatRequest {
    model: String,
    messages: Vec<ChatMessage>,
    max_tokens: u32,
}

#[derive(Serialize, Deserialize)]
struct ChatMessage {
    role: String,
    content: String,
}

#[derive(Deserialize)]
struct ChatResponse {
    choices: Vec<ChatChoice>,
}

#[derive(Deserialize)]
struct ChatChoice {
    message: ChatMessage,
}

fn call_llm(client: &Client, api_key: &str, prompt: &str) -> Option<String> {
    let request = ChatRequest {
        model: "gpt-4".to_string(),
        messages: vec![ChatMessage {
            role: "user".to_string(),
            content: prompt.to_string(),
        }],
        max_tokens: 500,
    };

    let response = client
        .post("https://api.openai.com/v1/chat/completions")
        .header("Authorization", format!("Bearer {}", api_key))
        .json(&request)
        .send()
        .ok()?;

    let chat_response: ChatResponse = response.json().ok()?;
    Some(chat_response.choices[0].message.content.clone())
}

Performance Considerations

Latency Budget

Typical robotics control loop at 100Hz (10ms cycle):

Sensor capture:  ~1-16ms  (hardware dependent)
ML inference:    ~5-50ms  (model dependent)
Topic transfer:  ~85ns    (HORUS shared memory)
Control logic:   ~1μs     (HORUS node tick)
Motor command:   ~1ms     (hardware actuator)

ML inference is typically the bottleneck. Strategies to manage this:

Throttle Inference

Process every Nth frame instead of every frame:

frame_count = 0

def ml_tick(node):
    global frame_count
    if node.has_msg("cam.image_raw"):
        frame_count += 1
        if frame_count % 5 == 0:  # Every 5th frame
            frame = node.recv("cam.image_raw")
            result = model.predict(frame)
            node.send("detections", result)

Offload Inference Off the Control Loop

Local inference is CPU-bound, so run it on a compute=True node. HORUS puts that node on a worker pool, and the other nodes keep their own rates while it works:

import time
import horus

def model_predict(frame):
    time.sleep(0.03)                     # stands in for a real model
    return {"boxes": [], "frame": frame["id"]}

def infer(node):
    if node.has_msg("cam.image_raw"):
        frame = node.recv("cam.image_raw")
        node.send("detections", model_predict(frame))

def control(node):
    if node.has_msg("detections"):
        node.recv("detections")

frames = {"n": 0}
def camera(node):
    frames["n"] += 1
    node.send("cam.image_raw", {"id": frames["n"]})

horus.run(
    horus.Node(name="camera", tick=camera, rate=30, pubs=["cam.image_raw"], order=0),
    horus.Node(name="infer", tick=infer, rate=30, order=1, compute=True,
               subs=["cam.image_raw"], pubs=["detections"]),
    horus.Node(name="control", tick=control, rate=100, order=2, subs=["detections"]),
    duration=1.0,
)

Each infer tick takes ~30 ms, yet camera still gets its 30 ticks and control its 100, each averaging under 130 µs of tick time. The control node reads the most recent detections and never waits.

If instead you call a remote inference API, the work is I/O-bound — use an async def tick, which HORUS runs on its async I/O executor:

import asyncio
import horus

async def remote_infer(node):
    if not node.has_msg("cam.image_raw"):
        return
    frame = node.recv("cam.image_raw")
    await asyncio.sleep(0.08)            # stands in for the HTTP round-trip
    node.send("detections", {"frame": frame["id"], "boxes": []})

horus.run(
    horus.Node(name="remote", tick=remote_infer, rate=30,
               subs=["cam.image_raw"], pubs=["detections"]),
    duration=1.0,
)

compute=True and an async tick are mutually exclusive — pick the one matching whether the bottleneck is CPU or I/O. See Async Nodes for the execution model.

Use Appropriate Models

Task	CPU Model	GPU Model	Cloud API
Object Detection	YOLOv8n (ONNX)	YOLOv8x	GPT-4 Vision
Classification	MobileNet (TFLite)	EfficientNet	Cloud Vision
Pose Estimation	MediaPipe	OpenPose	-
Task Planning	Phi-3 Mini	Llama 3	GPT-4 / Claude
Depth Estimation	MiDaS Small	MiDaS Large	-

Best Practices

Separate concerns: Keep AI inference in dedicated nodes. Don't mix ML code with control logic.

Handle failures gracefully: AI inference can fail. Always have a safe fallback:

def control_tick(node):
    if node.has_msg("detections"):
        detections = node.recv("detections")
        react_to(detections)
    else:
        # Safe default when no detections available
        node.send("cmd_vel", {"linear": 0.0, "angular": 0.0})

Monitor performance: Use horus monitor to watch node timing and message flow:
```
horus monitor  # See which nodes are slow
```
Start with Python: Prototype in Python first, then move performance-critical inference to Rust if needed.
Cache results: For cloud APIs, cache common responses to reduce latency and cost.