Memory System Integration

Memory systems enable agents to maintain context across conversations, remember user preferences, and learn from past interactions. This guide covers memory integration patterns from production agents.

Why Add Memory?

Agents with memory can:

Remember user preferences and context
Maintain conversation continuity across sessions
Learn from past interactions
Personalize responses based on history
Avoid repeating questions
Build long-term user relationships

Agno Memory v2 Pattern

Agno provides built-in memory capabilities with SQLite or Qdrant backends.

Basic Memory Setup

From memory_agents/agno_memory_agent/main.py:

from agno.agent import Agent
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory
from agno.models.nebius import Nebius
from agno.storage.sqlite import SqliteStorage
import os

# Database file for memory and storage
db_file = "tmp/agent.db"

# Initialize memory.v2
memory = Memory(
    model=Nebius(
        id="deepseek-ai/DeepSeek-V3-0324",
        api_key=os.getenv("NEBIUS_API_KEY")
    ),
    db=SqliteMemoryDb(
        table_name="user_memories",
        db_file=db_file
    ),
)

# Initialize storage for chat history
storage = SqliteStorage(
    table_name="agent_sessions",
    db_file=db_file
)

# Create agent with memory
memory_agent = Agent(
    model=Nebius(
        id="deepseek-ai/DeepSeek-V3-0324",
        api_key=os.getenv("NEBIUS_API_KEY")
    ),
    # Store memories in database
    memory=memory,
    # Agent can update memories autonomously
    enable_agentic_memory=True,
    # Run MemoryManager after each response
    enable_user_memories=True,
    # Store chat history in database
    storage=storage,
    # Add chat history to messages
    add_history_to_messages=True,
    # Number of history runs to include
    num_history_runs=3,
    markdown=True,
)

Using Memory in Conversations

user_id = "john_doe"

# First interaction - agent learns preferences
memory_agent.print_response(
    "My name is John and I support Arsenal.",
    user_id=user_id,
    stream=True,
)

# Check stored memories
from rich.pretty import pprint
print("Memories about John:")
pprint(memory.get_user_memories(user_id=user_id))

# Second interaction - agent remembers context
memory_agent.print_response(
    "I live in London, where should I move within a 4 hour drive?",
    user_id=user_id,
    stream=True,
)

# Later interaction - agent recalls user info
memory_agent.print_response(
    "Tell me about myself",
    user_id=user_id,
    stream=True,
)
# Agent will recall: "John lives in London and supports Arsenal"

Memory Management

# Get all memories for a user
user_memories = memory.get_user_memories(user_id="john_doe")

# Clear all memories for a user
memory.clear(user_id="john_doe")

# Clear all memories (all users)
memory.clear()

# Search memories
relevant_memories = memory.search(
    query="football teams",
    user_id="john_doe",
    limit=5
)

GibsonAI Memori Integration

Memori is a cloud-based memory service that provides semantic memory storage and retrieval.

Setup with Agno

from agno.agent import Agent
from agno.memory.memori import Memori
from agno.models.nebius import Nebius

# Initialize Memori
memory = Memori(
    user_id="user_123",
    api_key=os.getenv("MEMORI_API_KEY")
)

# Create agent with Memori
agent = Agent(
    model=Nebius(id="deepseek-ai/DeepSeek-V3-0324"),
    memory=memory,
    enable_agentic_memory=True,
    markdown=True,
)

# Use agent - memories automatically stored in Memori
response = agent.run(
    "I prefer technical documentation over tutorials",
    user_id="user_123"
)

Memori with AWS Strands

From course/aws_strands/ examples:

from strands import Agent
from strands.memory.memori import MemoriProvider

# Configure Memori
memory_provider = MemoriProvider(
    api_key=os.getenv("MEMORI_API_KEY"),
    agent_id="customer_support_agent",
    user_id="customer_456"
)

# Create agent with memory
agent = Agent(
    model=model,
    system_prompt="You are a helpful customer support agent",
    memory_provider=memory_provider
)

# Memory automatically retrieved and stored
response = agent("I'm having issues with my order")

Session-Based Memory

Maintain context within a single conversation session.

Using Session IDs

from agno.agent import Agent
from agno.storage.sqlite import SqliteStorage
import uuid

