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,
) -> list[MemoryEntry]

Retrieve the most relevant memory entries for a query.

Steps: 1. Embed the query via embedding_provider. 2. Search vector_backend for candidate nodes (2x top_k). 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,
) -> list[MemoryEntry]:
    """Retrieve the most relevant memory entries for a query.

    Steps:
    1. Embed the query via embedding_provider.
    2. Search vector_backend for candidate nodes (2x top_k).
    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)

    # Step 2: Vector search for candidates (2x for re-ranking headroom)
    candidates: list[CandidateNode] = []
    if query_embedding:
        vector_results = await self._vector.search(scope, query_embedding, top_k=k * 2)
        candidates.extend(self._normalize_vector_results(vector_results, scope))

    # Step 3: Graph candidates from enabled layers
    active_layers = layers or self._config.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)])
    if layers:
        filters.filters.append(LayerFilter(layers=frozenset(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)),
            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 entries

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)),
                    node_id=node.id,
                    importance=node.importance,
                    created_at=node.created_at,
                )
            )
    return entries