1# Memory Systems: Technical Reference
2 
3This document provides implementation details for memory system components.
4 
5## Vector Store Implementation
6 
7### Basic Vector Store
8 
9```python
10import numpy as np
11from typing import List, Dict, Any
12import json
13 
14 
15def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
16    """Compute cosine similarity between two vectors."""
17    norm_a = np.linalg.norm(a)
18    norm_b = np.linalg.norm(b)
19    if norm_a == 0 or norm_b == 0:
20        return 0.0
21    return float(np.dot(a, b) / (norm_a * norm_b))
22 
23 
24class VectorStore:
25    def __init__(self, dimension=768):
26        self.dimension = dimension
27        self.vectors = []
28        self.metadata = []
29        self.texts = []
30 
31    def add(self, text: str, metadata: Dict[str, Any] = None):
32        """Add document to store."""
33        embedding = self._embed(text)
34        self.vectors.append(embedding)
35        self.metadata.append(metadata or {})
36        self.texts.append(text)
37        return len(self.vectors) - 1
38    
39    def search(self, query: str, limit: int = 5, 
40               filters: Dict[str, Any] = None) -> List[Dict]:
41        """Search for similar documents."""
42        query_embedding = self._embed(query)
43        
44        scores = []
45        for i, vec in enumerate(self.vectors):
46            score = cosine_similarity(query_embedding, vec)
47            
48            # Apply filters
49            if filters and not self._matches_filters(self.metadata[i], filters):
50                score = -1  # Exclude
51            
52            scores.append((i, score))
53        
54        # Sort by score
55        scores.sort(key=lambda x: x[1], reverse=True)
56        
57        # Return top k
58        results = []
59        for idx, score in scores[:limit]:
60            if score > 0:  # Only include positive matches
61                results.append({
62                    "index": idx,
63                    "score": score,
64                    "text": self._get_text(idx),
65                    "metadata": self.metadata[idx]
66                })
67        
68        return results
69    
70    def _embed(self, text: str) -> np.ndarray:
71        """Generate deterministic pseudo-embedding for demonstration.
72        In production, replace with actual embedding model."""
73        np.random.seed(hash(text) % (2**32))
74        vec = np.random.randn(self.dimension)
75        return vec / (np.linalg.norm(vec) + 1e-8)
76    
77    def _matches_filters(self, metadata: Dict, filters: Dict) -> bool:
78        """Check if metadata matches filters."""
79        for key, value in filters.items():
80            if key not in metadata:
81                return False
82            if isinstance(value, list):
83                if metadata[key] not in value:
84                    return False
85            elif metadata[key] != value:
86                return False
87        return True
88    
89    def _get_text(self, index: int) -> str:
90        """Retrieve original text for index."""
91        return self.texts[index] if index < len(self.texts) else ""
92```
93 
94### Metadata-Enhanced Vector Store
95 
96```python
97class MetadataVectorStore(VectorStore):
98    def __init__(self, dimension=768):
99        super().__init__(dimension)
100        self.entity_index = {}  # entity -> [indices]
101        self.time_index = {}    # time_range -> [indices]
102    
103    def add(self, text: str, metadata: Dict[str, Any] = None):
104        """Add with enhanced indexing."""
105        metadata = metadata or {}
106        index = super().add(text, metadata)
107 
108        # Index by entity
109        if "entity" in metadata:
110            entity = metadata["entity"]
111            if entity not in self.entity_index:
112                self.entity_index[entity] = []
113            self.entity_index[entity].append(index)
114        
115        # Index by time
116        if "valid_from" in metadata:
117            time_key = self._time_range_key(
118                metadata.get("valid_from"),
119                metadata.get("valid_until")
120            )
121            if time_key not in self.time_index:
122                self.time_index[time_key] = []
123            self.time_index[time_key].append(index)
124        
125        return index
126    
127    def search_by_entity(self, query: str, entity: str, limit: int = 5) -> List[Dict]:
128        """Search within specific entity."""
129        indices = self.entity_index.get(entity, [])
130        filtered = [self.metadata[i] for i in indices]
131        
132        # Score and rank
133        query_embedding = self._embed(query)
134        scored = []
135        for i, meta in zip(indices, filtered):
136            vec = self.vectors[i]
137            score = cosine_similarity(query_embedding, vec)
138            scored.append((i, score, meta))
139        
140        scored.sort(key=lambda x: x[1], reverse=True)
141        
142        return [{
143            "index": idx,
144            "score": score,
145            "metadata": meta
146        } for idx, score, meta in scored[:limit]]
147```
148 
149## Knowledge Graph Implementation
150 
151### Property Graph Storage
152 
153```python
154from typing import Dict, List, Optional
155import uuid
156 
157class PropertyGraph:
158    def __init__(self):
159        self.nodes = {}  # id -> properties
160        self.edges = []  # list of edge dicts
161        self.entity_registry = {}  # name -> node_id (maintains identity)
162        self.indexes = {
163            "node_label": {},  # label -> [node_ids]
164            "edge_type": {}    # type -> [edge_ids]
165        }
166 
167    def get_or_create_node(self, name: str, label: str, properties: Dict = None) -> str:
168        """Get existing node by name, or create a new one.
169        Uses entity_registry to ensure identity across interactions."""
170        if name in self.entity_registry:
171            return self.entity_registry[name]
172        node_id = self.create_node(label, {**(properties or {}), "name": name})
173        self.entity_registry[name] = node_id
174        return node_id
175 
176    def create_node(self, label: str, properties: Dict = None) -> str:
177        """Create node with label and properties."""
178        node_id = str(uuid.uuid4())
179        self.nodes[node_id] = {
180            "label": label,
181            "properties": properties or {}
182        }
183 
184        # Index by label
185        if label not in self.indexes["node_label"]:
186            self.indexes["node_label"][label] = []
187        self.indexes["node_label"][label].append(node_id)
188 
189        return node_id
190    
191    def create_relationship(self, source_id: str, rel_type: str, 
192                           target_id: str, properties: Dict = None) -> str:
193        """Create directed relationship between nodes."""
194        edge_id = str(uuid.uuid4())
195        self.edges.append({
196            "id": edge_id,
197            "source": source_id,
198            "target": target_id,
199            "type": rel_type,
200            "properties": properties or {}
201        })
202        
203        # Index by type
204        if rel_type not in self.indexes["edge_type"]:
205            self.indexes["edge_type"][rel_type] = []
206        self.indexes["edge_type"][rel_type].append(edge_id)
207        
208        return edge_id
209    
210    def query(self, cypher_like: str, params: Dict = None) -> List[Dict]:
211        """
212        Simple query matching.
213        
214        Supports patterns like:
215        MATCH (e)-[r]->(o) WHERE e.id = $id RETURN r
216        """
217        # In production, use actual graph database
218        # This is a simplified pattern matcher
219        results = []
220        
221        if cypher_like.startswith("MATCH"):
222            # Parse basic pattern
223            pattern = self._parse_pattern(cypher_like)
224            results = self._match_pattern(pattern, params or {})
225        
226        return results
227    
228    def _parse_pattern(self, query: str) -> Dict:
229        """Parse simplified MATCH pattern."""
230        # Simplified parser for demonstration
231        return {
232            "source_label": self._extract_label(query, "source"),
233            "rel_type": self._extract_type(query),
234            "target_label": self._extract_label(query, "target"),
235            "where": self._extract_where(query)
236        }
237    
238    def _match_pattern(self, pattern: Dict, params: Dict) -> List[Dict]:
239        """Match pattern against graph."""
240        results = []
241        
242        for edge in self.edges:
243            # Match relationship type
244            if pattern["rel_type"] and edge["type"] != pattern["rel_type"]:
245                continue
246            
247            source = self.nodes.get(edge["source"], {})
248            target = self.nodes.get(edge["target"], {})
249            
250            # Match labels
251            if pattern["source_label"] and source.get("label") != pattern["source_label"]:
252                continue
253            if pattern["target_label"] and target.get("label") != pattern["target_label"]:
254                continue
255            
256            # Match where clause
257            if pattern["where"] and not self._match_where(edge, source, target, params):
258                continue
259            
260            results.append({
261                "source": source,
262                "relationship": edge,
263                "target": target
264            })
265        
266        return results
267```
268 
269## Temporal Knowledge Graph
270 
271```python
272from datetime import datetime
273from typing import Optional
274 
275class TemporalKnowledgeGraph(PropertyGraph):
276    def __init__(self):
277        super().__init__()
278        self.temporal_index = {}  # time_range -> [edge_ids]
279    
280    def create_temporal_relationship(
281        self, 
282        source_id: str, 
283        rel_type: str, 
284        target_id: str,
285        valid_from: datetime,
286        valid_until: Optional[datetime] = None,
287        properties: Dict = None
288    ) -> str:
289        """Create relationship with temporal validity."""
290        edge_id = super().create_relationship(
291            source_id, rel_type, target_id, properties
292        )
293        
294        # Index temporally
295        time_key = self._time_range_key(valid_from, valid_until)
296        if time_key not in self.temporal_index:
297            self.temporal_index[time_key] = []
298        self.temporal_index[time_key].append(edge_id)
299        
300        # Store validity on edge
301        edge = self._get_edge(edge_id)
302        edge["valid_from"] = valid_from.isoformat()
303        edge["valid_until"] = valid_until.isoformat() if valid_until else None
304        
305        return edge_id
306    
307    def query_at_time(self, query: str, query_time: datetime) -> List[Dict]:
308        """Query graph state at specific time."""
309        # Find edges valid at query time
310        valid_edges = []
311        for edge in self.edges:
312            valid_from = datetime.fromisoformat(edge.get("valid_from", "1970-01-01"))
313            valid_until = edge.get("valid_until")
314            
315            if valid_from <= query_time:
316                if valid_until is None or datetime.fromisoformat(valid_until) > query_time:
317                    valid_edges.append(edge)
318        
319        # Match against pattern
320        pattern = self._parse_pattern(query)
321        results = []
322        
323        for edge in valid_edges:
324            if pattern["rel_type"] and edge["type"] != pattern["rel_type"]:
325                continue
326            
327            source = self.nodes.get(edge["source"], {})
328            target = self.nodes.get(edge["target"], {})
329            
330            results.append({
331                "source": source,
332                "relationship": edge,
333                "target": target
334            })
335        
336        return results
337    
338    def _time_range_key(self, start: datetime, end: Optional[datetime]) -> str:
339        """Create time range key for indexing."""
340        start_str = start.isoformat()
341        end_str = end.isoformat() if end else "infinity"
342        return f"{start_str}::{end_str}"
343```
344 
345## Memory Consolidation
346 
347```python
348class MemoryConsolidator:
349    def __init__(self, graph: PropertyGraph, vector_store: VectorStore):
350        self.graph = graph
351        self.vector_store = vector_store
352        self.consolidation_threshold = 1000  # memories before consolidation
353    
354    def should_consolidate(self) -> bool:
355        """Check if consolidation should trigger."""
356        total_memories = len(self.graph.nodes) + len(self.graph.edges)
357        return total_memories > self.consolidation_threshold
358    
359    def consolidate(self):
360        """Run consolidation process."""
361        # Step 1: Identify duplicate or merged facts
362        duplicates = self.find_duplicates()
363        
364        # Step 2: Merge related facts
365        for group in duplicates:
366            self.merge_fact_group(group)
367        
368        # Step 3: Update validity periods
369        self.update_validity_periods()
370        
371        # Step 4: Rebuild indexes
372        self.rebuild_indexes()
373    
374    def find_duplicates(self) -> List[List]:
375        """Find groups of potentially duplicate facts."""
376        # Group by subject and predicate
377        groups = {}
378        
379        for edge in self.graph.edges:
380            key = (edge["source"], edge["type"])
381            if key not in groups:
382                groups[key] = []
383            groups[key].append(edge)
384        
385        # Return groups with multiple edges
386        return [edges for edges in groups.values() if len(edges) > 1]
387    
388    def merge_fact_group(self, edges: List[Dict]):
389        """Merge group of duplicate edges."""
390        if len(edges) == 1:
391            return
392        
393        # Keep most recent/relevant
394        keeper = max(edges, key=lambda e: e.get("properties", {}).get("confidence", 0))
395        
396        # Merge metadata
397        for edge in edges:
398            if edge["id"] != keeper["id"]:
399                self.merge_properties(keeper, edge)
400                self.graph.edges.remove(edge)
401    
402    def merge_properties(self, target: Dict, source: Dict):
403        """Merge properties from source into target."""
404        for key, value in source.get("properties", {}).items():
405            if key not in target["properties"]:
406                target["properties"][key] = value
407            elif isinstance(value, list):
408                target["properties"][key].extend(value)
409```
410 
411## Memory-Context Integration
412 
413```python
414class MemoryContextIntegrator:
415    def __init__(self, memory_system, context_limit=100000):
416        self.memory_system = memory_system
417        self.context_limit = context_limit
418    
419    def build_context(self, task: str, current_context: str = "") -> str:
420        """Build context including relevant memories."""
421        # Extract entities from task
422        entities = self._extract_entities(task)
423        
424        # Retrieve memories for each entity
425        memories = []
426        for entity in entities:
427            entity_memories = self.memory_system.retrieve_entity(entity)
428            memories.extend(entity_memories)
429        
430        # Format memories for context
431        memory_section = self._format_memories(memories)
432        
433        # Combine with current context
434        combined = current_context + "\n\n" + memory_section
435        
436        # Check limit and truncate if needed
437        if self._token_count(combined) > self.context_limit:
438            combined = self._truncate_context(combined, self.context_limit)
439        
440        return combined
441    
442    def _extract_entities(self, task: str) -> List[str]:
443        """Extract entity mentions from task."""
444        # In production, use NER or entity extraction
445        import re
446        pattern = r"\[([^\]]+)\]"  # [[entity_name]] convention
447        return re.findall(pattern, task)
448    
449    def _format_memories(self, memories: List[Dict]) -> str:
450        """Format memories for context injection."""
451        sections = ["## Relevant Memories"]
452        
453        for memory in memories:
454            formatted = f"- {memory.get('content', '')}"
455            if "source" in memory:
456                formatted += f" (Source: {memory['source']})"
457            if "timestamp" in memory:
458                formatted += f" [Time: {memory['timestamp']}]"
459            sections.append(formatted)
460        
461        return "\n".join(sections)
462    
463    def _token_count(self, text: str) -> int:
464        """Estimate token count."""
465        return len(text) // 4  # Rough approximation
466    
467    def _truncate_context(self, context: str, limit: int) -> str:
468        """Truncate context to fit limit."""
469        tokens = context.split()
470        truncated = []
471        count = 0
472 
473        for token in tokens:
474            if count + 1 > limit:
475                break
476            truncated.append(token)
477            count += 1
478 
479        return " ".join(truncated)
480```
481 
482## Framework Integration Examples
483 
484### Mem0 Quick Start
485 
486```python
487from mem0 import Memory
488 
489# Initialize with default config (uses local storage)
490m = Memory()
491 
492# Store memories with user scoping
493m.add("Prefers Python 3.12 with type hints", user_id="dev-alice")
494m.add("Working on microservices migration", user_id="dev-alice")
495 
496# Search with natural language
497results = m.search("What language does the user prefer?", user_id="dev-alice")
498 
499# Batch operations
500m.add([
501    "Sprint goal: complete auth service",
502    "Blocked on database schema review"
503], user_id="dev-alice")
504```
505 
506### Graphiti (Zep's Open-Source Temporal KG Engine)
507 
508```python
509from graphiti_core import Graphiti
510from graphiti_core.nodes import EpisodeType
511 
512# Initialize with Neo4j backend
513graphiti = Graphiti("bolt://localhost:7687", "neo4j", "password")
514 
515# Add episodes (conversations, events)
516await graphiti.add_episode(
517    name="user_conversation_42",
518    episode_body="Alice mentioned she moved to Berlin in January.",
519    source=EpisodeType.message,
520    source_description="Chat with Alice"
521)
522 
523# Search combines semantic, keyword, and graph traversal
524results = await graphiti.search("Where does Alice live?")
525```
526 
527### Cognee (Open-Source Knowledge Engine for AI Memory)
528 
529```python
530import cognee
531from cognee.modules.search.types import SearchType
532 
533# ECL pipeline: add → cognify → memify → search
534await cognee.add("./docs/")
535await cognee.add("any-data")
536await cognee.cognify()
537await cognee.memify()
538 
539# Graph-aware retrieval (default: GRAPH_COMPLETION)
540results = await cognee.search(
541    query_text="any query to search in memory",
542    query_type=SearchType.GRAPH_COMPLETION,
543)
544 
545# Raw chunks when agent reasons over text itself
546chunks = await cognee.search(
547    query_text="any query to search in memory",
548    query_type=SearchType.CHUNKS,
549)
550```
551 
552