Skip to main content

Building a Memory-Augmented Agent

You've learned the patterns. Now let's build a complete agent with persistent memory—one that remembers users across sessions and uses that memory to provide personalized assistance.

In this lesson, you'll build a Task Manager Agent using OpenAI's Agents SDK with Mem0 for memory. By the end, you'll have a working agent that remembers user preferences, tracks task history, and provides context-aware suggestions.

Project Setup

Project Structure

memory-agent/
├── agent.py # Agent definition with memory tools
├── memory_tools.py # Mem0-powered memory functions
├── main.py # Runner and conversation loop
├── requirements.txt
└── test_memory.py # Multi-session memory tests

Dependencies

# requirements.txt
openai-agents>=0.1.0
mem0ai>=0.1.0
python-dotenv>=1.0.0

Installation

pip install openai-agents mem0ai python-dotenv

Environment Setup

# .env
OPENAI_API_KEY=your-openai-api-key

Building Memory Tools

First, let's create the memory tools that our agent will use.

Create memory_tools.py:

from mem0 import Memory
from agents import function_tool
from typing import Optional

# Initialize Mem0 - uses OpenAI for embeddings by default
memory = Memory()

@function_tool
def recall_memories(query: str, user_id: str, category: Optional[str] = None) -> str:
"""
Search your memory for relevant past interactions and user information.

Args:
query: What to search for (e.g., "user preferences", "past tasks")
user_id: The user whose memories to search
category: Optional filter - "preferences", "tasks", "projects", or "general"

Returns:
Formatted string of relevant memories with dates
"""
filters = {"user_id": user_id}
if category:
filters["category"] = category

results = memory.search(
query=query,
user_id=user_id,
filters=filters if category else None,
limit=5
)

memories = results.get('results', [])
if not memories:
return "No relevant memories found."

formatted = []
for mem in memories:
score = mem.get('score', 0)
text = mem.get('memory', '')
created = mem.get('created_at', 'unknown date')[:10]
formatted.append(f"• [{created}] {text} (relevance: {score:.2f})")

return "Relevant memories:\n" + "\n".join(formatted)


@function_tool
def store_memory(
content: str,
user_id: str,
category: str = "general"
) -> str:
"""
Store important information about the user for future reference.

Args:
content: The information to remember (user preference, task outcome, etc.)
user_id: The user this memory belongs to
category: One of "preferences", "tasks", "projects", or "general"

Returns:
Confirmation of what was stored
"""
result = memory.add(
[{"role": "system", "content": content}],
user_id=user_id,
metadata={"category": category}
)

stored = result.get('results', [])
if stored:
return f"Stored: {stored[0].get('memory', content)}"
return f"Stored: {content}"


@function_tool
def list_user_memories(user_id: str, limit: int = 10) -> str:
"""
List all memories for a user, organized by category.

Args:
user_id: The user whose memories to list
limit: Maximum number of memories to return

Returns:
Categorized list of user memories
"""
# Search with empty query returns recent memories
results = memory.search(
query="",
user_id=user_id,
limit=limit
)

memories = results.get('results', [])
if not memories:
return "No memories stored for this user."

# Group by category
by_category = {}
for mem in memories:
cat = mem.get('metadata', {}).get('category', 'general')
if cat not in by_category:
by_category[cat] = []
by_category[cat].append(mem.get('memory', ''))

formatted = []
for cat, mems in by_category.items():
formatted.append(f"\n{cat.upper()}:")
for m in mems[:3]: # Show top 3 per category
formatted.append(f" • {m}")

return "User memories:" + "".join(formatted)


@function_tool
def forget_memory(memory_text: str, user_id: str) -> str:
"""
Delete a specific memory (for privacy or corrections).

Args:
memory_text: Text to search for and delete
user_id: The user whose memory to delete

Returns:
Confirmation of deletion
"""
# Find matching memory
results = memory.search(
query=memory_text,
user_id=user_id,
limit=1
)

memories = results.get('results', [])
if not memories:
return "No matching memory found to delete."

memory_id = memories[0].get('id')
memory.delete(memory_id=memory_id)

return f"Deleted memory: {memories[0].get('memory', memory_text)}"

Output when testing recall_memories:

Relevant memories:
• [2025-01-20] Prefers tasks scheduled in morning (relevance: 0.89)
• [2025-01-18] Currently working on Phoenix project (relevance: 0.76)
• [2025-01-15] Uses pytest for testing (relevance: 0.65)

Creating the Agent

Now let's create the Task Manager Agent with memory capabilities.

Create agent.py:

from agents import Agent
from memory_tools import (
recall_memories,
store_memory,
list_user_memories,
forget_memory
)

TASK_MANAGER_INSTRUCTIONS = """You are a helpful Task Manager Agent with persistent memory.

## Your Capabilities
- Create, track, and manage tasks for users
- Remember user preferences and apply them automatically
- Recall past tasks and project context
- Learn from user feedback over time

## Memory Guidelines
1. At conversation START: Always recall relevant memories about the user
2. During conversation: Store important new information (preferences, decisions, outcomes)
3. Use memories NATURALLY - reference them as context, don't just list them
4. Ask before storing sensitive information

## Task Management
When users mention tasks:
- Check for relevant project context in memory
- Apply known preferences (timing, priority patterns)
- Suggest based on past patterns

## Example Memory Usage
User: "Schedule a code review"
You: [Recall memories about code reviews and preferences]
Response: "I'll schedule that for morning since you prefer focused work then.
Should I add it to the Phoenix project you've been working on?"

## Privacy
- Never store passwords, API keys, or sensitive credentials
- If user says "don't remember this" - use forget_memory
- Respect when users want things kept private

Remember: You're building a relationship over time. Each conversation adds to shared context.
"""

def create_task_manager_agent() -> Agent:
"""Create a Task Manager Agent with memory tools."""
return Agent(
name="TaskManager",
instructions=TASK_MANAGER_INSTRUCTIONS,
tools=[
recall_memories,
store_memory,
list_user_memories,
forget_memory
],
model="gpt-4o-mini" # Cost-effective for task management
)

Running the Agent

Create the main conversation loop.

Create main.py:

import asyncio
from dotenv import load_dotenv
from agents import Runner
from agent import create_task_manager_agent

load_dotenv()

async def chat_session(user_id: str):
"""Run an interactive chat session with the Task Manager."""
agent = create_task_manager_agent()
runner = Runner()

print(f"\n{'='*60}")
print(f"Task Manager Agent - Session for {user_id}")
print(f"{'='*60}")
print("Type 'quit' to exit, 'memories' to see stored memories\n")

# Initial context retrieval
initial_context = f"New session started for user {user_id}. " \
f"Recall any relevant memories about this user."

result = await runner.run(agent, initial_context)
print(f"Agent: {result.final_output}\n")

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

if user_input.lower() == 'quit':
# Store session end
await runner.run(
agent,
f"User {user_id} is ending the session. "
f"Store any important information from this conversation."
)
print("Session ended. Memories saved.")
break

if user_input.lower() == 'memories':
result = await runner.run(
agent,
f"List all memories for user {user_id}"
)
print(f"Agent: {result.final_output}\n")
continue

# Include user_id in context for memory operations
context = f"User {user_id} says: {user_input}"
result = await runner.run(agent, context)
print(f"Agent: {result.final_output}\n")


async def main():
user_id = input("Enter your user ID: ").strip() or "demo_user"
await chat_session(user_id)


if __name__ == "__main__":
asyncio.run(main())

Example Conversation

Session 1: First Interaction

Task Manager Agent - Session for alex
============================================================
Type 'quit' to exit, 'memories' to see stored memories

Agent: Hello! I don't have any previous memories of you yet.
I'm your Task Manager Agent - I can help you manage tasks and
I'll remember your preferences for future sessions. What would
you like to work on?

You: I'm working on the Phoenix project. I prefer doing code
reviews in the morning when I'm fresh.

Agent: Got it! I've noted that you're working on the Phoenix
project and prefer morning code reviews. I'll remember these
for our future conversations.

[Stored: Working on Phoenix project]
[Stored: Prefers code reviews in morning when fresh]

You: Create a task to review the authentication PR

Agent: I'll create that code review task for the Phoenix project.
Since you prefer mornings for reviews, I'm scheduling it for
tomorrow morning.

