AI Agent Memory Architecture: Redis Vector Store Guide

┌────────────────────────────────────────────────────┐
│                  AI Agent                           │
├────────────────────────────────────────────────────┤
│                                                     │
│  ┌────────────────┐  ┌─────────────┐  ┌────────┐  │
│  │  Short-Term    │  │  Episodic   │  │Semantic│  │
│  │  Memory        │  │  Memory     │  │Memory  │  │
│  ├────────────────┤  ├─────────────┤  ├────────┤  │
│  │Current session │  │Past sessions│  │Facts   │  │
│  │Buffer          │  │Vector search│  │User    │  │
│  │TTL: 1 hour     │  │TTL: 30 days │  │prefs   │  │
│  │Redis Streams   │  │Redis VSS    │  │Graph   │  │
│  └────────────────┘  └─────────────┘  └────────┘  │
│                                                     │
└────────────────────────────────────────────────────┘

# Option 1: Docker (recommended)
docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack

# Option 2: Production (Kubernetes)
helm repo add redis https://charts.bitnami.com/bitnami
helm install redis-stack redis/bitnami/redis \
  --set image.tag=7.4.0 \
  --set architecture=standalone \
  --set master.persistence.enabled=true

pip install redis redisvl openai  # redisvl = Redis Vector Library

┌─────────────────────────────────────┐
│     Agent Session (Per User)        │
├─────────────────────────────────────┤
│                                     │
│  Redis Stream: agent:session:{id}   │
│                                     │
│  Message 1: user "Hello"           │
│  Message 2: assis "Hi, how can I?" │
│  Message 3: user "I need help..."  │
│  ...                                │
│  Message N: (trimmed to last 50)   │
│                                     │
└─────────────────────────────────────┘

import redis
import json
from datetime import datetime

class ShortTermMemory:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.MAX_STREAM_LEN = 50  # Keep last 50 messages per session
        self.TTL_SECONDS = 3600   # 1 hour

    def add_message(self, session_id: str, role: str, content: str):
        """Add message to short-term memory stream."""
        stream_key = f"agent:session:{session_id}"

        message = {
            "role": role,
            "content": content,
            "timestamp": datetime.utcnow().isoformat()
        }

        # Add to stream and trim to max length
        self.redis.xadd(stream_key, message, maxlen=self.MAX_STREAM_LEN)
        # Set TTL on first message (extends each time)
        self.redis.expire(stream_key, self.TTL_SECONDS)

    def get_recent_messages(self, session_id: str, count: int = 20) -> list:
        """Get last N messages from stream."""
        stream_key = f"agent:session:{session_id}"

        # Get all messages, keep only last `count`
        all_msgs = self.redis.xrevrange(stream_key, max="+", min="-", count=count)
        messages = []
        for msg_id, msg_data in reversed(all_msgs):
            messages.append({
                "role": msg_data[b"role"].decode(),
                "content": msg_data[b"content"].decode(),
                "timestamp": msg_data.get(b"timestamp", b"").decode()
            })
        return messages

    def clear_session(self, session_id: str):
        """Clear session memory."""
        self.redis.delete(f"agent:session:{session_id}")

    def format_for_llm(self, session_id: str, max_tokens: int = 4000) -> str:
        """Format recent conversation for LLM context window."""
        messages = self.get_recent_messages(session_id)
        formatted = []
        token_count = 0

        for msg in reversed(messages):  # Process newest first
            line = f"{msg['role'].upper()}: {msg['content']}"
            tokens_approx = len(line.split()) * 1.3
            if token_count + tokens_approx > max_tokens:
                break
            formatted.insert(0, line)
            token_count += tokens_approx

        return "
".join(formatted)

Stream trimming ensures bounded memory:

Strategy 1: Fixed length (recommended)
  Keep last N messages (e.g., 50)
  Max 50 messages × ~100 tokens each = 5K tokens
  Fits in most context windows with room for other memory

Strategy 2: Token-budget
  Calculate token count per message
  Trim until total fits in budget (e.g., 4K tokens)
  More precise but slightly more complex

