Memory Systems

Memory System Design

Memory provides the persistence layer that allows agents to maintain continuity across sessions and reason over accumulated knowledge. Simple agents rely entirely on context for memory, losing all state when sessions end. Sophisticated agents implement layered memory architectures that balance immediate context needs with long-term knowledge retention. The evolution from vector stores to knowledge graphs to temporal knowledge graphs represents increasing investment in structured memory for improved retrieval and reasoning.

When to Activate

Activate this skill when:

• Building agents that must persist knowledge across sessions
• Choosing between memory frameworks (Mem0, Zep/Graphiti, Letta, LangMem, Cognee)
• Needing to maintain entity consistency across conversations
• Implementing reasoning over accumulated knowledge
• Designing memory architectures that scale in production
• Evaluating memory systems against benchmarks (LoCoMo, LongMemEval, DMR)
• Building dynamic memory with automatic entity/relationship extraction and self-improving memory (Cognee)

Do not activate this skill for adjacent work owned by other skills:

• File-backed scratchpads, run logs, and tool-output offloading: filesystem-context.
• Conversation compaction or human-readable handoff summaries: context-compression.
• Masking, prefix caching, token budgets, or retrieval scoping inside one trajectory: context-optimization.
• Formal belief/desire/intention models over RDF state: bdi-mental-states.

Core Concepts

Think of memory as a spectrum from volatile context window to persistent storage. Default to the simplest layer that meets retrieval needs, because benchmark evidence suggests tool complexity matters less than reliable retrieval for some memory workloads (claim-memory-locomo-filesystem-baseline). Add structure (graphs, temporal validity) only when retrieval quality degrades or the agent needs multi-hop reasoning, relationship traversal, or time-travel queries.

Detailed Topics

Production Framework Landscape

Select a framework based on the dominant retrieval pattern the agent requires. Use this table to narrow the shortlist, then validate with the benchmark data below.

Choose Zep/Graphiti when the agent needs bi-temporal modeling (tracking both when events occurred and when they were ingested) because its three-tier knowledge graph (episode, semantic entity, community subgraphs) excels at temporal queries. Choose Mem0 when the priority is fast time-to-production with managed infrastructure. Choose Letta when the agent needs deep self-introspection through its Agent Development Environment. Choose Cognee when the agent must build dense multi-layer semantic graphs — it layers text chunks and entity types as nodes with detailed relationship edges, and every core piece (ingestion, entity extraction, post-processing, retrieval) is customizable.

Benchmark Performance Comparison

Consult these benchmarks to set expectations, but treat them as source-specific signals for retrieval dimensions rather than absolute rankings. No single benchmark is definitive.

Key takeaway: compare memory systems by retrieval shape, not brand. Use benchmark numbers as dated evidence that must be rechecked before making product claims; the stable design rule is to start shallow, measure retrieval quality, then add semantic or graph structure only when a simpler layer fails.

Memory Layers (Decision Points)

Pick the shallowest memory layer that satisfies the persistence requirement. Each deeper layer adds infrastructure cost and operational complexity, so only escalate when the shallower layer cannot meet the retrieval or durability need.

Retrieval Strategies

Match the retrieval strategy to the query shape. Semantic search handles direct factual lookups well but degrades on multi-hop reasoning; entity-based traversal handles "everything about X" queries but requires graph structure; temporal filtering handles changing facts but requires validity metadata. When accuracy is paramount and infrastructure budget allows, combine strategies into hybrid retrieval.

Hybrid approaches reduce active context by retrieving only relevant subgraphs or memories. Cognee implements hybrid retrieval through multiple search modes across graph, vector, and relational stores, letting agents select the retrieval strategy that fits the query type rather than using a one-size-fits-all approach.

Memory Consolidation

Run consolidation periodically to prevent unbounded growth, because unchecked memory accumulation degrades retrieval quality over time. Invalidate but do not discard — preserving history matters for temporal queries that need to reconstruct past states. Trigger consolidation on memory count thresholds, degraded retrieval quality, or scheduled intervals. See Implementation Reference for working consolidation code.

Practical Guidance

Choosing a Memory Architecture

Start with the simplest viable layer and add complexity only when retrieval quality degrades. Most agents do not need a temporal knowledge graph on day one. Follow this escalation path:

1. Prototype: Use file-system memory. Store facts as structured JSON with timestamps. This validates agent behavior before committing to infrastructure. 2. Scale: Move to Mem0 or a vector store with metadata when the agent needs semantic search and multi-tenant isolation, because file-based lookup cannot handle similarity queries. 3. Complex reasoning: Add Zep/Graphiti when the agent needs relationship traversal, temporal validity, or cross-session synthesis. Graphiti uses structured ties with generic relations, keeping graphs simple and easy to reason about; Cognee builds denser multi-layer semantic graphs with detailed relationship edges — choose based on whether the agent needs temporal bi-modeling (Graphiti) or richer interconnected knowledge structures (Cognee). 4. Full control: Use Letta or Cognee when the agent must self-manage its own memory with deep introspection, because these frameworks expose memory operations as first-class agent actions.

Integration with Context

Load memories just-in-time rather than preloading everything, because large context payloads are expensive and degrade attention quality. Place retrieved memories in attention-favored positions (beginning or end of context) to maximize their influence on generation.

Error Recovery

Handle retrieval failures gracefully because memory systems are inherently noisy. Apply these recovery strategies in order:

• Empty retrieval: Fall back to broader search (remove entity filter, widen time range). If still empty, prompt user for clarification.
• Stale results: Check valid_until timestamps. If most results are expired, trigger consolidation before retrying.
• Conflicting facts: Prefer the fact with the most recent valid_from. Surface the conflict to the user if confidence is low.
• Storage failure: Queue writes for retry. Never block the agent's response on a memory write.

Examples

Example 1: Mem0 Integration

from mem0 import Memory

m = Memory()
m.add("User prefers dark mode and Python 3.12", user_id="alice")
m.add("User switched to light mode", user_id="alice")

# Retrieves current preference (light mode), not outdated one
results = m.search("What theme does the user prefer?", user_id="alice")

Example 2: Temporal Query

# Track entity with validity periods
graph.create_temporal_relationship(
    source_id=user_node,
    rel_type="LIVES_AT",
    target_id=address_node,
    valid_from=datetime(2024, 1, 15),
    valid_until=datetime(2024, 9, 1),  # moved out
)

# Query: Where did user live on March 1, 2024?
results = graph.query_at_time(
    {"type": "LIVES_AT", "source_label": "User"},
    query_time=datetime(2024, 3, 1)
)

Example 3: Cognee Memory Ingestion and Search

import cognee
from cognee.modules.search.types import SearchType

# Ingest and build knowledge graph
await cognee.add("./docs/")
await cognee.add("any data")
await cognee.cognify()

# Enrich memory
await cognee.memify()

# Agent retrieves relationship-aware context
results = await cognee.search(
    query_text="Any query for your memory",
    query_type=SearchType.GRAPH_COMPLETION,
)

Guidelines

1. Start with file-system memory; add complexity only when retrieval quality demands it 2. Track temporal validity for any fact that can change over time 3. Use hybrid retrieval (semantic + keyword + graph) for best accuracy 4. Consolidate memories periodically — invalidate but don't discard 5. Design for retrieval failure: always have a fallback when memory lookup returns nothing 6. Consider privacy implications of persistent memory (retention policies, deletion rights) 7. Benchmark your memory system against LoCoMo or LongMemEval before and after changes 8. Monitor memory growth and retrieval latency in production

Gotchas

1. Stuffing everything into context: Loading all available memories into the prompt is expensive and degrades attention quality. Use just-in-time retrieval with relevance filtering instead. 2. Ignoring temporal validity: Facts go stale. Without validity tracking, outdated information poisons the context and the agent acts on wrong assumptions. 3. Over-engineering early: Simple filesystem-backed memory can outperform more specialized tooling on some benchmarks (claim-memory-locomo-filesystem-baseline). Add sophistication only when simple approaches demonstrably fail. 4. No consolidation strategy: Unbounded memory growth degrades retrieval quality over time. Set memory count thresholds or scheduled intervals to trigger consolidation. 5. Embedding model mismatch: Writing memories with one embedding model and reading with another produces poor retrieval because vector spaces are not interchangeable. Pin a single embedding model for each memory store and re-embed all entries if the model changes. 6. Graph schema rigidity: Over-structured graph schemas (rigid node types, fixed relationship labels) break when the domain evolves. Prefer generic relation types and flexible property bags so new entity kinds do not require schema migrations. 7. Stale memory poisoning: Old memories that contradict the current state corrupt agent behavior silently. Implement expiry policies or confidence decay so the agent deprioritizes aged facts, and surface contradictions explicitly when detected. 8. Memory-context mismatch: Retrieving memories that are topically related but contextually wrong (e.g., a memory about "Python" the snake when the agent is discussing Python the language). Mitigate by including session or domain metadata in memory entries and filtering on it during retrieval.

