View Notebook on Github
Memory Operations with Mem0
This example demonstrates how to use Mem0’s memory management capabilities with both synchronous and asynchronous operations to store, search, and manage conversational context and user preferences.
Overview
This example showcases practical memory management operations where we:
- Initialize Mem0 Memory instances for both sync and async operations
- Store conversation history and user preferences with metadata
- Search memories using natural language queries
- Compare performance between synchronous and asynchronous memory operations
By using async operations, you can perform multiple memory operations simultaneously instead of waiting for each one to complete sequentially. This is particularly beneficial when dealing with multiple memory additions or searches.
Setup and Imports
Installation
pip install agentops mem0ai python-dotenv
Import the required libraries for local memory management with Mem0. We’ll use both Memory and AsyncMemory classes to demonstrate different execution patterns for memory operations.
from mem0 import Memory, AsyncMemory
import os
import asyncio
import logging
from dotenv import load_dotenv
import agentops
Environment Configuration
Set up environment variables for API keys. These are essential for authenticating with AgentOps for tracing and OpenAI for the language model used in memory operations.
os.environ["AGENTOPS_API_KEY"] = os.getenv("AGENTOPS_API_KEY")
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
Configuration and Sample Data
Set up the configuration for local memory storage and define sample user data. The configuration specifies the LLM provider and model settings for processing memories.
local_config = {
"llm": {
"provider": "openai",
"config": {
"model": "gpt-4o-mini",
"temperature": 0.1,
"max_tokens": 2000,
},
}
}
user_id = "alice_demo"
agent_id = "assistant_demo"
run_id = "session_001"
sample_messages = [
{"role": "user", "content": "I'm planning to watch a movie tonight. Any recommendations?"},
{"role": "assistant", "content": "How about a thriller? They can be quite engaging."},
{"role": "user", "content": "I'm not a big fan of thriller movies but I love sci-fi movies."},
{
"role": "assistant",
"content": "Got it! I'll avoid thriller recommendations and suggest sci-fi movies in the future.",
},
]
sample_preferences = [
"I prefer dark roast coffee over light roast",
"I exercise every morning at 6 AM",
"I'm vegetarian and avoid all meat products",
"I love reading science fiction novels",
"I work in software engineering",
]
Synchronous Memory Operations
This function demonstrates sequential memory operations using the synchronous Memory class. While straightforward to implement, each operation must complete before the next begins, which can impact performance.
def demonstrate_sync_memory(local_config, sample_messages, sample_preferences, user_id):
"""
Demonstrate synchronous Memory class operations.
"""
agentops.start_trace("mem0_memory_example", tags=["mem0_memory_example"])
try:
memory = Memory.from_config(local_config)
result = memory.add(
sample_messages, user_id=user_id, metadata={"category": "movie_preferences", "session": "demo"}
)
for i, preference in enumerate(sample_preferences):
result = memory.add(preference, user_id=user_id, metadata={"type": "preference", "index": i})
search_queries = [
"What movies does the user like?",
"What are the user's food preferences?",
"When does the user exercise?",
]
for query in search_queries:
results = memory.search(query, user_id=user_id)
if results and "results" in results:
for j, result in enumerate(results):
print(f"Result {j+1}: {result.get('memory', 'N/A')}")
else:
print("No results found")
all_memories = memory.get_all(user_id=user_id)
if all_memories and "results" in all_memories:
print(f"Total memories: {len(all_memories['results'])}")
delete_all_result = memory.delete_all(user_id=user_id)
print(f"Delete all result: {delete_all_result}")
agentops.end_trace(end_state="success")
except Exception as e:
agentops.end_trace(end_state="error")
Asynchronous Memory Operations
This function showcases concurrent memory operations using AsyncMemory. By leveraging asyncio.gather(), multiple operations execute simultaneously, significantly reducing total execution time for I/O-bound tasks.
async def demonstrate_async_memory(local_config, sample_messages, sample_preferences, user_id):
"""
Demonstrate asynchronous Memory class operations with concurrent execution.
"""
agentops.start_trace("mem0_memory_async_example", tags=["mem0_memory_async_example"])
try:
async_memory = await AsyncMemory.from_config(local_config)
result = await async_memory.add(
sample_messages, user_id=user_id, metadata={"category": "async_movie_preferences", "session": "async_demo"}
)
async def add_preference(preference, index):
"""Helper function to add a single preference asynchronously."""
return await async_memory.add(
preference, user_id=user_id, metadata={"type": "async_preference", "index": index}
)
tasks = [add_preference(pref, i) for i, pref in enumerate(sample_preferences)]
results = await asyncio.gather(*tasks)
for i, result in enumerate(results):
print(f"Added async preference {i+1}: {result}")
search_queries = [
"What movies does the user like?",
"What are the user's dietary restrictions?",
"What does the user do for work?",
]
async def search_memory(query):
"""Helper function to perform async memory search."""
return await async_memory.search(query, user_id=user_id), query
search_tasks = [search_memory(query) for query in search_queries]
search_results = await asyncio.gather(*search_tasks)
for result, query in search_results:
if result and "results" in result:
for j, res in enumerate(result["results"]):
print(f"Result {j+1}: {res.get('memory', 'N/A')}")
else:
print("No results found")
all_memories = await async_memory.get_all(user_id=user_id)
if all_memories and "results" in all_memories:
print(f"Total async memories: {len(all_memories['results'])}")
delete_all_result = await async_memory.delete_all(user_id=user_id)
print(f"Delete all result: {delete_all_result}")
agentops.end_trace(end_state="success")
except Exception as e:
agentops.end_trace(end_state="error")
Execute Demonstrations
Run both synchronous and asynchronous demonstrations to compare their execution patterns and performance. The async version demonstrates the benefits of concurrent execution for multiple memory operations.
# Execute both sync and async demonstrations
demonstrate_sync_memory(local_config, sample_messages, sample_preferences, user_id)
await demonstrate_async_memory(local_config, sample_messages, sample_preferences, user_id)
