File size: 5,108 Bytes
1d8ce4e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """
Mem0 In-Memory Layer for Health Insurance AI Copilot.
Keeps a lightweight semantic memory PER SESSION, stored entirely in RAM.
- No disk persistence — resets cleanly on server restart / HF Space refresh.
- Namespaced by session_id so sessions never see each other's memories.
- Mem0 auto-extracts key facts (plan tier, drugs, preferences) from the
conversation so the agent stays context-aware throughout the session.
Each session gets its own Memory instance (created in SessionManager).
This module provides helpers that operate on a given Memory instance.
"""
import os
import sys
import logging
from typing import Optional
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
logger = logging.getLogger(__name__)
def create_session_memory():
"""
Create a fresh in-memory Mem0 Memory instance for a single session.
Uses Qdrant in-memory vector store — no files written, no API key needed.
Falls back gracefully if mem0 isn't installed or MEM0_ENABLED=false.
Returns:
A Mem0 Memory instance, or None if mem0 is unavailable/disabled.
"""
from config import MEM0_ENABLED, MEM0_LLM_MODEL, MEM0_EMBEDDER_MODEL
if not MEM0_ENABLED:
logger.info("ℹ️ Mem0 disabled via MEM0_ENABLED=false")
return None
try:
from mem0 import Memory
config = {
# In-memory Qdrant — lives only as long as this Python object
"vector_store": {
"provider": "qdrant",
"config": {
"collection_name": "session_memories",
"in_memory": True,
},
},
# Cheap fast model for fact extraction
"llm": {
"provider": "openai",
"config": {
"model": MEM0_LLM_MODEL,
"temperature": 0,
"api_key": os.getenv("OPENAI_API_KEY"),
},
},
# Small fast embedder
"embedder": {
"provider": "openai",
"config": {
"model": MEM0_EMBEDDER_MODEL,
"api_key": os.getenv("OPENAI_API_KEY"),
},
},
}
mem = Memory.from_config(config)
logger.info("✅ Mem0 session memory created (in-memory / ephemeral)")
return mem
except ImportError:
logger.warning("⚠️ mem0ai not installed — memory features disabled")
return None
except Exception as e:
logger.warning(f"⚠️ Mem0 init failed: {e} — memory features disabled")
return None
# ── Per-session helpers ───────────────────────────────────────────────────────
SESSION_USER = "session" # Fixed user_id within each isolated Memory instance
def search_memories(mem, query: str, limit: int = 5) -> str:
"""
Search this session's memory for facts relevant to the current query.
Args:
mem: The session Memory instance (from create_session_memory)
query: Current user question
limit: Max facts to return
Returns:
Formatted string of facts, or empty string if none found.
"""
if mem is None:
return ""
try:
results = mem.search(query, user_id=SESSION_USER, limit=limit)
facts = [r["memory"] for r in results.get("results", [])]
if not facts:
return ""
lines = ["📋 Relevant facts recalled from this session:"]
for i, fact in enumerate(facts, 1):
lines.append(f" {i}. {fact}")
return "\n".join(lines)
except Exception as e:
logger.debug(f"Mem0 search skipped: {e}")
return ""
def add_memory(mem, query: str, answer: str) -> list:
"""
Extract and store key facts from this Q&A turn into session memory.
Args:
mem: The session Memory instance
query: User's question
answer: AI's answer
Returns:
List of extracted fact strings (for logging).
"""
if mem is None:
return []
try:
messages = [
{"role": "user", "content": query},
{"role": "assistant", "content": answer},
]
result = mem.add(messages, user_id=SESSION_USER)
stored = []
if isinstance(result, dict):
for item in result.get("results", []):
if isinstance(item, dict) and "memory" in item:
stored.append(item["memory"])
return stored
except Exception as e:
logger.debug(f"Mem0 add skipped: {e}")
return []
def get_all_memories(mem) -> list:
"""Return all stored facts for the session (for /memory debug endpoint)."""
if mem is None:
return []
try:
results = mem.get_all(user_id=SESSION_USER)
return results.get("results", [])
except Exception as e:
logger.debug(f"Mem0 get_all skipped: {e}")
return []
|