mem0-fastapi-integration

What this skill does

# 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:**
```python
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:**
```python
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:**
```python
@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`

```python
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`

```python
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`

```python
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`

```python
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`

```python
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:**
```python
from fastapi import BackgroundTasks

@router.