Strategy 3: Time-based (TTL)
  Auto-expire after 1 hour
  Short enough for short-term, long enough for conversation
  Combined with stream maxlen for safety

┌─────────────────────────────────────────────────────┐
│              Episodic Memory Index                    │
├─────────────────────────────────────────────────────┤
│                                                      │
│  Index: idx:episodic                                 │
│                                                      │
│  ┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐         │
│  │Embed │   │Embed │   │Embed │   │Embed │         │
│  │conv #1│   │conv #2│   │conv #3│   │conv #4│  ... │
│  └──┬───┘   └──┬───┘   └──┬───┘   └──┬───┘         │
│     │           │           │           │            │
│     └───────────┴───────────┴───────────┘            │
│                         │                            │
│              Similarity Search                      │
│                         │                            │
│                "When did we discuss                   │
│                 vector databases?"                   │
└─────────────────────────────────────────────────────┘

import numpy as np
from redisvl.index import SearchIndex
from redisvl.query import VectorQuery
from sentence_transformers import SentenceTransformer

class EpisodicMemory:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.embedder = SentenceTransformer("all-MiniLM-L6-v2")
        self._setup_index()

    def _setup_index(self):
        """Create or connect to vector search index."""
        schema = {
            "index": {"name": "idx:episodic", "prefix": "ep:memory:"},
            "fields": [
                {"name": "user_id", "type": "tag"},
                {"name": "session_id", "type": "tag"},
                {"name": "timestamp", "type": "numeric"},
                {"name": "summary", "type": "text"},
                {"name": "embedding", "type": "vector",
                 "attrs": {
                     "algorithm": "FLAT",
                     "dims": 384,
                     "distance_metric": "cosine",
                     "datatype": "float32"
                 }}
            ]
        }
        self.index = SearchIndex(self.redis, schema)
        self.index.create(overwrite=False)

    def store_session(self, user_id: str, session_id: str, conversation: str):
        """Store conversation summary as vector memory."""
        # Generate embedding
        embedding = self.embedder.encode(conversation).astype(np.float32).tobytes()

        # Generate summary (could use LLM for better quality)
        summary = conversation[:500]  # Simple truncation

        # Store in Redis
        key = f"ep:memory:{user_id}:{session_id}"
        self.redis.hset(key, mapping={
            "user_id": user_id,
            "session_id": session_id,
            "timestamp": datetime.utcnow().timestamp(),
            "summary": summary,
            "embedding": embedding
        })
        # Expire after 30 days
        self.redis.expire(key, 30 * 24 * 60 * 60)

    def search_memory(self, user_id: str, query: str, k: int = 5) -> list:
        """Search episodic memory by semantic similarity."""
        query_embedding = self.embedder.encode(query).astype(np.float32).tobytes()

        vector_query = VectorQuery(
            vector=query_embedding,
            vector_field_name="embedding",
            return_fields=["user_id", "session_id", "summary", "timestamp"],
            num_results=k,
            filters={"user_id": user_id}  # Filter by user
        )

        results = self.index.query(vector_query)
        return [
            {
                "session_id": r["session_id"],
                "summary": r["summary"],
                "timestamp": r["timestamp"],
                "score": r.get("vector_distance", 0)
            }
            for r in results
        ]

    def format_for_llm(self, memories: list) -> str:
        """Format retrieved memories for LLM context."""
        if not memories:
            return ""

        lines = ["Relevant past conversations:"]
        for mem in memories:
            timestamp = datetime.fromtimestamp(float(mem["timestamp"]))
            lines.append(
                f"[{timestamp.strftime('%Y-%m-%d %H:%M')}] "
                f"(similarity: {1 - float(mem['score']):.2f}): "
                f"{mem['summary']}"
            )
        return "
".join(lines)

# Search with both semantic AND time filter
vector_query = VectorQuery(
    vector=query_embedding,
    vector_field_name="embedding",
    return_fields=["summary", "timestamp"],
    num_results=10,
    filters={
        "user_id": user_id,
        # Only last 7 days
        "timestamp": (timestamp_7days_ago, float('inf'))
    }
)

