Neo4j Memory Systems Design Patterns Catalog¶

def hybrid_search(query, kg, top_k=10):
    # Stage 1: Semantic search (vector similarity)
    query_embedding = embed(query)
    semantic_results = kg.vector_search(query_embedding, top_k=50)

    # Stage 2: Structural search (graph traversal)
    entities = extract_entities(query)
    structural_results = kg.graph_query("""
        MATCH (e:Entity)-[*1..2]-(related)
        WHERE e.name IN $entities
        RETURN related
    """, entities=entities)

    # Stage 3: Temporal filtering (recency boost)
    recent_threshold = datetime.now() - timedelta(days=30)
    temporal_results = kg.query("""
        MATCH (ep:Episode)-[:MENTIONS]->(e:Entity)
        WHERE ep.timestamp > $threshold
        RETURN e
    """, threshold=recent_threshold)

    # Stage 4: Reciprocal Rank Fusion (RRF)
    def rrf_score(item, rank_lists, k=60):
        score = 0
        for rank_list in rank_lists:
            if item in rank_list:
                rank = rank_list.index(item)
                score += 1 / (k + rank)
        return score

    all_results = set(semantic_results + structural_results + temporal_results)
    ranked = sorted(all_results,
                   key=lambda x: rrf_score(x, [semantic_results, structural_results, temporal_results]),
                   reverse=True)

    return ranked[:top_k]

class IncrementalGraphUpdater:
    def update_file(self, file_path, new_content, old_content=None):
        # Parse both versions
        new_ast = parse_file(file_path, new_content)
        old_ast = parse_file(file_path, old_content) if old_content else None

        # Extract entities from both
        new_entities = extract_entities(new_ast)
        old_entities = extract_entities(old_ast) if old_ast else []

        # Compute diff
        added = [e for e in new_entities if e not in old_entities]
        removed = [e for e in old_entities if e not in new_entities]
        modified = [e for e in new_entities if e in old_entities and changed(e)]

        # Apply updates atomically
        with self.db.transaction():
            # Remove deleted entities
            for entity in removed:
                self.db.delete_node(entity.id)

            # Add new entities
            for entity in added:
                self.db.create_node(entity.type, entity.properties)

            # Update modified entities
            for entity in modified:
                self.db.update_node(entity.id, entity.properties)

            # Recompute relationships only for affected entities
            affected = added + modified
            self.update_relationships(affected)

class MultiModalMemory:
    def __init__(self):
        # Specialized memory stores
        self.core = CoreMemory()           # Persistent facts (agent + user identity)
        self.episodic = EpisodicMemory()   # Time-stamped events
        self.semantic = SemanticMemory()   # Entity relationships
        self.procedural = ProceduralMemory()  # How-to knowledge
        self.resource = ResourceMemory()   # Documents, code files

        # Meta-manager routes events to appropriate stores
        self.meta_manager = MetaMemoryManager()

    def process_event(self, event):
        # Route to appropriate memory stores
        routing = self.meta_manager.route(event)
        # Example: conversation → episodic + semantic
        #          code_change → resource + semantic + episodic
        #          error_resolution → procedural + episodic

        for component, instructions in routing.items():
            memory = getattr(self, component)
            memory.update(event, instructions)

    def retrieve(self, query):
        # Parallel retrieval from all components
        results = {
            "core": self.core.retrieve(query),
            "episodic": self.episodic.retrieve(query),
            "semantic": self.semantic.retrieve(query),
            "procedural": self.procedural.retrieve(query),
            "resource": self.resource.retrieve(query)
        }

        # Tag by source and format for LLM
        return self.format_for_llm(results)

