Variables

AbstractVariable

The supertype of the variables of a factor graph: RandomVariable, a latent quantity; DataVariable, an observed one; and ConstVariable, a constant.

A subtype implements:

• ReactiveMP.degree, the number of connections to factor nodes;
• ReactiveMP.israndom, ReactiveMP.isdata and ReactiveMP.isconst;
• ReactiveMP.create_new_stream_of_inbound_messages!, which a node's interface calls once per connection;
• ReactiveMP.get_stream_of_inbound_messages(variable, index), the stream of the messages the node on connection index sends to the variable, and ReactiveMP.get_stream_of_outbound_messages(variable, index), that of the messages the variable sends to it;
• ReactiveMP.get_stream_of_marginals and ReactiveMP.set_stream_of_marginals!, ReactiveMP.get_stream_of_predictions and ReactiveMP.set_stream_of_predictions!;
• ReactiveMP.activate!(variable, options), when it has streams to wire.

A variable has a label field, which the callback events show.

ReactiveMP.activate!(randomvar::RandomVariable, options::RandomVariableActivationOptions)
ReactiveMP.activate!(datavar::DataVariable, options::DataVariableActivationOptions)
ReactiveMP.activate!(factornode::FactorNode, options::FactorNodeActivationOptions)

Wire the lazy streams of a built graph into a live one. A graph is activated in order, once every node is created: its random and data variables first, then the initial marginals and messages, then its factor nodes, whose rules read the variables' streams. A constant needs no activation. After it, each new observation propagates through the graph.

See each method: ReactiveMP.activate!(::RandomVariable, ::RandomVariableActivationOptions), ReactiveMP.activate!(::DataVariable, ::DataVariableActivationOptions), ReactiveMP.activate!(::FactorNode, ::ReactiveMP.FactorNodeActivationOptions).

ReactiveMP.israndom(variable::AbstractVariable) -> Bool
ReactiveMP.israndom(variables::AbstractArray{<:AbstractVariable}) -> Bool

Whether variable is a RandomVariable; for an array, whether every element is. A node interface answers for its variable.

ReactiveMP.isdata(variable::AbstractVariable) -> Bool
ReactiveMP.isdata(variables::AbstractArray{<:AbstractVariable}) -> Bool

Whether variable is a DataVariable; for an array, whether every element is. A node interface answers for its variable.

ReactiveMP.isconst(variable::AbstractVariable) -> Bool
ReactiveMP.isconst(variables::AbstractArray{<:AbstractVariable}) -> Bool

Whether variable is a ConstVariable; for an array, whether every element is. A node interface answers for its variable.

ReactiveMP.degree(variable::AbstractVariable) -> Int

The number of factor node connections of variable: for a random or a data variable the number of its inbound message streams, for a constant the number of nodes connected to it.

ReactiveMP.get_stream_of_marginals(variable::AbstractVariable) -> MarginalObservable

The stream of the marginals of variable, a ReactiveMP.MarginalObservable: the posterior of a random variable, the latest observation of a data variable, the constant of a constant. Subscribe to it to receive every update:

subscribe!(ReactiveMP.get_stream_of_marginals(x), (q) -> println(mean(q)))

See also ReactiveMP.set_initial_marginal!, ReactiveMP.set_stream_of_marginals!.

ReactiveMP.set_stream_of_marginals!(variable::AbstractVariable, stream)

Connect the marginal stream of variable to stream, which activation does.

Throws

• ErrorException for a ConstVariable, whose marginal is fixed.

See also ReactiveMP.get_stream_of_marginals.

ReactiveMP.set_initial_marginal!(variable::AbstractVariable, marginal)
ReactiveMP.set_initial_marginal!(variables::AbstractArray{<:AbstractVariable}, marginals)

Seed the marginal stream of variable with marginal, a distribution, as an initial Marginal: what a rule that depends on it reads before inference computes one. For an array, a single PointMass or Distribution seeds every variable, and a collection seeds each variable with its element.

The initial marginal carries no log scale, even where log scales are tracked.

Throws

• AssertionError when an array of variables and a collection of marginals differ in length.

See also ReactiveMP.set_initial_message!.