┌──────────────────────────────────────────┐
│         Semantic Knowledge Graph         │
├──────────────────────────────────────────┤
│                                          │
│  [User: Alice] ──lives_in──→ [City: Tokyo]│
│       │                                       │
│       ├──prefers──→ [Topic: Python]           │
│       │                                       │
│       ├──has_account──→ [Exchange: Binance]   │
│       │                                       │
│       └──mentioned──→ [Interest: AI Agents]  │
│                                          │
│  [User: Alice] ──asked_about──→ [Query:    │
│    "What's the best GPU for ML?"]         │
│                                          │
└──────────────────────────────────────────┘

from redisgraph import Node, Edge, Graph

class SemanticMemory:
    def __init__(self, redis_client: redis.Redis):
        self.graph = Graph("agent:semantic", redis_client)
        self._init_schema()

    def _init_schema(self):
        """Create constraints for graph schema."""
        # Note: RedisGraph supports schema-less, but labeling is useful
        pass

    def store_fact(self, user_id: str, subject: str, relation: str, obj: str):
        """Store a fact in the knowledge graph."""
        query = f"""
        MERGE (s:Entity {{name: $subject, user_id: $user_id}})
        MERGE (o:Entity {{name: $obj, user_id: $user_id}})
        MERGE (s)-[r:{relation}]->(o)
        SET r.created_at = timestamp()
        RETURN s.name, type(r), o.name
        """
        params = {
            "subject": subject.replace(" ", "_"),
            "obj": obj.replace(" ", "_"),
            "user_id": user_id
        }
        self.graph.query(query, params)

    def get_user_facts(self, user_id: str) -> list:
        """Get all known facts about a user."""
        query = """
        MATCH (s:Entity {user_id: $user_id})-[r]->(o:Entity {user_id: $user_id})
        RETURN s.name, type(r) as relation, o.name
        """
        result = self.graph.query(query, {"user_id": user_id})
        return [
            {"subject": r[0], "relation": r[1], "object": r[2]}
            for r in result.result_set
        ]

    def query_fact(self, user_id: str, query_text: str) -> list:
        """Query knowledge graph by relationship pattern."""
        # Example: "who lives in Tokyo"
        query = """
        MATCH (s:Entity {user_id: $user_id})-[r]->(o:Entity)
        WHERE o.name CONTAINS $query
           OR s.name CONTAINS $query
        RETURN s.name, type(r), o.name
        """
        result = self.graph.query(query, {
            "user_id": user_id,
            "query": query_text
        })
        return [
            {"subject": r[0], "relation": r[1], "object": r[2]}
            for r in result.result_set
        ]

    def extract_and_store(self, user_id: str, conversation: str, llm_client):
        """Use LLM to extract facts from conversation and store them."""
        prompt = f"""
        Extract factual statements about the user from this conversation.
        Return as JSON array: [{{"subject": "...", "relation": "...", "object": "..."}}]
        Use relations like: lives_in, prefers, has_account, owns, works_at, interested_in

        Conversation: {conversation}
        """
        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )

        import json
        facts = json.loads(response.choices[0].message.content)

        for fact in facts.get("facts", facts if isinstance(facts, list) else []):
            self.store_fact(
                user_id,
                fact.get("subject", "User"),
                fact.get("relation", "mentioned"),
                fact.get("object", "")
            )

class AgentMemory:
    def __init__(self, redis_client: redis.Redis):
        self.short_term = ShortTermMemory(redis_client)
        self.episodic = EpisodicMemory(redis_client)
        self.semantic = SemanticMemory(redis_client)

    def before_llm_call(self, user_id: str, session_id: str, query: str) -> dict:
        """Build memory context before LLM call."""
        # 1. Get short-term (recent conversation)
        conversation_history = self.short_term.format_for_llm(session_id)

        # 2. Search episodic memory for relevant past sessions
        episodic_memories = self.episodic.search_memory(user_id, query)
        episodic_context = self.episodic.format_for_llm(episodic_memories)

        # 3. Get relevant semantic facts
        facts = self.semantic.get_user_facts(user_id)
        fact_context = "User facts:
