import os import sqlite3 import logging import uuid from contextlib import contextmanager from pathlib import Path from typing import List, Dict, Any, Optional, Union from datetime import datetime from langchain_core.messages import BaseMessage, HumanMessage log = logging.getLogger(__name__) # Constants LILITH_HOME = Path(os.getenv("LILITH_HOME", ".lilith")) MEMORY_DB_PATH = LILITH_HOME / "long_term_memory.sqlite" MEMORY_CONTEXT_CHAR_BUDGET = 3000 MIN_MESSAGES_FOR_EXTRACTION = 2 MEMORY_LOG_PREVIEW_CHARS = 300 def _content_to_text(content: Any) -> str: if content is None: return "" if isinstance(content, str): return content if isinstance(content, list): parts = [] for item in content: if isinstance(item, str): parts.append(item) elif isinstance(item, dict): text = item.get("text") if text is not None: parts.append(str(text)) elif "content" in item: nested = _content_to_text(item["content"]) if nested: parts.append(nested) else: parts.append(str(item)) return "\n".join(part for part in parts if part) return str(content) def _memory_content_to_text(content: Any) -> str: if hasattr(content, "content"): return _content_to_text(content.content) return _content_to_text(content) def _preview_for_log(text: str, limit: int = MEMORY_LOG_PREVIEW_CHARS) -> str: compact = " ".join(str(text).split()) if len(compact) <= limit: return compact return compact[: limit - 1] + "…" def _print_memory_log(message: str, *args: Any) -> None: text = message % args if args else message print(text, flush=True) class MemoryStore: def __init__(self, db_path: Union[Path, str] = MEMORY_DB_PATH): self._in_memory = (str(db_path) == ":memory:") self.db_path = db_path if self._in_memory else Path(db_path) self._mem_conn: Optional[sqlite3.Connection] = None self._init_db() def _connect(self) -> sqlite3.Connection: if self._in_memory: if self._mem_conn is None: self._mem_conn = sqlite3.connect(":memory:", check_same_thread=False) return self._mem_conn return sqlite3.connect(str(self.db_path)) def close(self): if self._mem_conn is not None: self._mem_conn.close() self._mem_conn = None def _init_db(self): if not self._in_memory: Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) conn = self._connect() with conn: conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT NOT NULL, type TEXT DEFAULT 'fact', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS episodes ( id TEXT PRIMARY KEY, task TEXT NOT NULL, summary TEXT NOT NULL, outcome TEXT, created_at TEXT NOT NULL ) """) if not self._in_memory: conn.close() def get_all_memories(self) -> List[Dict[str, Any]]: conn = self._connect() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT * FROM memories ORDER BY updated_at DESC") rows = [dict(r) for r in cur.fetchall()] if not self._in_memory: conn.close() return rows def save_memories(self, memories: List[Dict[str, Any]], allow_empty: bool = False): """Replaces the memories table with the provided list (active compression).""" if not memories: existing = self.get_all_memories() if existing and not allow_empty: _print_memory_log("[memory] save_memories called with empty list while %d facts exist — refusing to wipe", len(existing)) log.warning("[memory] save_memories called with empty list while %d facts exist — refusing to wipe", len(existing)) return conn = self._connect() with conn: conn.execute("DELETE FROM memories") for m in memories: now = datetime.now().isoformat() created_at = m.get("created_at", now) updated_at = m.get("updated_at", now) conn.execute( "INSERT INTO memories (id, content, type, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", (m.get("id", str(uuid.uuid4())), m["content"], m.get("type", "fact"), created_at, updated_at) ) if not self._in_memory: conn.close() def delete_memory_prefix(self, prefix: str) -> int: if not prefix: return 0 matching_ids = [ row["id"] for row in self.get_all_memories() if row["id"].startswith(prefix) ] if not matching_ids: return 0 conn = self._connect() with conn: for memory_id in matching_ids: conn.execute("DELETE FROM memories WHERE id = ?", (memory_id,)) if not self._in_memory: conn.close() return len(matching_ids) def add_episode(self, task: str, summary: str, outcome: str): conn = self._connect() with conn: now = datetime.now().isoformat() conn.execute( "INSERT INTO episodes (id, task, summary, outcome, created_at) VALUES (?, ?, ?, ?, ?)", (str(uuid.uuid4()), task, summary, outcome, now) ) if not self._in_memory: conn.close() def get_recent_episodes(self, limit: int = 3) -> List[Dict[str, Any]]: conn = self._connect() conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT * FROM episodes ORDER BY created_at DESC LIMIT ?", (limit,)) rows = [dict(r) for r in cur.fetchall()] if not self._in_memory: conn.close() return rows _store = MemoryStore() def _set_store(store: MemoryStore) -> MemoryStore: """Swap the module-level store and return the previous one.""" global _store prev, _store = _store, store return prev @contextmanager def ephemeral_memory(db_path: Union[str, Path] = ":memory:"): """Context manager that replaces the global _store with a fresh, isolated MemoryStore for the duration of the block. On exit the store is closed and the previous store is restored. Use db_path=':memory:' (default) for tests or GAIA benchmarking where cross-contamination must be prevented. Example:: with ephemeral_memory(): result = graph.invoke(state, config) # any memory writes here are discarded on exit """ fresh = MemoryStore(db_path) prev = _set_store(fresh) try: yield fresh finally: _set_store(prev) fresh.close() def _is_remove_doc(content: Any) -> bool: return hasattr(content, "__repr_name__") and content.__repr_name__() == "RemoveDoc" def extract_and_compress_facts(messages: List[BaseMessage], model) -> None: """ Extracts new facts and merges them with existing ones using langmem's manager. Implements professional reflection and conflict resolution. """ from langmem import create_memory_manager from langmem.knowledge.extraction import Memory _print_memory_log("[memory] Running langmem memory management...") log.info("[memory] Running langmem memory management...") try: # 1. Get existing memories from our local store existing_rows = _store.get_all_memories() existing_memories = [(m["id"], Memory(content=m["content"])) for m in existing_rows] existing_by_content = {m["content"]: m for m in existing_rows} existing_by_id = {m["id"]: m for m in existing_rows} # 2. Initialize langmem manager (it's a Runnable) # We use the default schema which is basically content strings manager = create_memory_manager(model, enable_deletes=True) # 3. Invoke manager with history and existing knowledge # The manager will return the full updated list of memories result = manager.invoke({ "messages": messages, "existing": existing_memories }) # 4. Save results back to our local persistent SQLite if isinstance(result, list): # The result is a list of ExtractedMemory objects or tuples depending on version # Usually it's (id, content) or just objects. We'll be robust here. updated_facts = [] removed_ids = set() for item in result: item_id = getattr(item, "id", None) content = getattr(item, "content", None) if isinstance(item, tuple): if len(item) > 0: item_id = item[0] if len(item) > 1 and content is None: content = item[1] elif isinstance(item, dict): item_id = item.get("id", item_id) content = item.get("content", content) stable_id = str(item_id) if item_id else None if _is_remove_doc(content): if stable_id: removed_ids.add(stable_id) continue if content: text = _memory_content_to_text(content) existing = existing_by_id.get(stable_id) if stable_id else existing_by_content.get(text) fact = {"content": text} if stable_id: fact["id"] = stable_id elif existing: fact["id"] = existing["id"] if existing: fact["created_at"] = existing["created_at"] if existing["content"] == text: fact["updated_at"] = existing["updated_at"] if fact.get("id") not in removed_ids: updated_facts.append(fact) if updated_facts or removed_ids: _store.save_memories(updated_facts, allow_empty=bool(removed_ids)) _print_memory_log("[memory] langmem updated store to %d facts.", len(updated_facts)) log.info("[memory] langmem updated store to %d facts.", len(updated_facts)) else: _print_memory_log("[memory] langmem returned empty result — keeping existing facts") log.info("[memory] langmem returned empty result — keeping existing facts") summarize_episode(messages, model) except Exception as exc: _print_memory_log("[memory] langmem extraction failed: %s: %s", type(exc).__name__, exc) log.exception("[memory] langmem extraction failed") # Fallback to summarize episode if manager fails summarize_episode(messages, model) def summarize_episode(messages: List[BaseMessage], model) -> None: """Summarizes the trajectory to help avoid future mistakes.""" _print_memory_log("[memory] Summarizing task episode...") log.info("[memory] Summarizing task episode...") try: initial_question = "" outcome = "success" for m in messages: content = _content_to_text(m.content) if isinstance(m, HumanMessage) and not initial_question: initial_question = content if "ERROR" in content.upper(): outcome = "failed/struggled" prompt = f""" Summarize this task trajectory for Lilith's 'Episodic Memory'. Initial Question: {initial_question} Outcome: {outcome} Briefly explain: 1. What was the goal? 2. What tools worked? What failed? 3. What is the 'lesson learned' for next time? Keep it under 150 words. """ response = model.invoke(prompt) summary = _content_to_text(response.content) _store.add_episode(initial_question, summary, outcome) _print_memory_log( "[memory] Episode saved: task=%r outcome=%r summary=%r", _preview_for_log(initial_question), outcome, _preview_for_log(summary), ) log.info( "[memory] Episode saved: task=%r outcome=%r summary=%r", _preview_for_log(initial_question), outcome, _preview_for_log(summary), ) except Exception as exc: _print_memory_log("[memory] Summarization failed: %s: %s", type(exc).__name__, exc) log.exception("[memory] Summarization failed") def _relevance_score(text: str, query_tokens: set) -> float: """Simple word-overlap score between a fact and the query (Jaccard-like).""" if not query_tokens: return 0.0 fact_tokens = set(text.lower().split()) return len(fact_tokens & query_tokens) / (len(fact_tokens | query_tokens) or 1) def retrieve_relevant_context(query: str, char_budget: int = MEMORY_CONTEXT_CHAR_BUDGET) -> str: """Fetches facts and recent episodes ranked by relevance, capped by char_budget. If truncated, appends a note instructing the agent to call search_memory for more.""" try: query_tokens = set(query.lower().split()) facts = _store.get_all_memories() episodes = _store.get_recent_episodes(limit=5) facts.sort(key=lambda m: _relevance_score(m["content"], query_tokens), reverse=True) context_parts = [] budget_remaining = char_budget if facts: included, total = [], len(facts) for m in facts: line = f"- {m['content']}" if budget_remaining - len(line) - 1 > 0: included.append(line) budget_remaining -= len(line) + 1 else: break fact_block = "\n" + "\n".join(included) + "\n" omitted = total - len(included) if omitted: fact_block += f"\n" context_parts.append(fact_block) if episodes and budget_remaining > 0: included_ep = [] for e in episodes: line = f"Task: {e['task']}\nSummary: {e['summary']}" if budget_remaining - len(line) - 2 > 0: included_ep.append(line) budget_remaining -= len(line) + 2 else: break if included_ep: ep_block = "\n" + "\n\n".join(included_ep) + "\n" omitted_ep = len(episodes) - len(included_ep) if omitted_ep: ep_block += f"\n" context_parts.append(ep_block) return "\n\n".join(context_parts) except Exception as exc: _print_memory_log("[memory] Retrieval failed: %s: %s", type(exc).__name__, exc) log.exception("[memory] Retrieval failed") return "" def search_memory_store(query: str, max_results: int = 10) -> str: """Keyword search across all facts and episodes. Called by the agent when the system-prompt injection was truncated or the query needs deeper lookup.""" try: query_tokens = set(query.lower().split()) facts = _store.get_all_memories() episodes = _store.get_recent_episodes(limit=50) scored_facts = sorted( [(f, _relevance_score(f["content"], query_tokens)) for f in facts], key=lambda x: x[1], reverse=True ) scored_eps = sorted( [(e, _relevance_score(e["task"] + " " + e["summary"], query_tokens)) for e in episodes], key=lambda x: x[1], reverse=True ) results = [] for m, score in scored_facts[:max_results]: if score > 0: results.append((score, f"[fact] {m['content']}")) for e, score in scored_eps[:max_results]: if score > 0: results.append((score, f"[episode] Task: {e['task']}\n Summary: {e['summary']}")) results.sort(key=lambda x: x[0], reverse=True) parts = [text for _, text in results[:max_results]] if not parts: return "No matching memories found." return "\n\n".join(parts) except Exception as exc: _print_memory_log("[memory] search_memory_store failed: %s: %s", type(exc).__name__, exc) log.exception("[memory] search_memory_store failed") return "Memory search failed."