feat(consolidation): semantic dedup of near-duplicate observations (create + update)

hindsight-api-slim/hindsight_api/config.py

@@ -422,6 +422,7 @@ def normalize_config_dict(config: dict[str, Any]) -> dict[str, Any]:
 ENV_CONSOLIDATION_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE"
 ENV_CONSOLIDATION_MAX_MEMORIES_PER_ROUND = "HINDSIGHT_API_CONSOLIDATION_MAX_MEMORIES_PER_ROUND"
 ENV_CONSOLIDATION_LLM_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE"
+ENV_CONSOLIDATION_DEDUP_THRESHOLD = "HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD"
 ENV_CONSOLIDATION_LLM_PARALLELISM = "HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM"
 ENV_CONSOLIDATION_MAX_TOKENS = "HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS"
 ENV_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS = "HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS"
@@ -813,6 +814,10 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
     100  # Max memories per consolidation round (0 = unlimited). Limits how long one bank holds a worker slot.
 )
 DEFAULT_CONSOLIDATION_LLM_BATCH_SIZE = 8  # Facts per LLM call (1 = no batching; >1 = batch mode)
+# Cosine >= this between a newly-created observation and an existing one triggers a focused
+# 1-by-1 LLM "merge or keep" pass (the LLM reads both, so numbers/negation/entities are
+# respected). 1.0 disables it (no obs is ever >=1.0 to a *different* one after the exact guard).
+DEFAULT_CONSOLIDATION_DEDUP_THRESHOLD = 1.0
 DEFAULT_CONSOLIDATION_LLM_PARALLELISM = (
     4  # Max tag groups consolidated concurrently per op. Locks on overlapping write
     # scopes degrade to sequential automatically; matches retain_max_concurrent.
@@ -1376,6 +1381,7 @@ class HindsightConfig:
     enable_mental_model_history: bool
     mental_model_history_max_entries: int
     consolidation_batch_size: int
+    consolidation_dedup_threshold: float
     consolidation_max_memories_per_round: int
     consolidation_llm_batch_size: int
     consolidation_llm_parallelism: int
@@ -2205,6 +2211,9 @@ def from_env(cls) -> "HindsightConfig":
                     str(DEFAULT_CONSOLIDATION_MAX_MEMORIES_PER_ROUND),
                 )
             ),
+            consolidation_dedup_threshold=float(
+                os.getenv(ENV_CONSOLIDATION_DEDUP_THRESHOLD, str(DEFAULT_CONSOLIDATION_DEDUP_THRESHOLD))
+            ),
             consolidation_llm_batch_size=int(
                 os.getenv(ENV_CONSOLIDATION_LLM_BATCH_SIZE, str(DEFAULT_CONSOLIDATION_LLM_BATCH_SIZE))
             ),

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -25,7 +25,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from itertools import combinations
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 from pydantic import BaseModel, field_validator
 
