symfonic.memory.graph.scoring

scoring

RetrievalScorer -- multi-signal scoring engine.

Computes composite scores using four signals

semantic_similarity * w1 + recency * w2 + frequency * w3 + graph_proximity * w4

where weights come from ScoringWeights configuration.

RetrievalScorer

RetrievalScorer(
    weights: ScoringWeights,
    half_life_days: float = _DEFAULT_HALF_LIFE_DAYS,
    *,
    scope_blend: bool = False,
)

Multi-signal scoring engine for memory retrieval.

Combines semantic similarity, recency, frequency, and graph proximity into a single composite score using configurable weights.

Source code in src/symfonic/memory/graph/scoring.py

def __init__(
    self,
    weights: ScoringWeights,
    half_life_days: float = _DEFAULT_HALF_LIFE_DAYS,
    *,
    scope_blend: bool = False,
) -> None:
    self._weights = weights
    self._half_life = half_life_days
    # v8.0: when True, apply the MULTIPLICATIVE scope-specificity gate
    # (design §4.c) so ancestor-level candidates are attenuated relative
    # to the query tip — the "blend" mode.  Default False keeps the
    # pre-v8.0 composite byte-stable (the per-layer dispatch path and any
    # caller that does not opt into hierarchical blending).
    self._scope_blend = scope_blend

score async

score(
    node: MemoryNode,
    query_embedding: list[float],
    query_context_node: NodeId | None,
    traversal: GraphTraversal,
    scope: TenantScope,
    vector_similarity_score: float | None = None,
) -> float

Compute composite score for a single node.

Parameters:

Name	Type	Description	Default
node	MemoryNode	The node to score.	required
query_embedding	list[float]	Embedding vector for the query.	required
query_context_node	NodeId | None	Optional context node for graph proximity.	required
traversal	GraphTraversal	GraphTraversal instance for proximity computation.	required
scope	TenantScope	Tenant scope for isolation.	required
vector_similarity_score	float | None	Optional pre-computed cosine similarity between node.embedding and query_embedding. When the RetrievalEngine already computed this via the vector backend (CandidateNode.vector_similarity_score), pass it through to avoid the duplicate cosine recompute. When None (default), the scorer falls back to the 7.2.1 hotfix path -- compute cosine locally from node.embedding if present. See roadmap §12.2.	None

Returns:

Type	Description
float	Composite score as a float.

Source code in src/symfonic/memory/graph/scoring.py

async def score(
    self,
    node: MemoryNode,
    query_embedding: list[float],
    query_context_node: NodeId | None,
    traversal: GraphTraversal,
    scope: TenantScope,
    vector_similarity_score: float | None = None,
) -> float:
    """Compute composite score for a single node.

    Args:
        node: The node to score.
        query_embedding: Embedding vector for the query.
        query_context_node: Optional context node for graph proximity.
        traversal: GraphTraversal instance for proximity computation.
        scope: Tenant scope for isolation.
        vector_similarity_score: Optional pre-computed cosine similarity
            between ``node.embedding`` and ``query_embedding``. When the
            ``RetrievalEngine`` already computed this via the vector
            backend (``CandidateNode.vector_similarity_score``), pass it
            through to avoid the duplicate cosine recompute. When
            ``None`` (default), the scorer falls back to the 7.2.1
            hotfix path -- compute cosine locally from
            ``node.embedding`` if present. See roadmap §12.2.

    Returns:
        Composite score as a float.
    """
    # Semantic similarity -- prefer the pre-computed value supplied by
    # the engine; fall back to local cosine for callers that bypass the
    # engine (e.g. ad-hoc scorer.score() invocations and the 7.2.1
    # postgres-embedding-roundtrip regression guard).
    if vector_similarity_score is not None:
        sim = vector_similarity_score
    elif node.embedding:
        sim = _cosine_similarity(node.embedding, query_embedding)
    else:
        sim = 0.0

    # Recency
    rec = _recency_score(node.updated_at, self._half_life)

    # Frequency
    freq = _frequency_score(node.access_count)

    # Graph proximity
    prox = 0.0
    if query_context_node is not None:
        prox = await traversal.graph_proximity(scope, query_context_node, node.id)

    composite = (
        self._weights.semantic_similarity * sim
        + self._weights.recency * rec
        + self._weights.frequency * freq
        + self._weights.graph_proximity * prox
    )

    # v8.0 — MULTIPLICATIVE scope-specificity gate (design §4.c Shape 1).
    # Opt-in (blend mode); off by default → byte-stable composite.  At the
    # query tip (distance 0) the multiplier is 1.0 so tip memories are
    # unaffected; ancestors are attenuated by exp-decay over hierarchy
    # distance.  This sidesteps the sum-to-1.0 validator entirely — scope
    # distance is a relevance attenuator, not a competing additive signal.
    if self._scope_blend:
        distance = _scope_distance(node, scope)
        composite *= scope_specificity_score(
            distance,
            self._weights.scope_half_life_levels,
            self._weights.scope_specificity_floor,
        )
    return composite