Task created:
- Title: Review authentication PR
- Project: Phoenix
- Scheduled: Tomorrow morning
- Priority: Normal

Would you like me to remember anything specific about this PR?

You: quit
Session ended. Memories saved.

Session 2: Memory in Action (after restart)

Task Manager Agent - Session for alex
============================================================
Type 'quit' to exit, 'memories' to see stored memories

Agent: Welcome back, Alex! I remember you're working on the
Phoenix project and prefer code reviews in the morning. Last
time you created a task to review the authentication PR.
How can I help today?

You: How did the auth review go?

Agent: I don't have a record of the outcome yet. How did it go?
I can store the result for future reference.

You: It went well. Found two small issues that were easy fixes.

Agent: Great to hear! I'll remember that the auth PR review was
successful with minor issues found. This helps me understand
your project's progress.

[Stored: Auth PR review completed - found 2 small issues, easy fixes]

You: memories
Agent: Here are your stored memories:

PROJECTS:
• Working on Phoenix project
• Auth PR review completed - found 2 small issues, easy fixes

PREFERENCES:
• Prefers code reviews in morning when fresh

TASKS:
• Created task to review authentication PR

Testing Memory Persistence

Verify that memory works across completely independent sessions.

Create test_memory.py:

import asyncio
from dotenv import load_dotenv
from agents import Runner
from agent import create_task_manager_agent
from mem0 import Memory

load_dotenv()

async def test_memory_persistence():
"""Test that memories persist across agent sessions."""
user_id = "test_user_persistence"

print("=" * 60)
print("Testing Memory Persistence")
print("=" * 60)

# Clean start
memory = Memory()

# === Session 1: Store information ===
print("\n[Session 1] Storing information...")
agent1 = create_task_manager_agent()
runner1 = Runner()

result = await runner1.run(
agent1,
f"User {user_id} says: I'm a Python developer who loves FastAPI. "
f"Remember this preference."
)
print(f"Agent: {result.final_output}")

# Verify storage
stored = memory.search("Python FastAPI", user_id=user_id)
assert len(stored.get('results', [])) > 0, "Memory not stored!"
print("✓ Memory stored successfully")

# === Session 2: Retrieve in NEW agent instance ===
print("\n[Session 2] New agent instance - retrieving...")

# Create completely new agent and runner
agent2 = create_task_manager_agent()
runner2 = Runner()

result = await runner2.run(
agent2,
f"User {user_id} is starting a new session. "
f"What do you remember about them?"
)
print(f"Agent: {result.final_output}")

# Check that response includes stored preference
assert "Python" in result.final_output or "FastAPI" in result.final_output, \
"Agent didn't recall stored memory!"
print("✓ Memory retrieved in new session")

# === Cleanup ===
print("\n[Cleanup] Removing test memories...")
all_mems = memory.search("", user_id=user_id, limit=100)
for mem in all_mems.get('results', []):
memory.delete(memory_id=mem['id'])
print("✓ Test memories cleaned up")

print("\n" + "=" * 60)
print("All memory persistence tests passed!")
print("=" * 60)


async def test_memory_categories():
"""Test that memory categories work correctly."""
user_id = "test_user_categories"
memory = Memory()

print("\n[Testing Categories]")

agent = create_task_manager_agent()
runner = Runner()

# Store different categories
await runner.run(
agent,
f"User {user_id} says: Store these - "
f"Preference: I like TypeScript. "
f"Project: Working on EcommerceApp. "
f"Task: Need to fix the payment bug."
)

# Query specific category
result = await runner.run(
agent,
f"What projects is user {user_id} working on? "
f"Only check project memories."
)
print(f"Projects: {result.final_output}")

# Cleanup
all_mems = memory.search("", user_id=user_id, limit=100)
for mem in all_mems.get('results', []):
memory.delete(memory_id=mem['id'])

print("✓ Category filtering works")


if __name__ == "__main__":
asyncio.run(test_memory_persistence())
asyncio.run(test_memory_categories())

Output:

============================================================
Testing Memory Persistence
============================================================

[Session 1] Storing information...
Agent: I've noted that you're a Python developer who loves FastAPI.
I'll remember this for our future conversations!
✓ Memory stored successfully

