Mem0 Docs

Mem0 home page

Search...

Ctrl KAsk AI

Search...

Navigation

Ops & Automations

Automated Email Intelligence

Welcome Mem0 Platform Open Source Cookbooks Integrations API Reference Release Notes

Documentation

Getting Started

Overview

Essentials

Companion Playbooks

Ops & Automations

Integrations & Platforms

Frameworks & Multimodal

On this page

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

Platform
Open Source

Before you begin, ensure you have the required dependencies installed:

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:

pip install mem0ai qdrant-client openai ollama

Then start Qdrant and pull the Ollama models:

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:

Platform
Open Source

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")

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
        """
        # In OSS, user_id is an explicit parameter (not inside filters)
        if not sender:
            results = self.memory.search(
                query=query,
                user_id=user_id,
                filters={"memory_category": "email"},
            )
        else:
            results = self.memory.search(
                query=query,
                user_id=user_id,
                filters={
                    "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
        """
        # In OSS, user_id is an explicit parameter
        thread = self.memory.get_all(
            user_id=user_id,
            filters={
                "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")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

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:

Platform
Open Source

meeting_emails = processor.search_emails("meeting schedule", user_id)
for m in meeting_emails['results']:
    print(m['memory'])

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.

Tag and Organize Memories \ \ Categorize email threads by sender, topic, and priority for faster retrieval.

Support Inbox with Mem0 \ \ Build customer support agents that remember context across tickets.

Was this page helpful?

YesNo

Suggest edits Raise issue

Memory-Powered Support Agent\ \ Previous Content Creation Workflow\ \ Next

Ctrl+I

Assistant

Responses are generated using AI and may contain mistakes.

Suggestions

How do I set up email automation?How do I categorize emails?What can Mem0 do with emails?

Contact support

Getting Started ​

Essentials ​

Companion Playbooks ​

Ops & Automations ​

Integrations & Platforms ​

Frameworks & Multimodal ​

​ Overview ​

​ Setup ​

​ Installation ​

​ Implementation ​

​ Basic Email Memory System ​

​ Fetching Memories ​

​ Key Features and Benefits ​

​ Conclusion ​

Getting Started

Essentials

Companion Playbooks

Ops & Automations

Integrations & Platforms

Frameworks & Multimodal

Overview

Setup

Installation

Implementation

Basic Email Memory System

Fetching Memories

Key Features and Benefits

Conclusion