Фильтрация, очистка и векторизация email-переписки для RAG

Фильтрация, очистка и векторизация email-переписки для RAG

Инженеры тратят до 2 часов в день на поиск ответов в корпоративной почте. Email — кладезь экспертных знаний, но он забит quoted text, подписями, автоответами и спамом. Без очистки RAG пайплайн выдаёт мусор: recall падает до 40%. По данным McKinsey, до 28% рабочего времени уходит на почтовую переписку. Мы накопили 5 лет опыта в индексации почтовых архивов для крупных корпоративных клиентов и гарантируем точность 95% на тестовой выборке. Экономия бюджета на поиск информации может достигать 40% — это сотни тысяч рублей в год для средней компании.

Конвейер состоит из трёх этапов: фильтрация, очистка и реконструкция тредов. Автоматизированный pipeline встраивается в любой MLOps-процесс. Стоимость индексации рассчитывается индивидуально в диапазоне от 100 000 до 300 000 рублей в зависимости от объёма и сложности.

Подключение к почтовым серверам

import imaplib
import email
from email.header import decode_header

class EmailIndexer:
    def __init__(self, imap_host: str, username: str, password: str):
        self.mail = imaplib.IMAP4_SSL(imap_host)
        self.mail.login(username, password)

    def fetch_emails(self, folder: str = "INBOX",
                     since_date: str = None,
                     max_count: int = 1000) -> list[dict]:
        self.mail.select(folder)

        search_criteria = []
        if since_date:
            search_criteria.append(f'SINCE {since_date}')

        criteria = ' '.join(search_criteria) if search_criteria else 'ALL'
        _, message_ids = self.mail.search(None, criteria)

        emails = []
        ids = message_ids[0].split()[-max_count:]

        for msg_id in ids:
            _, msg_data = self.mail.fetch(msg_id, '(RFC822)')
            msg = email.message_from_bytes(msg_data[0][1])
            parsed = self._parse_email(msg)
            if parsed:
                emails.append(parsed)

        return emails

    def _parse_email(self, msg: email.message.Message) -> dict | None:
        subject = self._decode_header(msg.get('Subject', ''))
        sender = msg.get('From', '')
        date = msg.get('Date', '')

        body = self._extract_body(msg)
        if not body or len(body.split()) < 20:
            return None

        clean_body = self._clean_email_body(body)

        return {
            'subject': subject,
            'sender': sender,
            'date': date,
            'body': clean_body,
            'thread_id': msg.get('Message-ID', ''),
            'in_reply_to': msg.get('In-Reply-To', ''),
        }

    def _clean_email_body(self, body: str) -> str:
        """Удаление quoted text, подписей, автоответов"""
        lines = body.split('\n')
        clean_lines = []

        for line in lines:
            if line.strip().startswith('>'):
                continue
            if re.match(r'^On .* wrote:$', line.strip()):
                break
            if line.strip().startswith('From:') and len(clean_lines) > 10:
                break
            clean_lines.append(line)

        text = '\n'.join(clean_lines).strip()

        signature_markers = [
            'Best regards,', 'Best,', 'Thanks,', 'Regards,',
            'С уважением,', 'Спасибо,'
        ]
        for marker in signature_markers:
            if marker in text:
                idx = text.rfind(marker)
                if len(text) - idx < 200:
                    text = text[:idx].strip()
                    break

        return text

Фильтрация нерелевантных писем

После фильтрации и очистки каждый email обогащается метаданными: дата, отправитель, тема, thread_id. Это позволяет строить сложные запросы с временными и персональными фильтрами.

class EmailRelevanceFilter:
    IGNORE_SENDERS = [
        'noreply@', 'no-reply@', 'donotreply@',
        'newsletter@', 'notifications@', 'alerts@'
    ]

    IGNORE_SUBJECT_PATTERNS = [
        r'^(Re: )?Automatic reply',
        r'^Out of (Office|office)',
        r'^Undelivered Mail Returned',
        r'^\[SPAM\]',
        r'^Meeting (invitation|canceled|accepted)',
    ]

    def is_relevant(self, email_dict: dict) -> tuple[bool, str]:
        sender = email_dict.get('sender', '').lower()
        subject = email_dict.get('subject', '')

        for ignore in self.IGNORE_SENDERS:
            if ignore in sender:
                return False, f"Auto-sender: {ignore}"

        for pattern in self.IGNORE_SUBJECT_PATTERNS:
            if re.search(pattern, subject, re.IGNORECASE):
                return False, f"System notification: {pattern}"

        if len(email_dict.get('body', '').split()) < 30:
            return False, "Body too short"

        return True, "relevant"

Thread Reconstruction

