symfonic.memory.retrieval.engine

engine

RetrievalEngine -- multi-layer query planning and retrieval.

Coordinates vector search, graph traversal, and multi-signal scoring to retrieve the most relevant memory entries across enabled layers.

RetrievalEngine

RetrievalEngine(graph_store: GraphMemoryStore, vector_backend: VectorBackend, scorer: RetrievalScorer, traversal: GraphTraversal, config: OrchestratorConfig)

Cross-layer retrieval engine with multi-signal scoring.

Combines vector similarity search with graph-aware re-ranking to surface the most relevant memory entries.

Source code in src/symfonic/memory/retrieval/engine.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    vector_backend: VectorBackend,
    scorer: RetrievalScorer,
    traversal: GraphTraversal,
    config: OrchestratorConfig,
) -> None:
    self._graph = graph_store
    self._vector = vector_backend
    self._scorer = scorer
    self._traversal = traversal
    self._config = config

retrieve async

retrieve(scope: TenantScope, query: str, layers: set[MemoryLayer] | None = None, top_k: int | None = None, embedding_provider: EmbeddingProvider | None = None, query_context_node: NodeId | None = None, _vector_read_budget: int | None = None) -> list[MemoryEntry]

Retrieve the most relevant memory entries for a query.

Steps: 1. Embed the query via embedding_provider. 2. Search bounded vector prefixes until eligible headroom is found. 3. Normalize all results into CandidateNode models. 4. Score candidates via the RetrievalScorer. 5. Filter by layers if specified. 6. Return top_k results as MemoryEntry list.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	Natural language query.	required
layers	set[MemoryLayer] | None	Optional set of layers to filter results.	None
top_k	int | None	Number of results to return (defaults to config.default_top_k).	None
embedding_provider	EmbeddingProvider | None	Provider for query embedding.	None
query_context_node	NodeId | None	Optional node for graph proximity scoring.	None

Returns:

Type	Description
list[MemoryEntry]	Sorted list of MemoryEntry instances.

Source code in src/symfonic/memory/retrieval/engine.py