Integration

This skill owns persistent semantic memory. Adjacent skills own scratch storage, compaction, and context tactics:

• filesystem-context: file-backed scratchpads, logs, and simple run state before semantic retrieval is needed.
• context-compression: summaries and handoffs that preserve session state in prose.
• context-optimization: just-in-time memory loading and retrieval scoping inside active context budgets.
• context-degradation: stale or conflicting memories as context poisoning or clash.
• bdi-mental-states: formal mental-state modeling when beliefs, desires, intentions, and provenance chains matter.
• multi-agent-patterns: shared memory across agents.
• evaluation: memory quality, retrieval correctness, and benchmark selection.

References

Internal references:

• Implementation Reference - Read when: implementing vector stores, property graphs, temporal queries, or memory consolidation logic from scratch

Related skills in this collection:

• context-fundamentals - Read when: designing the context layer that memory feeds into
• multi-agent-patterns - Read when: multiple agents need to share or coordinate memory state

External resources:

• Zep temporal knowledge graph paper (arXiv:2501.13956) - Read when: evaluating bi-temporal modeling or Graphiti's architecture
• Mem0 production architecture paper (arXiv:2504.19413) - Read when: assessing managed memory infrastructure trade-offs
• Cognee optimized knowledge graph + LLM reasoning paper (arXiv:2505.24478) - Read when: comparing multi-layer semantic graph approaches
• LoCoMo benchmark (Snap Research) - Read when: evaluating long-conversation memory retention
• MemBench evaluation framework (ACL 2025) - Read when: designing memory evaluation suites
• Graphiti open-source temporal KG engine (github.com/getzep/graphiti) - Read when: implementing temporal knowledge graphs
• Cognee open-source knowledge graph memory (github.com/topoteretes/cognee) - Read when: building customizable ECL pipelines for memory
• Cognee comparison: Form vs Function - Read when: comparing graph structures across Mem0, Graphiti, LightRAG, Cognee

---

Skill Metadata

Created: 2025-12-20 Last Updated: 2026-05-15 Author: Agent Skills for Context Engineering Contributors Version: 4.1.0

Memory Systems: Technical Reference

This document provides implementation details for memory system components.

Vector Store Implementation

Basic Vector Store

import numpy as np
from typing import List, Dict, Any
import json


def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """Compute cosine similarity between two vectors."""
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return float(np.dot(a, b) / (norm_a * norm_b))


class VectorStore:
    def __init__(self, dimension=768):
        self.dimension = dimension
        self.vectors = []
        self.metadata = []
        self.texts = []

    def add(self, text: str, metadata: Dict[str, Any] = None):
        """Add document to store."""
        embedding = self._embed(text)
        self.vectors.append(embedding)
        self.metadata.append(metadata or {})
        self.texts.append(text)
        return len(self.vectors) - 1
    
    def search(self, query: str, limit: int = 5, 
               filters: Dict[str, Any] = None) -> List[Dict]:
        """Search for similar documents."""
        query_embedding = self._embed(query)
        
        scores = []
        for i, vec in enumerate(self.vectors):
            score = cosine_similarity(query_embedding, vec)
            
            # Apply filters
            if filters and not self._matches_filters(self.metadata[i], filters):
                score = -1  # Exclude
            
            scores.append((i, score))
        
        # Sort by score
        scores.sort(key=lambda x: x[1], reverse=True)
        
        # Return top k
        results = []
        for idx, score in scores[:limit]:
            if score > 0:  # Only include positive matches
                results.append({
                    "index": idx,
                    "score": score,
                    "text": self._get_text(idx),
                    "metadata": self.metadata[idx]
                })
        
        return results
    
    def _embed(self, text: str) -> np.ndarray:
        """Generate deterministic pseudo-embedding for demonstration.
        In production, replace with actual embedding model."""
        np.random.seed(hash(text) % (2**32))
        vec = np.random.randn(self.dimension)
        return vec / (np.linalg.norm(vec) + 1e-8)
    
    def _matches_filters(self, metadata: Dict, filters: Dict) -> bool:
        """Check if metadata matches filters."""
        for key, value in filters.items():
            if key not in metadata:
                return False
            if isinstance(value, list):
                if metadata[key] not in value:
                    return False
            elif metadata[key] != value:
                return False
        return True
    
    def _get_text(self, index: int) -> str:
        """Retrieve original text for index."""
        return self.texts[index] if index < len(self.texts) else ""