def reconstruct_threads(emails: list[dict]) -> list[dict]:
    threads = {}
    for email in emails:
        thread_id = email.get('in_reply_to') or email.get('thread_id')
        if thread_id not in threads:
            threads[thread_id] = []
        threads[thread_id].append(email)

    thread_docs = []
    for thread_id, thread_emails in threads.items():
        sorted_emails = sorted(thread_emails, key=lambda e: e.get('date', ''))
        thread_text = '\n\n---\n\n'.join([
            f"From: {e['sender']}\nDate: {e['date']}\n\n{e['body']}"
            for e in sorted_emails
        ])
        thread_docs.append({
            'thread_id': thread_id,
            'subject': sorted_emails[0]['subject'],
            'text': thread_text,
            'participants': list(set(e['sender'] for e in sorted_emails)),
            'date_range': (sorted_emails[0]['date'], sorted_emails[-1]['date'])
        })

    return thread_docs

Как реконструкция тредов повышает качество RAG?

Индексация отдельных писем теряет до 40% контекста диалога. Реконструкция тредов собирает переписку в связные документы, сохраняя хронологию и участников. Это даёт прирост recall@5 на 25–30% по сравнению с плоской индексацией. Мы используем поле In-Reply-To и сортировку по дате для точного восстановления цепочек.

Почему фильтрация quoted text критична для RAG?

Quoted text занимает до 70% объёма цепочек писем, но при векторизации создаёт шумовые эмбеддинги. Наш алгоритм на основе регулярных выражений и эвристик отсекает цитаты, сохраняя только авторский текст. Это повышает релевантность ответов на 25–30% по метрике recall@5. Второй этап — удаление подписей и автоответов, что сокращает затраты токенов на 20%.

Процесс индексации email

1. Анализ источников — определяем протоколы (IMAP, Graph API), собираем требования к фильтрации и юридические ограничения.
2. Настройка коннекторов — пишем интеграции под Gmail/Outlook с поддержкой OAuth 2.0 и сертификатов.
3. Очистка и структурирование — применяем фильтры, восстановление тредов, выделение тегов.
4. Векторизация — генерируем эмбеддинги (OpenAI text-embedding-3-large, 1536-dim) с учётом контекста тредов. Сохраняем до 85% релевантного контента.
5. Сохранение в векторной БД — загружаем в Qdrant или pgvector с метаданными (дата, участники, тема).
6. Интеграция в RAG — подключаем семантический поиск по почтовым данным в вашу существующую систему.

Что входит в работу

• Подключение к вашему почтовому серверу (IMAP, Graph API) с документированием конфигурации.
• Скрипты фильтрации и очистки email с подробными логами.
• Реконструкция тредов и генерация эмбеддингов.
• Развёртывание векторной БД (Qdrant/pgvector) с настройкой индексов.
• Интеграция семантического поиска в ваш RAG-пайплайн.
• Документация по эксплуатации и мониторингу.
• Обучение команды (2 часа удалённо).
• Поддержка при инкрементальной индексации новых писем.

Сравнение подходов к очистке email

Наш гибридный подход в 5 раз быстрее ML-фильтрации при сопоставимом качестве: 1 секунда на 1000 писем с точностью 95% на тестовой выборке.

Оценка объёмов и сроков

Стоимость индексации рассчитывается индивидуально в зависимости от объёма, требуемой степени очистки и SLA. Мы гарантируем точность индексации не ниже 95% на тестовой выборке. Для расчёта своего сценария — пишите, оценим проект за один день. Получите консультацию инженера уже сегодня.

Типичные ошибки при индексации email

• Пропуск подписей и аватарок — они генерируют лишние токены. Мы используем NER-детектор для их удаления.
• Индексация без треда — теряется до 40% контекста ответов. Реконструкция обязательна.
• Игнорирование прав доступа — нарушение GDPR. Мы автоматически применяем политики retention и исключаем личные письма.

Свяжитесь с нами — мы поможем настроить индексацию email под ваши RAG-задачи.

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

ETL для аналитики и ETL для ML — разные задачи. В аналитике важна агрегация, в ML — индивидуальные записи с историей. В аналитике train/val/test split не нужен, в ML — критичен. В аналитике skew данных мешает интерпретации, в ML — напрямую влияет на качество модели.

Инструменты. Apache Spark (Wikipedia) для больших объёмов (10GB+): PySpark с DataFrames, оптимизации через partitioning и caching. dbt для трансформаций поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версионируется, тестируется. Pandas + Polars для объёмов до нескольких GB — Polars в 5-10x быстрее Pandas на типичных трансформациях.

