Mem0 FastAPI Integration Patterns

Purpose: Provide complete Mem0 integration templates, memory service patterns, user tracking implementations, and conversation persistence strategies for building FastAPI applications with intelligent AI memory.

Activation Triggers:

• Integrating Mem0 memory layer into FastAPI

• Building chat applications with conversation history

• Implementing user context and personalization

• Adding memory to AI agents

• Creating stateful AI interactions

• User preference management

Key Resources:

• templates/memory_service.py

• Complete Mem0 service implementation

• templates/memory_middleware.py

• Request-scoped memory middleware

• templates/memory_client.py

• Mem0 client configuration

• templates/memory_routes.py

• API routes for memory operations

• scripts/setup-mem0.sh

• Mem0 installation and configuration

• scripts/test-memory.sh

• Memory service testing utility

• examples/chat_with_memory.py

• Complete chat implementation

• examples/user_preferences.py

• User preference management

Core Mem0 Integration

1. Client Configuration

Template: templates/memory_client.py

Workflow:

from mem0 import Memory, AsyncMemory, MemoryClient from mem0.configs.base import MemoryConfig

Hosted Mem0 Platform

client = MemoryClient(api_key=settings.MEM0_API_KEY)

Self-Hosted Configuration

config = MemoryConfig( vector_store={ "provider": "qdrant", "config": { "host": settings.QDRANT_HOST, "port": settings.QDRANT_PORT, "api_key": settings.QDRANT_API_KEY } }, llm={ "provider": "openai", "config": { "model": "gpt-4", "api_key": settings.OPENAI_API_KEY } }, embedder={ "provider": "openai", "config": { "model": "text-embedding-3-small", "api_key": settings.OPENAI_API_KEY } } ) memory = AsyncMemory(config)

2. Memory Service Pattern

Template: templates/memory_service.py

Key Operations:

class MemoryService: async def add_conversation( user_id: str, messages: List[Dict[str, str]], metadata: Optional[Dict] = None ) -> Dict

async def search_memories(
    query: str,
    user_id: str,
    limit: int = 5
) -> List[Dict]

async def get_user_summary(user_id: str) -> Dict

async def add_user_preference(
    user_id: str,
    preference: str,
    category: str = "general"
) -> bool

Initialization:

@asynccontextmanager async def lifespan(app: FastAPI): # Startup memory_service = MemoryService() app.state.memory_service = memory_service yield # Shutdown

Memory Patterns

1. Conversation Persistence

When to use: Chat applications, conversational AI

Template: templates/memory_service.py#add_conversation

async def add_conversation( self, user_id: str, messages: List[Dict[str, str]], metadata: Optional[Dict[str, Any]] = None ) -> Optional[Dict]: enhanced_metadata = { "timestamp": datetime.now().isoformat(), "conversation_type": "chat", **(metadata or {}) }

if self.client:
    result = self.client.add(
        messages=messages,
        user_id=user_id,
        metadata=enhanced_metadata
    )
elif self.memory:
    result = await self.memory.add(
        messages=messages,
        user_id=user_id,
        metadata=enhanced_metadata
    )

return result

Best for: Chat history, conversation context, multi-turn interactions

2. Semantic Memory Search

When to use: Context retrieval, relevant history lookup

Template: templates/memory_service.py#search_memories

async def search_memories( self, query: str, user_id: str, limit: int = 5, filters: Optional[Dict] = None ) -> List[Dict]: if self.client: result = self.client.search( query=query, user_id=user_id, limit=limit ) elif self.memory: result = await self.memory.search( query=query, user_id=user_id, limit=limit )

memories = result.get('results', [])
return memories

Best for: Finding relevant past conversations, context-aware responses

3. User Preference Storage

When to use: Personalization, user settings, behavioral tracking

Template: templates/memory_service.py#add_user_preference

async def add_user_preference( self, user_id: str, preference: str, category: str = "general" ) -> bool: preference_message = { "role": "system", "content": f"User preference ({category}): {preference}" }

metadata = {
    "type": "preference",
    "category": category,
    "timestamp": datetime.now().isoformat()
}

if self.client:
    self.client.add(
        messages=[preference_message],
        user_id=user_id,
        metadata=metadata
    )
elif self.memory:
    await self.memory.add(
        messages=[preference_message],
        user_id=user_id,
        metadata=metadata
    )

return True

Best for: User customization, learning user behavior, preference management

API Routes Integration

1. Memory Management Endpoints

Template: templates/memory_routes.py

from fastapi import APIRouter, Depends, BackgroundTasks from app.api.deps import get_current_user, get_memory_service from app.services.memory_service import MemoryService

router = APIRouter()

