Email Processing with FAISS and OpenAI Embeddings

Description

This Python script performs the following tasks:

Loads email data from a JSON file (full_emails.json).
Extracts relevant fields such as content, ID, and timestamp using a schema.
Cleans the email content by removing URLs, extra spaces, and unnecessary text.
Converts the cleaned content into Document objects for further processing.
Generates embeddings for the email documents using OpenAIEmbeddings.
Stores the embeddings in a FAISS vector database for fast similarity search.
Saves the FAISS index to disk for future use.

#!/usr/bin/env python

import json
import re
import os
from datetime import datetime
from langchain_community.document_loaders import JSONLoader
from langchain_openai import OpenAIEmbeddings  # Updated import
from langchain_community.vectorstores import FAISS
from langchain.schema import Document  # Import Document class

# Define jq schema to extract relevant fields from the JSON
jq_schema = ".[] | {page_content: .content, metadata: {id: .id, date: .date}}"

# Initialize the JSONLoader with the schema
loader = JSONLoader("full_emails.json", jq_schema=jq_schema, text_content=False)
documents = loader.load()

# Initialize OpenAI embeddings
embeddings = OpenAIEmbeddings()

# Clean email content function
def clean_content(text):
    text = re.sub(r'http\S+', '', text)  # Remove URLs
    text = re.sub(r'-{2,}', '-', text)  # Replace sequences of dashes with one dash
    text = re.sub(r'(unsubscribe.*|learn why we included.*|you are receiving.*)', ''
                                                    , text, flags=re.IGNORECASE)
    text = re.sub(r'\[image:.*?\]|\&\w+;', '', text)  # Remove HTML entities
    return re.sub(r'\s+', ' ', text).strip()  # Remove extra spaces

# Convert the cleaned content into Document objects
processed_docs = []
for doc in documents:
    try:
        content_data = json.loads(doc.page_content)  # Parse content
        content = clean_content(content_data.get("page_content", "No Content"))
        metadata = content_data.get("metadata", {})
        email_id = metadata.get("id", "Unknown ID")
        timestamp = int(metadata.get("date", 0)) / 1000  # Convert to seconds
        date = datetime.fromtimestamp(timestamp).isoformat() if timestamp > 0 else "Unknown"

        # Create Document object
        processed_docs.append(
            Document(page_content=content, metadata={"id": email_id, "date": date})
        )

    except json.JSONDecodeError as e:
        print(f"Error parsing JSON content: {e}")

# Create FAISS vector store with the Document objects
vector_store = FAISS.from_documents(processed_docs, embeddings)

# Save the FAISS index to disk
faiss_index_path = "faiss_index"
vector_store.save_local(faiss_index_path)
print(f"FAISS index saved to '{faiss_index_path}'")

Explanation of the Code

This script processes a set of emails and builds a vector index for efficient search and retrieval using FAISS.

JSONLoader: Loads email data from the JSON file using the provided schema to extract content, ID, and timestamp.
Cleaning Content: The clean_content() function removes unwanted elements like URLs, extra dashes, and boilerplate text (e.g., unsubscribe messages).
Document Objects: Each email is converted into a Document object, which stores both the cleaned content and metadata.
OpenAI Embeddings: Generates vector embeddings for each document to enable similarity search.
FAISS Vector Store: Stores the documents' embeddings in a FAISS index for fast searching.
Saving the Index: The FAISS index is saved locally for future searches.

Output

When the script is run, it displays the following message:


FAISS index saved to 'faiss_index'

Conclusion

This script leverages OpenAI embeddings and the FAISS vector store to process and index email data, allowing for efficient similarity searches. This setup can be extended to support various document types or integrate with other machine learning models for advanced use cases.

1. What the script does

This script is the indexing half of a semantic email search tool. It takes the JSON dump produced by the Gmail fetch step, cleans each message, converts the text to dense vectors with an OpenAI embedding model, and stores those vectors in a FAISS index on disk. Once the index exists, a question such as "when did the landlord confirm the lease renewal" can be embedded and compared against every email, returning the most similar messages even when they share no exact keywords with the question. The saved index is also the retriever used by the RetrievalQA page, where retrieved emails are passed to a chat model to generate an answer.

2. How it works

LangChain's JSONLoader applies a jq expression to the file, producing one record per email with the body as content and the ID and date as metadata. Because text_content=False, each record's content is a JSON string, so the loop parses it again, cleans the body, converts the millisecond timestamp to ISO format, and builds a Document. FAISS.from_documents sends the texts to the embedding API in batches, stacks the returned vectors into a matrix, adds them to a FAISS index, and keeps a docstore that maps each vector position back to its Document. save_local writes the index file and a pickle of the docstore.

3. The code explained

The clean_content function matters more than it looks. Marketing emails are full of tracking URLs, unsubscribe footers and HTML entities; left in, they dominate the embeddings and make unrelated newsletters look similar to each other. The regexes strip URLs, collapse dashes, remove common footer phrases and image placeholders, and normalise whitespace. Metadata is kept small (ID and date) so search results can be traced back to the original message with the Gmail API. Errors in individual records are logged and skipped rather than aborting the whole run.

4. Usage: querying the index

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings()          # must match the model used at index time
store = FAISS.load_local("faiss_index", embeddings,
                         allow_dangerous_deserialization=True)   # docstore is a pickle you created

for doc, score in store.similarity_search_with_score("lease renewal from landlord", k=5):
    print(f"{score:.3f}", doc.metadata["date"], doc.page_content[:100])

# Optional: restrict by metadata after retrieval
recent = [d for d in store.similarity_search("invoice", k=20) if d.metadata["date"] >= "2025-01-01"]

# Add new emails without rebuilding
from langchain.schema import Document
store.add_documents([Document(page_content="new message text", metadata={"id": "x", "date": "2026-10-01"})])
store.save_local("faiss_index")

Scores from the default index are L2 distances, so lower is more similar. The allow_dangerous_deserialization flag is required because the docstore is pickled; only load index directories you produced yourself.

5. Trade-offs: index types and chunking

Choice	Benefit	Cost
IndexFlatL2 / IndexFlatIP	Exact results, no training	Linear scan; fine up to roughly a million vectors on one machine
IVF (inverted lists)	Much faster search	Needs training; approximate recall
HNSW graph	Fast, high recall, no training	More memory; slower inserts
One vector per email	Simple, metadata maps 1:1	Long emails get a diluted embedding
Chunked emails	Precise matches in long threads	More vectors and API cost; need to group results by email

6. Pitfalls

Embedding the whole mailbox sends every email body to a third-party API; consider data sensitivity and use a local embedding model if needed.
Very long emails can exceed the embedding model's input limit; chunk them with a text splitter.
Changing embedding models requires re-indexing everything.
Timestamps of 0 produce "Unknown" dates; filters must handle that value.