Build an AI Agent with Persistent Memory in Python (Full Guide)

An AI agent with persistent memory remembers users between sessions: their preferences, past issues, ongoing tasks, and conversational history. Without persistent memory, every conversation starts from zero. This guide builds a complete stateful AI agent using Feather DB for memory storage and retrieval and OpenAI for responses — covering installation, memory ingestion, context retrieval, session management, and production patterns. All code is working and production-ready.

Prerequisites

pip install feather-db openai

# Set your OpenAI API key
export OPENAI_API_KEY="your-key-here"

Architecture Overview

The agent has three components:

1. Memory store: Feather DB stores all memories as vector-embedded facts. It handles semantic search, temporal decay, and graph traversal.
2. Context builder: Before each LLM call, retrieves the most relevant memories and formats them into a context block.
3. LLM interface: Sends the context block + current message to GPT-4o and receives the response.

Between sessions, memories persist in the .feather file. When the session starts, the agent loads the user's profile from memory and is immediately personalized.

Step 1: Memory Manager

import feather_db as fdb
from datetime import datetime
from typing import Optional

class MemoryManager:
    """Handles all memory operations for the AI agent."""

    def __init__(self, storage_path: str = "agent_memories.feather"):
        self.db = fdb.FeatherDB(storage_path)

    def store(self, text: str, user_id: str, memory_type: str = "episodic",
              importance: float = 0.6, session_id: Optional[str] = None) -> str:
        """Store a memory and return its ID."""
        half_life_map = {
            "episodic": 30,     # Session events fade in ~1 month
            "semantic": 180,    # User facts persist ~6 months
            "procedural": 365,  # How-to knowledge persists ~1 year
        }
        node_id = self.db.add(
            text=text,
            metadata={
                "user_id": user_id,
                "memory_type": memory_type,
                "session_id": session_id or "unknown",
                "stored_at": datetime.now().isoformat()
            },
            importance=importance,
            half_life_days=half_life_map.get(memory_type, 60)
        )
        return node_id

    def link(self, source_id: str, target_id: str,
             edge_type: str = "related", weight: float = 0.8):
        """Create a typed edge between two memories."""
        self.db.link_nodes(source_id, target_id, edge_type, weight)

    def recall(self, query: str, user_id: str, top_k: int = 8,
               use_graph: bool = True) -> list:
        """Retrieve the most relevant memories for a query."""
        if use_graph:
            return self.db.context_chain(
                query=query,
                hops=2,
                top_k=top_k,
                filters={"user_id": user_id}
            )
        return self.db.search(
            query=query,
            top_k=top_k,
            filters={"user_id": user_id}
        )

    def get_profile(self, user_id: str) -> list:
        """Get user's semantic profile — called once at session start."""
        return self.db.search(
            query="user background preferences goals tech stack",
            top_k=10,
            filters={"user_id": user_id, "memory_type": "semantic"}
        )

    def format_context(self, memories: list) -> str:
        """Format retrieved memories as a context block."""
        if not memories:
            return "No relevant context found."
        return "\n".join([f"- {m['text']}" for m in memories])

    def count(self, user_id: str) -> int:
        """Count total memories for a user."""
        return self.db.count(filters={"user_id": user_id})

Step 2: Memory Extractor

Before storing a memory, extract the key fact from the conversation turn. This keeps memories atomic and searchable:

from openai import OpenAI
import json

client = OpenAI()

def extract_memories(conversation_turn: str) -> list[dict]:
    """
    Extract memorable facts from a conversation turn.
    Returns list of dicts with 'text', 'memory_type', 'importance'.
    """
    extraction_prompt = """Extract memorable facts from this conversation excerpt.
Return a JSON array. Each item has:
- text: the fact as a declarative statement
- memory_type: "semantic" (general user fact/preference) or "episodic" (specific event)
- importance: 0.0-1.0 (how important this fact is to remember)

Only extract genuinely memorable facts. Skip pleasantries and filler.
Conversation: """ + conversation_turn

    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Cheap model for extraction
        messages=[{"role": "user", "content": extraction_prompt}],
        response_format={"type": "json_object"}
    )

    result = json.loads(response.choices[0].message.content)
    return result.get("memories", [])


