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

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

Інженери витрачають до 2 годин на день на пошук відповідей у корпоративній пошті. Email — скарбниця експертних знань, але він забитий quoted text, підписами, авто-відповідями та спамом. Без очищення RAG пайплайн видає сміття: recall падає до 40%. За даними McKinsey, до 28% робочого часу йде на поштове листування. Ми накопичили 5 років досвіду в індексації поштових архівів для великих корпоративних клієнтів і гарантуємо точність 95% на тестовій вибірці. Економія бюджету на пошук інформації може досягати 40%.

Конвеєр складається з трьох етапів: фільтрація, очищення та реконструкція тредів. Автоматизований pipeline вбудовується в будь-який MLOps-процес. Вартість індексації розраховується індивідуально залежно від обсягу та складності.

Підключення до поштових серверів

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-задачі.

Чому дата-інжиніринг визначає успіх ML-моделі

Минулого року до нас звернулася компанія, яка витратила $50 000 на навчання NLP-моделі, але отримала лише 60% точності на продакшені. Причина — data leakage через випадковий split часових даних. Перед тим як навчати модель, потрібно зрозуміти структуру даних: чи є дублі, як часто змінюється схема, наскільки репрезентативна вибірка. Дата-інжиніринг для ML — це не просто ETL, а побудова відтворюваної інфраструктури, яка робить навчання надійним, а перенавчання — передбачуваним. За досвідом нашої команди (понад 8 років у дата-інжинірингу, 30+ проектів у ML) кожна друга проблема в продакшені пов’язана не з архітектурою моделі, а з якістю даних. Замовте аудит ваших даних — оцінимо поточний пайплайн безкоштовно.

Як ETL-пайплайни для ML відрізняються від BI

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

Інструменти. Apache Spark для великих обсягів (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% якості моделі на валідації. Temporal split best practices (scikit-learn docs)

Інкрементальні пайплайни. Модель перенавчається щотижня на нових даних. Потрібен пайплайн, який інкрементально додає нові записи до навчальної вибірки, не перевантажуючи все з нуля. 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 — економія склала $15 000 на 100 000 розмічених прикладів.

Синтетичні дані. Коли реальних даних мало або отримати їх дорого. Для 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.

Типові помилки при побудові пайплайнів

• Пропуск перевірки свіжості даних. Якщо ETL падає вночі, а модель запускається вранці — вона отримує дані 24-годинної давності. Рішення: алерт при затримці > 30 хвилин.
• Відсутність версіонування даних. Не можна відтворити експеримент, бо дані змінилися. DVC або Delta Lake time travel виправляють це.
• Забувають про schema evolution. Нове поле з’являється, а пайплайн падає. Автоматичне виявлення змін схеми через Great Expectations.

Active learning дозволяє скоротити бюджет на розмітку до 50–70%. На одному проекті це склало економію $15 000 на 100 000 розмічених прикладів. Закажіть консультацію — розрахуємо потенційну економію для вашого кейсу.

Що входить у проект з дата-інжинірингу для ML

Ми надаємо повний цикл:

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

Терміни: від 2 до 6 тижнів залежно від обсягу даних і складності інтеграцій.

Як ми будуємо пайплайн: покроково

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

Чому варто довірити це нам

Ми займаємося дата-інжинірингом та ML з понад 8-річним досвідом. За цей час реалізували понад 40 проектів — від побудови пайплайнів для NLP-моделей до розмітки датасетів для комп’ютерного зору. Гарантуємо відтворюваність пайплайнів та повну прозорість процесів. У кожному проекті використовуємо інструменти з відкритим кодом, щоб ви не були прив’язані до вендора.

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