class UnifiedMemoryGraph:
    def __init__(self, neo4j_driver):
        self.driver = neo4j_driver
        self.vector_db = VectorDatabase()

    def ingest_event(self, event):
        # 1. Store raw episode (non-lossy)
        episode = self.create_episode(event)

        # 2. Extract entities (semantic layer)
        entities = self.extract_entities(event.content)
        for entity in entities:
            existing = self.find_or_create_entity(entity)
            # Link episode to entity
            self.link(episode, "MENTIONS", existing)

        # 3. Extract relationships (semantic layer)
        relationships = self.extract_relationships(entities)
        for rel in relationships:
            self.create_relationship(rel)

        # 4. Update communities (incremental)
        affected_communities = self.find_communities(entities)
        for community in affected_communities:
            self.update_community_summary(community)

    def retrieve(self, query, top_k=10):
        # Multi-stage retrieval

        # Stage 1: Semantic search (vector similarity)
        embedding = self.embed(query)
        candidate_entities = self.vector_db.search(embedding, top_k=50)

        # Stage 2: Graph traversal (structural)
        expanded = self.graph_expand(candidate_entities, depth=2)

        # Stage 3: Temporal filtering
        recent = self.filter_by_recency(expanded)

        # Stage 4: Episode retrieval
        episodes = self.get_episodes(recent)

        # Stage 5: Reranking
        reranked = self.rerank(
            entities=recent,
            episodes=episodes,
            query=query
        )

        return reranked[:top_k]

class FederatedMemory:
    def __init__(self):
        # Separate stores optimized for different access patterns
        self.core = InMemoryStore()        # Fast, small, persistent
        self.episodic = TimeSeriesDB()     # Time-ordered, append-only
        self.semantic = Neo4jGraph()       # Graph queries, relationships
        self.procedural = DocumentDB()     # Full-text search, workflows
        self.resource = ObjectStore()      # Large files, S3/filesystem

        # Federation layer coordinates queries
        self.federation = FederationLayer()

    def query(self, query_text):
        # Parse query to determine relevant stores
        query_plan = self.federation.plan(query_text)

        # Parallel queries to relevant stores
        results = {}
        with ThreadPoolExecutor() as executor:
            futures = {
                executor.submit(store.query, query_text): store_name
                for store_name, store in query_plan.stores.items()
            }

            for future in as_completed(futures):
                store_name = futures[future]
                results[store_name] = future.result()

        # Merge and rank results
        return self.federation.merge(results, query_text)

class CodeMemoryIntegration:
    def __init__(self, codebase_path):
        self.episodic = EpisodicMemory()
        self.semantic = SemanticMemory()
        self.procedural = ProceduralMemory()

        # Integrate blarify for code parsing
        self.code_graph = CodeGraph(codebase_path, use_scip=True)

    def on_file_change(self, file_path, new_content):
        # 1. Update code graph (incremental)
        self.code_graph.update_file(file_path, new_content)

        # 2. Create episode
        episode = self.episodic.record({
            "type": "code_change",
            "file": file_path,
            "timestamp": datetime.now()
        })

        # 3. Link episode to affected functions
        affected_functions = self.code_graph.extract_entities(file_path)
        for func in affected_functions:
            self.link(episode, "MODIFIED", func)

    def on_error(self, error):
        # 1. Record episode
        episode = self.episodic.record(error)

        # 2. Link to code location
        if error.file and error.line:
            func = self.code_graph.find_function_at(error.file, error.line)
            self.link(episode, "OCCURRED_IN", func)

        # 3. Find applicable procedure
        procedure = self.procedural.find_by_error_type(error.type)

        return {
            "procedure": procedure,
            "similar_errors": self.episodic.find_similar(error),
            "affected_code": func
        }

def create_indexes(driver):
    """Create indexes for optimal query performance"""
    indexes = [
        # Node property indexes (exact match)
        "CREATE INDEX entity_name IF NOT EXISTS FOR (e:Entity) ON (e.name)",
        "CREATE INDEX episode_type IF NOT EXISTS FOR (ep:Episode) ON (ep.type)",
        "CREATE INDEX function_file IF NOT EXISTS FOR (f:Function) ON (f.file_path)",

        # Composite indexes (multiple properties)
        "CREATE INDEX entity_name_type IF NOT EXISTS FOR (e:Entity) ON (e.name, e.type)",

        # Full-text indexes (text search)
        "CREATE FULLTEXT INDEX entity_content IF NOT EXISTS FOR (e:Entity) ON EACH [e.name, e.summary, e.description]",

        # Range indexes (temporal queries)
        "CREATE INDEX episode_timestamp IF NOT EXISTS FOR (ep:Episode) ON (ep.timestamp)",
    ]

    for index in indexes:
        driver.execute_query(index)