# Example output:
# [
#   {"text": "User builds AI agents for enterprise clients",
#    "memory_type": "semantic", "importance": 0.8},
#   {"text": "User reported latency issue with their agent in session 42",
#    "memory_type": "episodic", "importance": 0.6}
# ]

Step 3: The Stateful Agent

from openai import OpenAI
import feather_db as fdb
from datetime import datetime
from typing import Optional

client = OpenAI()

class StatefulAgent:
    """
    An AI agent that remembers users across sessions.
    Uses Feather DB for persistent memory storage and retrieval.
    """

    def __init__(self, user_id: str, memory_path: str = "agent_memories.feather"):
        self.user_id = user_id
        self.session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        self.memory = MemoryManager(memory_path)
        self.conversation_history = []  # In-session working memory

        # Load user profile at session start
        self.user_profile = self.memory.get_profile(user_id)
        self.session_context = self.memory.format_context(self.user_profile)

        print(f"Session {self.session_id} started for user {user_id}")
        print(f"Loaded {len(self.user_profile)} profile memories")
        print(f"Total memories: {self.memory.count(user_id)}")

    def chat(self, user_message: str) -> str:
        """Process a user message and return the agent's response."""

        # 1. Retrieve relevant memories for this specific query
        turn_memories = self.memory.recall(
            query=user_message,
            user_id=self.user_id,
            top_k=8,
            use_graph=True
        )
        turn_context = self.memory.format_context(turn_memories)

        # 2. Build the system prompt with both profile and turn context
        system_prompt = f"""You are a helpful AI assistant with persistent memory of past interactions.

User profile:
{self.session_context}

Relevant memories for this query:
{turn_context}

Use this context to give personalized responses. If you're referencing a past event, mention it naturally."""

        # 3. Maintain a short in-session conversation buffer (last 6 turns)
        self.conversation_history.append({"role": "user", "content": user_message})
        messages = [
            {"role": "system", "content": system_prompt}
        ] + self.conversation_history[-6:]  # Last 6 turns only

        # 4. LLM call
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )

        assistant_reply = response.choices[0].message.content
        self.conversation_history.append(
            {"role": "assistant", "content": assistant_reply}
        )

        return assistant_reply

    def learn_from_turn(self, user_message: str, assistant_reply: str):
        """Extract and store memories from a conversation turn."""
        turn_text = f"User: {user_message}\nAssistant: {assistant_reply}"
        extracted = extract_memories(turn_text)

        stored_ids = []
        for memory in extracted:
            memory_id = self.memory.store(
                text=memory["text"],
                user_id=self.user_id,
                memory_type=memory["memory_type"],
                importance=memory["importance"],
                session_id=self.session_id
            )
            stored_ids.append(memory_id)

        # Link related memories within this turn
        for i in range(len(stored_ids) - 1):
            self.memory.link(
                stored_ids[i],
                stored_ids[i + 1],
                edge_type="same_session",
                weight=0.6
            )

        return len(stored_ids)

    def chat_and_learn(self, user_message: str) -> str:
        """Chat with the user and automatically store memories from the turn."""
        reply = self.chat(user_message)
        memories_stored = self.learn_from_turn(user_message, reply)
        print(f"  [Memory: {memories_stored} facts stored]")
        return reply

Step 4: Running the Agent

def main():
    print("AI Agent with Persistent Memory")
    print("=" * 40)

    user_id = input("Enter your user ID: ").strip() or "user_001"

    # Initialize agent — loads memory from previous sessions automatically
    agent = StatefulAgent(user_id=user_id)

    print("\nChat with the agent. Type 'quit' to exit.")
    print("Your memories persist between sessions.\n")

    while True:
        user_input = input(f"You: ").strip()

        if user_input.lower() in ["quit", "exit", "q"]:
            print(f"Session ended. Total memories: {agent.memory.count(user_id)}")
            break

        if not user_input:
            continue

        reply = agent.chat_and_learn(user_input)
        print(f"Agent: {reply}\n")

if __name__ == "__main__":
    main()

