Asynchronous Updates

[navidivan]

In any hierarchical network, we have also a hierarchy of timescales. The lower-level vision system gets updated with incoming data at a much higher frequency than association cortex, which binds information between different sub-networks. Many active inference papers implement this sort of thing by controlling the message passing dynamics.
What is the most routine/standard way of implementing asynchronous updates and message passing in a factor graph using ReactiveBayes? assume we have two streams of data, one at 10 Hz and the other at 100 Hz. These two implement Filtering, and feed forward to a higher node that operates at 1 Hz.
Is there an implementable with current libraries? Do we have an example of this sort of thing?!

[bvdmitri]

Hey @navidivan, no, not yet. We have plans to make it possible within RxInfer but currently this is not implemented

[navidivan]

Thank you @bvdmitri
Would it make sense to do a hacky way of aggregation for now, and control the message passing in a manual way?
For example, there could be an outer loop running, and two inner loops. We just split the task manually.

psuedo code:

// Initialize engines
initialize sensor1_engine using sensor1_model  // (e.g., for variable x)
initialize sensor2_engine using sensor2_model  // (e.g., for variable w)
initialize aggregator_engine using aggregator_model  // (e.g., for Y = f(x,w))

for aggregator_step = 1 to TOTAL_AGGREGATOR_STEPS do
    // Fast sensor loop: 100 Hz (e.g., 100 updates per aggregator step)
    for fast_step = 1 to NUM_FAST_STEPS do
        data1 = acquireSensor1Data()         // get one sensor1 sample
        sensor1_engine.update(data1)           // update inference for sensor1
    end for

    // Slow sensor loop: 10 Hz (e.g., 10 updates per aggregator step)
    for slow_step = 1 to NUM_SLOW_STEPS do
        data2 = acquireSensor2Data()         // get one sensor2 sample
        sensor2_engine.update(data2)           // update inference for sensor2
    end for

    // Aggregator step (1 Hz)
    x_estimate = sensor1_engine.getPosteriorMean("x")
    w_estimate = sensor2_engine.getPosteriorMean("w")
    aggregator_data = { "x_input": x_estimate, "w_input": w_estimate }
    aggregator_engine.update(aggregator_data)
    record aggregator_engine.getPosterior("Y")
end for

[navidivan]

my question then becomes, if we can manually control the data consumption and updates like this. what do we miss if we do this?

[navidivan]

Maybe I don't fully understand how this reactive framework works. Chatgpt gives me this code:

