File size: 7,681 Bytes
aa9003d
 
 
 
 
 
 
 
 
 
 
 
 
6d2a17c
aa9003d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8f17704
aa9003d
 
8f17704
aa9003d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
# ────────────────────────────── memo/persistent.py ──────────────────────────────
"""
Persistent Memory System

MongoDB-based persistent memory storage with semantic search capabilities.
"""

import os
import uuid
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime, timezone

from utils.logger import get_logger
from utils.rag.embeddings import EmbeddingClient

logger = get_logger("PERSISTENT_MEMORY", __name__)

class PersistentMemory:
    """MongoDB-based persistent memory system with semantic search"""
    
    def __init__(self, mongo_uri: str, db_name: str, embedder: EmbeddingClient):
        self.mongo_uri = mongo_uri
        self.db_name = db_name
        self.embedder = embedder
        
        # MongoDB connection
        try:
            from pymongo import MongoClient
            self.client = MongoClient(mongo_uri)
            self.db = self.client[db_name]
            self.memories = self.db["memories"]
            
            # Create indexes for efficient querying
            self.memories.create_index([("user_id", 1), ("memory_type", 1)])
            self.memories.create_index([("user_id", 1), ("created_at", -1)])
            self.memories.create_index([("user_id", 1), ("project_id", 1)])
            
            logger.info(f"[PERSISTENT_MEMORY] Connected to MongoDB: {db_name}")
        except Exception as e:
            logger.error(f"[PERSISTENT_MEMORY] Failed to connect to MongoDB: {e}")
            raise
    
    def add_memory(self, user_id: str, content: str, memory_type: str, 
                  project_id: str = None, importance: str = "medium",
                  tags: List[str] = None, metadata: Dict[str, Any] = None) -> str:
        """Add a memory entry to the persistent system"""
        try:
            # Generate embedding for semantic search
            embedding = self.embedder.embed([content])[0] if content else None
            
            # Create summary
            summary = content[:200] + "..." if len(content) > 200 else content
            
            memory_entry = {
                "id": str(uuid.uuid4()),
                "user_id": user_id,
                "project_id": project_id,
                "memory_type": memory_type,
                "content": content,
                "summary": summary,
                "importance": importance,
                "tags": tags or [],
                "created_at": datetime.now(timezone.utc),
                "updated_at": datetime.now(timezone.utc),
                "last_accessed": datetime.now(timezone.utc),
                "access_count": 0,
                "embedding": embedding,
                "metadata": metadata or {}
            }
            
            # Store in MongoDB
            self.memories.insert_one(memory_entry)
            logger.info(f"[PERSISTENT_MEMORY] Added {memory_type} memory for user {user_id}")
            return memory_entry["id"]
            
        except Exception as e:
            logger.error(f"[PERSISTENT_MEMORY] Failed to add memory: {e}")
            raise
    
    def get_memories(self, user_id: str, memory_type: str = None, 
                    project_id: str = None, limit: int = 50) -> List[Dict[str, Any]]:
        """Get memories for a user with optional filtering"""
        try:
            query = {"user_id": user_id}
            
            if memory_type:
                query["memory_type"] = memory_type
            
            if project_id:
                query["project_id"] = project_id
            
            cursor = self.memories.find(query).sort("created_at", -1).limit(limit)
            return list(cursor)
            
        except Exception as e:
            logger.error(f"[PERSISTENT_MEMORY] Failed to get memories: {e}")
            return []
    
    def search_memories(self, user_id: str, query: str, memory_types: List[str] = None,
                       project_id: str = None, limit: int = 10) -> List[Tuple[Dict[str, Any], float]]:
        """Search memories using semantic similarity"""
        try:
            # Generate query embedding
            query_embedding = self.embedder.embed([query])[0]
            
            # Build MongoDB query
            mongo_query = {
                "user_id": user_id,
                "embedding": {"$exists": True}
            }
            
            if memory_types:
                mongo_query["memory_type"] = {"$in": memory_types}
            
            if project_id:
                mongo_query["project_id"] = project_id
            
            # Get all matching memories
            cursor = self.memories.find(mongo_query)
            
            # Calculate similarities
            results = []
            for doc in cursor:
                try:
                    if doc.get("embedding"):
                        # Calculate cosine similarity
                        similarity = self._cosine_similarity(query_embedding, doc["embedding"])
                        results.append((doc, similarity))
                except Exception as e:
                    logger.warning(f"[PERSISTENT_MEMORY] Failed to process memory for search: {e}")
                    continue
            
            # Sort by similarity and return top results
            results.sort(key=lambda x: x[1], reverse=True)
            return results[:limit]
            
        except Exception as e:
            logger.error(f"[PERSISTENT_MEMORY] Failed to search memories: {e}")
            return []
    
    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        """Calculate cosine similarity between two vectors"""
        try:
            import numpy as np
            from memo.context import cosine_similarity
            a_np = np.array(a)
            b_np = np.array(b)
            return cosine_similarity(a_np, b_np)
        except Exception:
            return 0.0
    
    def clear_user_memories(self, user_id: str) -> int:
        """Clear all memories for a user"""
        try:
            result = self.memories.delete_many({"user_id": user_id})
            logger.info(f"[PERSISTENT_MEMORY] Cleared {result.deleted_count} memories for user {user_id}")
            return result.deleted_count
        except Exception as e:
            logger.error(f"[PERSISTENT_MEMORY] Failed to clear user memories: {e}")
            return 0
    
    def get_memory_stats(self, user_id: str) -> Dict[str, Any]:
        """Get memory statistics for a user"""
        try:
            stats = {
                "total_memories": self.memories.count_documents({"user_id": user_id}),
                "by_type": {},
                "recent_activity": 0
            }
            
            # Count by memory type
            pipeline = [
                {"$match": {"user_id": user_id}},
                {"$group": {"_id": "$memory_type", "count": {"$sum": 1}}}
            ]
            
            for result in self.memories.aggregate(pipeline):
                stats["by_type"][result["_id"]] = result["count"]
            
            # Recent activity (last 7 days)
            from datetime import timedelta
            week_ago = datetime.now(timezone.utc) - timedelta(days=7)
            stats["recent_activity"] = self.memories.count_documents({
                "user_id": user_id,
                "created_at": {"$gte": week_ago}
            })
            
            return stats
            
        except Exception as e:
            logger.error(f"[PERSISTENT_MEMORY] Failed to get memory stats: {e}")
            return {}