[Session 2] New agent instance - retrieving...
Agent: Welcome back! I remember you're a Python developer who loves
FastAPI. How can I help with your development work today?
✓ Memory retrieved in new session

[Cleanup] Removing test memories...
✓ Test memories cleaned up

============================================================
All memory persistence tests passed!
============================================================

Architecture Summary

┌─────────────────────────────────────────────────────────────┐
│ MEMORY-AUGMENTED AGENT │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ OpenAI Agents SDK │ │
│ │ │ │
│ │ Agent │ │
│ │ ├── Instructions (memory-aware prompts) │ │
│ │ └── Tools │ │
│ │ ├── recall_memories (Mem0 search) │ │
│ │ ├── store_memory (Mem0 add) │ │
│ │ ├── list_user_memories (Mem0 list) │ │
│ │ └── forget_memory (Mem0 delete) │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Mem0 SDK │ │
│ │ │ │
│ │ Memory() │ │
│ │ ├── add() → Extract facts, embed, store │ │
│ │ ├── search() → Semantic retrieval │ │
│ │ ├── delete() → Remove memories │ │
│ │ └── update() → Modify memories │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Storage Layer │ │
│ │ │ │
│ │ Embeddings (OpenAI) → Vector Store (Qdrant) │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘

Key Patterns Demonstrated

PatternImplementation
Memory as Tools@function_tool decorators expose Mem0 to agent
User ScopingEvery operation includes user_id parameter
Category FilteringMetadata enables targeted retrieval
Session PersistenceMem0's disk storage survives restarts
Natural IntegrationInstructions guide memory usage in conversation
Privacy Controlsforget_memory tool for user control

Try With AI

Use these prompts to extend your memory-augmented agent.

Prompt 1: Add Memory Summarization

I have a memory-augmented agent using OpenAI Agents SDK and Mem0.
When a user has many memories (>20), retrieval becomes noisy.

Design a memory summarization strategy:
1. When to trigger summarization (memory count threshold)
2. How to group related memories for summarization
3. How to preserve important details while reducing volume
4. How to handle the summarize → delete old → store summary flow

Show the implementation as a new tool: summarize_memories(user_id, category)

What you're learning: Memory systems need maintenance. Summarization prevents context overload while preserving valuable information—a key pattern for long-running agents.

Prompt 2: Entity-Based Memory

My task manager agent stores unstructured memories. I want to add
entity extraction so memories are linked to specific entities
(projects, people, deadlines).

Design an entity memory system:
1. Extract entities from conversation (Project: Phoenix, Person: Alex)
2. Link memories to entities in metadata
3. Enable queries like "What do I know about the Phoenix project?"
4. Handle entity relationships (Alex works on Phoenix)

Show how to modify store_memory and add a new query_entity tool.

What you're learning: Entity linking transforms flat memories into a knowledge graph. This enables more precise retrieval and relationship-based reasoning.

Prompt 3: Memory Confidence Scoring

My agent sometimes retrieves irrelevant memories that confuse responses.
I want to add confidence scoring that considers:
- Semantic similarity (from Mem0)
- Recency (newer = more relevant for tasks)
- Frequency (oft-referenced memories = important)
- User feedback (memories the user corrected = lower confidence)

Design a confidence scoring system:
1. How to track these signals in metadata
2. How to compute combined confidence score
3. How to filter low-confidence memories from retrieval
4. How to let users boost/demote memory confidence

Show the updated recall_memories implementation with confidence filtering.

What you're learning: Raw semantic similarity isn't enough for production memory systems. Multi-factor confidence scoring improves retrieval precision and user trust.

What You've Built

In this lesson, you created a complete memory-augmented agent:

ComponentPurpose
memory_tools.pyMem0-powered tools for recall, store, list, forget
agent.pyTask Manager Agent with memory-aware instructions
main.pyInteractive conversation loop
test_memory.pyMulti-session persistence verification

This agent demonstrates the full memory lifecycle:

  1. Retrieval at session start (what do I know about this user?)
  2. Storage during conversation (user shared something important)
  3. Application in responses (use past context naturally)
  4. Persistence across sessions (memory survives restarts)

Next, you'll experience memory from the other side—as a user of a memory-augmented AI assistant.