symfonic.capabilities.memory.recall

recall

The vocabulary a deployment needs to compose memory recall.

memory_capabilities(..., activation=...) accepts a :class:SpreadingActivation, which follows graph associations after the direct query recall. This surface publishes the values needed to construct that collaborator without inspecting memory internals: :class:Association describes an edge, :class:AssociationSource supplies a frontier's edges, and :class:SpreadingActivation applies the bounded traversal.

The conversation-window types live beside them because they are the other public recall collaborator. A kernel-native host supplies its own :class:ConversationSource to form a :class:WorkingWindow; both the working window and graph expansion contribute to one hydrated recall.

Retrieval and hydration own ranking, floors, scope blend, spreading activation, and provenance that the former facade decided inline. This is a curated public surface rather than a mirror of memory internals, so a composition imports the collaborators it needs without making the broad hub grow for each recall seam.

The names remain available from :mod:symfonic.capabilities.memory for compatibility. New compositions should import this focused surface.

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.
    """
    ...

ConversationSource

Bases: Protocol

The working layer, as hydration needs it.

recent async

recent(scope: MemoryScope, limit: int) -> Sequence[ConversationTurn]

The last limit turns at scope, oldest first.

Raises :class:~.errors.MemoryUnavailable when the working store cannot be reached; :class:WorkingWindow degrades rather than failing the turn.

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

async def recent(self, scope: MemoryScope, limit: int) -> Sequence[ConversationTurn]:
    """The last ``limit`` turns at ``scope``, oldest first.

    Raises :class:`~.errors.MemoryUnavailable` when the working store cannot
    be reached; :class:`WorkingWindow` degrades rather than failing the turn.
    """
    ...

ConversationTurn dataclass

ConversationTurn(turn_id: str, speaker: str = '', text: str = '', turn: int = 0)

One thing that was said, as the working layer holds it.

line

line() -> str

The rendered form: layer prefix, speaker, single-line text.

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

def line(self) -> str:
    """The rendered form: layer prefix, speaker, single-line text."""
    body = flatten(self.text).strip()
    prefix = f"[{MemoryLayer.WORKING.value}]"
    return f"{prefix} {self.speaker}: {body}" if self.speaker else f"{prefix} {body}"

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)

WorkingContext dataclass

WorkingContext(turns: tuple[ConversationTurn, ...] = (), dropped: tuple[tuple[str, str], ...] = (), degraded: bool = False)

The conversation window as it will render, and what it left out.

render

render() -> str

One turn per line, oldest first.

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

def render(self) -> str:
    """One turn per line, oldest first."""
    return "\n".join(turn.line() for turn in self.turns)

WorkingWindow dataclass

WorkingWindow(source: ConversationSource, recent_turns: int = 0, exclude_speakers: frozenset[str] = frozenset())

Reads the last few turns of a conversation, ungated.

read async

read(scope: MemoryScope) -> WorkingContext

Read the window at scope, dropping only what would corrupt it.

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

async def read(self, scope: MemoryScope) -> WorkingContext:
    """Read the window at ``scope``, dropping only what would corrupt it."""
    if self.recent_turns == 0:
        return WorkingContext()
    try:
        recent = await self.source.recent(scope, self.recent_turns)
    except MemoryUnavailable:
        return WorkingContext(degraded=True)

    state = _Window()
    for turn in recent[-self.recent_turns :]:
        if not turn.text.strip():
            # An empty row renders as a bare "[working] " bullet, which reads
            # as a turn in which nobody said anything.
            state.dropped.append((turn.turn_id, "the turn carries no text"))
            continue
        if turn.speaker and turn.speaker in self.exclude_speakers:
            state.dropped.append(
                (turn.turn_id, f"speaker {turn.speaker!r} is excluded")
            )
            continue
        state.kept.append(turn)
    return WorkingContext(turns=tuple(state.kept), dropped=tuple(state.dropped))