symfonic.core.learning.phases_semantic_merge

phases_semantic_merge

Phase 13: LLM-judged semantic-duplicate merge (true supersede).

Closes the write-time dedup's documented misses: restatements that reword a fact's OPENING ("Uses pytest and ruff" vs "The user uses pytest and ruff"), front rewording, and clause reordering all escape the prefix-gated string match in symfonic.memory.dedup -- see tests/memory/test_restatement_corpus.py for the full matrix. Those classes need semantic-equivalence judgment, which is deliberately kept OFF the per-turn write path and run here, offline, during consolidation.

Safety model (why an LLM judge and not an embedding threshold): cosine similarity cannot distinguish "deploys Tuesdays, never Fridays" from "deploys Fridays, never Tuesdays", and a false merge OVERWRITES a real memory. Embeddings (or a lexical ratio when no embeddings are available) only PREFILTER candidate pairs; every merge requires the judge to confirm the two phrasings state the same fact and to produce the merged phrasing.

Supersede mechanics: the canonical node (higher importance, else the older row) absorbs the pair -- merged phrasing, max importance, union of tags -- and the absorbed node is SOFT-RETRACTED with the namespaced marker (symfonic.memory.retraction), so it vanishes from every read path immediately, remains auditable, and is hard-deleted by the existing prune_retracted phase after the grace window.

Exclusions: only nodes materialised at exactly the run scope are eligible (visibility != ownership); AGENT_IDENTITY and entity nodes are never merged (they have their own identity rules).

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,
) -> 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.

Source code in src/symfonic/core/learning/phases_semantic_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,
) -> 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.
    """
    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)

    merges = 0
    judged_pairs = 0
    absorbed_ids: set[str] = set()
    for i, a in enumerate(eligible):
        if str(a.id) in absorbed_ids:
            continue
        for b in eligible[i + 1 :]:
            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
            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 {})
                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
                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