Metadata-Enhanced Vector Store

class MetadataVectorStore(VectorStore):
    def __init__(self, dimension=768):
        super().__init__(dimension)
        self.entity_index = {}  # entity -> [indices]
        self.time_index = {}    # time_range -> [indices]
    
    def add(self, text: str, metadata: Dict[str, Any] = None):
        """Add with enhanced indexing."""
        metadata = metadata or {}
        index = super().add(text, metadata)

        # Index by entity
        if "entity" in metadata:
            entity = metadata["entity"]
            if entity not in self.entity_index:
                self.entity_index[entity] = []
            self.entity_index[entity].append(index)
        
        # Index by time
        if "valid_from" in metadata:
            time_key = self._time_range_key(
                metadata.get("valid_from"),
                metadata.get("valid_until")
            )
            if time_key not in self.time_index:
                self.time_index[time_key] = []
            self.time_index[time_key].append(index)
        
        return index
    
    def search_by_entity(self, query: str, entity: str, limit: int = 5) -> List[Dict]:
        """Search within specific entity."""
        indices = self.entity_index.get(entity, [])
        filtered = [self.metadata[i] for i in indices]
        
        # Score and rank
        query_embedding = self._embed(query)
        scored = []
        for i, meta in zip(indices, filtered):
            vec = self.vectors[i]
            score = cosine_similarity(query_embedding, vec)
            scored.append((i, score, meta))
        
        scored.sort(key=lambda x: x[1], reverse=True)
        
        return [{
            "index": idx,
            "score": score,
            "metadata": meta
        } for idx, score, meta in scored[:limit]]

Knowledge Graph Implementation

Property Graph Storage

from typing import Dict, List, Optional
import uuid

class PropertyGraph:
    def __init__(self):
        self.nodes = {}  # id -> properties
        self.edges = []  # list of edge dicts
        self.entity_registry = {}  # name -> node_id (maintains identity)
        self.indexes = {
            "node_label": {},  # label -> [node_ids]
            "edge_type": {}    # type -> [edge_ids]
        }

    def get_or_create_node(self, name: str, label: str, properties: Dict = None) -> str:
        """Get existing node by name, or create a new one.
        Uses entity_registry to ensure identity across interactions."""
        if name in self.entity_registry:
            return self.entity_registry[name]
        node_id = self.create_node(label, {**(properties or {}), "name": name})
        self.entity_registry[name] = node_id
        return node_id

    def create_node(self, label: str, properties: Dict = None) -> str:
        """Create node with label and properties."""
        node_id = str(uuid.uuid4())
        self.nodes[node_id] = {
            "label": label,
            "properties": properties or {}
        }

        # Index by label
        if label not in self.indexes["node_label"]:
            self.indexes["node_label"][label] = []
        self.indexes["node_label"][label].append(node_id)

        return node_id
    
    def create_relationship(self, source_id: str, rel_type: str, 
                           target_id: str, properties: Dict = None) -> str:
        """Create directed relationship between nodes."""
        edge_id = str(uuid.uuid4())
        self.edges.append({
            "id": edge_id,
            "source": source_id,
            "target": target_id,
            "type": rel_type,
            "properties": properties or {}
        })
        
        # Index by type
        if rel_type not in self.indexes["edge_type"]:
            self.indexes["edge_type"][rel_type] = []
        self.indexes["edge_type"][rel_type].append(edge_id)
        
        return edge_id
    
    def query(self, cypher_like: str, params: Dict = None) -> List[Dict]:
        """
        Simple query matching.
        
        Supports patterns like:
        MATCH (e)-[r]->(o) WHERE e.id = $id RETURN r
        """
        # In production, use actual graph database
        # This is a simplified pattern matcher
        results = []
        
        if cypher_like.startswith("MATCH"):
            # Parse basic pattern
            pattern = self._parse_pattern(cypher_like)
            results = self._match_pattern(pattern, params or {})
        
        return results
    
    def _parse_pattern(self, query: str) -> Dict:
        """Parse simplified MATCH pattern."""
        # Simplified parser for demonstration
        return {
            "source_label": self._extract_label(query, "source"),
            "rel_type": self._extract_type(query),
            "target_label": self._extract_label(query, "target"),
            "where": self._extract_where(query)
        }
    
    def _match_pattern(self, pattern: Dict, params: Dict) -> List[Dict]:
        """Match pattern against graph."""
        results = []
        
        for edge in self.edges:
            # Match relationship type
            if pattern["rel_type"] and edge["type"] != pattern["rel_type"]:
                continue
            
            source = self.nodes.get(edge["source"], {})
            target = self.nodes.get(edge["target"], {})
            
            # Match labels
            if pattern["source_label"] and source.get("label") != pattern["source_label"]:
                continue
            if pattern["target_label"] and target.get("label") != pattern["target_label"]:
                continue
            
            # Match where clause
            if pattern["where"] and not self._match_where(edge, source, target, params):
                continue
            
            results.append({
                "source": source,
                "relationship": edge,
                "target": target
            })
        
        return results

