GitHub

Annotations

A message or a marginal holds a distribution. Its annotations are optional metadata that travel with it: values keyed by Symbol, such as a record of the inputs a message was computed from. They cost nothing when unused. A message or marginal that nothing may annotate shares one frozen, empty dictionary, and the engine gives a fresh one only to what may write to it: the annotation processors, and a rule that takes the ann slot. A message's log scale is not an annotation, but part of the message.

The examples use the Gaussian node of The example node.

The annotation dictionary

Every message and marginal holds a ReactiveMP.AnnotationDict, which getannotations returns:

ann = ReactiveMP.AnnotationDict()
ReactiveMP.annotate!(ann, :source, "sensor 1")
ReactiveMP.has_annotation(ann, :source), ReactiveMP.get_annotation(ann, String, :source)
(true, "sensor 1")

The engine's functions read and write it. A rule does the same through the base package's functions, getannotation, hasannotation and annotate!, which the engine implements for it.

ReactiveMP.AnnotationDict — Type
ReactiveMP.AnnotationDict()
ReactiveMP.AnnotationDict(other::AnnotationDict)

The annotations of a message or a marginal: values keyed by Symbol, metadata about how it was computed. It allocates nothing until the first write. The second form is a shallow copy of other.

A message or marginal that nothing may annotate shares one frozen, empty AnnotationDict with every other such one, which ReactiveMP.annotate! refuses to write to. The engine hands a fresh one to whatever may write: the annotation processors of a mapping or a product, and a rule whose body takes the ann slot.

A rule reads and writes it through the base package's functions, getannotation, hasannotation and annotate!; the engine's own are ReactiveMP.get_annotation, ReactiveMP.has_annotation and ReactiveMP.annotate!.

Examples

julia> ann = ReactiveMP.AnnotationDict();

julia> ReactiveMP.annotate!(ann, :count, 1);

julia> ReactiveMP.has_annotation(ann, :count), ReactiveMP.get_annotation(ann, :count)
(true, 1)
source
MessagePassingRulesBase.annotate! — Method
ReactiveMP.annotate!(ann::AnnotationDict, key::Symbol, value) -> Nothing

Store value under key in ann, replacing any value already there.

Throws

  • ArgumentError when ann is the shared empty annotations of a message or marginal that nothing may annotate (see ReactiveMP.AnnotationDict).
source
ReactiveMP.get_annotation — Function
ReactiveMP.get_annotation(ann::AnnotationDict, key::Symbol)
ReactiveMP.get_annotation(ann::AnnotationDict, ::Type{T}, key::Symbol) -> T

The value stored under key in ann; with a type T, converted to T.

Throws

  • KeyError when ann holds nothing under key.
source

Annotation processors

An annotation processor, a subtype of ReactiveMP.AbstractAnnotations, decides what is written. The engine calls it at three points:

You give a processor to both places it acts in: to the factor nodes, as the activation option annotations of ReactiveMP.FactorNodeActivationOptions, and to the variables, as the annotations of their ReactiveMP.MessageProductContexts. Without the second, a product's annotations are empty.

ReactiveMP.AbstractAnnotations — Type
ReactiveMP.AbstractAnnotations

The supertype of annotation processors, which write annotations on messages. A processor implements three hooks:

Processors reach the rules through the activation option annotations of ReactiveMP.FactorNodeActivationOptions, and the products through the annotations of a variable's ReactiveMP.MessageProductContext: a processor is given to both. The built-in processor is InputArgumentsAnnotations.

source
ReactiveMP.pre_rule_annotations! — Function
ReactiveMP.pre_rule_annotations!(processor::AbstractAnnotations, ann::AnnotationDict, mapping, messages, marginals)

The hook of processor before a message rule runs: write into ann, the new message's annotations, from the ReactiveMP.MessageMapping and the rule's inbound messages and marginals. It runs for a missing message too. Its return value is ignored.

source
ReactiveMP.post_rule_annotations! — Function
ReactiveMP.post_rule_annotations!(processor::AbstractAnnotations, ann::AnnotationDict, mapping, messages, marginals, result)

The hook of processor after a message rule ran: write into ann, the new message's annotations, from the ReactiveMP.MessageMapping, the rule's inputs and result, the message's data. It is skipped for a missing message, when no rule ran. Its return value is ignored.

source
ReactiveMP.post_product_annotations! — Function
ReactiveMP.post_product_annotations!(processor::AbstractAnnotations, merged::AnnotationDict, left_ann::AnnotationDict, right_ann::AnnotationDict, new_dist, left_dist, right_dist)

The hook of processor at a product of two messages: write into merged, the product's empty annotations, from the two messages' annotations and their distributions. Its return value is ignored.

