symfonic.capabilities.memory.retrieval

retrieval

Public vocabulary for composing native memory retrieval.

The hub keeps the broad established memory API. This smaller surface is for an adopter that supplies a candidate source or needs to state its ranking policy explicitly.

Candidate dataclass

Candidate(record: MemoryRecord, signals: RelevanceSignals = RelevanceSignals(), pre_ranked: bool = False)

One memory a store found, with whatever it observed about it.

The signals are the store's honest report, including its silences: a layer with no embedding leaves cue at None, and the coordinator treats that as "cannot answer" rather than "answered zero".

CandidateSource

Bases: Protocol

A store, as the retrieval coordinator needs it.

search async

search(query: MemoryQuery, layers: frozenset[MemoryLayer]) -> Sequence[Candidate]

Return what this store can find for query within layers.

layers is the effective set — the deployment's enabled layers narrowed by the query's — passed so a store can skip work rather than return rows the coordinator will drop. The coordinator re-checks what comes back regardless: a source that answers outside the set is a bug this seam should surface, not one it should hide.

Ranking, limits, and ceilings are not this method's business. Return everything plausible; the coordinator decides what reaches a prompt.

Raises :class:~.errors.MemoryUnavailable when the store is unreachable. Returning an empty sequence instead would be indistinguishable from a scope that genuinely remembers nothing.

Source code in src/symfonic/capabilities/memory/sources.py

async def search(
    self, query: MemoryQuery, layers: frozenset[MemoryLayer]
) -> Sequence[Candidate]:
    """Return what this store can find for ``query`` within ``layers``.

    ``layers`` is the *effective* set — the deployment's enabled layers
    narrowed by the query's — passed so a store can skip work rather than
    return rows the coordinator will drop. The coordinator re-checks what
    comes back regardless: a source that answers outside the set is a bug
    this seam should surface, not one it should hide.

    Ranking, limits, and ceilings are not this method's business. Return
    everything plausible; the coordinator decides what reaches a prompt.

    Raises :class:`~.errors.MemoryUnavailable` when the store is unreachable.
    Returning an empty sequence instead would be indistinguishable from a
    scope that genuinely remembers nothing.
    """
    ...

LexicalCandidateSource dataclass

LexicalCandidateSource(records: tuple[MemoryRecord, ...] = ())

A complete candidate source over records held in this process.

Scores cue overlap with token arithmetic rather than an embedding, because this is the zero-dependency default: a source that needed a model would make "retrieval works out of the box" false. Salience is left to the coordinator, which reads it off the record — a source should not restate what the record already carries.

RelevanceSignals dataclass

RelevanceSignals(cue: float | None = None, salience: float | None = None, recency: float | None = None, frequency: float | None = None, proximity: float | None = None)

What a backend observed about one memory. None means "no such signal".

The distinction between 0.0 and None is the whole point: 0.0 is an observation ("this memory does not match the cue"), None is the absence of one ("this layer cannot answer that question").

present

present() -> Mapping[str, float]

The signals this source actually produced, in declaration order.

Source code in src/symfonic/capabilities/memory/relevance.py

def present(self) -> Mapping[str, float]:
    """The signals this source actually produced, in declaration order."""
    return {
        name: value
        for name in SIGNAL_NAMES
        if (value := getattr(self, name)) is not None
    }

RelevanceWeights dataclass

RelevanceWeights(cue: float = 1.0, salience: float = 0.5, recency: float = 0.3, frequency: float = 0.2, proximity: float = 0.2)

What a deployment believes each signal is worth.

Defaults lead with the cue, because the question a user just asked is the strongest evidence about what they want remembered, and keep salience second so a cold-start turn with no cue still surfaces a profile rather than noise.

compose

compose(signals: RelevanceSignals) -> float

The weighted mean of the signals that are present, in [0, 1].

Signals the source did not produce are absent from both the numerator and the denominator, so a layer that can only report salience is scored on salience rather than punished for its silence.

Source code in src/symfonic/capabilities/memory/relevance.py

def compose(self, signals: RelevanceSignals) -> float:
    """The weighted mean of the signals that are present, in ``[0, 1]``.

    Signals the source did not produce are absent from *both* the numerator
    and the denominator, so a layer that can only report salience is scored
    on salience rather than punished for its silence.
    """
    total = 0.0
    weighted = 0.0
    for name, value in signals.present().items():
        weight = getattr(self, name)
        total += weight
        weighted += weight * value
    return weighted / total if total > 0.0 else 0.0

RetrievalCoordinator

RetrievalCoordinator(*, source: CandidateSource, policy: RetrievalPolicy | None = None, activation: SpreadingActivation | None = None)

Owns layers, scoring, gates, scope blending, and activation.

Source code in src/symfonic/capabilities/memory/coordinator.py

def __init__(
    self,
    *,
    source: CandidateSource,
    policy: RetrievalPolicy | None = None,
    activation: SpreadingActivation | None = None,
) -> None:
    self._source = source
    self._policy = policy or RetrievalPolicy()
    self._activation = activation

coordinate async

coordinate(query: MemoryQuery) -> CoordinatedRetrieval

Retrieve for query, returning the result and its provenance.

Source code in src/symfonic/capabilities/memory/coordinator.py

