symfonic.capabilities.memory.vector_recall

vector_recall

The vector half of recall: what is embedded, searched, and merged back.

Composed vector backend and embedder add semantic recall to the graph store. Metadata rides with each vector and is reconstructed into a memory record. Every hit is checked against graph visibility after backend selection.

VectorRecall

VectorRecall(vectors: Any, embedder: Any)

The embedder and the vector backend, as one collaborator of the store.

Together rather than separately, because neither is useful alone: an embedder with nowhere to put a vector writes nothing, and a vector backend with no embedder cannot be queried. Composing them as a pair is what makes "this deployment does semantic recall" a single yes or no.

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

def __init__(self, vectors: Any, embedder: Any) -> None:
    self._vectors = vectors
    self._embedder = embedder

forget async

forget(scope: Any, record_ids: tuple[str, ...]) -> None

Remove vector copies for record_ids under scope.

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

async def forget(self, scope: Any, record_ids: tuple[str, ...]) -> None:
    """Remove vector copies for ``record_ids`` under ``scope``."""
    if record_ids:
        await self._vectors.delete(scope, list(record_ids))

search async

search(scope: Any, query: MemoryQuery, visible: Any) -> tuple[tuple[RetrievedMemory, ...], bool]

Semantically similar memories for query, and whether it degraded.

Parameters:

Name	Type	Description	Default
visible	Any	a predicate answering whether a record's scope is one this query may read. Passed in rather than recomputed, so the vector path and the graph path apply one isolation rule -- two copies of that rule is how they start to disagree.	required

Returns ((), True) when the backend or the embedder could not be reached. Degraded, and reported as such, rather than an empty answer that reads like "nothing is similar".

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

async def search(
    self, scope: Any, query: MemoryQuery, visible: Any
) -> tuple[tuple[RetrievedMemory, ...], bool]:
    """Semantically similar memories for ``query``, and whether it degraded.

    Args:
        visible: a predicate answering whether a record's scope is one this
            query may read. Passed in rather than recomputed, so the vector
            path and the graph path apply *one* isolation rule -- two
            copies of that rule is how they start to disagree.

    Returns ``((), True)`` when the backend or the embedder could not be
    reached. Degraded, and reported as such, rather than an empty answer
    that reads like "nothing is similar".
    """
    admission = query.read_admission
    if admission is not None and not getattr(self._vectors, "supports_read_admission", False):
        raise MemoryContractError("vector backend does not support exact read admission")
    if not query.cue.strip():
        return (), False
    try:
        embedding = await self._embedder.embed(query.cue)
    except Exception as exc:  # noqa: BLE001 - default route degrades
        if admission is not None:
            raise MemoryContractError(
                f"vector embed failed under exact read admission: {exc}"
            ) from exc
        logger.warning("vector embed failed; recall stays lexical", exc_info=True)
        return (), True

    options = {}
    if admission is not None:
        options["allowed_scope_paths"] = tuple(
            storage_scope_path(scope_from_path(path))
            for path in admission.allowed_scope_paths
        )
    try:
        hits = await self._vectors.search(
            scope, list(embedding), top_k=query.candidate_limit, **options
        )
    except Exception as exc:  # noqa: BLE001 - strict mode must refuse
        if admission is not None:
            raise MemoryContractError(
                f"vector backend cannot enforce exact read admission: {exc}"
            ) from exc
        logger.warning("vector search failed; recall stays lexical", exc_info=True)
        return (), True

    found: list[RetrievedMemory] = []
    for hit in (hits or ())[:query.candidate_limit]:
        record = _record_from_hit(hit)
        if record is None or record.layer not in query.layers:
            continue
        distance = visible(record)
        if distance < 0:
            # The backend filtered by scope in SQL and is checked anyway:
            # the component that might be wrong is the one below, and a
            # vector route that answered without this check would be a
            # way around the isolation the graph enforces.
            continue
        similarity = float(hit.get("score", 0.0) or 0.0)
        found.append(
            RetrievedMemory(
                record=record,
                score=similarity * _SIMILARITY_WEIGHT,
                scope_distance=distance,
            )
        )
    return tuple(found), False

merge

merge(graph: tuple[RetrievedMemory, ...], vector: tuple[RetrievedMemory, ...]) -> tuple[tuple[RetrievedMemory, ...], int]

One memory per record_id, under :data:MERGE_RULE.

Returns the merged candidates and how many were the same memory seen twice. A memory the graph holds and the vector index also holds is one memory: emitting it twice would spend two lines of a budgeted block on one fact and let a duplicate outrank a distinct memory beneath it.

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

def merge(
    graph: tuple[RetrievedMemory, ...], vector: tuple[RetrievedMemory, ...]
) -> tuple[tuple[RetrievedMemory, ...], int]:
    """One memory per ``record_id``, under :data:`MERGE_RULE`.

    Returns the merged candidates and how many were the same memory seen
    twice. A memory the graph holds and the vector index also holds is one
    memory: emitting it twice would spend two lines of a budgeted block on
    one fact and let a duplicate outrank a distinct memory beneath it.
    """
    by_id: dict[str, RetrievedMemory] = {}
    order: list[str] = []
    for memory in graph:
        by_id[memory.record.record_id] = memory
        order.append(memory.record.record_id)
    duplicates = 0
    for memory in vector:
        key = memory.record.record_id
        existing = by_id.get(key)
        if existing is None:
            by_id[key] = memory
            order.append(key)
            continue
        duplicates += 1
        if (memory.score or 0.0) > (existing.score or 0.0):
            by_id[key] = RetrievedMemory(
                record=existing.record,
                score=memory.score,
                scope_distance=existing.scope_distance,
                selected_publication=existing.selected_publication,
            )
    return tuple(by_id[key] for key in order), duplicates

vector_metadata

vector_metadata(record: MemoryRecord) -> dict[str, Any]

Everything about a record that has to survive the round trip.

Named fields rather than a blob: a reader of a stored row can tell what the capability meant by each, and a field the capability stops using disappears from new rows instead of lingering as an uninterpreted key.

metadata is nested under its own key so a producer's vocabulary can never collide with the capability's -- a record whose metadata had an origin key would otherwise silently overwrite the record's.

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

def vector_metadata(record: MemoryRecord) -> dict[str, Any]:
    """Everything about a record that has to survive the round trip.

    Named fields rather than a blob: a reader of a stored row can tell what
    the capability meant by each, and a field the capability stops using
    disappears from new rows instead of lingering as an uninterpreted key.

    ``metadata`` is nested under its own key so a producer's vocabulary can
    never collide with the capability's -- a record whose metadata had an
    ``origin`` key would otherwise silently overwrite the record's.
    """
    return {
        "record_id": record.record_id,
        "layer": record.layer.value,
        "scope_path": storage_scope_path(record.scope),
        "origin": record.origin,
        "importance": salience_to_importance(record.salience),
        "revision": record.revision,
        # Stripped of the capability's own names for the same reason the
        # graph payload is: a producer that could write ``scope_path`` here
        # would be naming the scope its memory is recalled into.
        "metadata": producer_metadata(record.metadata),
    }