" + "
".join(
            f"- {f['subject']} {f['relation']} {f['object']}"
            for f in facts
        )

        return {
            "short_term": conversation_history,
            "episodic": episodic_context,
            "semantic": fact_context
        }

    def after_llm_call(self, user_id: str, session_id: str,
                       query: str, response: str):
        """Store memories after LLM response."""
        # 1. Store in short term
        self.short_term.add_message(session_id, "user", query)
        self.short_term.add_message(session_id, "assistant", response)

    def end_session(self, user_id: str, session_id: str):
        """Archive session to episodic memory."""
        # 1. Get full conversation
        messages = self.short_term.get_recent_messages(session_id, count=100)
        conversation = "
".join(
            f"{m['role']}: {m['content']}" for m in messages
        )

        # 2. Store as episodic memory
        self.episodic.store_session(user_id, session_id, conversation)

        # 3. Extract semantic facts (optional, could use async job)
        # self.semantic.extract_and_store(user_id, conversation, llm_client)

        # 4. Clear short-term (optional — TTL handles this)
        self.short_term.clear_session(session_id)

System Prompt + Memory:
────────────────────────
You are an AI assistant with memory of the user.

[SHORT-TERM MEMORY — Recent conversation]
USER: Hi, I need help with Redis.
ASSISTANT: I can help! What specifically about Redis?
USER: I want to use it for vector search.

[EPISODIC MEMORY — Relevant past sessions]
2026-06-15 (similarity: 0.92):
User asked about vector databases and FAISS vs Pinecone.
Discussed HNSW index parameters.

[SEMANTIC MEMORY — Known facts]
- User lives_in Tokyo
- User prefers Python
- User works_at AI Startup

CURRENT USER QUERY: How do I set up Redis vector search?
────────────────────────

1. ALWAYS inject short-term memory (last N turns)
   - Maximum: 4K tokens or 20 messages
   - Priority: recency

2. Search episodic memory if query length > 10 chars
   - Retrieve top 3-5 most similar past sessions
   - Filter: same user, last 30 days
   - Priority: similarity score

3. Query semantic memory for known facts
   - Always retrieve user facts if user identified
   - Priority: by relation type (preferences > facts > history)

4. Merge and trim to context window
   - Short-term: 30% of budget
   - Episodic: 40% of budget
   - Semantic: 10% of budget
   - Remaining 20% for system prompt + current query

┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│ Agent Node 1 │  │ Agent Node 2 │  │ Agent Node 3 │
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       │                 │                 │
       └─────────────────┼─────────────────┘
                         │
              ┌──────────▼──────────┐
              │  Redis Cluster      │
              │  (Shared Memory)    │
              │                     │
              │  • Agent session    │
              │  • Episodic index   │
              │  • Semantic graph   │
              └──────────────────────┘

# Minimal production cluster (3 nodes)
redis-cli --cluster create \
  10.0.0.1:6379 10.0.0.2:6379 10.0.0.3:6379 \
  --cluster-replicas 1

# Memory limits per node
maxmemory 8gb
maxmemory-policy allkeys-lru  # Ephemeral memory eviction

□ Redis Stack 7.4+ deployed (with RediSearch + RedisJSON)
□ Vector index on SSD-backed persistent storage (AOF + RDB)
□ Memory limit per node (prevent OOM)
□ TTL on all temporal data (short-term: 1h, episodic: 30d)
□ Cluster mode for >50K concurrent users
□ Read replicas for episodic search (vector queries are read-heavy)
□ Backup schedule: RDB every 6h, AOF every 1s
□ Monitoring: latency P95 <20ms for vector search
□ Rate limiting: per-user memory writes (prevent abuse)

Redis wins for agent memory because agents need:
1. Multiple data types in ONE system (not just vectors)
2. Sub-5ms latency (agents can't wait 50ms+ for memory)
3. TTL-based memory expiration (critical for short-term)
4. Streams for conversation history
5. ACID transactions for critical memory updates
6. Battle-tested deployment (90%+ of infra already has Redis)