# Create storage
storage = SqliteStorage(
    table_name="agent_sessions",
    db_file="tmp/sessions.db"
)

# Create agent with storage
agent = Agent(
    model=model,
    storage=storage,
    add_history_to_messages=True,
    num_history_runs=10,  # Include last 10 interactions
)

# Start a new session
session_id = str(uuid.uuid4())

# Conversation within session
response1 = agent.run(
    "What are the best Python frameworks?",
    session_id=session_id
)

response2 = agent.run(
    "Tell me more about the first one you mentioned",
    session_id=session_id  # Agent has context from previous message
)

# New session - no shared context
new_session_id = str(uuid.uuid4())
response3 = agent.run(
    "What were we talking about?",
    session_id=new_session_id  # Agent won't remember previous session
)

Session Cleanup

# Clear specific session
storage.clear_session(session_id=session_id)

# Clear all sessions for a user
storage.clear_user_sessions(user_id="john_doe")

# Clear old sessions (older than 30 days)
from datetime import datetime, timedelta

cutoff_date = datetime.now() - timedelta(days=30)
storage.clear_sessions_before(cutoff_date)

Custom Memory Backend

Implement custom memory storage for specific requirements.

Memory Interface

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from pydantic import BaseModel

class MemoryEntry(BaseModel):
    """Single memory entry."""
    content: str
    timestamp: datetime
    metadata: Dict[str, Any] = {}
    embedding: List[float] | None = None

class MemoryBackend(ABC):
    """Abstract memory backend interface."""
    
    @abstractmethod
    def store(self, user_id: str, entry: MemoryEntry) -> str:
        """Store a memory entry."""
        pass
    
    @abstractmethod
    def retrieve(self, user_id: str, query: str, limit: int = 5) -> List[MemoryEntry]:
        """Retrieve relevant memories."""
        pass
    
    @abstractmethod
    def delete(self, user_id: str, entry_id: str) -> bool:
        """Delete a specific memory."""
        pass
    
    @abstractmethod
    def clear(self, user_id: str) -> int:
        """Clear all memories for a user."""
        pass

PostgreSQL Implementation

import psycopg2
from pgvector.psycopg2 import register_vector
import numpy as np
from openai import OpenAI

