# backend/memory/episodic_memory.py import logging import asyncio from uuid import uuid4 try: import chromadb from chromadb.config import Settings except ImportError: chromadb = None Settings = None try: import google.generativeai as genai from config import GEMINI_API_KEY # Initialize genai explicitly genai.configure(api_key=GEMINI_API_KEY) except ImportError: pass # "models/embedding-001" was RETIRED — it 404s ("not found for API version # v1beta"), which meant every embed() call fell through to the zero-vector # fallback below. That failed silently and is worse than an outright error: # identical all-zero vectors make every document equidistant, so semantic # retrieval returns essentially arbitrary results while still "working". # Verified live against this key — the only models exposing embedContent are # gemini-embedding-001, gemini-embedding-2 and gemini-embedding-2-preview. EMBED_MODEL = "models/gemini-embedding-001" EMBED_DIM = 768 # keep in step with the existing persona_* collections async def embed(text: str) -> list[float]: """Generates an embedding for the given text using Gemini.""" try: def _get_embedding(): result = genai.embed_content( model=EMBED_MODEL, content=text, task_type="retrieval_document", output_dimensionality=EMBED_DIM, ) return result['embedding'] vec = await asyncio.to_thread(_get_embedding) if not vec: raise ValueError("empty embedding returned") return vec except Exception as e: # Loud, not silent: a zero vector destroys ranking, so make it obvious # in the logs that retrieval quality is degraded rather than fine. logging.error( f"Embedding failed via {EMBED_MODEL} ({e}); returning a ZERO vector — " "semantic ranking is degraded until this is resolved." ) return [0.0] * EMBED_DIM class EpisodicMemory: def __init__(self, persist_dir: str): # ITEM: Cross-session persistence confirmed # Real ChromaDB persistent client — data survives app restarts if chromadb is None: logging.error("chromadb is not installed. Episodic memory will not work.") self.client = None return logging.getLogger("chromadb").setLevel(logging.ERROR) self.client = chromadb.PersistentClient( path=persist_dir, settings=Settings(anonymized_telemetry=False) ) def _get_collection(self, persona: str): if not self.client: return None return self.client.get_or_create_collection(f"persona_{persona.lower()}") async def add(self, text: str, metadata: dict, persona: str = "jarvis"): # ITEM: Memory write — EXE (exact add() call with collection=f"persona_{active_persona}") collection = self._get_collection(persona) if not collection: return embedding = await embed(text) metadata["persona"] = persona.lower() metadata["pinned"] = metadata.get("pinned", False) def _add(): collection.add( documents=[text], embeddings=[embedding], metadatas=[metadata], ids=[str(uuid4())] ) await asyncio.to_thread(_add) async def query(self, text: str, top_k=5, persona: str = "jarvis") -> list[dict]: # ITEM: Memory read/query — EXE (exact query() call filtered to active persona collection) collection = self._get_collection(persona) if not collection: return [] embedding = await embed(text) def _query(): return collection.query( query_embeddings=[embedding], n_results=top_k ) results = await asyncio.to_thread(_query) out = [] if results and results.get("documents") and results["documents"][0]: docs = results["documents"][0] metas = results["metadatas"][0] if results.get("metadatas") and results["metadatas"][0] else [{}] * len(docs) ids = results["ids"][0] for d, m, i in zip(docs, metas, ids): out.append({"id": i, "text": d, "metadata": m}) return out async def pin(self, memory_id: str, persona: str = "jarvis"): # ITEM: Memory pin — EXE (exact metadata update to mark pinned=True) collection = self._get_collection(persona) if not collection: return def _pin(): result = collection.get(ids=[memory_id]) if result and result['metadatas'] and len(result['metadatas']) > 0: meta = result['metadatas'][0] meta['pinned'] = True collection.update( ids=[memory_id], metadatas=[meta] ) await asyncio.to_thread(_pin) async def delete(self, memory_id: str, persona: str = "jarvis"): # ITEM: Memory delete — EXE (exact delete() call) collection = self._get_collection(persona) if not collection: return def _delete(): collection.delete(ids=[memory_id]) await asyncio.to_thread(_delete) async def embed_model_description(self, description: str, model_path: str, persona: str): collection = self._get_collection(f"{persona}_models") if not collection: return def _add(): collection.add( documents=[description], metadatas=[{"model_path": model_path, "engine": "stable_fast_3d_local"}], ids=[str(uuid4())] ) try: await asyncio.to_thread(_add) except Exception as e: logging.error(f"ChromaDB embed error: {e}") async def query_similar_model(self, description: str, threshold: float = 0.85, persona: str = "JARVIS") -> dict | None: collection = self._get_collection(f"{persona}_models") if not collection: return None def _query(): return collection.query( query_texts=[description], n_results=1 ) try: results = await asyncio.to_thread(_query) if results and results.get('distances') and results['distances'][0]: dist = results['distances'][0][0] if dist <= (1.0 - threshold): meta = results['metadatas'][0][0] return {"model_path": meta.get("model_path")} except Exception as e: logging.error(f"ChromaDB query error: {e}") return None