@router.post("/conversation") async def add_conversation( request: ConversationRequest, background_tasks: BackgroundTasks, user_id: str = Depends(get_current_user), memory_service: MemoryService = Depends(get_memory_service) ): # Add to memory in background background_tasks.add_task( memory_service.add_conversation, user_id, request.messages, request.metadata )

return {
    "status": "success",
    "message": "Conversation added to memory"
}

@router.post("/search") async def search_memories( request: SearchRequest, user_id: str = Depends(get_current_user), memory_service: MemoryService = Depends(get_memory_service) ): memories = await memory_service.search_memories( query=request.query, user_id=user_id, limit=request.limit )

return {
    "query": request.query,
    "results": memories,
    "count": len(memories)
}

@router.get("/summary") async def get_memory_summary( user_id: str = Depends(get_current_user), memory_service: MemoryService = Depends(get_memory_service) ): summary = await memory_service.get_user_summary(user_id) return {"user_id": user_id, "summary": summary}

2. Request Models

Template: templates/memory_routes.py#models

from pydantic import BaseModel, Field from typing import List, Dict, Optional, Any

class ConversationRequest(BaseModel): messages: List[Dict[str, str]] = Field(..., description="Conversation messages") session_id: Optional[str] = Field(None, description="Session identifier") metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata")

class SearchRequest(BaseModel): query: str = Field(..., description="Search query") limit: int = Field(5, ge=1, le=20, description="Number of results") filters: Optional[Dict[str, Any]] = Field(None, description="Search filters")

class PreferenceRequest(BaseModel): preference: str = Field(..., description="User preference") category: str = Field("general", description="Preference category")

Background Task Integration

1. Async Memory Storage

Pattern:

from fastapi import BackgroundTasks

@router.post("/chat") async def chat( request: ChatRequest, background_tasks: BackgroundTasks, memory_service: MemoryService = Depends(get_memory_service) ): # Generate AI response response = await ai_service.generate(request.message)

# Store conversation in background
background_tasks.add_task(
    memory_service.add_conversation,
    user_id=request.user_id,
    messages=[
        {"role": "user", "content": request.message},
        {"role": "assistant", "content": response}
    ]
)

return {"response": response}

Benefits:

• Non-blocking memory operations

• Faster response times

• Graceful failure handling

2. Batch Memory Updates

Pattern:

async def batch_update_memories( conversations: List[Conversation], memory_service: MemoryService ): tasks = [ memory_service.add_conversation( user_id=conv.user_id, messages=conv.messages ) for conv in conversations ]

results = await asyncio.gather(*tasks, return_exceptions=True)
return results

Dependency Injection Pattern

1. Memory Service Dependency

Template: templates/memory_middleware.py#deps

from fastapi import Request

def get_memory_service(request: Request) -> MemoryService: """Get memory service from app state""" return request.app.state.memory_service

Usage in routes

@router.get("/") async def endpoint( memory_service: MemoryService = Depends(get_memory_service) ): memories = await memory_service.search_memories(...) return memories

2. User Context Injection

Pattern:

async def get_user_context( user_id: str = Depends(get_current_user), memory_service: MemoryService = Depends(get_memory_service) ) -> Dict[str, Any]: """Get enriched user context from memory""" summary = await memory_service.get_user_summary(user_id) return { "user_id": user_id, "preferences": summary.get("user_preferences", []), "total_conversations": summary.get("total_memories", 0) }

Usage

@router.post("/personalized-response") async def personalized( request: QueryRequest, context: Dict = Depends(get_user_context) ): # Use context for personalized AI response pass

Implementation Workflow

Step 1: Setup Mem0

Install and configure Mem0

./scripts/setup-mem0.sh

Creates:

• Mem0 configuration file

• Environment variable template

• Vector database setup (if self-hosted)

Step 2: Configure Memory Service

Decision tree:

• Hosted Mem0: Use MemoryClient with API key (simple, managed)

• Self-Hosted: Use AsyncMemory with config (full control, cost-effective)

• Hybrid: Support both with fallback logic

Step 3: Integrate with FastAPI Lifespan

Pattern:

from contextlib import asynccontextmanager

@asynccontextmanager async def lifespan(app: FastAPI): # Startup memory_service = MemoryService() app.state.memory_service = memory_service logger.info("Memory service initialized")

yield

# Shutdown
logger.info("Shutting down memory service")

Step 4: Add Memory Routes

Use template: templates/memory_routes.py

Register:

app.include_router( memory.router, prefix=f"{settings.API_V1_STR}/memory", tags=["memory"] )

Step 5: Test Memory Service

Run memory service tests

./scripts/test-memory.sh

Tests:

• Add conversation

• Search memories

• Get user summary

• Add preferences

• Error handling