class PostgresMemory(MemoryBackend):
    """PostgreSQL + pgvector memory backend."""
    
    def __init__(self, connection_string: str, embedding_model: str = "text-embedding-3-small"):
        self.conn = psycopg2.connect(connection_string)
        register_vector(self.conn)
        self.embedding_client = OpenAI()
        self.embedding_model = embedding_model
        self._create_tables()
    
    def _create_tables(self):
        """Create memories table with vector support."""
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS memories (
                    id SERIAL PRIMARY KEY,
                    user_id VARCHAR(255) NOT NULL,
                    content TEXT NOT NULL,
                    timestamp TIMESTAMP NOT NULL,
                    metadata JSONB,
                    embedding vector(1536)
                );
                
                CREATE INDEX IF NOT EXISTS memories_user_id_idx ON memories(user_id);
                CREATE INDEX IF NOT EXISTS memories_embedding_idx ON memories 
                    USING ivfflat (embedding vector_cosine_ops);
            """)
            self.conn.commit()
    
    def _get_embedding(self, text: str) -> List[float]:
        """Generate embedding for text."""
        response = self.embedding_client.embeddings.create(
            model=self.embedding_model,
            input=text
        )
        return response.data[0].embedding
    
    def store(self, user_id: str, entry: MemoryEntry) -> str:
        """Store memory with vector embedding."""
        # Generate embedding if not provided
        if entry.embedding is None:
            entry.embedding = self._get_embedding(entry.content)
        
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO memories (user_id, content, timestamp, metadata, embedding)
                VALUES (%s, %s, %s, %s, %s)
                RETURNING id
            """, (
                user_id,
                entry.content,
                entry.timestamp,
                json.dumps(entry.metadata),
                entry.embedding
            ))
            memory_id = cur.fetchone()[0]
            self.conn.commit()
        
        return str(memory_id)
    
    def retrieve(self, user_id: str, query: str, limit: int = 5) -> List[MemoryEntry]:
        """Retrieve memories using vector similarity search."""
        query_embedding = self._get_embedding(query)
        
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT content, timestamp, metadata, embedding
                FROM memories
                WHERE user_id = %s
                ORDER BY embedding <=> %s
                LIMIT %s
            """, (user_id, query_embedding, limit))
            
            results = []
            for row in cur.fetchall():
                results.append(MemoryEntry(
                    content=row[0],
                    timestamp=row[1],
                    metadata=json.loads(row[2]) if row[2] else {},
                    embedding=row[3]
                ))
            
            return results
    
    def delete(self, user_id: str, entry_id: str) -> bool:
        """Delete specific memory."""
        with self.conn.cursor() as cur:
            cur.execute(
                "DELETE FROM memories WHERE id = %s AND user_id = %s",
                (int(entry_id), user_id)
            )
            self.conn.commit()
            return cur.rowcount > 0
    
    def clear(self, user_id: str) -> int:
        """Clear all memories for user."""
        with self.conn.cursor() as cur:
            cur.execute("DELETE FROM memories WHERE user_id = %s", (user_id,))
            count = cur.rowcount
            self.conn.commit()
            return count

Memory-Enhanced Workflows

Combine memory with multi-agent workflows.

Workflow with Shared Memory

from agno.workflow import Workflow
from agno.agent import Agent
from agno.memory.v2.memory import Memory

class ConsultantWorkflow(Workflow):
    """Multi-agent workflow with shared memory."""
    
    def __init__(self, user_id: str, memory: Memory):
        self.user_id = user_id
        self.shared_memory = memory
        
        # All agents share the same memory
        self.researcher = Agent(
            name="Researcher",
            model=model,
            memory=memory,
            enable_agentic_memory=True,
        )
        
        self.analyst = Agent(
            name="Analyst",
            model=model,
            memory=memory,  # Same memory instance
            enable_agentic_memory=True,
        )
        
        self.advisor = Agent(
            name="Advisor",
            model=model,
            memory=memory,  # Same memory instance
            enable_agentic_memory=True,
        )
    
    def run(self, query: str) -> str:
        # Step 1: Research (stores findings in memory)
        research = self.researcher.run(
            query,
            user_id=self.user_id
        )
        
        # Step 2: Analysis (has access to research memories)
        analysis = self.analyst.run(
            f"Analyze this: {research.content}",
            user_id=self.user_id
        )
        
        # Step 3: Advice (has access to all previous memories)
        advice = self.advisor.run(
            f"Provide recommendations based on analysis: {analysis.content}",
            user_id=self.user_id
        )
        
        return advice.content

Best Practices

1. User-Scoped Memories

# ✅ Good: Always scope memories to specific users
memory.store(
    user_id="user_123",
    content="Prefers technical documentation"
)

# ❌ Bad: Global memories (privacy/security issue)
memory.store(content="User preferences")  # No user scope

2. Memory Expiration

# ✅ Good: Set expiration for temporary memories
class MemoryEntry(BaseModel):
    content: str
    timestamp: datetime
    expires_at: datetime | None = None
    
def clean_expired_memories(memory: MemoryBackend, user_id: str):
    """Remove expired memories."""
    memories = memory.get_all(user_id)
    now = datetime.now()
    
    for mem in memories:
        if mem.expires_at and now > mem.expires_at:
            memory.delete(user_id, mem.id)

# ❌ Bad: Memories accumulate forever
# No cleanup mechanism

3. Semantic vs. Factual Memory

# ✅ Good: Separate semantic and factual memories
class SemanticMemory:
    """Context and preferences (vector search)."""
    def retrieve(self, query: str) -> List[str]:
        return vector_search(query, limit=5)

class FactualMemory:
    """Structured facts (exact lookup)."""
    def get_fact(self, key: str) -> Any:
        return database.get(key)

agent = Agent(
    semantic_memory=SemanticMemory(),
    factual_memory=FactualMemory()
)

# ❌ Bad: Mix all memory types together
# Hard to query effectively

4. Privacy-Aware Storage

# ✅ Good: Encrypt sensitive memories
from cryptography.fernet import Fernet

class EncryptedMemory(MemoryBackend):
    def __init__(self, encryption_key: bytes):
        self.cipher = Fernet(encryption_key)
        self.backend = PostgresMemory(...)
    
    def store(self, user_id: str, entry: MemoryEntry):
        # Encrypt content before storing
        encrypted_content = self.cipher.encrypt(
            entry.content.encode()
        ).decode()
        
        entry.content = encrypted_content
        return self.backend.store(user_id, entry)
    
    def retrieve(self, user_id: str, query: str, limit: int = 5):
        memories = self.backend.retrieve(user_id, query, limit)
        
        # Decrypt content
        for mem in memories:
            mem.content = self.cipher.decrypt(
                mem.content.encode()
            ).decode()
        
        return memories

# ❌ Bad: Store sensitive data in plaintext

Real-World Examples

Study Coach Agent

Location: memory_agents/study_coach_agent/ Remembers:

Learning style preferences
Topics studied
Knowledge gaps
Progress over time

Customer Support Agent

Location: memory_agents/customer_support_voice_agent/ Remembers:

Past support tickets
Customer preferences
Product history
Previous resolutions

Job Search Agent

Location: memory_agents/job_search_agent/ Remembers:

Resume details
Job preferences
Application history
Interview feedback

Memory Configuration Comparison

Backend	Best For	Pros	Cons
SQLite	Development, single-user	Simple setup, no dependencies	Not scalable, single-file
PostgreSQL + pgvector	Production, multi-user	Scalable, vector search, ACID	Requires setup, more complex
Memori (Cloud)	Quick start, managed	Hosted, semantic search, easy	Requires API key, cost
Qdrant	Vector-heavy workloads	Fast vector search, scalable	Additional service to run
Redis	Session memory, caching	Fast, TTL support	Volatile (unless persisted)

Next Steps

RAG Workflows

Combine memory with retrieval-augmented generation

Multi-Agent Patterns

Share memory across multiple agents

Best Practices

Production patterns for memory management

Environment Setup

Set up local and cloud memory backends

Frameworks

Patterns

Development

Why Add Memory?

Agno Memory v2 Pattern

Basic Memory Setup

Using Memory in Conversations

Memory Management

GibsonAI Memori Integration

Setup with Agno

Memori with AWS Strands

Session-Based Memory

Using Session IDs

Session Cleanup

Custom Memory Backend

Memory Interface

PostgreSQL Implementation

Memory-Enhanced Workflows

Workflow with Shared Memory

Best Practices

1. User-Scoped Memories

2. Memory Expiration

3. Semantic vs. Factual Memory

4. Privacy-Aware Storage

Real-World Examples

Study Coach Agent

Customer Support Agent

Job Search Agent

Memory Configuration Comparison

Next Steps

RAG Workflows

Multi-Agent Patterns

Best Practices

Environment Setup

Build docs developers (and LLMs) love

Frameworks

Patterns

Development

​Why Add Memory?

​Agno Memory v2 Pattern

​Basic Memory Setup

​Using Memory in Conversations

​Memory Management

​GibsonAI Memori Integration

​Setup with Agno

​Memori with AWS Strands

​Session-Based Memory

​Using Session IDs

​Session Cleanup

​Custom Memory Backend

​Memory Interface

​PostgreSQL Implementation

​Memory-Enhanced Workflows

​Workflow with Shared Memory

​Best Practices

​1. User-Scoped Memories

​2. Memory Expiration

​3. Semantic vs. Factual Memory

​4. Privacy-Aware Storage

​Real-World Examples

​Study Coach Agent

​Customer Support Agent

​Job Search Agent

​Memory Configuration Comparison

​Next Steps

RAG Workflows

Multi-Agent Patterns

Best Practices

Environment Setup

Build docs developers (and LLMs) love

Why Add Memory?

Agno Memory v2 Pattern

Basic Memory Setup

Using Memory in Conversations

Memory Management

GibsonAI Memori Integration

Setup with Agno

Memori with AWS Strands

Session-Based Memory

Using Session IDs

Session Cleanup

Custom Memory Backend

Memory Interface

PostgreSQL Implementation

Memory-Enhanced Workflows

Workflow with Shared Memory

Best Practices

1. User-Scoped Memories

2. Memory Expiration

3. Semantic vs. Factual Memory

4. Privacy-Aware Storage

Real-World Examples

Study Coach Agent

Customer Support Agent

Job Search Agent

Memory Configuration Comparison

Next Steps