Step 5: Testing Persistence Across Sessions

Run the agent twice to verify memory persists:

# Session 1:
# You: My name is Priya and I build fintech apps in Python
# Agent: Nice to meet you, Priya! What kind of fintech applications are you working on?
# [Memory: 2 facts stored]
# You: I'm building a payment reconciliation system using FastAPI
# Agent: That sounds interesting — payment reconciliation has some tricky edge cases...
# [Memory: 1 facts stored]
# You: quit
# Session ended. Total memories: 3

# Session 2 (new Python process, same user_id):
# Loaded 3 profile memories
# You: Can you help me with my project?
# Agent: Of course, Priya! Based on what you told me, you're building a payment
#   reconciliation system in FastAPI. What specifically do you need help with?
# (Agent remembers from session 1 without being told again)

Production Patterns

Memory deduplication

def store_if_novel(memory: MemoryManager, text: str, user_id: str,
                   similarity_threshold: float = 0.92, **kwargs) -> Optional[str]:
    """Only store a memory if it's not already represented."""
    similar = memory.recall(text, user_id, top_k=1)
    if similar and similar[0]['score'] > similarity_threshold:
        # Update importance of existing memory instead of duplicating
        memory.db.update_importance(
            similar[0]['id'],
            similar[0]['importance'] + 0.05  # Bump importance on re-mention
        )
        return None  # Not stored (duplicate)
    return memory.store(text, user_id, **kwargs)

Multi-user deployment

# Option A: One file per user (recommended for clear isolation)
agent_alice = StatefulAgent(user_id="alice", memory_path="memories_alice.feather")
agent_bob = StatefulAgent(user_id="bob", memory_path="memories_bob.feather")

# Option B: Shared file with namespace isolation
# (All queries filter by user_id — never cross boundaries)
agent_alice = StatefulAgent(user_id="alice", memory_path="shared_memories.feather")
agent_bob = StatefulAgent(user_id="bob", memory_path="shared_memories.feather")

Async support for high-concurrency APIs

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    user_id: str
    message: str

agents: dict[str, StatefulAgent] = {}  # In production, use a proper cache

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    if request.user_id not in agents:
        agents[request.user_id] = StatefulAgent(request.user_id)

    agent = agents[request.user_id]
    reply = agent.chat_and_learn(request.message)

    return {
        "reply": reply,
        "total_memories": agent.memory.count(request.user_id)
    }

Cost at Scale

For a production deployment handling 1,000 sessions/day:

• Feather DB retrieval: $0 (in-process, no API cost)
• Extraction LLM (GPT-4o-mini, ~200 tokens/turn): ~$0.0003/turn
• Response LLM (GPT-4o, ~4,300 tokens/session): ~$0.022/session
• Total: ~$0.022/session vs $0.285/session with full-context stuffing
• Monthly savings at 1,000 sessions/day: ~$7,900/month

FAQ

How do I handle memory for a user who interacts with multiple agents?

Use a single shared .feather file for the user with agent-scoped metadata. All agents query the same memory store but can filter by agent_id for agent-specific memories, or omit the filter to access the full user context.

What happens if the agent stores a false memory?

False memories (hallucinated facts stored as memories) are the main failure mode. Mitigate by: (1) only extracting facts directly stated by the user, not inferred by the assistant; (2) setting lower importance scores for extracted facts (0.5 vs 0.9 for user-explicit statements); (3) implementing a contradiction check that marks conflicting memories with a contradicts edge.

How do I inspect what memories the agent has for a user?

Use db.search("*", filters={"user_id": user_id}, top_k=50) to retrieve all memories sorted by relevance, or db.get_all(filters={"user_id": user_id}) to retrieve every memory unsorted.

Does this work with Claude / Anthropic models?

Yes. Replace the OpenAI client with the Anthropic SDK and update the API calls. Feather DB is model-agnostic — it only affects the context block injected into the prompt, not the model itself.

How do I prevent sensitive memories from being stored?

Add a pre-storage filter using regex or a classifier to detect PII, financial data, or health information before calling memory.store(). Feather DB does not automatically redact sensitive content — this is an application-layer responsibility.