Optimization Strategies

1. Memory Caching

Pattern:

from functools import lru_cache from datetime import datetime, timedelta

class CachedMemoryService(MemoryService): def init(self): super().init() self._cache = {} self._cache_ttl = timedelta(minutes=5)

async def search_memories(self, query: str, user_id: str, limit: int = 5):
    cache_key = f"{user_id}:{query}:{limit}"

    if cache_key in self._cache:
        cached_time, cached_result = self._cache[cache_key]
        if datetime.now() - cached_time < self._cache_ttl:
            return cached_result

    result = await super().search_memories(query, user_id, limit)
    self._cache[cache_key] = (datetime.now(), result)
    return result

2. Batch Operations

Pattern:

async def batch_add_conversations( conversations: List[Tuple[str, List[Dict]]], memory_service: MemoryService ): """Add multiple conversations efficiently""" tasks = [ memory_service.add_conversation(user_id, messages) for user_id, messages in conversations ] return await asyncio.gather(*tasks)

3. Error Handling with Fallback

Pattern:

async def resilient_memory_add( user_id: str, messages: List[Dict], memory_service: MemoryService ): try: return await memory_service.add_conversation(user_id, messages) except Exception as e: logger.error(f"Memory add failed: {e}") # Fallback: Queue for retry or log to backup storage await backup_storage.save(user_id, messages) return None

Production Best Practices

1. Environment Configuration

Template: .env variables

Hosted Mem0

MEM0_API_KEY=your_mem0_api_key

Self-Hosted Configuration

QDRANT_HOST=localhost QDRANT_PORT=6333 QDRANT_API_KEY=your_qdrant_key OPENAI_API_KEY=your_openai_key

Memory Settings

MEMORY_CACHE_TTL_SECONDS=300 MEMORY_SEARCH_LIMIT_DEFAULT=5 MEMORY_SEARCH_LIMIT_MAX=20

2. Monitoring and Logging

Pattern:

import logging from datetime import datetime

logger = logging.getLogger(name)

class MonitoredMemoryService(MemoryService): async def add_conversation(self, user_id: str, messages: List[Dict], metadata=None): start_time = datetime.now() try: result = await super().add_conversation(user_id, messages, metadata) latency = (datetime.now() - start_time).total_seconds() logger.info(f"Memory add success: user={user_id}, latency={latency}s") return result except Exception as e: logger.error(f"Memory add failed: user={user_id}, error={e}") raise

3. Rate Limiting

Pattern:

from fastapi import HTTPException from slowapi import Limiter from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/search") @limiter.limit("10/minute") async def search_memories(request: Request, ...): # Rate-limited memory search pass

Common Memory Patterns

1. Chat with Memory

Example: examples/chat_with_memory.py

Complete chat implementation with conversation history and context retrieval

Features:

• Automatic conversation storage

• Context-aware responses

• Session management

• Memory search integration

2. User Preference Management

Example: examples/user_preferences.py

User preference tracking and personalization

Features:

• Preference categorization

• Preference retrieval

• Behavioral learning

• Personalized AI responses

3. Multi-User Memory Isolation

Pattern:

async def get_user_memories_isolated( user_id: str, memory_service: MemoryService ): """Ensure memory isolation between users""" # All memory operations scoped to user_id memories = await memory_service.search_memories( query="...", user_id=user_id # Always scoped ) return memories

Vector Database Support

Supported Vector Databases

1. Qdrant (Recommended for Self-Hosted)

vector_store={ "provider": "qdrant", "config": { "host": "localhost", "port": 6333, "api_key": settings.QDRANT_API_KEY } }

2. Pinecone (Fully Managed)

vector_store={ "provider": "pinecone", "config": { "api_key": settings.PINECONE_API_KEY, "environment": "us-west1-gcp" } }

3. Chroma (Local Development)

vector_store={ "provider": "chroma", "config": { "host": "localhost", "port": 8000 } }

Resources

Scripts:

• setup-mem0.sh

• Install and configure Mem0

• test-memory.sh

• Test memory service functionality

Templates:

• memory_service.py

• Complete MemoryService implementation

• memory_client.py

• Mem0 client configuration

• memory_middleware.py

• Middleware and dependencies

• memory_routes.py

• FastAPI routes for memory operations

Examples:

• chat_with_memory.py

• Complete chat with memory

• user_preferences.py

• User preference management

• memory_analytics.py

• Memory usage analytics

Supported Configurations: Hosted Mem0, Self-Hosted (Qdrant, Pinecone, Chroma) FastAPI Version: 0.100+ Mem0 Version: 0.1.0+ Python Version: 3.9+

Best Practice: Start with hosted Mem0 for simplicity, migrate to self-hosted for cost optimization