@@ -88,6 +88,209 @@ def _duplicate_create_target(
     return None
 
 
+# Top-K existing observations probed (by the new observation's own embedding) when
+# semantic dedup is enabled. Small: we only need the nearest few candidates.
+_DEDUP_TOP_K = 5
+
+
+class _DedupDecision(BaseModel):
+    """Focused 1-by-1 verdict for whether a new observation duplicates an existing one."""
+
+    action: Literal["merge", "keep"]
+    text: str = ""  # the synthesized merged observation (when action == "merge")
+    reason: str = ""
+
+
+_DEDUP_PROMPT = """You reconcile long-term memory observations. A NEW observation is about to be \
+stored, and it is highly similar to an EXISTING one:
+
+[NEW] {new}
+[EXISTING] {existing}
+
+If they assert the SAME fact (wording aside), respond action="merge" and provide `text`: a single \
+observation that preserves EVERY detail from both. If they differ in ANY important detail — a \
+number/quantity, a named entity or language, a negation, or a condition — respond action="keep"."""
+
+
+@dataclass
+class _DedupOutcome:
+    """Result of probing one observation against its in-scope neighbours.
+
+    ``best_id`` is the nearest observation at/above the threshold (None if none),
+    ``merged_text`` is the LLM-synthesized union text (set only when ``should_merge``).
+    """
+
+    best_id: str | None
+    merged_text: str
+    should_merge: bool
+
+
+async def _dedup_adjudicate(
+    conn: "Connection",
+    memory_engine: "MemoryEngine",
+    bank_id: str,
+    config: Any,
+    dedup_llm_config: Any,
+    anchor_text: str,
+    anchor_emb_str: str | None,
+    tags: list[str] | None,
+    exclude_id: str | None,
+) -> _DedupOutcome:
+    """Probe one observation's embedding against in-scope observations and adjudicate a merge.
+
+    Anchored on the observation text — the correct obs<->obs comparison, unlike consolidation
+    recall which is anchored on the raw fact. Returns the nearest observation at/above
+    ``consolidation_dedup_threshold`` and, when found, the LLM's focused 1-by-1 merge-or-keep
+    verdict (scope ``consolidation_dedup``): the LLM reads both texts, so a word-level difference
+    (number / negation / entity) is respected. ``exclude_id`` skips the anchor observation itself
+    (used by the UPDATE path, where the anchor row already exists and would self-match at 1.0).
+    ``anchor_emb_str`` reuses an already-computed embedding (the UPDATE path just embedded it);
+    pass None to embed ``anchor_text`` here (the CREATE path).
+    """
+    from ..search.retrieval import retrieve_semantic_bm25_combined
+
+    threshold = config.consolidation_dedup_threshold
+    if anchor_emb_str is None:
+        embs = await embedding_utils.generate_embeddings_batch(memory_engine.embeddings, [anchor_text])
+        if not embs:
+            return _DedupOutcome(best_id=None, merged_text="", should_merge=False)
+        anchor_emb_str = str(embs[0])
+    tags_match = "all_strict" if tags else "any"
+    grouped = await retrieve_semantic_bm25_combined(
+        conn, anchor_emb_str, anchor_text, bank_id, ["observation"], _DEDUP_TOP_K, tags=tags, tags_match=tags_match
+    )
+    results = grouped.get("observation", ([], []))[0]
+    best_id: str | None = None
+    best_text = ""
+    best_sim = threshold  # only candidates at/above the threshold are considered
+    for r in results:
+        rid = str(r.id)
+        if exclude_id is not None and rid == exclude_id:
+            continue  # never match the anchor observation against itself
+        sim = r.similarity or 0.0
+        if sim >= best_sim:
+            best_id, best_text, best_sim = rid, r.text, sim
+
+    if best_id is None:
+        return _DedupOutcome(best_id=None, merged_text="", should_merge=False)
+
+    decision: _DedupDecision = await dedup_llm_config.call(
+        messages=[{"role": "user", "content": _DEDUP_PROMPT.format(new=anchor_text, existing=best_text)}],
+        response_format=_DedupDecision,
+        scope="consolidation_dedup",
+    )
+    if decision.action != "merge":
+        return _DedupOutcome(best_id=best_id, merged_text="", should_merge=False)
+    return _DedupOutcome(best_id=best_id, merged_text=decision.text.strip() or best_text, should_merge=True)
+
+
+async def _dedup_reconcile_create(
+    conn: "Connection",
+    memory_engine: "MemoryEngine",
+    bank_id: str,
+    config: Any,
+    dedup_llm_config: Any,
+    create_text: str,
+    create_source_ids: list[uuid.UUID],
+    tags: list[str] | None,
+) -> str | None:
+    """Semantic dedup for a single CREATE (create-time, focused 1-by-1).
+
+    On "merge", folds the new source facts + the synthesized text into the existing
+    observation and returns its id (caller skips the CREATE). Returns None when there is
+    no near twin or the LLM keeps them distinct.
+    """
+    outcome = await _dedup_adjudicate(
+        conn, memory_engine, bank_id, config, dedup_llm_config, create_text, None, tags, exclude_id=None
+    )
+    if not outcome.should_merge or outcome.best_id is None:
+        return None
+
+    # Fold the new source facts into the twin and persist the merged text. We keep the twin's
+    # existing embedding: the merged text is >= threshold similar, so the stored vector stays
+    # representative and we avoid a re-embed + a dialect-specific vector UPDATE.
+    await conn.execute(
+        f"""
+        UPDATE {fq_table("memory_units")}
+        SET text = $1,
+            source_memory_ids = (SELECT array_agg(DISTINCT e) FROM unnest(source_memory_ids || $2::uuid[]) e),
+            proof_count = (SELECT count(DISTINCT e) FROM unnest(source_memory_ids || $2::uuid[]) e),
+            updated_at = now()
+        WHERE id = $3::uuid
+        """,
+        outcome.merged_text,
+        create_source_ids,
+        uuid.UUID(outcome.best_id),
+    )
+    return outcome.best_id
+
+
+async def _dedup_reconcile_update(
+    conn: "Connection",
+    memory_engine: "MemoryEngine",
+    bank_id: str,
+    config: Any,
+    dedup_llm_config: Any,
+    updated_id: str,
+    updated_text: str,
+    updated_emb_str: str | None,
+    tags: list[str] | None,
+) -> None:
+    """Semantic dedup for an UPDATE (after the observation was rewritten + re-embedded).
+
+    An UPDATE rewrites an observation's text and re-embeds it, so its vector can drift to
+    within threshold of a DIFFERENT existing observation. The create-time guard never sees
+    this (it only runs on CREATE), so without this the two persist as a near-duplicate pair —
+    the measured residual-duplicate source. Probe the updated observation's new embedding
+    against the others (excluding itself); on "merge", fold the just-updated observation's
+    sources into the twin, persist the merged text, and DELETE the updated row. Unlike the
+    CREATE path the row already exists, so reconciliation is a fold-and-delete, not a skip.
+    """
+    outcome = await _dedup_adjudicate(
+        conn,
+        memory_engine,
+        bank_id,
+        config,
+        dedup_llm_config,
+        updated_text,
+        updated_emb_str,
+        tags,
+        exclude_id=updated_id,
+    )
+    if not outcome.should_merge or outcome.best_id is None:
+        return
+
+    # Fold the updated observation's sources into the twin (keeping the twin's embedding, as in
+    # the create path) then delete the now-redundant updated row. The all_strict/any tag match
+    # guarantees twin and updated share scope, so dropping the updated row's tags loses no
+    # visibility. Temporal fields follow the surviving twin (minimal scope; matches create).
+    await conn.execute(
+        f"""
+        UPDATE {fq_table("memory_units")} t
+        SET text = $1,
+            source_memory_ids = (
+                SELECT array_agg(DISTINCT e) FROM unnest(t.source_memory_ids || u.source_memory_ids) e
+            ),
+            proof_count = (
+                SELECT count(DISTINCT e) FROM unnest(t.source_memory_ids || u.source_memory_ids) e
+            ),
+            updated_at = now()
+        FROM {fq_table("memory_units")} u
+        WHERE t.id = $2::uuid AND u.id = $3::uuid
+        """,
+        outcome.merged_text,
+        uuid.UUID(outcome.best_id),
+        uuid.UUID(updated_id),
+    )
+    await _execute_delete_action(conn, bank_id, updated_id)
+    logger.info(
+        "[CONSOLIDATION] dedup-merged updated observation %s into %s (cosine>=%.2f)",
+        updated_id[:8],
+        outcome.best_id[:8],
+        config.consolidation_dedup_threshold,
+    )
+
+
 @dataclass
 class _BatchDeltas:
     """Per-LLM-batch deltas, merged into the job's running stats after dispatch.
@@ -1141,6 +1344,20 @@ async def _process_memory_batch(
 
     mem_by_id = {str(m["id"]): m for m in memories}
 
+    # Semantic dedup: when enabled, an observation that is >= the threshold cosine to a DIFFERENT
+    # existing observation is reconciled by a focused 1-by-1 LLM merge (anchored on the observation
+    # text, not the source fact). It runs on both CREATE (a near-dup emitted despite the twin being
+    # in context — weak-model failure mode) and UPDATE (a rewrite+re-embed that drifts an existing
+    # observation into a twin — the create-time guard can't see this). The trace operation/scope is
+    # "consolidation_dedup" (routes through the consolidation concurrency bucket via llm_wrapper's
+    # "consolidation" prefix; recorded distinctly in llm_requests).
+    dedup_enabled = config is not None and getattr(config, "consolidation_dedup_threshold", 1.0) < 1.0
+    dedup_llm_config = (
+        memory_engine._consolidation_llm_config.with_config(config, bank_id=bank_id, operation="consolidation_dedup")
+        if dedup_enabled
+        else None
+    )
+
     # Execute deletes first to free observation slots before creates consume them
     deleted_count = 0
     for delete in llm_result.deletes:
@@ -1165,7 +1382,7 @@ async def _process_memory_batch(
             )
             continue
         agg = _aggregate_source_fields(source_mems, tags=fact_tags)
-        await _execute_update_action(
+        updated_emb_str = await _execute_update_action(
             conn=conn,
             memory_engine=memory_engine,
             bank_id=bank_id,
@@ -1181,6 +1398,21 @@ async def _process_memory_batch(
         )
         for m in source_mems:
             per_memory_updated.add(str(m["id"]))
+        # Reconcile the rewritten observation against its neighbours: the re-embed may have
+        # drifted it into a near-twin of another existing observation (the residual-duplicate
+        # source). updated_emb_str is None when the update was skipped — nothing to reconcile.
+        if dedup_enabled and updated_emb_str is not None:
+            await _dedup_reconcile_update(
+                conn,
+                memory_engine,
+                bank_id,
+                config,
+                dedup_llm_config,
+                update.observation_id,
+                update.text,
+                updated_emb_str,
+                agg.tags,
+            )
 
     # Deterministic dedup guard: map the observations the LLM was SHOWN by their
     # normalised text. The model intermittently emits a CREATE whose text is identical
@@ -1215,6 +1447,22 @@ async def _process_memory_batch(
             )
             continue
 
+        # Semantic near-duplicate reconciliation: merge this CREATE into an existing
+        # near-identical observation (LLM-adjudicated, 1-by-1) instead of inserting a dup.
+        if dedup_enabled:
+            merged_into = await _dedup_reconcile_create(
+                conn, memory_engine, bank_id, config, dedup_llm_config, create.text, create_source_ids, agg.tags
+            )
+            if merged_into is not None:
+                logger.info(
+                    "[CONSOLIDATION] dedup-merged observation CREATE into %s (cosine>=%.2f)",
+                    merged_into[:8],
+                    config.consolidation_dedup_threshold,
+                )
+                for m in source_mems:
+                    per_memory_created.add(str(m["id"]))
+                continue
+
         await _execute_create_action(
             conn=conn,
             memory_engine=memory_engine,
@@ -1272,12 +1520,15 @@ async def _execute_update_action(
     source_occurred_end: datetime | None = None,
     source_mentioned_at: datetime | None = None,
     perf: ConsolidationPerfLog | None = None,
-) -> None:
+) -> str | None:
     """
     Update an existing observation.
 
     Extends source_memory_ids with all contributing memories, updates temporal fields
     (LEAST for occurred_start, GREATEST for occurred_end / mentioned_at), and merges tags.
+
+    Returns the observation's freshly-computed embedding (pgvector literal) so the caller can
+    run UPDATE-path dedup without re-embedding, or None when the update was skipped.
     """
     model = next((m for m in observations if str(m.id) == observation_id), None)
     if not model:
@@ -1374,6 +1625,7 @@ async def _execute_update_action(
     # Map the updated observation onto the consolidation trace as a produced memory.
     record_created_memory_ids([observation_id])
     logger.debug(f"Updated observation {observation_id} from {len(source_memory_ids)} source memories")
+    return embedding_str
 
 
 async def _execute_create_action(