def multi_stage_retrieval(query, kg, top_k=10):
    # Stage 1: Broad semantic search (cast wide net)
    embedding = embed(query)
    candidates = kg.vector_search(embedding, top_k=100)

    # Stage 2: Graph expansion (find related entities)
    expanded = kg.graph_query("""
        MATCH (e:Entity)-[*1..2]-(related)
        WHERE id(e) IN $candidate_ids
        RETURN DISTINCT related
    """, candidate_ids=[c.id for c in candidates])

    # Stage 3: Temporal filtering (recency boost)
    recent = [e for e in expanded
              if e.updated_at > datetime.now() - timedelta(days=30)]

    # Stage 4: Episode retrieval (get context)
    episodes = kg.query("""
        MATCH (ep:Episode)-[:MENTIONS]->(e:Entity)
        WHERE id(e) IN $entity_ids
        RETURN ep, e
        ORDER BY ep.timestamp DESC
    """, entity_ids=[e.id for e in recent])

    # Stage 5: Reranking (combine signals)
    scored = []
    for entity in recent:
        score = (
            0.5 * semantic_score(entity, query) +
            0.3 * recency_score(entity) +
            0.2 * frequency_score(entity, episodes)
        )
        scored.append((entity, score))

    return sorted(scored, key=lambda x: x[1], reverse=True)[:top_k]

def handle_new_fact(new_fact, kg):
    # Find potentially contradicting facts
    related_facts = kg.query("""
        MATCH (e1:Entity)<-[:ABOUT]-(f:Fact)-[:ABOUT]->(e2:Entity)
        WHERE e1.id = $entity1 AND e2.id = $entity2
          AND f.t_invalid IS NULL  // Only active facts
        RETURN f
    """, entity1=new_fact.entity1, entity2=new_fact.entity2)

    # Check for contradictions
    for old_fact in related_facts:
        if contradicts(new_fact, old_fact):
            # Temporal invalidation (don't delete)
            old_fact.t_invalid = new_fact.t_valid
            old_fact.invalidated_by = new_fact.id
            kg.update(old_fact)

            # Log contradiction
            kg.create_node("Contradiction", {
                "old_fact": old_fact.id,
                "new_fact": new_fact.id,
                "detected_at": datetime.now(),
                "resolution": "temporal_invalidation"
            })

    # Add new fact
    kg.add(new_fact)

def multi_hop_reasoning(query, kg, max_hops=3):
    # Extract seed entities from query
    seed_entities = extract_entities(query)

    # Iteratively expand
    results = []
    current_entities = seed_entities
    visited = set()

    for hop in range(max_hops):
        # Find related entities
        related = kg.query("""
            MATCH (e:Entity)-[r]-(related:Entity)
            WHERE e.id IN $entities
              AND NOT related.id IN $visited
            RETURN related, r, e, type(r) as rel_type
        """, entities=[e.id for e in current_entities],
             visited=list(visited))

        # Score by relevance (decay by distance)
        decay_factor = 0.7 ** hop
        for rel in related:
            score = (
                rel.r.strength * decay_factor *
                relationship_relevance(rel.rel_type, query)
            )
            results.append((rel.related, score, hop))
            visited.add(rel.related.id)

        # Update current entities for next hop
        current_entities = [r.related for r in related]

        # Early stopping if no new entities
        if not current_entities:
            break

    # Rerank by combined score
    return sorted(results, key=lambda x: x[1], reverse=True)

class ContextInjectionAgent:
    def __init__(self, memory):
        self.memory = memory
        # Pre-load context at agent initialization
        self.context = self.memory.get_recent_context(limit=100)

    def process_query(self, query):
        # Context already loaded
        response = self.llm(
            system_prompt=self.build_system_prompt(self.context),
            user_query=query
        )
        return response

class QueryBasedAgent:
    def __init__(self, memory):
        self.memory = memory

    def process_query(self, query):
        # Retrieve context only when needed
        relevant_context = self.memory.retrieve(query, top_k=10)

        response = self.llm(
            system_prompt=self.build_system_prompt(relevant_context),
            user_query=query
        )
        return response

