Feat/benchmark query expansion

benchmarks/longmemeval/longmemeval_runner.py

@@ -355,11 +355,13 @@ async def _answer_question(
         question_date = instance.get("question_date") or ""
 
         retrieval_t0 = time.perf_counter()
+        use_expansion = question_type in ("multi-session", "temporal-reasoning")
         search_results = await self.vektori_client.search(
             query=question,
             user_id=user_id,
             depth=self.config.retrieval_depth,
             reference_date=_parse_date(question_date) if question_date else None,
+            expand=use_expansion,
         )
         retrieval_ms = (time.perf_counter() - retrieval_t0) * 1000
 

test_conflict.py

@@ -0,0 +1,43 @@
+import asyncio
+from vektori.storage.memory import MemoryBackend
+from vektori.ingestion.extractor import FactExtractor
+from vektori.models.base import LLMProvider, EmbeddingProvider
+
+class MockEmbedder(EmbeddingProvider):
+    @property
+    def dimension(self): return 2
+    async def embed(self, text):
+        if "hate" in text: return [0.5, -0.5]
+        return [0.5, 0.5]
+    async def embed_batch(self, texts):
+        return [await self.embed(t) for t in texts]
+
+class MockLLM(LLMProvider):
+    async def generate(self, prompt, max_tokens=1000, **kwargs):
+        pass
+
+async def test():
+    db = MemoryBackend()
+    await db.initialize()
+    embedder = MockEmbedder()
+    llm = MockLLM()
+    extractor = FactExtractor(db, embedder, llm)
+
+    # insert first fact
+    await extractor._process_facts(
+        [{"text": "User loves apples"}],
+        session_id="s1", user_id="u1", agent_id=None, conversation="User: I love apples"
+    )
+
+    # insert second fact
+    await extractor._process_facts(
+        [{"text": "User hates apples"}],
+        session_id="s2", user_id="u1", agent_id=None, conversation="User: Actually I hate apples"
+    )
+
+    facts = await db.get_active_facts("u1")
+    print("Active facts count:", len(facts))
+    for f in facts:
+        print(f["text"])
+
+asyncio.run(test())

test_conflict2.py

@@ -0,0 +1,39 @@
+import asyncio
+from vektori.storage.memory import MemoryBackend
+from vektori.ingestion.extractor import FactExtractor
+from vektori.models.base import LLMProvider, EmbeddingProvider
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+
+class MockEmbedder(EmbeddingProvider):
+    @property
+    def dimension(self): return 2
+    async def embed(self, text):
+        if "hate" in text: return [0.5, 0.4]  # High similarity to trigger check
+        return [0.5, 0.5]
+    async def embed_batch(self, texts):
+        return [await self.embed(t) for t in texts]
+
+class MockLLM(LLMProvider):
+    async def generate(self, prompt, max_tokens=1000, **kwargs):
+        if "loves apples" in prompt and "hates apples" in prompt:
+            old_id = prompt.split("- [")[1].split("]")[0]
+            return f'{{"supersedes_id": "{old_id}"}}'
+        return "{}"
+
+async def test():
+    db = MemoryBackend()
+    await db.initialize()
+    embedder = MockEmbedder()
+    llm = MockLLM()
+    extractor = FactExtractor(db, embedder, llm)
+
+    await extractor._process_facts([{"text": "User loves apples"}], "s1", "u1", None, "")
+    await extractor._process_facts([{"text": "User hates apples"}], "s2", "u1", None, "")
+
+    facts = await db.get_active_facts("u1")
+    print("Active facts count:", len(facts))
+    for f in facts: print("-", f["text"])
+
+asyncio.run(test())

test_llm_dedup_check.py

@@ -0,0 +1,33 @@
+import asyncio
+from vektori.storage.memory import MemoryBackend
+from vektori.ingestion.extractor import FactExtractor
+from vektori.models.base import LLMProvider, EmbeddingProvider
+
+class MockEmbedder(EmbeddingProvider):
+    @property
+    def dimension(self): return 2
+    async def embed(self, text):
+        if "hate" in text: return [0.5, 0.4] # Give sim around 0.8 to test LLM
+        return [0.5, 0.5]
+    async def embed_batch(self, texts):
+        return [await self.embed(t) for t in texts]
+
+class MockLLM(LLMProvider):
+    async def generate(self, prompt, max_tokens=1000, **kwargs):
+        if "contradict" in prompt.lower():
+            # If it has "hate", it contradicts the "love" one
+            if "hates apples" in prompt and "loves apples" in prompt:
+                return '{"contradicts": "1"}'
+            return '{"contradicts": null}'
+        pass
+
+async def test():
+    db = MemoryBackend()
+    await db.initialize()
+    embedder = MockEmbedder()
+    llm = MockLLM()
+    extractor = FactExtractor(db, embedder, llm)
+
+    # Needs actual logic in Extractor...
+
+asyncio.run(test())

test_synthesis.py

@@ -0,0 +1,26 @@
+import asyncio
+from vektori import Vektori
+
+async def test():
+    v = Vektori(storage_backend="memory")
+    await v._ensure_initialized()
+    # Mock LLM to just return a synthesis fact
+    async def mock_gen(*args, **kwargs):
+        return '{"facts": [{"text": "User loves fruit overall."}]}'
+    async def mock_emb(*args, **kwargs):
+        return [[0.1, 0.2]]
+    v.llm.generate = mock_gen
+    v.embedder.embed_batch = mock_emb
+
+    # add dummy facts so synthesize triggers (needs >=5 base facts)
+    for i in range(5):
+        await v.db.insert_fact(f"fact {i}", [0.1, 0.2], "u1", confidence=1.0)
+
+    n = await v.synthesize("u1")
+    print("New synthesized facts:", n)
+    facts = await v.db.get_active_facts("u1", limit=100)
+    for f in facts:
+        if f.get("metadata", {}).get("source") == "synthesis":
+            print("Synthesized:", f["text"])
+
+asyncio.run(test())

vektori/client.py

@@ -39,6 +39,7 @@ def __init__(
         context_window: int = 3,
         temporal_decay_rate: float = 0.001,
         async_extraction: bool = True,
+        synthesis_interval: int | None = 5,
         qdrant_api_key: str | None = None,
         milvus_token: str | None = None,
         agent_type: str = "general",
@@ -61,6 +62,7 @@ def __init__(
                 context_window=context_window,
                 temporal_decay_rate=temporal_decay_rate,
                 async_extraction=async_extraction,
+                synthesis_interval=synthesis_interval,
                 qdrant_api_key=qdrant_api_key,
                 milvus_token=milvus_token,
                 extraction_config=ec,
@@ -101,6 +103,13 @@ async def _initialize(self) -> None:
             max_output_tokens=self.config.max_extraction_output_tokens,
             extraction_config=self.config.extraction_config,
         )
+        # Import syntheizer lazily or at module level. Let's do it inside the file at the top or here.
+        from vektori.ingestion.synthesis import Synthesizer
+        self._synthesizer = Synthesizer(
+            db=self.db,
+            embedder=self.embedder,
+            llm=self.llm,
+        )
         self._search = SearchPipeline(
             db=self.db,
             embedder=self.embedder,
@@ -155,10 +164,52 @@ async def add(
             {"status": "ok", "sentences_stored": N, "extraction": "queued"|"done"|"skipped"}
         """
         await self._ensure_initialized()
-        return await self._pipeline.ingest(
+
+        result = await self._pipeline.ingest(
             messages, session_id, user_id, agent_id, metadata, session_time=session_time
         )
 
+        n = self.config.synthesis_interval
+        if n is not None and n > 0:
+            # We don't want to block the caller on synthesis since it could be slow.
+            # However, if async_extraction is False, we should do it inline.
+            if getattr(self.config, "async_extraction", True):
+                import asyncio
+                if getattr(self, "_bg_tasks", None) is None:
+                    self._bg_tasks = set()
+                # Fire and forget safely
+                async def _bg_synthesize(uid: str, aid: str | None) -> None:
+                    try:
+                        count = await self.db.count_sessions(user_id=uid, agent_id=aid)
+                        if count > 0 and count % n == 0:
+                            await self.synthesize(user_id=uid, agent_id=aid)
+                    except Exception as e:
+                        logger.error("Auto-synthesis failed: %s", e)
+
+                task = asyncio.create_task(_bg_synthesize(user_id, agent_id))
+                self._bg_tasks.add(task)
+                task.add_done_callback(self._bg_tasks.discard)
+            else:
+                count = await self.db.count_sessions(user_id=user_id, agent_id=agent_id)
+                if count > 0 and count % n == 0:
+                    await self.synthesize(user_id=user_id, agent_id=agent_id)
+
+        return result
+
+    async def synthesize(self, user_id: str, agent_id: str | None = None) -> int:
+        """
+        Run a cross-session synthesis pass for the user.
+        Examines active facts across sessions to form aggregate/macro facts
+        (e.g., "User consistently prefers X", "User has done Y across multiple sessions").
+
+        Returns the number of new aggregate facts inserted.
+        """
+        await self._ensure_initialized()
+        if not hasattr(self, "_synthesizer"):
+            logger.warning("Synthesizer not initialized.")
+            return 0
+        return await self._synthesizer.synthesize(user_id, agent_id)
+
     async def search(
         self,
         query: str,

vektori/config.py

@@ -153,6 +153,10 @@ class VektoriConfig:
     # Token-threshold batching — fire extraction once buffered input exceeds this
     token_batch_threshold: int = 800  # ~800 tokens ≈ 3-4 turns before extraction fires
 
+    # Automatically synthesize aggregate cross-session facts every N sessions
+    # (None disables auto-synthesis)
+    synthesis_interval: int | None = 5
+
     # Hard token limits for extraction LLM calls (API-level, not prompt hints)
     max_extraction_input_tokens: int = 4000  # truncate conversation before sending to LLM
     max_extraction_output_tokens: int = (

vektori/ingestion/extractor.py

@@ -685,7 +685,7 @@ async def _process_facts(
 
                 # Write-time semantic dedup
                 dedup = await self._check_dedup(
-                    fact_embedding, session_id, user_id, agent_id, subject
+                    fact_data["text"], fact_embedding, session_id, user_id, agent_id, subject
                 )
 
                 if dedup is not None:
@@ -787,7 +787,9 @@ async def replay_facts(
 
                 # Dedup check — in a fresh user context this always returns None,
                 # but we run it anyway for correctness if sessions overlap.
-                dedup = await self._check_dedup(fact_emb, session_id, user_id, agent_id, subject)
+                dedup = await self._check_dedup(
+                    fact_data["text"], fact_emb, session_id, user_id, agent_id, subject
+                )
                 if dedup is not None:
                     existing_id, same_session = dedup
                     await self.db.increment_fact_mentions(existing_id)
@@ -834,6 +836,7 @@ async def replay_facts(
 
     async def _check_dedup(
         self,
+        fact_text: str,
         fact_embedding: list[float],
         session_id: str,
         user_id: str,
@@ -869,6 +872,26 @@ async def _check_dedup(
                 return (best["id"], True)
             if not same_session and sim > 0.85:
                 return (best["id"], False)
+
+            # Contradiction LLM check for near-misses
+            # If the fact wasn't deduped by bare similarity, check if it contradicts.
+            # Only consider same-subject facts that are reasonably close (sim > 0.65).
+            plausible_conflicts = [
+                c for c in candidates
+                if (1.0 - c.get("distance", 1.0)) > 0.65
+            ]
+            if plausible_conflicts:
+                facts_text = "\n".join(f"- [{c['id']}] {c['text']}" for c in plausible_conflicts)
+                prompt = CONTRADICTION_PROMPT.format(fact_text=fact_text, existing_facts=facts_text)
+                try:
+                    response = await self.llm.generate(prompt, max_tokens=100)
+                    data = _parse_json_response(response)
+                    supersedes_id = data.get("supersedes_id")
+                    if supersedes_id and any(c["id"] == supersedes_id for c in plausible_conflicts):
+                        return (supersedes_id, False)
+                except Exception as e:
+                    logger.debug("Contradiction check failed: %s", e)
+
         except Exception as e:
             logger.warning("Dedup lookup failed: %s", e)
         return None
@@ -1042,3 +1065,17 @@ def _parse_json_response(response: str) -> dict[str, Any]:
     except json.JSONDecodeError as e:
         logger.error("Failed to parse extraction JSON: %s\nResponse: %.500s", e, response)
         return {"facts": []}
+
+# ── Contradiction prompt ──────────────────────────────────────────────────────
+CONTRADICTION_PROMPT = """Do any of the existing facts contradict or get superseded by the new fact?
+New fact: "{fact_text}"
+
+Existing facts:
+{existing_facts}
+
+If the new fact updates, contradicts, or replaces an existing fact, return its ID. Otherwise return null.
+
+Return ONLY JSON:
+{{
+  "supersedes_id": "fact_id_here" or null
+}}"""