Adding conflict resolution

Author: Aayush Kataria

Docs/concepts.md

@@ -114,6 +114,31 @@ Prompts for summarization and fact extraction live in `azure_functions/prompts/`
 
 ---
 
+## Memory Reconciliation
+
+The `reconcile_memories(user_id, n=50)` pipeline step reads up to N most-recent active facts for a user and asks the LLM to identify two orthogonal outcomes in one pass:
+
+- **Duplicates** — two or more facts that restate the same claim in different words. Resolution: collapse into one merged fact; the originals are soft-deleted with `supersede_reason="duplicate"` and `superseded_by` set to the merged fact's id.
+- **Contradictions** — two facts that assert opposing claims about the same subject. Resolution: keep the winner (more recent first, higher confidence as tiebreaker), soft-delete the loser with `supersede_reason="contradiction"` and `superseded_by` set to the winner.
+
+### Why one pass
+
+Detecting contradictions semantically requires the LLM to see the candidate pool as a whole — paraphrased ("user prefers aisle seats") and contradictory ("user is vegetarian" vs "user loves steak") facts often have very different embedding vectors and would never co-occur in any cosine cluster. Putting all N candidates into one prompt lets the LLM do the semantic reasoning across both axes simultaneously. The pipeline returns `{"kept": int, "merged": int, "contradicted": int}`.
+
+### Loser preservation
+
+Soft-deleted facts stay in the container with their `supersede_reason`, `superseded_at`, and `superseded_by` fields populated. Default reads (`get_memories`, `search_cosmos`) filter them out via `superseded_by IS NULL`. To inspect the audit trail (e.g. "show everything that ever applied to this user"), opt out of the filter at the query level.
+
+### Write-time exact dedup
+
+Each fact written by `extract_memories` carries a `content_hash` (SHA-256 of normalized content, truncated to 32 hex chars; lowercase, whitespace-collapsed). Before upserting a freshly-extracted fact, the pipeline checks the hash against existing active facts and short-circuits if a match exists, incrementing the `exact_dedup_skipped` metric. This catches identical re-extractions cheaply without an LLM call.
+
+### Tunable
+
+`DEDUP_EVERY_N` (default 5) controls how often `reconcile_memories` runs in the auto-trigger path. Set to `0` to disable. The candidate cap `n` (default 50) is tunable per call; larger values give the LLM a wider view at higher token cost.
+
+---
+
 ## Automatic Processing (Change Feed)
 
 In addition to on-demand processing via the SDK, the toolkit includes a Cosmos DB change feed trigger that **automatically** starts processing orchestrations when enough new turns have been written.

README.md

