GitHub

Messages

A message is what flows along an edge of the factor graph: a summary, about the variable on that edge, of the part of the graph it comes from. In belief propagation it is an unnormalised function. In variational message passing it is the exponent of an expected log-factor (see Message passing).

The examples use the Gaussian node of The example node.

Messages as distributions

A message is usually a normalised probability distribution. A univariate normal is two numbers, which is all that passes along the edge:

message = Message(NormalMeanVariance(0.0, 1.0), false, false)
mean(message), var(message), logpdf(message, 1.0)
(0.0, 1.0, -1.4189385332046727)

The message forwards the statistics to its distribution. The constant the normalisation leaves out is the message's log scale, which the engine tracks when you ask it to (see Log scales).

The message type

Every message is a Message. It holds its data and records two flags:

  • clamped: the message is computed from constants and observations only;
  • initial: the message was set before inference, or computed from initial values.

The product of messages derives both flags from its sides'. A message also carries its log scale and a ReactiveMP.AnnotationDict of optional metadata (see Annotations).

tracked = Message(NormalMeanVariance(0.0, 1.0), false, false, ReactiveMP.AnnotationDict(), -1.5)
tracked == message, getlogscale(tracked)
(true, -1.5)
Equality ignores annotations and log scales

== on Message and on Marginal compares the data and the two flags, not the annotations or the log scale. These describe how a value was computed, not the belief it holds. Compare getannotations and getlogscale explicitly where they matter, such as in a cache key.

ReactiveMP.Message — Type
Message(data, is_clamped::Bool, is_initial::Bool)
Message(data, is_clamped::Bool, is_initial::Bool, annotations::AnnotationDict)
Message(data, is_clamped::Bool, is_initial::Bool, annotations::AnnotationDict, logscale)

A message along an edge of the factor graph: its data, usually a distribution, with the flags and the metadata the engine tracks for it.

Arguments

  • data: the message itself, usually a distribution, and missing where it is not known, such as an unobserved data point;
  • is_clamped: whether the message comes from constants and observations alone. A clamped value is the result of computations on constants and observations alone: it never

changes during inference.

  • is_initial: whether the message was set before inference or computed from initial values only. An initial value was set before inference, with ReactiveMP.set_initial_message! or

ReactiveMP.set_initial_marginal!, or computed from clamped and initial values only, at least one of them initial.

  • annotations: the ReactiveMP.AnnotationDict of optional metadata (see ReactiveMP.AbstractAnnotations). By default the engine's shared empty annotations, which nothing may write to: a message meant to be annotated is given its own AnnotationDict();
  • logscale: the log of the constant the normalised data leaves out (see getlogscale): a number, an UndefinedLogScale with its reason, or nothing, the default, where log scales are not tracked.

The statistics of the data are forwarded: mean, median, mode, var, std, cov, invcov, precision, entropy, params, mean_cov, mean_var, mean_invcov, mean_precision, weightedmean and its variants, shape, scale, rate, probvec, logdetcov, length, ndims, size, pdf and logpdf.

== compares the data and the is_clamped and is_initial flags, not the annotations or the log scale: those describe how a value was computed, not the belief it holds. Compare getannotations and getlogscale explicitly where they matter.

Examples

julia> message = Message(Gamma(10.0, 2.0), false, true)
Message(Distributions.Gamma{Float64}(α=10.0, θ=2.0))

julia> mean(message)
20.0

julia> is_clamped(message), is_initial(message)
(false, true)

julia> a = Message(NormalMeanVariance(0.0, 1.0), false, false);

julia> b = Message(NormalMeanVariance(0.0, 1.0), false, false, ReactiveMP.AnnotationDict(), -0.5);

julia> a == b, getlogscale(b)
(true, -0.5)

See also Marginal, DeferredMessage, as_message.

source
ReactiveMP.is_clamped — Method
is_clamped(message::Message) -> Bool

Whether message comes from constants and observations alone. A clamped value is the result of computations on constants and observations alone: it never changes during inference.

source
ReactiveMP.as_message — Function
as_message(message::AbstractMessage) -> Message
as_message(marginal::Marginal) -> Message

The Message a value stands for: a Message itself; the message a DeferredMessage computes, computed on the first call and cached for the next; or a Marginal's data, flags, annotations and log scale as a message.

source

Message streams

The engine does not compute a message once and store it. Each connection between a variable and a node carries a stream of messages, a ReactiveMP.MessageObservable, which emits a new message whenever its inputs change. The stream is lazy until activation connects it to its source. Before that, ReactiveMP.set_initial_message! can seed it, so that a rule that reads it at the start has something to read. The stream keeps its latest message: a subscriber that joins late receives it at once.