source

A custom processor

The processor below counts the rule calls a message was computed from. After a rule, the count is one. A product adds the counts of its two sides:

import ReactiveMP: AbstractAnnotations, AnnotationDict, has_annotation, get_annotation, annotate!

struct CountAnnotations <: AbstractAnnotations end

ReactiveMP.pre_rule_annotations!(::CountAnnotations, ann::AnnotationDict, mapping, messages, marginals) = nothing

function ReactiveMP.post_rule_annotations!(::CountAnnotations, ann::AnnotationDict, mapping, messages, marginals, result)
    annotate!(ann, :count, 1)
    return nothing
end

function ReactiveMP.post_product_annotations!(::CountAnnotations, merged::AnnotationDict, left::AnnotationDict, right::AnnotationDict, new_dist, left_dist, right_dist)
    count(ann) = has_annotation(ann, :count) ? get_annotation(ann, Int, :count) : 0
    annotate!(merged, :count, count(left) + count(right))
    return nothing
end

The function below builds a latent x with a normal prior, observed through a second node, gives the processors to every node and to x, and returns the posterior's annotations:

function posterior_annotations(processors)
    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))])
    context = MessageProductContext(; annotations = processors)
    activate!(x, RandomVariableActivationOptions(nothing, context, context))
    activate!(y, DataVariableActivationOptions())
    foreach(n -> activate!(n, FactorNodeActivationOptions(; annotations = processors)), (prior, likelihood))
    posteriors = Marginal[]
    subscription = subscribe!(get_stream_of_marginals(x), (q) -> push!(posteriors, q))
    new_observation!(y, 2.0)
    unsubscribe!(subscription)
    return getannotations(last(posteriors))
end

posterior_annotations((CountAnnotations(),))
AnnotationDict(count => 2)

The posterior is the product of two messages, each computed by one rule call.

Input arguments

InputArgumentsAnnotations records what went into each rule call: the ReactiveMP.MessageMapping (the node, the target and the algorithm), the inbound messages, the marginals and the result. It carries the record through the products of messages, so the provenance of a message travels with it. After a rule runs, a RuleInputArgumentsRecord is stored under the key :rule_input_arguments. A product merges the two sides' records into a ProductInputArgumentsRecord, a flat list, however deeply the products nest. get_rule_input_arguments reads the record:

record = get_rule_input_arguments(posterior_annotations((InputArgumentsAnnotations(),)))
Product of 2 rule input arguments:
  [1]
    Rule input arguments:
      node:       Gaussian
      target:     :out
      algorithm:  MessagePassingRulesBase.DefaultAlgorithm()
      msg(μ) = Message(BayesBase.PointMass{Float64}(0.0)) with logscale = 0
      msg(v) = Message(BayesBase.PointMass{Float64}(10.0)) with logscale = 0
      result:     ExponentialFamily.NormalMeanVariance{Float64}(μ=0.0, v=10.0)
  [2]
    Rule input arguments:
      node:       Gaussian
      target:     :μ
      algorithm:  MessagePassingRulesBase.DefaultAlgorithm()
      msg(out) = Message(BayesBase.PointMass{Float64}(2.0)) with logscale = 0
      msg(v) = Message(BayesBase.PointMass{Float64}(1.0)) with logscale = 0
      result:     ExponentialFamily.NormalMeanVariance{Float64}(μ=2.0, v=1.0)

A ProductInputArgumentsRecord lists its rule calls in mappings, each a RuleInputArgumentsRecord:

[(r.mapping.target, r.result) for r in record.mappings]
2-element Vector{Tuple{MessagePassingRulesBase.Target, ExponentialFamily.NormalMeanVariance{Float64}}}:
 (:out, ExponentialFamily.NormalMeanVariance{Float64}(μ=0.0, v=10.0))
 (:μ, ExponentialFamily.NormalMeanVariance{Float64}(μ=2.0, v=1.0))
ReactiveMP.InputArgumentsAnnotations — Type
InputArgumentsAnnotations()

The annotation processor that records what each message rule call received and returned, and carries the record through the products of messages: a message's provenance, for debugging and for callbacks. After a rule runs, it stores a RuleInputArgumentsRecord under the key :rule_input_arguments; a product merges the two sides' records into a ProductInputArgumentsRecord, or keeps the one side's that has one. Read the record with get_rule_input_arguments.

It is given to the node's activation, FactorNodeActivationOptions(; annotations = (InputArgumentsAnnotations(),)), and to the variables' MessageProductContext(; annotations = (InputArgumentsAnnotations(),)), so that products carry the records on.

source