refactor: Separate syntheses layer from L0 facts and update locomo benchmarks

Author: laxmanclo

.gitignore

@@ -43,7 +43,7 @@ env/
 .coverage
 htmlcov/
 .tox/
-
+/data
 # Environment variables
 .env
 .env.local

benchmarks/locomo/locomo_runner.py

@@ -703,6 +703,12 @@ def _format_retrieved_context(search_results: Any) -> str:
         for i, ep in enumerate(episodes, 1):
             lines.append(f"{i}. {ep.get('text', str(ep))}")
 
+    syntheses = search_results.get("syntheses") or []
+    if syntheses:
+        lines.append("\n## Syntheses")
+        for i, sy in enumerate(syntheses, 1):
+            lines.append(f"{i}. {sy.get('text', str(sy))}")
+
     sentences = search_results.get("sentences") or []
     if sentences:
         session_sents: dict[str, list[dict[str, Any]]] = {}

test_synthesis.py

@@ -18,9 +18,8 @@ async def mock_emb(*args, **kwargs):
 
     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"])
+    syntheses = await v.db.search_syntheses([0.1, 0.2], "u1", limit=100)
+    for s in syntheses:
+        print("Synthesized:", s["text"])
 
 asyncio.run(test())

vektori/ingestion/synthesis.py

@@ -79,41 +79,37 @@ async def synthesize(self, user_id: str, agent_id: str | None = None) -> int:
         inserted = 0
 
         for fact_dict, emb in zip(new_facts, embeddings):