ReactiveMP.MessageObservable — Type
MessageObservable(M::Type = AbstractMessage)

The stream of the messages along one connection between a variable and a factor node, with values of type M. Every subscriber shares one upstream subscription, and the latest message is kept, so Rocket.getrecent returns it and a late subscriber receives it at once.

The stream is lazy: activation connects it to its source, the node's rule or the variable's product. Before that, ReactiveMP.set_initial_message! can seed it with an initial message, which is what a rule reads before any message has been computed.

A random and a data variable allocate one per connection (ReactiveMP.create_new_stream_of_inbound_messages!); a constant has one, shared by all its connections.

See also ReactiveMP.MarginalObservable.

source

Product of messages

A variable multiplies its inbound messages: all of them for its marginal, and all but one for each outbound message. The product of two messages is BayesBase.prod of their distributions, under a product strategy. It is in general not normalised, and its log scale accounts for the constant.

A ReactiveMP.MessageProductContext holds the settings: the product strategy, the form constraint and when it applies, the order of the fold, the annotation processors and the callbacks.

messages = [
    Message(NormalMeanVariance(0.0, 10.0), true, false),
    Message(NormalMeanVariance(2.0, 1.0), false, true),
]
product = ReactiveMP.compute_product_of_messages(randomvar(), MessageProductContext(), messages)
getdata(product), is_clamped(product), is_initial(product)
(ExponentialFamily.NormalWeightedMeanPrecision{Float64}(xi=2.0, w=1.1), false, true)

One side is clamped and the other initial, so the product is initial.

ReactiveMP.MessageProductContext — Type
MessageProductContext(; prod_constraint = GenericProd(), form_constraint = UnspecifiedFormConstraint(), form_constraint_check_strategy = FormConstraintCheckLast(), fold_strategy = MessagesProductFromLeftToRight(), annotations = nothing, callbacks = nothing)

How a variable multiplies messages: the settings ReactiveMP.compute_product_of_messages and ReactiveMP.compute_product_of_two_messages read. A random variable holds two, one for its outbound messages and one for its marginal (see RandomVariableActivationOptions).

Keywords

RxInfer builds it from a variable's form constraint, with default_prod_constraint and default_form_check_strategy of that constraint.

See also ReactiveMP.compute_product_of_messages, ReactiveMP.compute_product_of_two_messages.

source
ReactiveMP.compute_product_of_two_messages — Function
compute_product_of_two_messages(variable::AbstractVariable, context::MessageProductContext, left, right) -> Message

Multiply two messages for variable, with BayesBase.prod under context.prod_constraint. A DeferredMessage is computed first (see as_message). The product is a Message, not necessarily normalised; context.form_constraint applies to it here when the strategy is FormConstraintCheckEach.

The product is clamped when both messages are, and initial when it is not clamped and each side is clamped or initial.

Its log scale is the sum of the two messages' log scales and the product's own, BayesBase.compute_logscale(new, left, right). A missing side leaves the other side's; either side's nothing, log scales not tracked, gives nothing. It is undefined, an UndefinedLogScale, when either side's is, when the pair has no compute_logscale, and when a form constraint changes the product, returning something other than it was given.

Its annotations are merged by the processors in context.annotations (see ReactiveMP.post_product_annotations!).

variable names the variable the product is for; it reaches the callbacks, which receive ReactiveMP.BeforeProductOfTwoMessagesEvent and ReactiveMP.AfterProductOfTwoMessagesEvent, and, with the strategy FormConstraintCheckEach, the form constraint events.

See also ReactiveMP.MessageProductContext, ReactiveMP.compute_product_of_messages.

source
ReactiveMP.compute_product_of_messages — Function
compute_product_of_messages(variable::AbstractVariable, context::MessageProductContext, messages) -> Message

Multiply a collection of messages for variable, pairwise with ReactiveMP.compute_product_of_two_messages, in the order context.fold_strategy gives. With the strategy FormConstraintCheckLast, context.form_constraint applies once, to the whole product, and a form constraint that changes it leaves its log scale undefined.

The callbacks receive ReactiveMP.BeforeProductOfMessagesEvent and ReactiveMP.AfterProductOfMessagesEvent around the whole product.

Examples

julia> x = randomvar();

julia> messages = (Message(NormalMeanVariance(0.0, 1.0), false, false), Message(NormalMeanVariance(2.0, 1.0), false, false));

julia> product = ReactiveMP.compute_product_of_messages(x, ReactiveMP.MessageProductContext(), messages);

julia> mean(product) ≈ 1.0 && var(product) ≈ 0.5
true

See also ReactiveMP.MessageProductContext, ReactiveMP.MessagesProductFromLeftToRight.

source
compute_product_of_messages(f::Function, variable::AbstractVariable, context::MessageProductContext, messages)