ReactiveMP.set_initial_message!(variable::RandomVariable, message)
ReactiveMP.set_initial_message!(variables::AbstractArray{<:AbstractVariable}, messages)

Seed every message variable sends to its nodes with message, a distribution, as an initial Message: what a rule that depends on it reads before inference computes one. For an array, a single PointMass or Distribution seeds every variable, and a collection seeds each variable with its element.

A random variable's outbound message streams exist once it is activated, so this is called after ReactiveMP.activate! of the variable and before that of its nodes. The initial message's log scale is undefined.

Throws

• ArgumentError for a random variable not activated yet, and for a DataVariable or a ConstVariable, whose messages are their observations and their value;
• AssertionError when an array of variables and a collection of messages differ in length.

See also ReactiveMP.set_initial_marginal!.

ReactiveMP.get_stream_of_predictions(variable::AbstractVariable)

The stream of the predictions of variable, a ReactiveMP.MarginalObservable:

• for a DataVariable, the product of the messages the nodes send to it, what the model predicts for the observation without it; built only when the variable is activated with prediction = true (see DataVariableActivationOptions), and never emitting otherwise;
• for a RandomVariable and a ConstVariable, its marginal stream.

See also ReactiveMP.set_stream_of_predictions!.

ReactiveMP.set_stream_of_predictions!(variable::DataVariable, stream)

Connect the prediction stream of a data variable to stream, which activation does.

Throws

• ErrorException for a RandomVariable and a ConstVariable, whose predictions are their marginals.

See also ReactiveMP.get_stream_of_predictions.

RandomVariable <: AbstractVariable

A latent variable, the quantity inference computes a posterior for. It has a stream of inbound messages per connected factor node, as many outbound ones once it is activated, and the stream of its marginal, the product of its inbound messages. Create one with randomvar.

See also ConstVariable, DataVariable.

randomvar(; label = nothing) -> RandomVariable

Create a RandomVariable, with no connections yet. label names it in the callback events and in error messages, and is any value.

Examples

julia> x = randomvar(label = :x);

julia> ReactiveMP.israndom(x), ReactiveMP.degree(x)
(true, 0)

RandomVariableActivationOptions(stream_postprocessor, prod_context_for_message_computation::MessageProductContext, prod_context_for_marginal_computation::MessageProductContext)
RandomVariableActivationOptions()

What activating a RandomVariable needs, given positionally. With no arguments, no postprocessor and two default ReactiveMP.MessageProductContexts.

Fields

• stream_postprocessor: the stream postprocessor applied to the streams of the variable's ReactiveMP.EqualityChain and to its marginal stream, or nothing (see ReactiveMP.AbstractStreamPostprocessor);
• prod_context_for_message_computation: how an outbound message, the product of every inbound message but the one on its own connection, is computed;
• prod_context_for_marginal_computation: how the marginal, the product of every inbound message, is computed. Its callbacks also receive ReactiveMP.BeforeMarginalComputationEvent and ReactiveMP.AfterMarginalComputationEvent, and the variable's free-energy events, ReactiveMP.BeforeVariableBoundEntropyEvent and ReactiveMP.AfterVariableBoundEntropyEvent.

See also ReactiveMP.activate!.

ReactiveMP.activate!(randomvar::RandomVariable, options::RandomVariableActivationOptions)

Wire the streams of a random variable, after every factor node it connects to has been created.

1. Outbound messages: one stream per connection. With more than one connection, the message to connection i is the product of the inbound messages on every other connection, computed along a ReactiveMP.EqualityChain, which reuses the partial products. With one connection, the outbound message never emits: there is nothing to multiply.
2. Marginal: the product of all inbound messages, formed with as_marginal. It is computed once every inbound message has a value, and again whenever each has updated since.

A product that is initial does not consume its inputs, so the next update of any of them computes it again.

Throws

• ArgumentError for a variable with no connections.

See also RandomVariableActivationOptions.

DataVariable <: AbstractVariable

An observed variable, whose value arrives after the graph is built, with new_observation!, and may change: each observation propagates through the graph. Every connected node shares its one outbound stream, of the observations, and has a stream of the messages it sends back, whose product is the variable's prediction. Its marginal is its latest observation. Create one with datavar.