class HybridAgent:
    def __init__(self, memory):
        self.memory = memory
        # Pre-load critical context (core memory)
        self.core_context = memory.core.get_all()

    def process_query(self, query):
        # Combine core + query-specific context
        query_context = self.memory.retrieve(query, top_k=10)
        full_context = self.core_context + query_context

        response = self.llm(
            system_prompt=self.build_system_prompt(full_context),
            user_query=query
        )
        return response

class SyncMemoryAgent:
    def process_event(self, event):
        # Memory operations block execution
        memory_id = self.memory.store(event)
        entities = self.memory.extract_entities(event)
        self.memory.update_graph(entities)

        # Continue only after memory updated
        return self.process_with_memory(memory_id)

class AsyncMemoryAgent:
    async def process_event(self, event):
        # Fire-and-forget memory operations
        asyncio.create_task(self.memory.store(event))
        asyncio.create_task(self.memory.extract_entities(event))

        # Continue immediately without waiting
        return await self.process_without_blocking()

class HybridMemoryAgent:
    async def process_event(self, event):
        # Write asynchronously (fire-and-forget)
        asyncio.create_task(self.memory.store(event))

        # Read synchronously (need result)
        context = await self.memory.retrieve(event.query)

        return await self.process(context)

class MemoryAwareAgent:
    def __init__(self, session_id):
        # 1. INITIALIZATION: Load persistent context
        self.memory = get_memory_manager(session_id)
        self.context = self.memory.restore_session_context()

    def on_user_message(self, message):
        # 2. PRE-PROCESSING: Retrieve relevant context
        relevant_memories = self.memory.retrieve(message, top_k=10)

        # 3. PROCESSING: Use memories in decision-making
        response = self.generate_response(message, relevant_memories)

        # 4. POST-PROCESSING: Store interaction
        self.memory.store({
            "type": "conversation",
            "message": message,
            "response": response,
            "timestamp": datetime.now()
        })

        return response

    def on_error(self, error):
        # ERROR HANDLING: Learn from errors
        self.memory.record_error(error)
        procedure = self.memory.find_procedure(error)
        return procedure

    def on_success(self, task):
        # SUCCESS HANDLING: Learn patterns
        self.memory.record_success(task)
        self.memory.learn_procedure(task)

    def on_session_end(self):
        # 5. TEARDOWN: Persist session state
        self.memory.preserve_session_context(
            summary=self.summarize_session(),
            decisions=self.decisions_made,
            tasks=self.active_tasks
        )

class ErrorPatternLearner:
    def __init__(self, memory):
        self.memory = memory

    def handle_error(self, error):
        # 1. Check for known procedure
        procedure = self.memory.procedural.find_procedure(error)

        if procedure:
            return {
                "procedure": procedure,
                "confidence": procedure.success_rate,
                "times_used": procedure.times_used
            }
        else:
            # 2. Find similar past errors
            similar = self.memory.episodic.find_similar({
                "type": "error",
                "error_type": error.type,
                "message": error.message
            })

            return {
                "similar_cases": similar,
                "confidence": 0.3  # Lower confidence (no exact procedure)
            }

    def record_resolution(self, error_id, steps_taken, success):
        # Update episode with resolution
        self.memory.episodic.update(error_id, {
            "resolution_steps": steps_taken,
            "success": success,
            "resolved_at": datetime.now()
        })

        # Learn procedure if successful
        if success:
            error = self.memory.episodic.get(error_id)

            # Check if procedure exists
            procedure = self.memory.procedural.find_by_trigger(error.type)

            if procedure:
                # Update success rate (exponential moving average)
                alpha = 0.1
                procedure.success_rate = (
                    alpha * 1.0 +
                    (1 - alpha) * procedure.success_rate
                )
                procedure.times_used += 1
                self.memory.procedural.update(procedure)
            else:
                # Create new procedure
                self.memory.procedural.create({
                    "name": f"Fix {error.type}",
                    "trigger_pattern": error.type,
                    "steps": steps_taken,
                    "success_rate": 1.0,
                    "times_used": 1,
                    "learned_from": error_id
                })
        else:
            # Record failure for learning
            if procedure:
                alpha = 0.1
                procedure.success_rate = (
                    alpha * 0.0 +
                    (1 - alpha) * procedure.success_rate
                )
                self.memory.procedural.update(procedure)