The product under a custom fold strategy, context.fold_strategy = f: calls f(variable, context, messages), which multiplies the messages with ReactiveMP.compute_product_of_two_messages in whatever order it chooses and returns the product.

source

Deferred messages

A factor node emits its messages deferred. The engine computes each when a variable first reads it, and caches it, so that a message nobody reads is never computed. as_message computes a deferred message.

ReactiveMP.DeferredMessage — Type
DeferredMessage(messages, marginals, mapping)

A message computed on demand: what a factor node emits on an outbound stream, so that a message nobody reads is never computed. as_message computes it on the first call, as mapping(getrecent(messages), getrecent(marginals)), from the latest values of its inputs at that time, and caches the resulting Message for the next calls.

Arguments

  • messages: the stream of the rule's inbound messages, or nothing for none;
  • marginals: the stream of the rule's marginals, or nothing for none;
  • mapping: a function of the two, usually a ReactiveMP.MessageMapping, that returns the Message.

A variable computes the deferred messages it receives when it multiplies them.

source

Message mappings

A ReactiveMP.MessageMapping computes a node's message towards one interface. It resolves the rule for the latest inputs, checks it, and runs it. When an input is missing, it gives missing without running a rule. It keeps the scratch of the rule it runs between calls. The callback events of a rule call carry it, so a callback reads the node, the target and the algorithm from it:

x, y = randomvar(label = :x), datavar(label = :y)
prior = factornode(Gaussian, [(:out, x), (:μ, constvar(0.0)), (:v, constvar(10.0))])
likelihood = factornode(Gaussian, [(:out, y), (:μ, x), (:v, constvar(1.0))])

mappings = []
callbacks = (before_message_rule_call = (event) -> push!(mappings, event.mapping),)
activate!(x, RandomVariableActivationOptions())
activate!(y, DataVariableActivationOptions())
foreach(n -> activate!(n, FactorNodeActivationOptions(; callbacks)), (prior, likelihood))

subscription = subscribe!(get_stream_of_marginals(x), (q) -> nothing)
new_observation!(y, 2.0)
unsubscribe!(subscription)
mappings
2-element Vector{Any}:
 MessageMapping(Gaussian, :out, msgs=[:μ, :v], algorithm=MessagePassingRulesBase.DefaultAlgorithm())
 MessageMapping(Gaussian, :μ, msgs=[:out, :v], algorithm=MessagePassingRulesBase.DefaultAlgorithm())

Each mapping names its node, its target and the messages its rule takes.

ReactiveMP.MessageMapping — Type
MessageMapping

What computes a node's message towards one interface, called with the latest inbound messages and marginals the rule depends on: mapping(messages, marginals) -> Message. A factor node builds one per interface at activation and wraps it in each DeferredMessage it emits.

It holds the node type, the rule's target, the names of its inputs, the algorithm, the annotation processors, the factor node, the callbacks, the ReactiveMP.EngineDiagnostics, the rule context (see ReactiveMP.node_context), the rule fallback, the rule's scratch (reused from call to call) and whether log scales are tracked.

A call:

  1. returns a missing message, and runs no rule, when an input is missing, or a static input folded into the node's function is data observed as missing;
  2. resolves the rule with find_message_rule under the node's algorithm, from the inputs' types;
  3. where no rule matches, gives the rule fallback's message, if one is set, with an undefined log scale; otherwise throws a RuleNotFoundError listing the near misses;
  4. checks the rule: against the diagnostics, for a declared reading of log scales that are not tracked, and for a declared context service the node's context does not supply (check_services);
  5. runs it with execute_rule, with the annotation processors before and after it.

The message is clamped when every input is, and initial when it is not clamped and every input is clamped or initial. When log scales are tracked, its log scale is the one the rule declares.

The callbacks receive ReactiveMP.BeforeMessageRuleCallEvent and ReactiveMP.AfterMessageRuleCallEvent around each call, the latter also for a missing message, with what gave the message.

See also Message, ReactiveMP.rule_arguments.

source

A mapping hands the rule the data of the messages and marginals it depends on, keyed as the rule declares them, with their annotations alongside.

ReactiveMP.rule_arguments — Function
ReactiveMP.rule_arguments(messages_names, messages, marginals_names, marginals)
ReactiveMP.rule_arguments(messages_names, messages, marginals_names, marginals, logscales::Val)

The RuleArgs of a rule call: the data of the messages and marginals it depends on, keyed as the rule declares them, args.m[:x] and args.q[:x] (see ReactiveMP.rule_messages and ReactiveMP.rule_marginals). With logscales = Val(true) it also holds the log scales the messages arrived with, read as args.logscale.m[:x]; Val(false) holds none.

source