System Architecture Blueprint

RAG-008: Agentic Long-Term Memory — Engineering Blueprint

This specification details the mathematical formulations, system design, memory classification layers, and evaluation modules for RAG-008 (Agentic Long-Term Memory RAG). It transitions autonomous agents from static context processors into stateful, evolving partners.

1. System Specifications & Performance SLA

Memory Consolidation Latency: $< 150\text{ms}$ executed asynchronously on separate worker threads to avoid blocking real-time chat loops.

Semantic Retrieval SLA: $< 30\text{ms}$ query latency over $1,000,000$ active memory nodes (utilizing local HNSW pgvector indices).

Episodic Reconstruction Fidelity: $> 98\%$ retrieval accuracy on temporal query chains (e.g., "What was the user's primary goal three sessions ago?").

Memory Optimization Target: Dynamic consolidation merging rules to reduce raw chat history token bloat by up to $85\%$ in long-running agent loops.

1. Agentic memory Ingestion & Consolidation Topology

Standard agent pipelines dump entire chat histographic records into prompts, exceeding token boundaries and degrading LLM attention. RAG-008 separates memory into three distinct architectural spaces:

                      [ Raw Real-Time User Input ]
                                   │
                                   ▼
                      [ Working Memory (Session) ]
                       • RAM-bound local state
                       • Active context variables
                                   │
                ┌──────────────────┴──────────────────┐
                ▼                                     ▼
    [ Episodic Memory Pool ]              [ Semantic Memory Pool ]
     • Raw, sequenced events               • Abstracted facts & rules
     • Logs of past conversations          • Evolving user preferences
     • Stored in pgvector                  • Consolidated periodically
                │                                     │
                └──────────────────┬──────────────────┘
                                   ▼
                   [ Memory Optimization Router ]
                    • Applies Ebbinghaus Decay math
                    • Resolves mutation conflicts
                    • Purges stale contextual links

2.1 The Three Memory Domains

Working Memory: The active prompt envelope, holding current task definitions and the immediate chat history.

Episodic Memory: Chronological records of past sessions, captured as vectorized interaction logs ($1536$-dimensional vector space via pgvector or Supabase).

Semantic Memory: Abstracted, structured assertions (e.g., User prefers Python over TypeScript, User is planning a vacation in October). These represent consolidated behavior patterns, decoupled from specific conversation times.

1. Mathematical Formulations for memory Dynamics

An agentic memory layer cannot simply stack facts forever; it must decay low-importance events, reinforce recurring patterns, and resolve factual contradictions as they occur in real time.

3.1 Temporal Decay Model (Ebbinghaus Forgetting Curve)

The retrieval strength $S_{\text{retrieval}}(M, t)$ of a memory node $M$ at epoch time $t$ decreases over time unless reinforced. This decay is governed by the Ebbinghaus retention curve:

$$S_{\text{retrieval}}(M, t) = \cos(\mathbf{e}_Q, \mathbf{e}_M) \cdot e^{-\frac{t - t_0}{H(M)}}$$

Where:

$\mathbf{e}_Q$ and $\mathbf{e}_M$ are the query and memory vectors, respectively.

$t_0$ is the epoch timestamp when the memory node was last accessed or reinforced.

$H(M) \in (0, \infty)$ is the Memory Half-Life factor, which dictates how fast a memory is forgotten. Highly impactful memories are assigned high $H(M)$ values.

3.2 Cognitive Memory Reinforcement

When a memory node $M$ is accessed during interaction, its half-life $H(M)$ increases, reducing its future decay speed:

$$H(M)_{t+1} = H(M)_t + \eta \cdot (1.0 - e^{-\alpha \cdot C(M)})$$

Where:

$\eta$ is the reinforcement learning rate.

$C(M)$ is the count of lifetime retrievals of memory $M$.

$\alpha > 0$ is a calibration coefficient regulating reinforcement saturation.

3.3 Conflict Resolution (Fact Mutation)

If a new observation $M_{\text{new}}$ contradicts an existing semantic fact $M_{\text{old}}$ (measured via high semantic overlap but contradictory logical values, e.g., User lives in Seattle vs. User moved to Boston), the memory engine resolves the state. We calculate a resolution score $R$:

$$R = \arg\max_{m \in {M_{\text{new}}, M_{\text{old}}}} \left( \beta \cdot \text{Confidence}(m) + (1.0 - \beta) \cdot \text{Recency}(m) \right)$$

Where:

$\text{Confidence}(m)$ is the model's extraction confidence score.

$\text{Recency}(m)$ is normalized temporal proximity: $1.0 - \frac{t_{\text{current}} - t_m}{t_{\text{current}} - t_{\text{origin}}}$.

$\beta \in [0.0, 1.0]$ is the weight balancing truth credibility against historical updates. The lower-scoring node is archived into Episodic History, and the higher-scoring node updates the active Semantic Memory index.

1. Complete Implementation Specification

This self-contained Python module uses an embedded SQLite database to model the relational metadata, NumPy to calculate vector similarities, and custom consolidation logic to parse, mutate, reinforce, and decay memories.

import sqlite3 import json import time import numpy as np from typing import List, Dict, Any, Tuple, Optional

class AgentMemoryManager: def init(self, db_path: str = "agent_memory.db"): self.conn = sqlite3.connect(db_path) self.cursor = self.conn.cursor() self._initialize_schema()

def _initialize_schema(self):
    """Build relational tables for Episodic and Semantic memory nodes."""
    # SQLite lacks native vector types, so we simulate pgvector storing embeddings as BLOBs or JSON
    self.cursor.execute("""
        CREATE TABLE IF NOT EXISTS episodic_memories (
            id TEXT PRIMARY KEY,
            user_id TEXT NOT NULL,
            session_id TEXT NOT NULL,
            interaction_text TEXT NOT NULL,
            embedding_json TEXT NOT NULL,
            created_at REAL NOT NULL
        );
    """)
    self.cursor.execute("""
        CREATE TABLE IF NOT EXISTS semantic_memories (
            id TEXT PRIMARY KEY,
            user_id TEXT NOT NULL,
            fact_key TEXT NOT NULL,
            fact_value TEXT NOT NULL,
            embedding_json TEXT NOT NULL,
            half_life REAL NOT NULL,
            access_count INTEGER DEFAULT 1,
            last_accessed_at REAL NOT NULL,
            created_at REAL NOT NULL,
            UNIQUE(user_id, fact_key)
        );
    """)
    self.conn.commit()

def _cosine_similarity(self, v1: List[float], v2: List[float]) -> float:
    a = np.array(v1)
    b = np.array(v2)
    dot = np.dot(a, b)
    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(dot / (norm_a * norm_b))

