diff --git a/hindsight-api-slim/hindsight_api/config.py b/hindsight-api-slim/hindsight_api/config.py index 2d3d84dca7..be20b28dc6 100644 --- a/hindsight-api-slim/hindsight_api/config.py +++ b/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)) ), diff --git a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py index a0d99090d8..09c5fc010d 100644 --- a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py +++ b/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( diff --git a/hindsight-api-slim/tests/test_consolidation_dedup.py b/hindsight-api-slim/tests/test_consolidation_dedup.py index 4bdf973a89..63d30c52d3 100644 --- a/hindsight-api-slim/tests/test_consolidation_dedup.py +++ b/hindsight-api-slim/tests/test_consolidation_dedup.py @@ -5,9 +5,19 @@ the path stochastically. """ +import types +import uuid from dataclasses import dataclass +from unittest.mock import AsyncMock, patch -from hindsight_api.engine.consolidation.consolidator import _duplicate_create_target, _norm_obs_text +from hindsight_api.engine.consolidation.consolidator import ( + _dedup_reconcile_create, + _dedup_reconcile_update, + _DedupDecision, + _duplicate_create_target, + _norm_obs_text, +) +from hindsight_api.engine.search.types import RetrievalResult @dataclass @@ -51,3 +61,161 @@ def test_novel_create_is_not_duplicate() -> None: shown = _shown(_FakeObs(id="22222222-bbbb", text="User waters the herbs early in the morning.")) assert _duplicate_create_target("Rosemary is drought-tolerant.", shown, set()) is None assert _duplicate_create_target("", {}, set()) is None + + +# ── semantic dedup (_dedup_reconcile_create) ────────────────────────────────── +# +# Mocks the embedder, the obs-anchored ANN probe, and the LLM so the decision logic is +# tested without a DB or a real model. + +_TWIN_ID = "33333333-3333-4333-8333-333333333333" + + +def _obs(text: str, sim: float, oid: str = _TWIN_ID) -> RetrievalResult: + return RetrievalResult(id=oid, text=text, fact_type="observation", similarity=sim) + + +def _ctx(threshold: float = 0.97): + """Return (kwargs, conn_mock, llm_mock) for a _dedup_reconcile_create call.""" + conn = AsyncMock() + llm = types.SimpleNamespace(call=AsyncMock()) + kwargs = dict( + conn=conn, + memory_engine=types.SimpleNamespace(embeddings=object()), + bank_id="bank1", + config=types.SimpleNamespace(consolidation_dedup_threshold=threshold), + dedup_llm_config=llm, + create_text="YouTube content in Uzbek is very rich.", + create_source_ids=[uuid.uuid4()], + tags=["t1"], + ) + return kwargs, conn, llm + + +def _patch_probe(results): + return patch( + "hindsight_api.engine.search.retrieval.retrieve_semantic_bm25_combined", + AsyncMock(return_value={"observation": (results, [])}), + ) + + +def _patch_embed(): + return patch( + "hindsight_api.engine.retain.embedding_utils.generate_embeddings_batch", + AsyncMock(return_value=[[0.1, 0.2, 0.3]]), + ) + + +async def test_dedup_no_twin_above_threshold_returns_none() -> None: + kwargs, conn, llm = _ctx(threshold=0.97) + with _patch_embed(), _patch_probe([_obs("something loosely related", 0.81)]): + result = await _dedup_reconcile_create(**kwargs) + assert result is None + llm.call.assert_not_called() # below threshold → no LLM call + conn.execute.assert_not_called() # no merge + + +async def test_dedup_llm_keep_does_not_merge() -> None: + kwargs, conn, llm = _ctx() + llm.call.return_value = _DedupDecision(action="keep", reason="different language") + with _patch_embed(), _patch_probe([_obs("Uzbek content on YouTube is described as very rich.", 0.98)]): + result = await _dedup_reconcile_create(**kwargs) + assert result is None + llm.call.assert_awaited_once() + conn.execute.assert_not_called() # kept distinct → no merge + + +async def test_dedup_llm_merge_folds_into_twin() -> None: + kwargs, conn, llm = _ctx() + kwargs["create_source_ids"] = [uuid.uuid4(), uuid.uuid4()] + llm.call.return_value = _DedupDecision(action="merge", text="Uzbek content on YouTube is very rich.") + with _patch_embed(), _patch_probe([_obs("Uzbek content on YouTube is described as very rich.", 0.99)]): + result = await _dedup_reconcile_create(**kwargs) + assert result == _TWIN_ID # merged into the twin; caller skips the CREATE + conn.execute.assert_awaited_once() + args = conn.execute.await_args.args + assert args[1] == "Uzbek content on YouTube is very rich." # merged text persisted + assert args[2] == kwargs["create_source_ids"] # new source facts folded in + assert args[3] == uuid.UUID(_TWIN_ID) # onto the twin row + + +async def test_dedup_picks_highest_above_threshold_skips_below() -> None: + # Only the >=threshold candidate is considered; a 0.95 result is ignored at threshold 0.97. + kwargs, conn, llm = _ctx(threshold=0.97) + llm.call.return_value = _DedupDecision(action="keep") + with _patch_embed(), _patch_probe([_obs("near but distinct", 0.95), _obs("the real twin", 0.98)]): + await _dedup_reconcile_create(**kwargs) + # the twin passed to the LLM is the >=0.97 one, not the 0.95 + sent = llm.call.await_args.kwargs["messages"][0]["content"] + assert "the real twin" in sent + assert "near but distinct" not in sent + + +# ── UPDATE-path dedup (_dedup_reconcile_update) ─────────────────────────────── +# +# An UPDATE rewrites+re-embeds an observation, which can drift it into a near-twin of a +# DIFFERENT existing observation. These cover the fold-and-delete reconciliation (unlike +# CREATE, both rows already exist), the self-exclusion, and the keep/no-twin no-ops. + +_UPDATED_ID = "44444444-4444-4444-8444-444444444444" + + +def _update_ctx(threshold: float = 0.97): + """Return (kwargs, conn_mock, llm_mock) for a _dedup_reconcile_update call.""" + conn = AsyncMock() + llm = types.SimpleNamespace(call=AsyncMock()) + kwargs = dict( + conn=conn, + memory_engine=types.SimpleNamespace(embeddings=object()), + bank_id="bank1", + config=types.SimpleNamespace(consolidation_dedup_threshold=threshold), + dedup_llm_config=llm, + updated_id=_UPDATED_ID, + updated_text="Uzbek content on YouTube is very rich and growing.", + updated_emb_str="[0.1, 0.2, 0.3]", # already embedded by _execute_update_action + tags=["t1"], + ) + return kwargs, conn, llm + + +async def test_dedup_update_merge_folds_into_twin_and_deletes_updated() -> None: + kwargs, conn, llm = _update_ctx() + llm.call.return_value = _DedupDecision(action="merge", text="Uzbek YouTube content is very rich and growing.") + with _patch_probe([_obs("Uzbek content on YouTube is described as very rich.", 0.98)]): + await _dedup_reconcile_update(**kwargs) + llm.call.assert_awaited_once() + # Two writes: fold-into-twin UPDATE, then DELETE of the updated row. + assert conn.execute.await_count == 2 + fold_args = conn.execute.await_args_list[0].args + assert fold_args[1] == "Uzbek YouTube content is very rich and growing." # merged text on the twin + assert fold_args[2] == uuid.UUID(_TWIN_ID) # survivor = the twin + assert fold_args[3] == uuid.UUID(_UPDATED_ID) # folded-from = the updated row + delete_args = conn.execute.await_args_list[1].args + assert delete_args[1] == uuid.UUID(_UPDATED_ID) # the updated row is deleted + + +async def test_dedup_update_keep_does_not_merge() -> None: + kwargs, conn, llm = _update_ctx() + llm.call.return_value = _DedupDecision(action="keep", reason="different growth claim") + with _patch_probe([_obs("Uzbek content on YouTube is described as very rich.", 0.98)]): + await _dedup_reconcile_update(**kwargs) + llm.call.assert_awaited_once() + conn.execute.assert_not_called() # kept distinct → neither fold nor delete + + +async def test_dedup_update_excludes_self() -> None: + # The probe surfaces the updated observation itself at 1.0; it must be excluded so we don't + # "merge" a row into itself. With no other candidate, there is no twin → no LLM, no writes. + kwargs, conn, llm = _update_ctx() + with _patch_probe([_obs("its own current text", 1.0, oid=_UPDATED_ID)]): + await _dedup_reconcile_update(**kwargs) + llm.call.assert_not_called() + conn.execute.assert_not_called() + + +async def test_dedup_update_no_twin_above_threshold() -> None: + kwargs, conn, llm = _update_ctx(threshold=0.97) + with _patch_probe([_obs("loosely related", 0.8)]): + await _dedup_reconcile_update(**kwargs) + llm.call.assert_not_called() + conn.execute.assert_not_called() diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index a434702aff..d25f826103 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -1220,6 +1220,7 @@ Observations are deduplicated, evidence-grounded knowledge consolidated from mul | `HINDSIGHT_API_CONSOLIDATION_MAX_MEMORIES_PER_ROUND` | Maximum memories processed per consolidation round. When the limit is reached, the job yields its worker slot and re-queues itself so other banks get fair scheduling. Mental model refreshes only run on the final round. `0` = unlimited. Configurable per bank. | `100` | | `HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS` | Max tokens for recall when finding related observations during consolidation | `1024` | | `HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE` | Number of facts sent to the LLM in a single consolidation call. Higher values reduce LLM calls and improve throughput at the cost of larger prompts. Set to `1` to disable batching. Configurable per bank. | `8` | +| `HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD` | Cosine similarity at/above which a newly-created or freshly-updated observation is reconciled against an existing near-identical one via a focused 1-by-1 LLM "merge or keep" call (the model reads both texts, so a number/negation/entity difference is respected). Catches near-duplicate observations that weaker consolidation models emit even when shown the twin, as well as duplicates that arise when an update rewrites an observation into a near-twin of another. `1.0` disables it. Postgres only. | `1.0` (disabled) | | `HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM` | Maximum number of tag groups consolidated concurrently within one consolidation op. Each group acquires per-scope locks before processing, so groups whose write scopes overlap (e.g. under `per_tag` / `all_combinations` / explicit-list `observation_scopes`) automatically serialise on the overlapping scopes — actual concurrency may be lower than this cap when scopes contend. Set to `1` for fully sequential behaviour. Higher values raise peak LLM QPS and connection-pool usage during consolidation proportionally — tune down if your LLM provider rate-limits tightly or your DB pool is small. Configurable per bank. | `4` | | `HINDSIGHT_API_CONSOLIDATION_RECALL_BUDGET` | Budget level for the recall pass inside consolidation (`low`, `mid`, `high`). Lower budgets fetch fewer candidate rows, reducing peak memory usage on large banks. | `low` | | `HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS` | Total token budget for source facts included with observations in the consolidation prompt. `-1` = unlimited. Configurable per bank. | `4096` | diff --git a/hindsight-docs/docs/developer/observations.mdx b/hindsight-docs/docs/developer/observations.mdx index 4097292004..501bce9303 100644 --- a/hindsight-docs/docs/developer/observations.mdx +++ b/hindsight-docs/docs/developer/observations.mdx @@ -52,6 +52,14 @@ After `retain()` completes, the consolidation engine runs automatically: 3. **Observation creation/update** — New observations are created or existing ones refined 4. **Evidence tracking** — Each observation maintains references to supporting facts +### Near-Duplicate Reconciliation + +Consolidation can still produce two observations that say the same thing in slightly different words — for example when a weaker model writes a near-identical observation instead of refining the existing one, or when refining an observation reshapes its wording so it overlaps another one. Left alone, these near-duplicates clutter recall with redundant beliefs. + +When enabled, Hindsight reconciles them automatically. Whenever an observation is created **or** updated, it is compared against the existing observations it most closely resembles. If one is highly similar, a focused check decides whether to **merge** them into a single belief (folding both sets of supporting evidence together) or **keep** them separate. Because the check reads the full text of both, observations that differ in a meaningful detail — a number, a negation, a named entity or language — are correctly kept apart rather than collapsed. + +This is controlled by the [`HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD`](/developer/configuration#observations) setting: the cosine similarity at or above which two observations are reconciled. It is **disabled by default** (`1.0`) and is available on PostgreSQL deployments. Lower it (e.g. `0.97`) to enable reconciliation; a lower value reconciles more aggressively. + ### Disabling Auto-Consolidation Set `HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION=false` (or configure per-bank via the [bank config API](/developer/api/memory-banks#observations-configuration)) to prevent consolidation from running automatically after retain, delete, and update operations. When disabled, consolidation only runs when you explicitly call the [consolidate endpoint](#trigger-consolidation). diff --git a/skills/hindsight-docs/references/developer/configuration.md b/skills/hindsight-docs/references/developer/configuration.md index 9c75c6dc67..91e523b644 100644 --- a/skills/hindsight-docs/references/developer/configuration.md +++ b/skills/hindsight-docs/references/developer/configuration.md @@ -1220,6 +1220,7 @@ Observations are deduplicated, evidence-grounded knowledge consolidated from mul | `HINDSIGHT_API_CONSOLIDATION_MAX_MEMORIES_PER_ROUND` | Maximum memories processed per consolidation round. When the limit is reached, the job yields its worker slot and re-queues itself so other banks get fair scheduling. Mental model refreshes only run on the final round. `0` = unlimited. Configurable per bank. | `100` | | `HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS` | Max tokens for recall when finding related observations during consolidation | `1024` | | `HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE` | Number of facts sent to the LLM in a single consolidation call. Higher values reduce LLM calls and improve throughput at the cost of larger prompts. Set to `1` to disable batching. Configurable per bank. | `8` | +| `HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD` | Cosine similarity at/above which a newly-created or freshly-updated observation is reconciled against an existing near-identical one via a focused 1-by-1 LLM "merge or keep" call (the model reads both texts, so a number/negation/entity difference is respected). Catches near-duplicate observations that weaker consolidation models emit even when shown the twin, as well as duplicates that arise when an update rewrites an observation into a near-twin of another. `1.0` disables it. Postgres only. | `1.0` (disabled) | | `HINDSIGHT_API_CONSOLIDATION_LLM_PARALLELISM` | Maximum number of tag groups consolidated concurrently within one consolidation op. Each group acquires per-scope locks before processing, so groups whose write scopes overlap (e.g. under `per_tag` / `all_combinations` / explicit-list `observation_scopes`) automatically serialise on the overlapping scopes — actual concurrency may be lower than this cap when scopes contend. Set to `1` for fully sequential behaviour. Higher values raise peak LLM QPS and connection-pool usage during consolidation proportionally — tune down if your LLM provider rate-limits tightly or your DB pool is small. Configurable per bank. | `4` | | `HINDSIGHT_API_CONSOLIDATION_RECALL_BUDGET` | Budget level for the recall pass inside consolidation (`low`, `mid`, `high`). Lower budgets fetch fewer candidate rows, reducing peak memory usage on large banks. | `low` | | `HINDSIGHT_API_CONSOLIDATION_SOURCE_FACTS_MAX_TOKENS` | Total token budget for source facts included with observations in the consolidation prompt. `-1` = unlimited. Configurable per bank. | `4096` | diff --git a/skills/hindsight-docs/references/developer/observations.md b/skills/hindsight-docs/references/developer/observations.md index 98978fd2d0..dc8b62f4ed 100644 --- a/skills/hindsight-docs/references/developer/observations.md +++ b/skills/hindsight-docs/references/developer/observations.md @@ -46,6 +46,14 @@ After `retain()` completes, the consolidation engine runs automatically: 3. **Observation creation/update** — New observations are created or existing ones refined 4. **Evidence tracking** — Each observation maintains references to supporting facts +### Near-Duplicate Reconciliation + +Consolidation can still produce two observations that say the same thing in slightly different words — for example when a weaker model writes a near-identical observation instead of refining the existing one, or when refining an observation reshapes its wording so it overlaps another one. Left alone, these near-duplicates clutter recall with redundant beliefs. + +When enabled, Hindsight reconciles them automatically. Whenever an observation is created **or** updated, it is compared against the existing observations it most closely resembles. If one is highly similar, a focused check decides whether to **merge** them into a single belief (folding both sets of supporting evidence together) or **keep** them separate. Because the check reads the full text of both, observations that differ in a meaningful detail — a number, a negation, a named entity or language — are correctly kept apart rather than collapsed. + +This is controlled by the [`HINDSIGHT_API_CONSOLIDATION_DEDUP_THRESHOLD`](configuration.md#observations) setting: the cosine similarity at or above which two observations are reconciled. It is **disabled by default** (`1.0`) and is available on PostgreSQL deployments. Lower it (e.g. `0.97`) to enable reconciliation; a lower value reconciles more aggressively. + ### Disabling Auto-Consolidation Set `HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION=false` (or configure per-bank via the [bank config API](api/memory-banks.md#observations-configuration)) to prevent consolidation from running automatically after retain, delete, and update operations. When disabled, consolidation only runs when you explicitly call the [consolidate endpoint](#trigger-consolidation).