---
title: Automated Email Intelligence
description: "Capture, categorize, and recall inbox threads using persistent memories."
---
**Works with:** Mem0 OSS (`Memory`) and Mem0 Platform (`MemoryClient`)
This guide demonstrates how to build an intelligent email processing system using Mem0's memory capabilities. You'll learn how to store, categorize, retrieve, and analyze emails to create a smart email management solution.
## Overview
Email overload is a common challenge for many professionals. By leveraging Mem0's memory capabilities, you can build an intelligent system that:
- Stores emails as searchable memories
- Categorizes emails automatically
- Retrieves relevant past conversations
- Prioritizes messages based on importance
- Generates summaries and action items
## Setup
Before you begin, ensure you have the required dependencies installed:
```bash
pip install mem0ai openai
```
Here we use **Mem0 open source** (`Memory`): all local, no API keys needed for memory. Vectors in **Qdrant**, LLM and embeddings via **Ollama**.
### Installation
Install the required dependencies:
```bash
pip install mem0ai qdrant-client openai ollama
```
Then start Qdrant and pull the Ollama models:
```bash
docker run -d -p 6333:6333 qdrant/qdrant
ollama pull llama3.1:latest
ollama pull nomic-embed-text:latest
```
You can swap `nomic-embed-text` for any Ollama-supported embedding model (e.g., `snowflake-arctic-embed`, `mxbai-embed-large`). Just update the `model` in the `embedder` config and set `embedding_model_dims` in the Qdrant config to match the model's output dimensions (768 for `nomic-embed-text`).
## Implementation
### Basic Email Memory System
The following example shows how to create a basic email processing system with Mem0:
```python
import os
from mem0 import MemoryClient
from email.parser import Parser
# Configure API keys
os.environ["MEM0_API_KEY"] = "your-mem0-api-key"
# Initialize Mem0 client
client = MemoryClient()
class EmailProcessor:
def __init__(self):
"""Initialize the Email Processor with Mem0 memory client"""
self.client = client
def process_email(self, email_content, user_id):
"""
Process an email and store it in Mem0 memory
Args:
email_content (str): Raw email content
user_id (str): User identifier for memory association
"""
# Parse email
parser = Parser()
email = parser.parsestr(email_content)
# Extract email details
sender = email['from']
recipient = email['to']
subject = email['subject']
date = email['date']
body = self._get_email_body(email)
# Create message object for Mem0
message = {
"role": "user",
"content": f"Email from {sender}: {subject}\n\n{body}"
}
# Create metadata for better retrieval
metadata = {
"email_type": "incoming",
"sender": sender,
"recipient": recipient,
"subject": subject,
"date": date
}
# Store in Mem0 with appropriate categories
response = self.client.add(
messages=[message],
user_id=user_id,
metadata=metadata,
categories=["email", "correspondence"],
)
return response
def _get_email_body(self, email):
"""Extract the body content from an email"""
# Simplified extraction - in real-world, handle multipart emails
if email.is_multipart():
for part in email.walk():
if part.get_content_type() == "text/plain":
return part.get_payload(decode=True).decode()
else:
return email.get_payload(decode=True).decode()
def search_emails(self, query, user_id, sender=None):
"""
Search through stored emails
Args:
query (str): Search query
user_id (str): User identifier
sender (str, optional): Filter by sender email address
"""
# For Platform API, all filters including user_id go in filters object
if not sender:
# Simple filter - just user_id and category
filters = {
"AND": [
{"user_id": user_id},
{"categories": {"contains": "email"}}
]
}
results = self.client.search(query=query, filters=filters)
else:
# Advanced filter - add sender condition
filters = {
"AND": [
{"user_id": user_id},
{"categories": {"contains": "email"}},
{"sender": sender}
]
}
results = self.client.search(query=query, filters=filters)
return results
def get_email_thread(self, subject, user_id):
"""
Retrieve all emails in a thread based on subject
Args:
subject (str): Email subject to match
user_id (str): User identifier
"""
# For Platform API, user_id goes in the filters object
filters = {
"AND": [
{"user_id": user_id},
{"categories": {"contains": "email"}},
{"subject": {"icontains": subject}}
]
}
thread = self.client.get_all(filters=filters)
return thread
# Initialize the processor
processor = EmailProcessor()
# Example raw email
sample_email = """From: alice@example.com
To: bob@example.com
Subject: Meeting Schedule Update
Date: Mon, 15 Jul 2024 14:22:05 -0700
Hi Bob,
I wanted to update you on the schedule for our upcoming project meeting.
We'll be meeting this Thursday at 2pm instead of Friday.
Could you please prepare your section of the presentation?
Thanks,
Alice
"""
# Process and store the email
user_id = "bob@example.com"
processor.process_email(sample_email, user_id)
# Later, search for emails about meetings
meeting_emails = processor.search_emails("meeting schedule", user_id)
print(f"Found {len(meeting_emails['results'])} relevant emails")
```
```python
from mem0 import Memory
from email.parser import Parser
OLLAMA_URL = "http://localhost:11434"
# Set up Mem0 with local providers
memory = Memory.from_config({
"vector_store": {
"provider": "qdrant",
"config": {
"collection_name": "email_intelligence",
"host": "localhost",
"port": 6333,
"embedding_model_dims": 768,
},
},
"llm": {
"provider": "ollama",
"config": {
"model": "llama3.1:latest",
"temperature": 0,
"max_tokens": 2000,
"ollama_base_url": OLLAMA_URL,
},
},
"embedder": {
"provider": "ollama",
"config": {
"model": "nomic-embed-text:latest",
"ollama_base_url": OLLAMA_URL,
},
},
})
class EmailProcessor:
def __init__(self):
"""Initialize the Email Processor with Mem0 memory"""
self.memory = memory
def process_email(self, email_content, user_id):
"""
Process an email and store it in Mem0 memory
Args:
email_content (str): Raw email content
user_id (str): User identifier for memory association
"""
# Parse email
parser = Parser()
email = parser.parsestr(email_content)
# Extract email details
sender = email["from"]
recipient = email["to"]
subject = email["subject"]
date = email["date"]
body = self._get_email_body(email)
# Create message object for Mem0
message = {
"role": "user",
"content": f"Email from {sender}: {subject}\n\n{body}",
}
# Create metadata for better retrieval
# In OSS, categories are modeled as metadata fields
metadata = {
"email_type": "incoming",
"memory_category": "email",
"sender": sender,
"recipient": recipient,
"subject": subject,
"date": date,
}
# Store in Mem0
response = self.memory.add(
message,
user_id=user_id,
metadata=metadata,
)
return response
def _get_email_body(self, email):
"""Extract the body content from an email"""
if email.is_multipart():
for part in email.walk():
if part.get_content_type() == "text/plain":
return part.get_payload(decode=True).decode()
else:
return email.get_payload(decode=True).decode()
def search_emails(self, query, user_id, sender=None):
"""
Search through stored emails
Args:
query (str): Search query
user_id (str): User identifier
sender (str, optional): Filter by sender email address
"""
if not sender:
results = self.memory.search(
query=query,
filters={"user_id": user_id, "memory_category": "email"},
)
else:
results = self.memory.search(
query=query,
filters={
"user_id": user_id,
"AND": [
{"memory_category": "email"},
{"sender": sender},
]
},
)
return results
def get_email_thread(self, subject, user_id):
"""
Retrieve all emails in a thread based on subject
Args:
subject (str): Email subject to match
user_id (str): User identifier
"""
thread = self.memory.get_all(
filters={
"user_id": user_id,
"AND": [
{"memory_category": "email"},
{"subject": {"icontains": subject}},
]
},
)
return thread
# Initialize the processor
processor = EmailProcessor()
# Example raw email
sample_email = """From: alice@example.com
To: bob@example.com
Subject: Meeting Schedule Update
Date: Mon, 15 Jul 2024 14:22:05 -0700
Hi Bob,
I wanted to update you on the schedule for our upcoming project meeting.
We'll be meeting this Thursday at 2pm instead of Friday.
Could you please prepare your section of the presentation?
Thanks,
Alice
"""
# Process and store the email
user_id = "bob@example.com"
processor.process_email(sample_email, user_id)
# Later, search for emails about meetings
meeting_emails = processor.search_emails("meeting schedule", user_id)
print(f"Found {len(meeting_emails['results'])} relevant emails")
```
**Categories vs Metadata:** The Platform version uses `categories=["email"]` which are AI-assigned by Mem0. In open source, categories are modeled as `metadata` fields (e.g., `memory_category`) that you set on each `add` call and filter on during search.
### Fetching Memories
You can fetch all the memories at any point in time using the following code:
```python
meeting_emails = processor.search_emails("meeting schedule", user_id)
for m in meeting_emails['results']:
print(m['memory'])
```
```python
meeting_emails = processor.search_emails("meeting schedule", user_id)
for m in meeting_emails["results"]:
print(m["memory"])
```
## Key Features and Benefits
- **Long-term Email Memory**: Store and retrieve email conversations across long periods
- **Semantic Search**: Find relevant emails even if they don't contain exact keywords
- **Intelligent Categorization**: Automatically sort emails into meaningful categories
- **Action Item Extraction**: Identify and track tasks mentioned in emails
- **Priority Management**: Focus on important emails based on AI-determined priority
- **Context Awareness**: Maintain thread context for more relevant interactions
## Conclusion
By combining Mem0's memory capabilities with email processing, you can create intelligent email management systems that help users organize, prioritize, and act on their inbox effectively. The advanced capabilities like automatic categorization, action item extraction, and priority management can significantly reduce the time spent on email management, allowing users to focus on more important tasks.
---
Categorize email threads by sender, topic, and priority for faster retrieval.
Build customer support agents that remember context across tickets.