async def retrieve(
    self,
    scope: TenantScope,
    query: str,
    layers: set[MemoryLayer] | None = None,
    top_k: int | None = None,
    embedding_provider: EmbeddingProvider | None = None,
    query_context_node: NodeId | None = None,
    _vector_read_budget: int | None = None,
) -> list[MemoryEntry]:
    """Retrieve the most relevant memory entries for a query.

    Steps:
    1. Embed the query via embedding_provider.
    2. Search bounded vector prefixes until eligible headroom is found.
    3. Normalize all results into CandidateNode models.
    4. Score candidates via the RetrievalScorer.
    5. Filter by layers if specified.
    6. Return top_k results as MemoryEntry list.

    Args:
        scope: Tenant scope for isolation.
        query: Natural language query.
        layers: Optional set of layers to filter results.
        top_k: Number of results to return (defaults to config.default_top_k).
        embedding_provider: Provider for query embedding.
        query_context_node: Optional node for graph proximity scoring.

    Returns:
        Sorted list of MemoryEntry instances.
    """
    k = top_k or self._config.default_top_k

    # Step 1: Embed query
    query_embedding: list[float] = []
    if embedding_provider:
        query_embedding = await embedding_provider.embed(query)

    active_layers = layers or self._config.enabled_layers

    # Step 2: Search enough ranked vector hits to get eligible headroom.
    # Backend.search has no layer selector or cursor, so each read is a
    # larger prefix. Count every requested row against one finite budget.
    candidates: list[CandidateNode] = []
    vector_scan_exhausted = True
    vector_scan_truncated = False
    vector_scan_cost = 0
    if query_embedding:
        (
            vector_results, vector_scan_exhausted, vector_scan_truncated,
            vector_scan_cost,
        ) = await self._vector_candidates(
            scope, query_embedding, active_layers, k,
            budget=min(VECTOR_READ_BUDGET, _vector_read_budget or VECTOR_READ_BUDGET),
        )
        candidates.extend(self._normalize_vector_results(vector_results, scope))

    # Step 3: Graph candidates from enabled layers
    for layer in active_layers:
        graph_nodes = await self._graph.query_nodes(scope, layer=layer)
        for node in graph_nodes:
            if not any(str(c.node.id) == str(node.id) for c in candidates):
                candidates.append(
                    CandidateNode(
                        node=node,
                        # ``None`` leaves the cosine recompute path open
                        # for graph-only hits whose embedding was loaded
                        # by the 7.2.1 hotfix -- otherwise we would lose
                        # the semantic-similarity signal for them.
                        vector_similarity_score=None,
                        origin_layer=node.layer,
                    )
                )

    # Step 4: Score all candidates. Forward the pre-computed cosine
    # from the vector backend so RetrievalScorer skips the duplicate
    # recompute (roadmap §12.2 -- Item 13.2).
    nodes = [c.node for c in candidates]
    precomputed = [c.vector_similarity_score for c in candidates]
    scored = await self._scorer.score_batch(
        nodes,
        query_embedding,
        query_context_node,
        self._traversal,
        scope,
        precomputed_similarities=precomputed,
    )

    # Step 5: Filter by layers
    filters = CompositeFilter(filters=[TenantFilter(scope=scope)])
    # Vector hits can bypass the graph-layer loop above. Apply the effective
    # selector to both sources, including the configured default when the
    # caller leaves ``layers`` unset.
    filters.filters.append(LayerFilter(layers=frozenset(active_layers)))

    entries: list[MemoryEntry] = []
    for node, score in scored:
        entry = MemoryEntry(
            layer=node.layer,
            tenant_id=node.tenant_id,
            content=str(node.properties.get("content", node.label)),
            # TA8.74. Identity and provenance travel with the entry. This
            # carried ``node_id`` and nothing else -- no ``id``, no
            # properties -- so ``record_id``, ``scope_path`` and ``origin``
            # were gone by the time the capability layer read them, and a
            # record written as ``m1`` came back under a fresh uuid. A
            # store that cannot return what it was given cannot upsert,
            # dedupe or supersede anything.
            id=node.id,
            metadata=dict(node.properties or {}),
            node_id=node.id,
            score=score,
            importance=node.importance,
            created_at=node.created_at,
        )
        if filters.matches(entry):
            entries.append(entry)
        if len(entries) > k:
            break

    return _RetrievalPage(
        entries[:k], scan_exhausted=vector_scan_exhausted and len(entries) <= k,
        vector_scan_truncated=vector_scan_truncated,
        scan_cost=max(k, vector_scan_cost),
    )

retrieve_by_graph async

retrieve_by_graph(scope: TenantScope, start_node: NodeId, max_depth: int = 2) -> list[MemoryEntry]

Graph-first retrieval via BFS traversal.

Results are also normalized via CandidateNode for consistent scoring.

Source code in src/symfonic/memory/retrieval/engine.py

async def retrieve_by_graph(
    self,
    scope: TenantScope,
    start_node: NodeId,
    max_depth: int = 2,
) -> list[MemoryEntry]:
    """Graph-first retrieval via BFS traversal.

    Results are also normalized via CandidateNode for consistent scoring.
    """
    nodes = await self._traversal.bfs(scope, start_node, max_depth)
    entries: list[MemoryEntry] = []
    for node in nodes:
        if node.tenant_id == scope.tenant_id:
            entries.append(
                MemoryEntry(
                    layer=node.layer,
                    tenant_id=node.tenant_id,
                    content=str(node.properties.get("content", node.label)),
                    # Same reason as the scored path above.
                    id=node.id,
                    metadata=dict(node.properties or {}),
                    node_id=node.id,
                    importance=node.importance,
                    created_at=node.created_at,
                )
            )
    return entries