symfonic.capabilities.memory.activation

activation

Spreading activation: the memories a turn recalls without asking for them.

Every other retrieval path answers a question. This one follows edges the query never named, which makes it the most useful part of a graph memory and the most dangerous. Two consequences shape the whole module:

• It is the easiest way out of a tenant. A cue can only match what the store was asked for; an edge arrives from the backend and points wherever it points. So every activated memory's scope is re-checked here against the query's, and a crossing is a :class:~.errors.ScopeViolation — not a low score, not a dropped candidate.
• It has to be explainable. Every activated record leaves with its origin rewritten to name the seed and the hop that reached it, and the :class:~.provenance.ActivationLog keeps the edges and paths that got there. The vocabulary for both lives in :mod:.provenance.

Degradation is asymmetric on purpose. An unreachable association source returns an empty, degraded log and the turn keeps its recall: by the time expansion runs, the memories that answer the query are already in hand, and failing the turn to protect their neighbours trades the answer for the garnish. A scope violation still propagates — it is not a transport problem.

Determinism. Each hop's discoveries are admitted in (-score, record_id) order, so a backend that yields equally-weighted edges in a different order across two runs still produces the same log, the same caps, and the same prompt bytes. Without that, an unchanged turn changes its cache digest for no reason.

ActivatedNode dataclass

ActivatedNode(record_id: str, label: str, layer: MemoryLayer, score: float, hop: int = 0, source_id: str = '', relationship: str = '')

One memory the activation pass lit up, and how it got there.

of classmethod

of(record: MemoryRecord, *, score: float, hop: int = 0, source_id: str = '', relationship: str = '') -> ActivatedNode

Build a node from the memory it stands for, with a bounded label.

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

@classmethod
def of(
    cls,
    record: MemoryRecord,
    *,
    score: float,
    hop: int = 0,
    source_id: str = "",
    relationship: str = "",
) -> ActivatedNode:
    """Build a node from the memory it stands for, with a bounded label."""
    label = flatten(record.text).strip()
    if len(label) > MAX_LABEL_CHARS:
        label = label[: MAX_LABEL_CHARS - 1].rstrip() + "…"
    return cls(
        record_id=record.record_id,
        label=label,
        layer=record.layer,
        score=score,
        hop=hop,
        source_id=source_id,
        relationship=relationship,
    )

ActivationEdge dataclass

ActivationEdge(source_id: str, target_id: str, relationship: str = 'associated', weight: float = 1.0)

One traversal, kept so a later correction pass can see what was believed.

ActivationLog dataclass

ActivationLog(nodes: tuple[ActivatedNode, ...] = (), edges: tuple[ActivationEdge, ...] = (), paths: tuple[tuple[str, ...], ...] = (), truncated: bool = False, degraded: bool = False)

The provenance of one activation pass.

Returned from the walk rather than stored on it, so two concurrent turns cannot read each other's expansion.

Association dataclass

Association(source_id: str, target: MemoryRecord, relationship: str = 'associated', weight: float = 1.0)

One edge out of a memory, as a backend reports it.

AssociationSource

Bases: Protocol

The graph half of the memory system, as activation needs it.

neighbours async

neighbours(scope: MemoryScope, record_ids: tuple[str, ...]) -> Sequence[Association]

Every edge out of record_ids, one round trip per frontier.

Takes the whole frontier rather than one id so a hop costs one query instead of one per seed. Raises :class:~.errors.MemoryUnavailable when the graph cannot be reached.

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

async def neighbours(
    self, scope: MemoryScope, record_ids: tuple[str, ...]
) -> Sequence[Association]:
    """Every edge out of ``record_ids``, one round trip per frontier.

    Takes the whole frontier rather than one id so a hop costs one query
    instead of one per seed. Raises :class:`~.errors.MemoryUnavailable` when
    the graph cannot be reached.
    """
    ...

SpreadingActivation dataclass

SpreadingActivation(source: AssociationSource, max_hops: int = 1, decay: float = 0.5, max_nodes: int = 10)

Expands a recall through the association graph, with decay and a cap.

expand async

expand(scope: MemoryScope, seeds: Sequence[RetrievedMemory]) -> tuple[tuple[RetrievedMemory, ...], ActivationLog]

Walk out from seeds and return what lit up, plus the provenance.

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

async def expand(
    self, scope: MemoryScope, seeds: Sequence[RetrievedMemory]
) -> tuple[tuple[RetrievedMemory, ...], ActivationLog]:
    """Walk out from ``seeds`` and return what lit up, plus the provenance."""
    if not seeds or self.max_hops == 0:
        return (), ActivationLog()

    state = _Frontier(seen={m.record.record_id for m in seeds})
    for memory in seeds:
        state.nodes.append(ActivatedNode.of(memory.record, score=_unit(memory.score)))
        state.trail[memory.record.record_id] = (memory.record.record_id,)

    frontier = {m.record.record_id: _unit(m.score) for m in seeds}
    for hop in range(1, self.max_hops + 1):
        try:
            edges = await self.source.neighbours(scope, tuple(frontier))
        except MemoryUnavailable:
            return tuple(state.found), _log(state, degraded=True)
        frontier = self._admit(state, edges, frontier, hop, scope)
        if not frontier:
            break

    return tuple(state.found), _log(state)

activation_origin

activation_origin(source_id: str, hop: int) -> str

The provenance stamped on a memory that activation reached.

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

def activation_origin(source_id: str, hop: int) -> str:
    """The provenance stamped on a memory that activation reached."""
    return f"activation:{source_id}:hop{hop}"