`using RxInfer, Random, StableRNGs

Sensor 1: fast sensor (100 Hz)

@model function sensor1_model(y)
x_prev ~ Normal(0.0, 1e3)
x_current ~ Normal(x_prev, 1.0)
y ~ Normal(x_current, 10.0)
end
@Constraints function sensor1_constraints()
q(x_prev, x_current) = q(x_prev)q(x_current)
end
sensor1_init = @initialization begin
q(x_current) = NormalMeanVariance(0.0, 1e3)
end
sensor1_autoupdates = @autoupdates begin
x_prev_mean, x_prev_var = mean_var(q(x_current))
end

Sensor 2: slow sensor (10 Hz)

@model function sensor2_model(y)
w_prev ~ Normal(1.0, 1e3)
w_current ~ Normal(w_prev, 0.5)
y ~ Normal(w_current, 2.0)
end
@Constraints function sensor2_constraints()
q(w_prev, w_current) = q(w_prev)q(w_current)
end
sensor2_init = @initialization begin
q(w_current) = NormalMeanVariance(1.0, 1e3)
end
sensor2_autoupdates = @autoupdates begin
w_prev_mean, w_prev_var = mean_var(q(w_current))
end

Aggregator: fuses sensor outputs (1 Hz)

@model function aggregator_model(x_in, w_in)
y_agg ~ Normal(x_in + w_in, 1.0)
end
@Constraints function aggregator_constraints()
q(y_agg) = q(y_agg)
end
aggregator_init = @initialization begin
q(y_agg) = NormalMeanVariance(0.0, 10.0)
end

Create datastreams.

Sensor 1: 100 Hz (every 10 ms)

sensor1_stream = timer(10, 10) |> map(_ -> (y = rand(Normal(0,2.0)),)) |> take(300)

Sensor 2: 10 Hz (every 100 ms)

sensor2_stream = timer(100, 100) |> map(_ -> (y = rand(Normal(1.0,5.0)),)) |> take(30)

Aggregator: 1 Hz (every 1000 ms)

function aggregator_datapoint()
x_mean = mean(last(sensor1_engine.posteriors[:x_current]))
w_mean = mean(last(sensor2_engine.posteriors[:w_current]))
return (x_in = x_mean, w_in = w_mean)
end
aggregator_stream = timer(1000, 1000) |> map(_ -> aggregator_datapoint()) |> take(10)

Instantiate engines.

sensor1_engine = infer(
model = sensor1_model(),
constraints = sensor1_constraints(),
datastream = sensor1_stream,
autoupdates = sensor1_autoupdates,
initialization = sensor1_init,
returnvars = (x_current = KeepLast(),),
iterations = 5,
autostart = false
)
sensor2_engine = infer(
model = sensor2_model(),
constraints = sensor2_constraints(),
datastream = sensor2_stream,
autoupdates = sensor2_autoupdates,
initialization = sensor2_init,
returnvars = (w_current = KeepLast(),),
iterations = 5,
autostart = false
)
aggregator_engine = infer(
model = aggregator_model(),
constraints = aggregator_constraints(),
datastream = aggregator_stream,
initialization = aggregator_init,
returnvars = (y_agg = KeepLast(),),
iterations = 5,
autostart = false
)

Start the three engines concurrently.

RxInfer.start(sensor1_engine)
RxInfer.start(sensor2_engine)
RxInfer.start(aggregator_engine)
sleep(12)
RxInfer.stop(sensor1_engine)
RxInfer.stop(sensor2_engine)
RxInfer.stop(aggregator_engine)

final_estimate = mean(last(aggregator_engine.posteriors[:y_agg]))
println("Final aggregator estimate: ", final_estimate)
`

Reactive frameworks like RxInfer are built on asynchronous, event‐driven paradigms. Although the code that defines each engine is written sequentially, it doesn’t mean that one engine runs to completion before the next starts. Instead, each call to infer creates an independent reactive pipeline that “registers” callbacks (or subscriptions) to its own data stream (often defined via a timer or similar operator).

When autostart is set to false, nothing happens immediately; the engines merely exist as configured pipelines. Then, when you call RxInfer.start(engine) for each engine, each one is scheduled—typically on the same event loop or scheduler—to run concurrently. They yield control (i.e. pause) while waiting for their next data point and resume when an event (like a timer tick) fires. This means that although the engines are defined one after the other, they all run asynchronously, each processing its own stream at its specified rate (e.g. 100 Hz, 10 Hz, 1 Hz).

In summary, the reactive framework ensures that:

Each engine is an asynchronous process that updates whenever new data is available.
The operations yield (or suspend) while waiting for events, allowing other engines to run in the meantime.
Once started, all engines operate concurrently—sensor engines process their high-frequency data and the aggregator picks up the latest estimates as soon as its timer fires—without one blocking the others.
Thus, even though the engine definitions are written sequentially, their execution is concurrent and non-blocking.

[bvdmitri]

ChatGPT isn't really good when it comes to Julia programming language, I wouldn't try to use it. We know what do you want to achieve and we have it in our agenda (because we also want to implement that ourselves). It just isn't ready yet. As for now we are working for client-server architecture for RxInfer to also make it easier to call RxInfer from python. After that we plan to re-write the core of RxInfer to make it easier to use and debug but also to support case that you've described. So stay tuned :)