def batch_create_nodes_slow(nodes):
    # SLOW: Individual creates
    for node in nodes:
        driver.execute_query(
            "CREATE (n:Entity {id: $id, name: $name})",
            id=node.id, name=node.name
        )
    # 1000 nodes = 1000 network round-trips

def batch_create_nodes_fast(nodes):
    # FAST: Single query with UNWIND
    query = """
    UNWIND $batch as node
    CREATE (n:Entity)
    SET n = node
    """
    driver.execute_query(query, batch=nodes)
    # 1000 nodes = 1 network round-trip

# BAD: Concatenation (SQL injection, no caching)
query = f"MATCH (e:Entity {{name: '{name}'}}) RETURN e"
driver.execute_query(query)

# GOOD: Parameters (safe, cached)
query = "MATCH (e:Entity {name: $name}) RETURN e"
driver.execute_query(query, name=name)

class CachedMemoryRetrieval:
    def __init__(self, memory):
        self.memory = memory
        # L1: In-memory cache (fast, small)
        self.l1_cache = LRUCache(maxsize=100)
        # L2: Redis cache (medium, larger)
        self.l2_cache = RedisCache()
        # L3: Neo4j (slow, unlimited)
        self.l3_database = memory

    def retrieve(self, query):
        # L1: Check in-memory cache
        cache_key = hash_query(query)
        if cache_key in self.l1_cache:
            return self.l1_cache[cache_key]

        # L2: Check Redis cache
        result = self.l2_cache.get(cache_key)
        if result:
            self.l1_cache[cache_key] = result
            return result

        # L3: Query Neo4j
        result = self.memory.retrieve(query)

        # Populate caches
        self.l2_cache.set(cache_key, result, ttl=3600)
        self.l1_cache[cache_key] = result

        return result

    def invalidate(self, entity_id):
        # Invalidate relevant cache entries
        self.l1_cache.clear()  # Simple: clear all
        self.l2_cache.delete_pattern(f"*{entity_id}*")

class CommunityManager:
    def __init__(self, memory):
        self.memory = memory
        self.last_recompute = None
        self.recompute_interval = timedelta(hours=1)

    def update_entity(self, entity):
        # Update entity immediately
        self.memory.update(entity)

        # Schedule community recompute if needed
        if (not self.last_recompute or
            datetime.now() - self.last_recompute > self.recompute_interval):
            self.schedule_recompute()

    def schedule_recompute(self):
        # Run in background (celery, asyncio, etc.)
        asyncio.create_task(self.recompute_communities())

    async def recompute_communities(self):
        # Use graph algorithm (label propagation, louvain, etc.)
        query = """
        CALL gds.labelPropagation.stream({
            nodeProjection: 'Entity',
            relationshipProjection: 'RELATES_TO'
        })
        YIELD nodeId, communityId
        MATCH (e:Entity) WHERE id(e) = nodeId
        SET e.community_id = communityId
        """

        await self.memory.execute_query(query)
        self.last_recompute = datetime.now()

class SessionContinuityAgent:
    def __init__(self, session_id):
        self.session_id = session_id
        self.memory = get_memory_manager(session_id)

        # Restore previous session
        self.restore_session()

    def restore_session(self):
        """Restore session state from memory"""
        context = self.memory.restore_session_context(
            agent_id="orchestrator"
        )

        if context:
            # Restore conversation history
            self.conversation_history = context.get("conversation_summary", "")

            # Restore decisions
            self.decisions_made = context.get("key_decisions", [])

            # Restore active tasks
            self.active_tasks = context.get("active_tasks", [])

            # Restore agent states
            self.agent_states = context.get("agent_states", {})

            print(f"Restored session from {context['preserved_at']}")
        else:
            # New session
            self.conversation_history = ""
            self.decisions_made = []
            self.active_tasks = []
            self.agent_states = {}

    def on_session_end(self):
        """Preserve session state to memory"""
        self.memory.preserve_session_context(
            agent_id="orchestrator",
            summary=self.conversation_history,
            decisions=self.decisions_made,
            tasks=self.active_tasks,
            metadata={
                "agent_states": self.agent_states,
                "session_duration": self.get_session_duration(),
                "message_count": len(self.conversation_history)
            }
        )

