symfonic.core.learning.phases_semantic_merge

phases_semantic_merge

Phase 13, now owned by the memory capability.

merge_semantic_duplicates is on the quick roster -- v8.17 put it there because quick is the only cadence the engine schedules in-process -- so it moved to :mod:symfonic.capabilities.memory.phases.merge and is imported back here for the legacy consolidator to call. One phase, one implementation, for as long as both routes ship.

merge_semantic_duplicates async

merge_semantic_duplicates(graph: Any, scope: TenantScope, *, chat_model: Any, embedding_provider: Any | None = None, embedding_threshold: float = DEFAULT_EMBEDDING_THRESHOLD, lexical_threshold: float = DEFAULT_LEXICAL_THRESHOLD, max_pairs_per_run: int = 10, on_candidates: Callable[[Iterable[Any]], None] | None = None, on_supersede: Callable[[Any, Any], None] | None = None) -> int

Merge LLM-confirmed duplicate semantic facts; return merge count.

Per-pair failures are logged and skipped -- one bad judge response must never abort the phase.

on_candidates and on_supersede report node ids, and nothing else, to a caller keeping a cycle ledger. The supersede hook fires where the retraction is written rather than being derived from the return value, so the count stays true if one merge ever absorbs more than one row. Both are optional and the legacy consolidator passes neither.

Source code in src/symfonic/capabilities/memory/phases/merge.py

async def merge_semantic_duplicates(
    graph: Any,
    scope: TenantScope,
    *,
    chat_model: Any,
    embedding_provider: Any | None = None,
    embedding_threshold: float = DEFAULT_EMBEDDING_THRESHOLD,
    lexical_threshold: float = DEFAULT_LEXICAL_THRESHOLD,
    max_pairs_per_run: int = 10,
    on_candidates: Callable[[Iterable[Any]], None] | None = None,
    on_supersede: Callable[[Any, Any], None] | None = None,
) -> int:
    """Merge LLM-confirmed duplicate semantic facts; return merge count.

    Per-pair failures are logged and skipped -- one bad judge response
    must never abort the phase.

    ``on_candidates`` and ``on_supersede`` report node ids, and nothing else,
    to a caller keeping a cycle ledger. The supersede hook fires where the
    retraction is written rather than being derived from the return value, so
    the count stays true if one merge ever absorbs more than one row. Both are
    optional and the legacy consolidator passes neither.
    """
    if chat_model is None:
        return 0

    nodes = await graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    run_path = materialise_scope_path(scope)
    eligible = []
    for n in nodes:
        label = str(getattr(n, "label", "") or "")
        if label.startswith(_EXCLUDED_LABEL_PREFIXES):
            continue
        props = getattr(n, "properties", None) or {}
        if props.get("label_prefix") in _EXCLUDED_LABEL_PREFIXES:
            continue
        if stored_scope_path(props, getattr(n, "tenant_id", "")) != run_path:
            # Visible but ancestor-owned: not this scope's to merge.
            continue
        eligible.append(n)

    if on_candidates is not None:
        on_candidates([node.id for node in eligible])

    merges = 0
    judged_pairs = 0
    absorbed_ids: set[str] = set()
    for i, a in enumerate(eligible):
        if str(a.id) in absorbed_ids:
            continue
        peers = {str(node.id): node for node in eligible[i + 1 :]}
        if callable(getattr(type(graph), "related_candidates", None)):
            for node in await graph.related_candidates(scope, _node_text(a)):
                props = node.properties or {}
                if (str(node.id) != str(a.id)
                    and not node.label.startswith(_EXCLUDED_LABEL_PREFIXES)
                    and props.get("label_prefix") not in _EXCLUDED_LABEL_PREFIXES
                    and stored_scope_path(props, node.tenant_id) == run_path):
                    peers.setdefault(str(node.id), node)
        for b in peers.values():
            if judged_pairs >= max_pairs_per_run:
                logger.info(
                    "semantic_merge: pair budget (%d) reached; remaining "
                    "candidates deferred to the next run",
                    max_pairs_per_run,
                )
                return merges
            if str(b.id) in absorbed_ids or str(a.id) in absorbed_ids:
                continue
            if not carry_classification(a, b, {}):
                continue
            try:
                similarity, kind = await _pair_similarity(a, b, embedding_provider)
                threshold = embedding_threshold if kind == "embedding" else lexical_threshold
                if similarity < threshold:
                    continue
                judged_pairs += 1
                same, merged_fact = await _judge(chat_model, _node_text(a), _node_text(b))
                if not same:
                    continue
                # Canonical = higher importance, else the earlier row.
                canonical, absorbed = (
                    (a, b) if float(a.importance or 0) >= float(b.importance or 0) else (b, a)
                )
                phrasing = merged_fact or _node_text(canonical)
                merged_props = dict(canonical.properties or {})
                if not carry_classification(canonical, absorbed, merged_props):
                    continue
                merged_props["content"] = phrasing
                tags_a = list((canonical.properties or {}).get("tags") or [])
                tags_b = list((absorbed.properties or {}).get("tags") or [])
                if tags_a or tags_b:
                    merged_props["tags"] = sorted(set(tags_a) | set(tags_b))
                await graph.update_node(
                    scope,
                    canonical.id,
                    {
                        "label": phrasing[:100],
                        "importance": max(
                            float(canonical.importance or 1.0),
                            float(absorbed.importance or 1.0),
                        ),
                        "properties": merged_props,
                    },
                )
                await graph.update_node(
                    scope,
                    absorbed.id,
                    {
                        "properties": {
                            **(absorbed.properties or {}),
                            RETRACTED_KEY: True,
                            RETRACTED_AT_KEY: datetime.now(UTC).isoformat(),
                            RETRACTION_REASON_KEY: (
                                f"superseded by semantic merge into {canonical.id}"
                            )[:200],
                        },
                    },
                )
                absorbed_ids.add(str(absorbed.id))
                merges += 1
                if on_supersede is not None:
                    on_supersede(absorbed.id, canonical.id)
                logger.info(
                    "semantic_merge: %r absorbed into %s (%s sim=%.2f)",
                    _node_text(absorbed)[:60],
                    canonical.id,
                    kind,
                    similarity,
                )
            except Exception:
                logger.warning(
                    "semantic_merge: pair (%s, %s) failed; skipping",
                    a.id,
                    b.id,
                    exc_info=True,
                )
    return merges