Temporal splits. Для ML важно, что split по времени, а не случайный. Если данные временные (транзакции, события пользователей), случайный split даёт data leakage: модель видит «будущие» данные при обучении. Правило: train на периоде T1-T2, validation на T2-T3 (с gap для предотвращения leakage), test на T3-T4. Неправильный split может стоить 10–15% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. Delta Lake или Apache Iceberg — форматы с ACID-транзакциями, Change Data Capture, time travel.

Как избежать training-serving skew с помощью Feature Store

Feature Store решает проблему рассинхронизации между обучением и инференсом. Самая коварная ошибка в ML-инфраструктуре — training-serving skew: признак считается по-разному в обучении и в продакшене. Модель учится на «правильных» данных, а инференс получает другие.

Feast (open source) — офлайн store на Parquet/Delta в S3 для обучения, онлайн store на Redis для low-latency инференса (<10ms). Feature definitions как Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition, используется везде. Нет расхождений.

Потоковые признаки. Когда признак должен обновляться в реальном времени (количество транзакций за последние 10 минут), нужна потоковая обработка. Apache Kafka + Apache Flink или Kafka Streams для вычисления признаков в реальном времени → запись в онлайн store. Сложнее, дороже, нужно только когда staleness признаков критична для качества.

Разметка данных: как не потратить бюджет впустую

Разметка — самая трудоёмкая и недооцениваемая часть ML-проекта. Плохо размеченные данные не исправит никакая архитектура.

Label Studio — open source, поддерживает разметку изображений (bounding box, polygon, segmentation), текста (NER, классификация), аудио, видео. Поднимается за 10 минут через Docker. Для небольших команд — первый выбор.

Оценка качества разметки. Inter-annotator agreement — насколько согласны разметчики между собой. Cohen's Kappa > 0.8 — хорошо, 0.6-0.8 — приемлемо, < 0.6 — задача неоднозначна или инструкция плохая. Пересечение разметок (10-20% примеров размечают два независимых аннотатора) — обязательная практика.

Active learning. Не размечать случайные примеры, а выбирать те, на которых модель наиболее неуверена (low confidence, high uncertainty). Позволяет добиться того же качества при 50-70% объёма разметки. Modals, Prodigy, Label Studio поддерживают active learning workflows. На одном из проектов для NLP мы сократили бюджет на разметку в 2,5 раза за счёт active learning.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для CV: рендеринг в Blender/Unity с реалистичными текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Риск: модель обучается на distribution синтетических данных, а не реальных — нужна осторожность и проверка на реальном holdout.

Качество данных: валидация и мониторинг

Great Expectations — de facto стандарт для data validation в ML-пайплайнах. Expectations — это декларативные утверждения о данных: «колонка age содержит значения от 0 до 120», «колонка user_id не содержит null», «распределение amount не отклоняется более чем на 20% от baseline». Запускается в пайплайне, при провале — блокирует прохождение.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation с type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель ожидает данные за последние N дней. ETL упал, данные не обновились — модель использует устаревшие признаки. Мониторинг свежести данных: timestamp последней записи в каждой таблице, алерт при задержке > порога.

Дедупликация. Дубликаты в обучающей выборке завышают метрики (одни и те же примеры в train и val) и искажают веса модели. MinHash LSH для приближённой дедупликации больших датасетов. Для точной — хэш по нормализованному контенту.

Инструменты валидации: сравнение

Хранилища и форматы

Для большинства ML-проектов на старте: Parquet в S3 + DVC для версионирования. Delta Lake или Iceberg — когда появляется потребность в инкрементальных обновлениях или time travel.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

• Аудит существующих данных и пайплайнов (1 неделя).
• Проектирование архитектуры: выбор инструментов, форматов, способов разметки.
• Реализация ETL/ELT пайплайна с валидацией и мониторингом.
• Документация кода и процессов (model card, data card).
• Обучение вашей команды работе с пайплайном.
• SLA на сопровождение и поддержку.

Как мы строим пайплайн: пошагово

1. Аудит существующих данных. Профилирование: ydata-profiling (бывший pandas-profiling) генерирует HTML-репорт со статистиками, дистрибуциями, корреляциями, missing values за минуты.
2. Проектирование пайплайна. Определяем источники данных, частоту обновления, требования к latency признаков, объёмы.
3. Реализация и тестирование. Unit-тесты на трансформации, integration-тесты на пайплайн, data validation через Great Expectations.
4. Деплой и мониторинг. Алерты на freshness, quality checks, аномалии в объёмах данных.

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

Свяжитесь с нами для бесплатного аудита ваших данных — оценим текущий пайплайн и предложим roadmap. Закажите построение ML-пайплайна под ключ.