def write_episodic_record(self, user_id: str, session_id: str, interaction_text: str, embedding: List[float]) -> str:
    """Saves a raw chronological interaction segment into the episodic database log."""
    mem_id = f"epi_{uuid_like()}"
    now = time.time()
    self.cursor.execute("""
        INSERT INTO episodic_memories (id, user_id, session_id, interaction_text, embedding_json, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (mem_id, user_id, session_id, interaction_text, json.dumps(embedding), now))
    self.conn.commit()
    return mem_id

def upsert_semantic_fact(self, user_id: str, fact_key: str, fact_value: str, embedding: List[float]) -> Tuple[str, str]:
    """
    Inserts or mutates a consolidated semantic memory fact.
    Detects contradictions and manages conflict resolution rules.
    """
    now = time.time()
    fact_key_clean = fact_key.strip().lower()

    # Check if user already has an active semantic node matching this key
    self.cursor.execute("""
        SELECT id, fact_value, created_at FROM semantic_memories 
        WHERE user_id = ? AND fact_key = ?
    """, (user_id, fact_key_clean))
    row = self.cursor.fetchone()

    if row:
        old_id, old_value, old_created_at = row
        if old_value != fact_value:
            # Fact Mutation Conflict Detected! Execute mathematical resolution
            # For demonstration, we trust the newest value but update reinforcement metrics
            self.cursor.execute("""
                UPDATE semantic_memories
                SET fact_value = ?, embedding_json = ?, last_accessed_at = ?, half_life = half_life + 10.0, access_count = access_count + 1
                WHERE id = ?
            """, (fact_value, json.dumps(embedding), now, old_id))
            self.conn.commit()
            return old_id, "MUTATED"
        return old_id, "UNCHANGED"

    # Fresh memory node insertion
    new_id = f"sem_{uuid_like()}"
    initial_half_life = 86400.0 # Default decay boundary: 24 hours
    self.cursor.execute("""
        INSERT INTO semantic_memories (id, user_id, fact_key, fact_value, embedding_json, half_life, last_accessed_at, created_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """, (new_id, user_id, fact_key_clean, fact_value, json.dumps(embedding), initial_half_life, now, now))
    self.conn.commit()
    return new_id, "CREATED"

def retrieve_active_memories(self, user_id: str, query_embedding: List[float], similarity_threshold: float = 0.5) -> List[Dict[str, Any]]:
    """
    Retrieves active memory segments.
    Calculates Ebbinghaus Temporal Forgetting curves inside the matching loops.
    """
    now = time.time()
    self.cursor.execute("""
        SELECT id, fact_key, fact_value, embedding_json, half_life, last_accessed_at, access_count 
        FROM semantic_memories 
        WHERE user_id = ?
    """, (user_id,))
    rows = self.cursor.fetchall()

    scored_memories = []
    for row in rows:
        mem_id, key, val, emb_json, half_life, last_accessed, access_count = row
        emb = json.loads(emb_json)

        # 1. Base Semantic Similarity (Cosine)
        base_sim = self._cosine_similarity(query_embedding, emb)

        # 2. Apply Ebbinghaus Forgetting Curve: Sim * e^(-delta_t / half_life)
        delta_t = now - last_accessed
        decay_factor = np.exp(-delta_t / half_life)
        decayed_similarity = base_sim * decay_factor

        if decayed_similarity >= similarity_threshold:
            # Reinforce memory path during active access
            self._reinforce_memory(mem_id, half_life, access_count, now)
            scored_memories.append({
                "id": mem_id,
                "fact_key": key,
                "fact_value": val,
                "raw_score": base_sim,
                "decayed_score": decayed_similarity
            })

    # Sort descending by decayed score
    return sorted(scored_memories, key=lambda x: x["decayed_score"], reverse=True)

def _reinforce_memory(self, mem_id: str, current_half_life: float, access_count: int, now: float):
    """Reinforces memories when accessed, increasing their half-life."""
    # S_{t+1} = S_t + eta * (1 - e^(-alpha * count))
    learning_rate = 3600.0 * 12 # Add up to 12 hours of extra retention
    alpha = 0.25
    reinforcement = learning_rate * (1.0 - np.exp(-alpha * access_count))
    new_half_life = current_half_life + reinforcement

    self.cursor.execute("""
        UPDATE semantic_memories
        SET half_life = ?, last_accessed_at = ?, access_count = access_count + 1
        WHERE id = ?
    """, (new_half_life, now, mem_id))
    self.conn.commit()

def close(self):
    self.conn.close()

def uuid_like() -> str: """Helper to generate standard random string identifiers.""" import random return "".join(random.choices("abcdef0123456789", k=12))

1. Verification & Validation Protocols

5.1 Temporal Forgetting Assertions

Verify that the Ebbinghaus forgetting calculations run correctly, systematically scaling down older, un-reinforced memory vectors.

import pytest

def test_ebbinghaus_forgetting_decay(): manager = AgentMemoryManager(db_path=":memory:") user_id = "usr_01"

# Embedding template representing machine learning interests
query_vector = [1.0, 0.0, 0.0, 0.0]

# 1. Store a semantic fact (User is building ML projects)
new_id, status = manager.upsert_semantic_fact(
    user_id=user_id,
    fact_key="interest",
    fact_value="Building neural compilers",
    embedding=[0.95, 0.05, 0.0, 0.0]
)

# Assert successful database initialization
assert status == "CREATED"

# 2. Retrieve immediately. Score should be near absolute
active = manager.retrieve_active_memories(user_id, query_vector)
assert len(active) == 1
assert active[0]["decayed_score"] > 0.80

# 3. Simulate deep time elapse (e.g. 5 days of absolute silence)
now_simulated = time.time() + (86400.0 * 5)

# Manually adjust db timestamps to simulate time leap
manager.cursor.execute(
    "UPDATE semantic_memories SET last_accessed_at = ?", 
    (time.time() - (86400.0 * 5),)
)
manager.conn.commit()

# 4. Attempt retrieval under strict similarity metrics. 
# The decayed score should fall below standard threshold boundaries
inactive = manager.retrieve_active_memories(user_id, query_vector, similarity_threshold=0.80)
assert len(inactive) == 0

manager.close()

5.2 Conflict Resolution Integrity Testing

Assert that conflicting updates mutate key facts cleanly, updating access counts and memory half-lives without duplicating relational rows.

def test_conflict_mutation_deduplication(): manager = AgentMemoryManager(db_path=":memory:") user_id = "usr_02" fact_key = "current_location"

# Step A: User lives in Seattle
idx_1, status_1 = manager.upsert_semantic_fact(
    user_id=user_id,
    fact_key=fact_key,
    fact_value="Seattle, WA",
    embedding=[0.1, 0.2, 0.8]
)

# Step B: User moves to Boston
idx_2, status_2 = manager.upsert_semantic_fact(
    user_id=user_id,
    fact_key=fact_key,
    fact_value="Boston, MA",
    embedding=[0.1, 0.2, 0.8]
)

# Verify that conflict updates the same row instead of inserting duplicates
assert idx_1 == idx_2
assert status_2 == "MUTATED"

manager.cursor.execute(
    "SELECT count(*) FROM semantic_memories WHERE user_id = ? AND fact_key = ?", 
    (user_id, fact_key)
)
row_count = manager.cursor.fetchone()[0]
assert row_count == 1

manager.close()

Required Technical Skills

Advanced Engineering Skills Inventory: Agentic Long-Term Memory RAG (RAG-008)

This matrix details the specialized database security, cognitive architecture design, mathematical modeling, and asynchronous task-queue competencies required to build and maintain evolving long-term memory layers for autonomous AI agents.

1. Cognitive Memory Partitioning & Architecture Design

Moving past static chat logs requires designing systems that mimic human cognitive structures, partitioning memory dynamically to optimize context execution and query speed.

              [ Raw Agent Conversational Interactions ]
                                 │
              ┌──────────────────┴──────────────────┐
              ▼                                     ▼
  [ Episodic Memory Pool ]               [ Semantic Memory Pool ]
   • Sequenced chronological logs         • Consolidated abstract facts
   • Raw text + dense vector arrays       • Evolving preferences (K-V)
   • Decoupled background inputs          • Time-decayed index mappings

Cognitive Domain Boundaries:

Setting up the operational limits of Working Memory (short-term, RAM-bound context queues matching recent session turns) vs. Episodic Memory (immutable, indexed chronologies of individual user turns) vs. Semantic Memory (mutable, consolidated abstract factual records).

Structuring dynamic context assembly patterns to seamlessly stitch relevant semantic memories back into active working prompts during chat execution loops.

Memory Rollup Mechanics:

Engineering scheduled or criteria-driven triggers (such as interaction density, temporal windows, or NLP assertion cues like "I changed my mind") to move data from raw episodic tables to structured semantic indexes.

1. Secure Multi-Tenant Vector Database Operations (pgvector & Supabase)

Storing personal and behavioral preferences (home addresses, working hours, sensitive patterns) requires configuring strict logical security boundaries.

PostgreSQL Row-Level Security (RLS) Rules:

Implementing enterprise-grade RLS configurations on relational tables storing vector types (vector(1536)).

Binding database policies directly to incoming API JSON Web Token (JWT) claims, mapping auth.jwt_user_tenant_id() check constraints directly at the database driver engine.

Configuring custom Postgres schema indices (specifically HNSW indices with cosine operators) tuned to process multi-tenant isolated search queries efficiently.

Relational-Vector Operations:

Writing unified SQL queries that perform exact relational joins alongside dense cosine-similarity searches to merge user meta-records with semantic facts.

1. High-Throughput Asynchronous Consolidation Pipelines

Executing heavy language model assertion runs and embedding recalculations synchronously ruins real-time user experiences. You must know how to decouple ingestion from compilation.

[ Ingress Chat Gateway ] ──► [ Redis Ingestion Stream ] ──► [ Background Celery Pool ] • Structured fact extraction • Contradiction checks (vector) • Mutating state commits

Event-Driven Task Distribution:

Configuring message brokers (such as Redis Streams, RabbitMQ, or Amazon SQS) to buffer episodic transactions synchronously during chat turns.

Setting up background processing pools (such as Python Celery or task-running workers) to poll streams asynchronously, isolating heavy GPU/CPU workloads from fast API worker threads.

Conflict Detection & Mutation Management:

Designing multi-step vector verification runs: querying existing semantic records using a new fact's vector representation to identify logical contradictions.

Constructing thread-safe SQL writing blocks to safely mutate or archive outdated factual nodes without corrupting the historical episodic log.

1. Mathematical Modeling & Memory Decay Dynamics

An agent's memory cannot grow infinitely; it must systematically decay low-importance nodes, reinforce accessed assertions, and resolve contradictions mathematically.

Ebbinghaus Forgetting Curve Formulations:

Implementing time-decayed retrieval formulas where score retention scales down exponentially relative to memory half-lives ($H(M)$):

$$S_{\text{retrieval}}(M, t) = \cos(\mathbf{e}_Q, \mathbf{e}_M) \cdot e^{-\frac{t - t_0}{H(M)}}$$

Dynamically calculating decayed similarity scores inside query execution loops to prioritize fresh connections.

Cognitive Reinforcement Math:

Implementing reinforcing multipliers that adjust memory half-life values when nodes are retrieved during active sessions:

$$H(M)_{t+1} = H(M)_t + \eta \cdot (1.0 - e^{-\alpha \cdot C(M)})$$

Tuning hyperparameters ($\eta$, $\alpha$) to calibrate retention limits across various SaaS consumer tiers.

Conflict Resolution Formulas:

Applying weighted optimization formulas to resolve semantic updates by balancing confidence weights against normalized recency indicators:

$$R = \arg\max_{m \in {M_{\text{new}}, M_{\text{old}}}} \left( \beta \cdot \text{Confidence}(m) + (1.0 - \beta) \cdot \text{Recency}(m) \right)$$

1. Next.js Dynamic Memory State Visualizers

Exposing agent mental spaces to developers or B2B compliance administrators requires building highly performant UI components that can render real-time vector properties and simulate calculations on the client side.

Temporal Simulation Hooks:

Writing complex useMemo hooks in React to calculate simulated memory decay scores instantly inside browser threads when users interact with timeline sliders.

Implementing visual progress layouts to render dynamic memory properties (e.g., color shifts from emerald-green to rose-red when a memory decays past retrieval thresholds, adding strikethroughs, or fading opacity).

Interactive Fact Updates:

Building event-driven UI triggers (e.g., "Reinforce Fact" or "Inject Conflict") to dynamically update component states, resetting elapsed decay timers and modifying values with clean visual animations.

🚀 Quick Installation & Setup Guide

To run this project locally, follow these simple steps designed for both developers and users with no prior programming experience.

1. Download Python: Make sure you have Python 3.10+ installed on your computer. Download it from the official Python Website. During installation, make sure to check the box "Add Python to PATH".
2. Extract / Setup Code: Open the terminal (Command Prompt on Windows) and navigate to the project directory, or download/extract the project files.
3. Install Dependencies (pip): In Command Prompt, run the following command to install all necessary packages automatically:

   pip install -r requirements.txt

4. Setup API Keys / Environment: Open the .env file in a text editor (like Notepad) and paste your API keys:

   GEMINI_API_KEY=your_gemini_api_key_here

5. Launch the Application: Double-click the run_project.bat or run_pito.bat file, or run the following command:

   python app.py

   (or app_gui.py)