-            # Check dedup against existing synthesis facts
-            existing = await self.db.search_facts(
+            # Check dedup against existing synthesis
+            existing = await self.db.search_syntheses(
                 embedding=emb,
                 user_id=user_id,
                 agent_id=agent_id,
-                limit=1,
-                active_only=True
+                limit=1
             )
 
             skip = False
             if existing:
                 best = existing[0]
                 sim = 1.0 - best.get("distance", 1.0)
-                if sim > 0.85 and best.get("metadata", {}).get("source") == "synthesis":
-                    # We already have this synthesis fact, let's just update mentions
-                    await self.db.increment_fact_mentions(best["id"])
+                if sim > 0.85:
+                    # We already have this synthesis, let's just skip it
                     skip = True
 
             if skip:
                 continue
 
             try:
-                await self.db.insert_fact(
+                # Need to link the synthesis to the facts that generated it if we want graph traversal
+                # But for now we just insert it
+                synthesis_id = await self.db.insert_synthesis(
                     text=fact_dict["text"],
                     embedding=emb,
                     user_id=user_id,
                     agent_id=agent_id,
-                    session_id="synthesis", # Special session ID
-                    subject="user",
-                    confidence=fact_dict.get("confidence", 0.9),
-                    metadata={"source": "synthesis"},
-                    event_time=now
+                    session_id="synthesis"
                 )
                 inserted += 1
             except Exception as e:
-                logger.warning("Failed to insert synthesis fact: %s", e)
+                logger.warning("Failed to insert synthesis: %s", e)
 
         return inserted

vektori/retrieval/search.py

@@ -207,10 +207,10 @@ async def _search_stepped(
         if not seed_facts:
             logger.debug("search: no facts for user=%s, falling back to sentence search", user_id)
             if depth == "l0":
-                return {"facts": [], "episodes": [], "memory_found": False}
+                return {"facts": [], "episodes": [], "syntheses": [], "memory_found": False}
             return {
                 "facts": [],
-                "episodes": [],
+                "episodes": [], "syntheses": [],
                 "sentences": direct_sentences,
                 "memory_found": False,
             }
@@ -239,16 +239,19 @@ async def _search_stepped(
         if depth == "l0":
             # L0: graph traversal only — keep it light
             episodes = await self.db.get_episodes_for_facts(seed_fact_ids)
+            syntheses = await self.db.get_syntheses_for_facts(seed_fact_ids)
             top = _clean(top_k_facts)
-            return {"facts": top, "episodes": episodes, "memory_found": len(top) > 0}
+            return {"facts": top, "episodes": episodes, "syntheses": syntheses, "memory_found": len(top) > 0}
 
         # ── Step 2: Episodes (L1/L2) — graph traversal + vector search ────────
         # Run both concurrently: graph edges from matched facts, and direct
         # cosine search over episode embeddings. Merge by ID so the same
         # episode doesn't appear twice.
-        graph_episodes, vec_episodes = await asyncio.gather(
+        graph_episodes, vec_episodes, graph_syntheses, vec_syntheses = await asyncio.gather(
             self.db.get_episodes_for_facts(seed_fact_ids),
             self.db.search_episodes(query_embedding, user_id, agent_id),
+            self.db.get_syntheses_for_facts(seed_fact_ids),
+            self.db.search_syntheses(query_embedding, user_id, agent_id),
         )
         seen_episode_ids: set[str] = set()
         episodes: list[dict[str, Any]] = []
@@ -258,6 +261,14 @@ async def _search_stepped(
                 seen_episode_ids.add(iid)
                 episodes.append(ins)
 
+        seen_synthesis_ids: set[str] = set()
+        syntheses: list[dict[str, Any]] = []
+        for ins in (*graph_syntheses, *vec_syntheses):
+            iid = ins.get("id")
+            if iid and iid not in seen_synthesis_ids:
+                seen_synthesis_ids.add(iid)
+                syntheses.append(ins)
+
         # ── Step 3: Trace facts → source sentences ────────────────────────────
         source_sentence_ids = await self.db.get_source_sentences(seed_fact_ids)
 
@@ -289,7 +300,7 @@ async def _search_stepped(
         if not all_candidate_ids:
             return {
                 "facts": top_facts,
-                "episodes": episodes,
+                "episodes": episodes, "syntheses": syntheses,
                 "sentences": [],
                 "memory_found": memory_found,
             }
@@ -342,7 +353,7 @@ async def _search_stepped(
 
         return {
             "facts": top_facts,
-            "episodes": episodes,
+            "episodes": episodes, "syntheses": syntheses,
             "sentences": _dedup(result_sentences),
             "memory_found": memory_found,
         }
@@ -394,9 +405,11 @@ async def _search_l2_fast(
         top_facts = _clean(_diverse_top_k(scored_facts, top_k))
         top_fact_ids = [f["id"] for f in top_facts]
 
-        graph_episodes, vec_episodes = await asyncio.gather(
+        graph_episodes, vec_episodes, graph_syntheses, vec_syntheses = await asyncio.gather(
             self.db.get_episodes_for_facts(top_fact_ids),
             self.db.search_episodes(query_embedding, user_id, agent_id),
+            self.db.get_syntheses_for_facts(top_fact_ids),
+            self.db.search_syntheses(query_embedding, user_id, agent_id),
         )
         seen_episode_ids: set[str] = set()
         episodes: list[dict[str, Any]] = []
@@ -406,9 +419,17 @@ async def _search_l2_fast(
                 seen_episode_ids.add(iid)
                 episodes.append(ins)
 
+        seen_synthesis_ids: set[str] = set()
+        syntheses: list[dict[str, Any]] = []
+        for ins in (*graph_syntheses, *vec_syntheses):
+            iid = ins.get("id")
+            if iid and iid not in seen_synthesis_ids:
+                seen_synthesis_ids.add(iid)
+                syntheses.append(ins)
+
         return {
             "facts": top_facts,
-            "episodes": episodes,
+            "episodes": episodes, "syntheses": syntheses,
             "sentences": _dedup(raw.get("sentences", [])),
             "memory_found": len(top_facts) > 0,
         }
@@ -482,7 +503,7 @@ async def _search_one(embedding: list[float]) -> list[dict[str, Any]]:
             logger.debug("search_expanded: no facts found, falling back to sentence search")
             return {
                 "facts": [],
-                "episodes": [],
+                "episodes": [], "syntheses": [],
                 "sentences": direct_sentences,
                 "memory_found": False,
             }
@@ -508,9 +529,11 @@ async def _search_one(embedding: list[float]) -> list[dict[str, Any]]:
         top_facts = _diverse_top_k(scored_facts, top_k)
         fact_ids = [f["id"] for f in top_facts]
 
-        graph_episodes, vec_episodes, source_sentence_ids = await asyncio.gather(
+        graph_episodes, vec_episodes, graph_syntheses, vec_syntheses, source_sentence_ids = await asyncio.gather(
             self.db.get_episodes_for_facts(fact_ids),
             self.db.search_episodes(embeddings[0], user_id, agent_id),
+            self.db.get_syntheses_for_facts(fact_ids),
+            self.db.search_syntheses(embeddings[0], user_id, agent_id),
             self.db.get_source_sentences(fact_ids),
         )
         seen_episode_ids: set[str] = set()
@@ -521,6 +544,14 @@ async def _search_one(embedding: list[float]) -> list[dict[str, Any]]:
                 seen_episode_ids.add(iid)
                 episodes.append(ins)
 
+        seen_synthesis_ids: set[str] = set()
+        syntheses: list[dict[str, Any]] = []
+        for ins in (*graph_syntheses, *vec_syntheses):
+            iid = ins.get("id")
+            if iid and iid not in seen_synthesis_ids:
+                seen_synthesis_ids.add(iid)
+                syntheses.append(ins)
+
         # Session scoring: merge direct search + fact-linked sentences
         direct_sent_by_id: dict[str, dict[str, Any]] = {s["id"]: s for s in direct_sentences}
         missing_ids = [sid for sid in source_sentence_ids if sid not in direct_sent_by_id]
@@ -574,7 +605,7 @@ async def _search_one(embedding: list[float]) -> list[dict[str, Any]]:
 
         return {
             "facts": _clean(top_facts),
-            "episodes": episodes,
+            "episodes": episodes, "syntheses": syntheses,
             "sentences": _dedup(result_sentences),
             "memory_found": len(top_facts) > 0,
         }
@@ -596,23 +627,23 @@ async def _sentence_fallback(
         since they're stored synchronously in add().
         """
         if depth == "l0":
-            return {"facts": [], "episodes": [], "memory_found": False}
+            return {"facts": [], "episodes": [], "syntheses": [], "memory_found": False}
 
         sentences = await self.db.search_sentences(
             embedding=query_embedding,
             user_id=user_id,
             agent_id=agent_id,
             limit=top_k,
         )
-        return {"facts": [], "episodes": [], "sentences": sentences, "memory_found": False}
+        return {"facts": [], "episodes": [], "syntheses": [], "sentences": sentences, "memory_found": False}
 
 
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
 
 def _empty(depth: str) -> dict[str, Any]:
     """Return the correct empty structure for a given depth."""
-    result: dict[str, Any] = {"facts": [], "episodes": []}
+    result: dict[str, Any] = {"facts": [], "episodes": [], "syntheses": []}
     if depth in ("l1", "l2"):
         result["sentences"] = []
     return result

vektori/storage/base.py

@@ -196,6 +196,64 @@ async def insert_fact_sources(self, pairs: list[tuple[str, str]]) -> None:
         for fact_id, sentence_id in pairs:
             await self.insert_fact_source(fact_id, sentence_id)
 
+    # ── Syntheses ──
+
+    
+    async def insert_synthesis(self, text: str, embedding: list[float], user_id: str, agent_id: str | None = None, score: float = 1.0, mentions: int = 1, metadata: dict | None = None) -> str:
+        raise NotImplementedError()
+
+    
+    async def insert_synthesis_fact(self, synthesis_id: str, fact_id: str) -> None:
+        raise NotImplementedError()
+
+    
+    async def get_syntheses_for_facts(self, fact_ids: list[str]) -> list[dict[str, Any]]:
+        return []
+
+    
+    async def search_syntheses(self, embedding: list[float], user_id: str, agent_id: str | None = None, limit: int = 10, threshold: float = 0.0) -> list[dict[str, Any]]:
+        return []
+
+    @abstractmethod
+    async def get_source_sentences(self, fact_ids: list[str]) -> list[str]:
+        """Return sentence IDs that are sources for the given facts (via fact_sources)."""
+        ...
+
+    @abstractmethod
+    async def get_sentences_by_ids(self, sentence_ids: list[str]) -> list[dict[str, Any]]:
+        """Fetch full sentence rows for the given IDs."""
+        ...
+
+
+    # ── Syntheses ──
+
+    
+    async def insert_synthesis(self, text: str, embedding: list[float], user_id: str, agent_id: str | None = None, score: float = 1.0, mentions: int = 1, metadata: dict | None = None) -> str:
+        raise NotImplementedError()
+
+    
+    async def insert_synthesis_fact(self, synthesis_id: str, fact_id: str) -> None:
+        raise NotImplementedError()
+
+    
+    async def get_syntheses_for_facts(self, fact_ids: list[str]) -> list[dict[str, Any]]:
+        return []
+
+    
+    async def search_syntheses(self, embedding: list[float], user_id: str, agent_id: str | None = None, limit: int = 10, threshold: float = 0.0) -> list[dict[str, Any]]:
+        return []
+
+    @abstractmethod
+    async def get_source_sentences(self, fact_ids: list[str]) -> list[str]:
+        """Return sentence IDs that are sources for the given facts (via fact_sources)."""
+        ...
+
+    @abstractmethod
+    async def get_sentences_by_ids(self, sentence_ids: list[str]) -> list[dict[str, Any]]:
+        """Fetch full sentence rows for the given IDs."""
+        ...
+
+
     # ── Episodes ──
 
     @abstractmethod