class WorkflowStateManager:
    def __init__(self, workflow_name, memory):
        self.workflow_name = workflow_name
        self.memory = memory

        # Restore workflow state if exists
        self.state = memory.restore_workflow_state(workflow_name)
        if not self.state:
            self.state = self.initialize_workflow()

    def initialize_workflow(self):
        """Start new workflow"""
        return {
            "workflow_name": self.workflow_name,
            "current_step": "init",
            "completed_steps": [],
            "pending_steps": [],
            "step_results": {},
            "started_at": datetime.now(),
            "metadata": {}
        }

    def complete_step(self, step_name, results):
        """Mark step as complete and advance workflow"""
        # Update state
        self.state["completed_steps"].append(step_name)
        self.state["step_results"][step_name] = results

        # Determine next step
        if self.state["pending_steps"]:
            self.state["current_step"] = self.state["pending_steps"].pop(0)
        else:
            self.state["current_step"] = "completed"

        # Persist to memory (checkpoint)
        self.memory.preserve_workflow_state(
            workflow_name=self.workflow_name,
            current_step=self.state["current_step"],
            completed_steps=self.state["completed_steps"],
            pending_steps=self.state["pending_steps"],
            step_results=self.state["step_results"],
            workflow_metadata=self.state["metadata"]
        )

    def get_progress(self):
        """Get workflow progress"""
        total = len(self.state["completed_steps"]) + len(self.state["pending_steps"]) + 1
        completed = len(self.state["completed_steps"])

        return {
            "workflow_name": self.workflow_name,
            "current_step": self.state["current_step"],
            "progress_percentage": (completed / total) * 100,
            "completed": self.state["completed_steps"],
            "pending": self.state["pending_steps"]
        }

class CollaborativeMemory:
    def share_insight(self, from_agent, to_agent, insight):
        """Share insight between agents"""
        insight_id = self.memory.store({
            "agent_id": from_agent,
            "title": f"Insight for {to_agent}: {insight['title']}",
            "content": insight['content'],
            "memory_type": MemoryType.CONTEXT,
            "tags": ["collaboration", "insight", to_agent, from_agent],
            "metadata": {
                "recipient": to_agent,
                "shared_at": datetime.now()
            }
        })

        return insight_id

    def get_insights_for_agent(self, agent_id):
        """Get insights shared with specific agent"""
        return self.memory.retrieve(
            tags=["collaboration", "insight", agent_id],
            memory_type=MemoryType.CONTEXT
        )

    def record_collaboration(self, agents, collaboration_type, outcome):
        """Record collaborative work"""
        collab_data = {
            "participating_agents": agents,
            "collaboration_type": collaboration_type,
            "outcome": outcome,
            "collaborated_at": datetime.now()
        }

        # Store for each participating agent
        memory_ids = []
        for agent_id in agents:
            memory_id = self.memory.store({
                "agent_id": agent_id,
                "title": f"Collaboration: {collaboration_type}",
                "content": json.dumps(collab_data),
                "memory_type": MemoryType.CONTEXT,
                "tags": ["collaboration", collaboration_type] + agents
            })
            memory_ids.append(memory_id)

        return memory_ids

# DON'T DO THIS
name = "login'; DROP DATABASE; --"
query = f"MATCH (e:Entity {{name: '{name}'}}) RETURN e"
driver.execute_query(query)

# DO THIS
name = "login"
query = "MATCH (e:Entity {name: $name}) RETURN e"
driver.execute_query(query, name=name)

def on_file_change(file_path):
    # DON'T: Rebuild entire codebase graph
    rebuild_entire_graph(codebase_path)  # Takes 5 minutes

def on_file_change(file_path):
    # DO: Update only affected entities
    update_file_entities(file_path)  # Takes <1 second