score_batch async

score_batch(
    nodes: list[MemoryNode],
    query_embedding: list[float],
    query_context_node: NodeId | None,
    traversal: GraphTraversal,
    scope: TenantScope,
    precomputed_similarities: list[float | None]
    | None = None,
) -> list[tuple[MemoryNode, float]]

Score a batch of nodes and return sorted results (descending).

Parameters:

Name	Type	Description	Default
nodes	list[MemoryNode]	Nodes to score.	required
query_embedding	list[float]	Embedding vector for the query.	required
query_context_node	NodeId | None	Optional context node for graph proximity.	required
traversal	GraphTraversal	GraphTraversal instance for proximity computation.	required
scope	TenantScope	Tenant scope for isolation.	required
precomputed_similarities	list[float | None] | None	Optional list aligned 1:1 with nodes carrying pre-computed cosine similarities. When supplied, each per-node call to :meth:score skips its local cosine recompute. Entries may be None to opt back into the fallback path on a per-node basis. Roadmap §12.2.	None

Returns:

Type	Description
list[tuple[MemoryNode, float]]	List of (node, score) tuples sorted by score descending.

Raises:

Type	Description
ValueError	if precomputed_similarities is supplied but its length does not match nodes.

Source code in src/symfonic/memory/graph/scoring.py

async def score_batch(
    self,
    nodes: list[MemoryNode],
    query_embedding: list[float],
    query_context_node: NodeId | None,
    traversal: GraphTraversal,
    scope: TenantScope,
    precomputed_similarities: list[float | None] | None = None,
) -> list[tuple[MemoryNode, float]]:
    """Score a batch of nodes and return sorted results (descending).

    Args:
        nodes: Nodes to score.
        query_embedding: Embedding vector for the query.
        query_context_node: Optional context node for graph proximity.
        traversal: GraphTraversal instance for proximity computation.
        scope: Tenant scope for isolation.
        precomputed_similarities: Optional list aligned 1:1 with
            ``nodes`` carrying pre-computed cosine similarities. When
            supplied, each per-node call to :meth:`score` skips its
            local cosine recompute. Entries may be ``None`` to opt back
            into the fallback path on a per-node basis. Roadmap §12.2.

    Returns:
        List of (node, score) tuples sorted by score descending.

    Raises:
        ValueError: if ``precomputed_similarities`` is supplied but its
            length does not match ``nodes``.
    """
    if precomputed_similarities is not None and len(precomputed_similarities) != len(nodes):
        raise ValueError(
            "precomputed_similarities length "
            f"({len(precomputed_similarities)}) must match nodes length "
            f"({len(nodes)})"
        )

    scored: list[tuple[MemoryNode, float]] = []
    for idx, node in enumerate(nodes):
        sim = (
            precomputed_similarities[idx]
            if precomputed_similarities is not None
            else None
        )
        s = await self.score(
            node,
            query_embedding,
            query_context_node,
            traversal,
            scope,
            vector_similarity_score=sim,
        )
        scored.append((node, s))
    scored.sort(key=lambda x: x[1], reverse=True)
    return scored

scope_specificity_score

scope_specificity_score(
    distance: int, half_life_levels: float, floor: float
) -> float

Multiplicative scope-specificity gate (design §4.c Shape 1 / §4.b).

Exponential decay over hierarchy distance: exp(-ln2/half_life * d), then linearly lifted off the floor. At distance 0 (tip) the result is always 1.0 (tip memories unaffected — byte-stable with pre-v8.0). With half_life_levels=1.0 and floor=0.0: tip 1.0, parent 0.5, grandparent 0.25.

Source code in src/symfonic/memory/graph/scoring.py

def scope_specificity_score(
    distance: int,
    half_life_levels: float,
    floor: float,
) -> float:
    """Multiplicative scope-specificity gate (design §4.c Shape 1 / §4.b).

    Exponential decay over hierarchy distance: ``exp(-ln2/half_life * d)``,
    then linearly lifted off the ``floor``.  At distance 0 (tip) the result is
    always 1.0 (tip memories unaffected — byte-stable with pre-v8.0).  With
    half_life_levels=1.0 and floor=0.0: tip 1.0, parent 0.5, grandparent 0.25.
    """
    if distance <= 0:
        return 1.0
    decay = math.exp(-_LN2 / half_life_levels * distance)
    return floor + (1.0 - floor) * decay