Temporal Knowledge Graph

from datetime import datetime
from typing import Optional

class TemporalKnowledgeGraph(PropertyGraph):
    def __init__(self):
        super().__init__()
        self.temporal_index = {}  # time_range -> [edge_ids]
    
    def create_temporal_relationship(
        self, 
        source_id: str, 
        rel_type: str, 
        target_id: str,
        valid_from: datetime,
        valid_until: Optional[datetime] = None,
        properties: Dict = None
    ) -> str:
        """Create relationship with temporal validity."""
        edge_id = super().create_relationship(
            source_id, rel_type, target_id, properties
        )
        
        # Index temporally
        time_key = self._time_range_key(valid_from, valid_until)
        if time_key not in self.temporal_index:
            self.temporal_index[time_key] = []
        self.temporal_index[time_key].append(edge_id)
        
        # Store validity on edge
        edge = self._get_edge(edge_id)
        edge["valid_from"] = valid_from.isoformat()
        edge["valid_until"] = valid_until.isoformat() if valid_until else None
        
        return edge_id
    
    def query_at_time(self, query: str, query_time: datetime) -> List[Dict]:
        """Query graph state at specific time."""
        # Find edges valid at query time
        valid_edges = []
        for edge in self.edges:
            valid_from = datetime.fromisoformat(edge.get("valid_from", "1970-01-01"))
            valid_until = edge.get("valid_until")
            
            if valid_from <= query_time:
                if valid_until is None or datetime.fromisoformat(valid_until) > query_time:
                    valid_edges.append(edge)
        
        # Match against pattern
        pattern = self._parse_pattern(query)
        results = []
        
        for edge in valid_edges:
            if pattern["rel_type"] and edge["type"] != pattern["rel_type"]:
                continue
            
            source = self.nodes.get(edge["source"], {})
            target = self.nodes.get(edge["target"], {})
            
            results.append({
                "source": source,
                "relationship": edge,
                "target": target
            })
        
        return results
    
    def _time_range_key(self, start: datetime, end: Optional[datetime]) -> str:
        """Create time range key for indexing."""
        start_str = start.isoformat()
        end_str = end.isoformat() if end else "infinity"
        return f"{start_str}::{end_str}"

Memory Consolidation

class MemoryConsolidator:
    def __init__(self, graph: PropertyGraph, vector_store: VectorStore):
        self.graph = graph
        self.vector_store = vector_store
        self.consolidation_threshold = 1000  # memories before consolidation
    
    def should_consolidate(self) -> bool:
        """Check if consolidation should trigger."""
        total_memories = len(self.graph.nodes) + len(self.graph.edges)
        return total_memories > self.consolidation_threshold
    
    def consolidate(self):
        """Run consolidation process."""
        # Step 1: Identify duplicate or merged facts
        duplicates = self.find_duplicates()
        
        # Step 2: Merge related facts
        for group in duplicates:
            self.merge_fact_group(group)
        
        # Step 3: Update validity periods
        self.update_validity_periods()
        
        # Step 4: Rebuild indexes
        self.rebuild_indexes()
    
    def find_duplicates(self) -> List[List]:
        """Find groups of potentially duplicate facts."""
        # Group by subject and predicate
        groups = {}
        
        for edge in self.graph.edges:
            key = (edge["source"], edge["type"])
            if key not in groups:
                groups[key] = []
            groups[key].append(edge)
        
        # Return groups with multiple edges
        return [edges for edges in groups.values() if len(edges) > 1]
    
    def merge_fact_group(self, edges: List[Dict]):
        """Merge group of duplicate edges."""
        if len(edges) == 1:
            return
        
        # Keep most recent/relevant
        keeper = max(edges, key=lambda e: e.get("properties", {}).get("confidence", 0))
        
        # Merge metadata
        for edge in edges:
            if edge["id"] != keeper["id"]:
                self.merge_properties(keeper, edge)
                self.graph.edges.remove(edge)
    
    def merge_properties(self, target: Dict, source: Dict):
        """Merge properties from source into target."""
        for key, value in source.get("properties", {}).items():
            if key not in target["properties"]:
                target["properties"][key] = value
            elif isinstance(value, list):
                target["properties"][key].extend(value)