See also RandomVariable, ConstVariable.

datavar(; label = nothing) -> DataVariable

Create a DataVariable, with no observation and no connections yet. label names it in the callback events and in the error of an invalid observation.

Examples

julia> y = datavar(label = :y);

julia> ReactiveMP.isdata(y), ReactiveMP.degree(y)
(true, 0)

new_observation!(datavar::DataVariable, data)
new_observation!(datavar::DataVariable, data::PointMass)
new_observation!(datavar::DataVariable, ::Missing)
new_observation!(datavars::AbstractArray{<:DataVariable}, data::AbstractArray)
new_observation!(datavars::AbstractArray{<:DataVariable}, ::Missing)

Observe data on a data variable, and propagate it through the graph: the variable sends it to its nodes as a PointMass message, with log scale zero.

• A real number, an array of real numbers or a UniformScaling is wrapped in a PointMass: these are the values for which PointMass defines a variate_form, and so a mean.
• A PointMass is sent as it is, without validation: this is how an observation of another kind, such as text a custom node reads with BayesBase.getpointmass, is given. See Non-standard observations.
• missing says the observation is not available: the message is missing, and the rules that depend on it give missing in turn.
• An array of data variables observes an array of values of the same size, element by element, or missing on every variable.

Throws

• ErrorException for any other value, a distribution included, naming the variable and the type;
• AssertionError when an array of data variables and an array of values differ in size.

Examples

julia> y = datavar();

julia> new_observation!(y, 1.5)

julia> new_observation!([datavar(), datavar()], [1.0, 2.0])

DataVariableActivationOptions(prediction::Bool, linked::Bool, transform, args)
DataVariableActivationOptions()

What activating a DataVariable needs, given positionally. With no arguments, no prediction and no link.

Fields

• prediction: whether to build the prediction stream, the product of the messages the nodes send to the variable (see ReactiveMP.get_stream_of_predictions);
• linked: whether the variable's observations are a function of other variables, rather than given with new_observation!;
• transform: for a linked variable, the function its observation is of;
• args: for a linked variable, the constants, variables and arrays of variables transform takes, in order. transform is given their point values: a constant, a data variable's observation (the payload of its PointMass, numeric or not) or, for an array, the vector of its variables' observations.

See also ReactiveMP.activate!.

ReactiveMP.activate!(datavar::DataVariable, options::DataVariableActivationOptions)

Wire the streams of a data variable, after every factor node it connects to has been created.

1. Prediction, with options.prediction: the product of the messages the nodes send to the variable, computed once every one has a value and again whenever each has updated since.
2. Link, with options.linked: whenever each of options.args has a value, and again when one updates, options.transform of their values is observed with new_observation!. Linking to a random variable fails at the first update, since its marginal is not a point value.
3. Marginal: always the latest observation, as a Marginal.

See also DataVariableActivationOptions.

ConstVariable <: AbstractVariable

A constant of the model, fixed when it is created. Its message and its marginal are a clamped PointMass of the constant, with log scale zero, wired at creation: a constant needs no activation, and every node connected to it shares its one message stream. A node's interface on a constant sends it no message. Create one with constvar.

See also RandomVariable, DataVariable.

constvar(constant; label = nothing) -> ConstVariable

Create a ConstVariable holding constant, a number, an array or any other value a PointMass holds. label names it in the callback events and in error messages.

Examples

julia> c = constvar(2.0);

julia> ReactiveMP.isconst(c), ReactiveMP.degree(c)
(true, 0)

ReactiveMP.create_new_stream_of_inbound_messages!(variable::AbstractVariable) -> (observable, index)

Allocate the stream of the messages a new factor node connection sends to variable, and return it with the connection's index. A ReactiveMP.NodeInterface calls it when it is created, and keeps the stream as its outbound message stream: the node's outbound message is the variable's inbound one. The stream is a lazy ReactiveMP.MessageObservable until the node is activated.

A random and a data variable allocate a new stream per connection, numbered from 1. A constant counts the connection and returns its one shared stream, of the constant's message, with index 1 every time.