@@ -205,14 +205,25 @@ results = memory.search_cosmos("user preferences", user_id="u1", min_confidence=
 high_conf_facts = memory.get_memories(user_id="u1", memory_type="fact", min_confidence=0.7)
 ```
 
+### Memory Reconciliation
+
+`reconcile_memories(user_id, n=50)` collapses paraphrased duplicates and resolves semantic contradictions in a single LLM pass over the N most-recent active facts. Both outcomes soft-delete the loser with a `supersede_reason` of `"duplicate"` or `"contradiction"`. See [docs/concepts.md](docs/concepts.md#memory-reconciliation) for details.
+
+| New `MemoryRecord` field | Meaning |
+|---|---|
+| `content_hash` | SHA-256 of normalized content; enables write-time exact-dedup short-circuit |
+| `supersede_reason` | `"duplicate"` or `"contradiction"` (None for live records) |
+| `superseded_at` | ISO timestamp when the supersede happened (None for live records) |
+| `superseded_by` | Id of the record that replaced this one (existing field) |
+
 ### Auto-trigger (per-turn extraction)
 
 By default, the **InProcess processor** runs each pipeline step independently as its own threshold trips inside `push_to_cosmos()`:
 
 | Env var | Default | Step that fires | Async behavior |
 |---|---|---|---|
 | `FACT_EXTRACTION_EVERY_N` | `1` (every turn) | `process_extract_memories` | scheduled via `asyncio.create_task` |
-| `DEDUP_EVERY_N` | `5` | `process_dedup` (fires every Nth extract → effectively every `FACT_EXTRACTION_EVERY_N × DEDUP_EVERY_N` turns) | scheduled via `asyncio.create_task` |
+| `DEDUP_EVERY_N` | `5` | `process_reconcile` (fires every Nth extract → effectively every `FACT_EXTRACTION_EVERY_N × DEDUP_EVERY_N` turns) | scheduled via `asyncio.create_task` |
 | `THREAD_SUMMARY_EVERY_N` | `10` | `process_thread_summary` | scheduled via `asyncio.create_task` |
 | `USER_SUMMARY_EVERY_N` | `20` | `process_user_summary` | scheduled via `asyncio.create_task` |
 

Samples/Demo.ipynb

@@ -16,7 +16,7 @@
     "5. **Memory Extraction** – `extract_memories()` (facts + episodic + procedural)\n",
     "6. **User Summary** – `generate_user_summary()` (cross-thread profile)\n",
     "7. **Vector / hybrid search** – `search_cosmos()`\n",
-    "8. **Tagging, salience & deduplication** – tag mutation, salience filter, `deduplicate_facts()`\n",
+    "8. **Tagging, salience & deduplication** – tag mutation, salience filter, `reconcile()`\n",
     "9. **Automatic processing (Change Feed)** – optional Azure Function for background processing\n",
     "\n",
     "> 💡 **Tip:** the synchronous `CosmosMemoryClient` accepts an optional `processor=` kwarg (defaults to `InProcessProcessor`). Pass `DurableFunctionProcessor()` to delegate summarization to the sibling Azure Function app — see `Samples/scenario_remote_processor.py`.\n"

Samples/scenario_memory_reconciliation.py

@@ -0,0 +1,178 @@
+"""Scenario: Memory reconciliation — duplicate merging and contradiction resolution.
+
+Demonstrates how `reconcile_memories` collapses paraphrased facts and resolves
+semantic contradictions in a single LLM pass:
+
+1. Seed paraphrased facts about a user (will be merged).
+2. Seed contradicting facts about the same subject (one will win, one will lose).
+3. Run reconcile() and print the {kept, merged, contradicted} stats.
+4. Show the live state — paraphrased duplicates collapsed, contradiction loser hidden.
+5. Show the audit trail (include_superseded=True) — soft-deleted records carry
+   supersede_reason, superseded_at, and superseded_by pointing at the survivor.
+
+Requirements:
+    - Azure Cosmos DB account with vector-search enabled.
+    - Azure AI Foundry endpoint with chat + embeddings deployments.
+    - Environment variables (.env supported via python-dotenv):
+        COSMOS_DB_ENDPOINT
+        COSMOS_DB_KEY                          (optional, falls back to DefaultAzureCredential)
+        AI_FOUNDRY_ENDPOINT
+        AI_FOUNDRY_EMBEDDING_DEPLOYMENT_NAME
+        AI_FOUNDRY_CHAT_DEPLOYMENT_NAME
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import uuid
+
+from dotenv import load_dotenv
+
+from agent_memory_toolkit import CosmosMemoryClient
+
+load_dotenv()
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+DIVIDER = "-" * 60
+
+
+def banner(title: str) -> None:
+    """Print a section banner."""
+    print(f"\n{DIVIDER}")
+    print(f"  {title}")
+    print(DIVIDER)
+
+
+def print_facts(facts: list[dict]) -> None:
+    """Pretty-print fact records (id + content)."""
+    if not facts:
+        print("  (none)")
+        return
+    for f in facts:
+        print(f"  id={f['id']}  content: {f.get('content', '')}")
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+PARAPHRASED_FACTS = [
+    "User prefers aisle seats on flights.",
+    "User likes aisle seats when flying.",
+    "User always picks aisle when booking flights.",
+]
+
+CONTRADICTING_FACTS = [
+    "User is strictly vegetarian and avoids all meat.",
+    "User loves a good ribeye steak.",
+]
+
+
+def main() -> None:
+    required = ["COSMOS_DB_ENDPOINT", "AI_FOUNDRY_ENDPOINT"]
+    missing = [v for v in required if not os.environ.get(v)]
+    if missing:
+        print(f"ERROR: missing env vars: {', '.join(missing)}")
+        sys.exit(1)
+
+    mem = CosmosMemoryClient(
+        cosmos_endpoint=os.environ["COSMOS_DB_ENDPOINT"],
+        cosmos_key=os.environ.get("COSMOS_DB_KEY") or None,
+        cosmos_database=os.environ.get("COSMOS_DB_DATABASE", "ai_memory"),
+        cosmos_container=os.environ.get("COSMOS_DB_CONTAINER", "memories"),
+        ai_foundry_endpoint=os.environ["AI_FOUNDRY_ENDPOINT"],
+        ai_foundry_api_key=os.environ.get("AI_FOUNDRY_API_KEY") or None,
+        embedding_deployment_name=os.environ.get("AI_FOUNDRY_EMBEDDING_DEPLOYMENT_NAME", "text-embedding-3-large"),
+        chat_deployment_name=os.environ.get("AI_FOUNDRY_CHAT_DEPLOYMENT_NAME", "gpt-4o-mini"),
+        use_default_credential=True,
+    )
+    print("Connected to Cosmos DB.")
+
+    unique_user_id = f"reconcile-demo-{uuid.uuid4().hex[:8]}"
+    unique_thread_id = f"reconcile-demo-thread-{uuid.uuid4().hex[:8]}"
+    print(f"User ID:   {unique_user_id}")
+    print(f"Thread ID: {unique_thread_id}")
+
+    try:
+        banner("1. Seeding paraphrased facts (duplicates)")
+        for content in PARAPHRASED_FACTS:
+            mem.add_cosmos(
+                user_id=unique_user_id,
+                role="user",
+                content=content,
+                memory_type="fact",
+                thread_id=unique_thread_id,
+                salience=0.7,
+            )
+            print(f"  + {content}")
+
+        banner("2. Seeding contradicting facts")
+        for content in CONTRADICTING_FACTS:
+            mem.add_cosmos(
+                user_id=unique_user_id,
+                role="user",
+                content=content,
+                memory_type="fact",
+                thread_id=unique_thread_id,
+                salience=0.7,
+            )
+            print(f"  + {content}")
+
+        banner("3. Active facts before reconcile")
+        before = mem.get_memories(user_id=unique_user_id, memory_type="fact")
+        print_facts(before)
+
+        banner("4. Running reconcile_memories")
+        stats = mem.reconcile(user_id=unique_user_id)
+        print(f"  stats: {dict(stats)}")
+
+        banner("5. Active facts after reconcile (duplicates merged, contradictions resolved)")
+        after = mem.get_memories(user_id=unique_user_id, memory_type="fact")
+        print_facts(after)
+
+        banner("6. Audit trail (soft-deleted records)")
+        all_facts = mem.get_memories(
+            user_id=unique_user_id,
+            memory_type="fact",
+            include_superseded=True,
+        )
+        soft_deleted = [f for f in all_facts if f.get("supersede_reason")]
+        if not soft_deleted:
+            print("  (no soft-deleted records)")
+        for f in soft_deleted:
+            print(
+                f"  id={f['id']}  reason={f.get('supersede_reason')}  "
+                f"superseded_at={f.get('superseded_at')}  "
+                f"superseded_by={f.get('superseded_by')}"
+            )
+            print(f"  content: {f.get('content', '')}")
+
+    finally:
+        banner("7. Cleanup")
+        try:
+            all_records = mem.get_memories(
+                user_id=unique_user_id,
+                include_superseded=True,
+            )
+            deleted = 0
+            for rec in all_records:
+                try:
+                    mem.delete_cosmos(
+                        memory_id=rec["id"],
+                        thread_id=rec.get("thread_id", unique_thread_id),
+                        user_id=unique_user_id,
+                    )
+                    deleted += 1
+                except Exception as exc:  # pragma: no cover - best effort cleanup
+                    print(f"  WARN: failed to delete {rec.get('id')}: {exc}")
+            print(f"  Deleted {deleted} record(s) for user {unique_user_id}")
+        except Exception as exc:  # pragma: no cover - best effort cleanup
+            print(f"  WARN: cleanup failed: {exc}")
+
+
+if __name__ == "__main__":
+    main()

agent_memory_toolkit/aio/cosmos_memory_client.py

@@ -1607,10 +1607,10 @@ async def _run_auto_trigger_steps(
 
             if fire_dedup:
                 try:
-                    await processor.process_dedup(user_id=user_id)
+                    await processor.process_reconcile(user_id=user_id)
                 except Exception as exc:
                     logger.warning(
-                        "Auto-trigger process_dedup failed for %s: %s",
+                        "Auto-trigger process_reconcile failed for %s: %s",
                         user_id,
                         exc,
                     )
@@ -1619,7 +1619,7 @@ async def _run_auto_trigger_steps(
                         thread_counter_id(user_id, thread_id),
                         user_id,
                         thread_id,
-                        f"process_dedup: {exc!r}",
+                        f"process_reconcile: {exc!r}",
                     )
 
             if fire_summary:
@@ -1721,20 +1721,19 @@ async def generate_user_summary(
         self._require_pipeline()
         return await asyncio.to_thread(self._pipeline.generate_user_summary, user_id, thread_ids, recent_k)
 
-    async def deduplicate_facts(
+    async def reconcile(
         self,
         user_id: str,
-        similarity_threshold: float = 0.9,
-        max_facts: int = 200,
+        n: int = 50,
     ) -> dict[str, int]:
-        """Run LLM-based dedup on user's facts.
+        """Reconcile a user's facts via the contradiction-aware dedup pass.
 
         Pipeline calls are dispatched to a worker thread via
         :func:`asyncio.to_thread` to avoid blocking the event loop.
         """
         await self._require_cosmos()
         self._require_pipeline()
-        return await asyncio.to_thread(self._pipeline.deduplicate_facts, user_id, similarity_threshold, max_facts)
+        return await asyncio.to_thread(self._pipeline.reconcile_memories, user_id, n)
 
     # ------------------------------------------------------------------
     # Processor delegation

agent_memory_toolkit/aio/processors/base.py

@@ -49,10 +49,11 @@ async def process_user_summary(
         thread_ids: Optional[list[str]] = None,
     ) -> UserSummaryResult: ...
 
-    async def process_dedup(
+    async def process_reconcile(
         self,
         *,
         user_id: str,
+        n: int = 50,
     ) -> int: ...
 
     async def generate_user_summary(

agent_memory_toolkit/aio/processors/durable.py

@@ -33,7 +33,7 @@ async def process_thread(
             thread_id,
             len(turns) if turns else 0,
         )
-        return ProcessThreadResult(thread_summary=None, extracted_counts={}, deduplicated_count=0, elapsed_ms=0)
+        return ProcessThreadResult(thread_summary=None, extracted_counts={}, reconciled_count=0, elapsed_ms=0)
 
     async def process_extract_memories(
         self,
@@ -73,8 +73,12 @@ async def process_user_summary(
         )
         return UserSummaryResult(summary=None)
 
-    async def process_dedup(self, *, user_id: str) -> int:
-        logger.debug("AsyncDurableFunctionProcessor.process_dedup no-op user_id=%s", user_id)
+    async def process_reconcile(self, *, user_id: str, n: int = 50) -> int:
+        logger.debug(
+            "AsyncDurableFunctionProcessor.process_reconcile no-op user_id=%s n=%d",
+            user_id,
+            n,
+        )
         return 0
 
     async def generate_user_summary(

agent_memory_toolkit/aio/processors/inprocess.py

@@ -66,9 +66,9 @@ async def process_thread(
 
         thread_summary = await asyncio.to_thread(self._pipeline.generate_thread_summary, user_id, thread_id)
         extracted = await asyncio.to_thread(self._pipeline.extract_memories, user_id, thread_id)
-        dedup = await asyncio.to_thread(self._pipeline.deduplicate_facts, user_id)
+        reconciled = await asyncio.to_thread(self._pipeline.reconcile_memories, user_id)
 
-        deduped_count = self._extract_dedup_count(dedup)
+        deduped_count = self._extract_reconcile_count(reconciled)
 
         extracted_counts: dict[str, int] = (
             {k: v for k, v in extracted.items() if isinstance(v, int)} if isinstance(extracted, dict) else {}
@@ -78,7 +78,7 @@ async def process_thread(
         return ProcessThreadResult(
             thread_summary=thread_summary if isinstance(thread_summary, dict) else None,
             extracted_counts=extracted_counts,
-            deduplicated_count=deduped_count,
+            reconciled_count=deduped_count,
             elapsed_ms=elapsed_ms,
         )
 
@@ -109,24 +109,24 @@ async def process_user_summary(
         summary = await asyncio.to_thread(self._pipeline.generate_user_summary, user_id, thread_ids)
         return UserSummaryResult(summary=summary if isinstance(summary, dict) else None)
 
-    async def process_dedup(self, *, user_id: str) -> int:
-        dedup = await asyncio.to_thread(self._pipeline.deduplicate_facts, user_id)
-        return self._extract_dedup_count(dedup)
+    async def process_reconcile(self, *, user_id: str, n: int = 50) -> int:
+        reconciled = await asyncio.to_thread(self._pipeline.reconcile_memories, user_id, n)
+        return self._extract_reconcile_count(reconciled)
 
     @staticmethod
-    def _extract_dedup_count(dedup: Any) -> int:
-        """Sum the merged + superseded facts from a ``deduplicate_facts`` result.
+    def _extract_reconcile_count(reconciled: Any) -> int:
+        """Sum ``merged + contradicted`` from a ``reconcile_memories`` result.
 
-        ``ProcessingPipeline.deduplicate_facts`` returns a dict with
-        ``{"kept", "merged", "superseded"}`` — both ``merged`` and
-        ``superseded`` represent facts that were consolidated, so they
-        contribute to the dedup-count metric.
+        ``ProcessingPipeline.reconcile_memories`` returns a dict with
+        ``{"kept", "merged", "contradicted"}`` — both ``merged`` and
+        ``contradicted`` represent facts that were consolidated or retired,
+        so they contribute to the dedup-count metric.
         """
-        if not isinstance(dedup, dict):
+        if not isinstance(reconciled, dict):
             return 0
-        merged = dedup.get("merged", 0) if isinstance(dedup.get("merged"), int) else 0
-        superseded = dedup.get("superseded", 0) if isinstance(dedup.get("superseded"), int) else 0
-        return merged + superseded
+        merged = reconciled.get("merged", 0) if isinstance(reconciled.get("merged"), int) else 0
+        contradicted = reconciled.get("contradicted", 0) if isinstance(reconciled.get("contradicted"), int) else 0
+        return merged + contradicted
 
     async def generate_user_summary(
         self,