async def coordinate(self, query: MemoryQuery) -> CoordinatedRetrieval:
    """Retrieve for ``query``, returning the result and its provenance."""
    layers = self._policy.effective_layers(query)
    if not layers:
        return CoordinatedRetrieval(
            result=RetrievalResult(dropped=((NO_RECORD, _no_layers(query)),)),
            layers=layers,
        )

    candidates = await self._source.search(query, layers)
    # A source may refuse rows before this coordinator ever sees them --
    # an adapter over legacy storage drops what it cannot attribute to a
    # scope. Those refusals belong in the same ledger as the ones the gate
    # makes, or a refused row is indistinguishable from a scope that
    # remembers nothing. Optional because ``CandidateSource`` does not
    # require it: a source with nothing to report simply has no drain.
    drain = getattr(self._source, "drain_dropped", None)
    refused = tuple(drain()) if callable(drain) else ()
    # The same shape for the route tally: optional, because
    # ``CandidateSource`` does not require it, and drained rather than read
    # so one turn's recall is never attributed to the next.
    tally = getattr(self._source, "drain_sources", None)
    sources, unavailable = tally() if callable(tally) else ({}, ())
    kept, dropped = self._gate(tuple(candidates), query, layers)
    found, log = await self._expand(query.scope, kept)

    ranked = select((*kept, *found), query)
    return CoordinatedRetrieval(
        result=RetrievalResult(
            memories=ranked.memories,
            dropped=(*refused, *dropped, *ranked.dropped),
            sources=sources,
            unavailable=unavailable,
        ),
        activation=log,
        layers=layers,
    )

retrieve async

retrieve(query: MemoryQuery) -> RetrievalResult

The :class:~.ports.MemoryRetrievalPort surface: just the result.

Source code in src/symfonic/capabilities/memory/coordinator.py

async def retrieve(self, query: MemoryQuery) -> RetrievalResult:
    """The :class:`~.ports.MemoryRetrievalPort` surface: just the result."""
    return (await self.coordinate(query)).result

RetrievalPolicy dataclass

RetrievalPolicy(enabled_layers: frozenset[MemoryLayer] = RETRIEVABLE_LAYERS, weights: RelevanceWeights = DEFAULT_WEIGHTS, blend: ScopeBlend = ScopeBlend(), min_relevance: float = 0.0, min_salience: float = 0.0, profile_slots: int = 1)

The deployment's retrieval decisions, in one value.

Frozen and validated at construction so a deployment that mis-states a gate fails at wiring time rather than by quietly retrieving nothing on every turn — the failure mode a floor of 1.5 actually has.

effective_layers

effective_layers(query: MemoryQuery) -> frozenset[MemoryLayer]

The layers actually searched: what is enabled, narrowed by the query.

Source code in src/symfonic/capabilities/memory/coordinator.py

def effective_layers(self, query: MemoryQuery) -> frozenset[MemoryLayer]:
    """The layers actually searched: what is enabled, narrowed by the query."""
    return self.enabled_layers & query.layers

ScopeBlend dataclass

ScopeBlend(mode: ScopeBlendMode = ScopeBlendMode.OFF, half_life_levels: float = 1.0, floor: float = 0.0)

How hierarchy distance is folded into the rank score.

gate

gate(distance: int) -> float

The multiplier applied to a memory distance levels above the query.

Source code in src/symfonic/capabilities/memory/relevance.py

def gate(self, distance: int) -> float:
    """The multiplier applied to a memory ``distance`` levels above the query."""
    if distance < 0:
        raise MemoryContractError(
            f"scope blend was asked to gate distance {distance}; a negative distance is "
            "MemoryScope.distance's 'not visible from here', and scoring it would rank a "
            "memory the query cannot see."
        )
    if distance == 0 or self.mode is ScopeBlendMode.OFF:
        return 1.0
    decay = math.exp(-_LN2 / self.half_life_levels * distance)
    return self.floor + (1.0 - self.floor) * decay

rank_score

rank_score(score: float, distance: int) -> float

Fold distance into score, staying inside [0, 1].

Under :attr:ScopeBlendMode.BACKFILL the two bands are disjoint — tips land in [0.5, 1] and ancestors in [0, 0.5) — so "every tip first" is a property of the number rather than of a second sort pass that some caller might skip. Ordering within each band is preserved, and the gate still runs inside the ancestor band so a parent outranks a grandparent.

Source code in src/symfonic/capabilities/memory/relevance.py

def rank_score(self, score: float, distance: int) -> float:
    """Fold ``distance`` into ``score``, staying inside ``[0, 1]``.

    Under :attr:`ScopeBlendMode.BACKFILL` the two bands are disjoint — tips
    land in ``[0.5, 1]`` and ancestors in ``[0, 0.5)`` — so "every tip
    first" is a property of the number rather than of a second sort pass
    that some caller might skip. Ordering *within* each band is preserved,
    and the gate still runs inside the ancestor band so a parent outranks a
    grandparent.
    """
    gated = score * self.gate(distance)
    if self.mode is not ScopeBlendMode.BACKFILL:
        return gated
    if distance == 0:
        return _TIP_BAND + _TIP_BAND * gated
    return _TIP_BAND * gated

ScopeBlendMode

Bases: StrEnum

How an ancestor-scoped memory competes with a memory written here.

BACKFILL class-attribute instance-attribute

BACKFILL = 'backfill'

Ancestors rank strictly below every tip memory: they fill leftover slots.

BLEND class-attribute instance-attribute

BLEND = 'blend'

Distance attenuates the score. A strong ancestor can still beat a weak tip.

OFF class-attribute instance-attribute

OFF = 'off'

Distance does not affect the score. A tenant fact competes on merit.