Memory-Context Integration

class MemoryContextIntegrator:
    def __init__(self, memory_system, context_limit=100000):
        self.memory_system = memory_system
        self.context_limit = context_limit
    
    def build_context(self, task: str, current_context: str = "") -> str:
        """Build context including relevant memories."""
        # Extract entities from task
        entities = self._extract_entities(task)
        
        # Retrieve memories for each entity
        memories = []
        for entity in entities:
            entity_memories = self.memory_system.retrieve_entity(entity)
            memories.extend(entity_memories)
        
        # Format memories for context
        memory_section = self._format_memories(memories)
        
        # Combine with current context
        combined = current_context + "\n\n" + memory_section
        
        # Check limit and truncate if needed
        if self._token_count(combined) > self.context_limit:
            combined = self._truncate_context(combined, self.context_limit)
        
        return combined
    
    def _extract_entities(self, task: str) -> List[str]:
        """Extract entity mentions from task."""
        # In production, use NER or entity extraction
        import re
        pattern = r"\[([^\]]+)\]"  # [[entity_name]] convention
        return re.findall(pattern, task)
    
    def _format_memories(self, memories: List[Dict]) -> str:
        """Format memories for context injection."""
        sections = ["## Relevant Memories"]
        
        for memory in memories:
            formatted = f"- {memory.get('content', '')}"
            if "source" in memory:
                formatted += f" (Source: {memory['source']})"
            if "timestamp" in memory:
                formatted += f" [Time: {memory['timestamp']}]"
            sections.append(formatted)
        
        return "\n".join(sections)
    
    def _token_count(self, text: str) -> int:
        """Estimate token count."""
        return len(text) // 4  # Rough approximation
    
    def _truncate_context(self, context: str, limit: int) -> str:
        """Truncate context to fit limit."""
        tokens = context.split()
        truncated = []
        count = 0

        for token in tokens:
            if count + 1 > limit:
                break
            truncated.append(token)
            count += 1

        return " ".join(truncated)

Framework Integration Examples

Mem0 Quick Start

from mem0 import Memory

# Initialize with default config (uses local storage)
m = Memory()

# Store memories with user scoping
m.add("Prefers Python 3.12 with type hints", user_id="dev-alice")
m.add("Working on microservices migration", user_id="dev-alice")

# Search with natural language
results = m.search("What language does the user prefer?", user_id="dev-alice")

# Batch operations
m.add([
    "Sprint goal: complete auth service",
    "Blocked on database schema review"
], user_id="dev-alice")

Graphiti (Zep's Open-Source Temporal KG Engine)

from graphiti_core import Graphiti
from graphiti_core.nodes import EpisodeType

# Initialize with Neo4j backend
graphiti = Graphiti("bolt://localhost:7687", "neo4j", "password")

# Add episodes (conversations, events)
await graphiti.add_episode(
    name="user_conversation_42",
    episode_body="Alice mentioned she moved to Berlin in January.",
    source=EpisodeType.message,
    source_description="Chat with Alice"
)

# Search combines semantic, keyword, and graph traversal
results = await graphiti.search("Where does Alice live?")

Cognee (Open-Source Knowledge Engine for AI Memory)

import cognee
from cognee.modules.search.types import SearchType

# ECL pipeline: add → cognify → memify → search
await cognee.add("./docs/")
await cognee.add("any-data")
await cognee.cognify()
await cognee.memify()

# Graph-aware retrieval (default: GRAPH_COMPLETION)
results = await cognee.search(
    query_text="any query to search in memory",
    query_type=SearchType.GRAPH_COMPLETION,
)

# Raw chunks when agent reasons over text itself
chunks = await cognee.search(
    query_text="any query to search in memory",
    query_type=SearchType.CHUNKS,
)