Разработка Embedding Pipeline для индексации данных

Разработка Embedding Pipeline для индексации данных

Без правильно построенного пайплайна эмбеддингов RAG-система выдаёт нерелевантные ответы или теряет смысл из-за некорректного chunking. Мы строим production-ready пайплайны, которые обеспечивают латентность p99 менее 50 мс при индексации 1 миллиона документов. Наш опыт — 5+ лет в ML-инженерии и более 15 внедрённых систем семантического поиска. При индексации через OpenAI API text-embedding-3-small затраты составляют около $2 на 1M документов, а self-hosted решения на A10G дают почти вдвое большую пропускную способность.

Почему важен правильный chunking?

Разбивка текста на чанки — критический этап. Слишком короткие чанки теряют контекст, слишком длинные — размывают семантику. Оптимальный размер — 256–512 токенов для английского, 128–256 для русского. Мы используем рекурсивное разбиение с перекрытием 10–20% для сохранения граничных смыслов. Без этого даже лучшие модели эмбеддингов дают мисклик до 30%.

Как выбрать модель эмбеддингов?

Self-hosted модели (BGE-M3) в 1.5–2 раза быстрее OpenAI API на больших объёмах, но требуют выделенного GPU. Для проектов с конфиденциальными данными это единственный вариант.

Как обеспечить качество эмбеддингов?

Мы комбинируем несколько техник:

• Дедупликация по хешу содержимого — исключаем двойную индексацию одинаковых документов.
• Усечение слишком длинных текстов — не более 8000 токенов на вход модели.
• Добавление метаданных — каждое векторное представление содержит текстовую превью и атрибуты для фильтрации.
• Мониторинг дрейфа — каждую неделю пересчитываем распределение эмбеддингов и при необходимости переиндексируем.

Как настроить Embedding Pipeline: 5 шагов

1. Выбор модели и хранилища. Определитесь, будете ли вы использовать API (OpenAI, Cohere) или self-hosted (BGE-M3). Выберите векторную БД: Qdrant, Pinecone, Weaviate.
2. Проектирование chunking. Настройте размер чанка и перекрытие под ваш язык и тип контента. Протестируйте на выборке из 1000 документов.
3. Реализация дедупликации. Добавьте хеширование SHA-256 для пропуска дубликатов при инкрементальной индексации.
4. Батчевая обработка. Настройте размер батча (100 для API, 500–1000 для self-hosted). Реализуйте retry с экспоненциальной задержкой.
5. Мониторинг и тестирование. Запустите нагрузочное тестирование на 100K+ запросов, измерьте p99 latency и процент ошибок.

Базовый пайплайн

import asyncio
import hashlib
from typing import Any
import numpy as np
from openai import AsyncOpenAI
import qdrant_client
from qdrant_client.models import Distance, VectorParams, PointStruct

class EmbeddingPipeline:
    def __init__(self, collection_name: str,
                 embedding_model: str = "text-embedding-3-small"):
        self.oai = AsyncOpenAI()
        self.qdrant = qdrant_client.QdrantClient(host="localhost", port=6333)
        self.model = embedding_model
        self.collection = collection_name
        self.batch_size = 100
        self._init_collection()

    def _init_collection(self):
        dims = {"text-embedding-3-small": 1536, "text-embedding-3-large": 3072}
        dim = dims.get(self.model, 1536)

        existing = [c.name for c in self.qdrant.get_collections().collections]
        if self.collection not in existing:
            self.qdrant.create_collection(
                collection_name=self.collection,
                vectors_config=VectorParams(size=dim, distance=Distance.COSINE)
            )

    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
        """Батч-генерация эмбеддингов с retry"""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = await self.oai.embeddings.create(
                    input=texts,
                    model=self.model
                )
                return [item.embedding for item in response.data]
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

    async def index_documents(self, documents: list[dict],
                               text_field: str = "content") -> dict:
        """
        documents: список словарей с текстом и метаданными
        text_field: ключ с текстом для эмбеддинга
        """
        total = len(documents)
        indexed = 0
        skipped = 0

        # Проверка уже проиндексированных (дедупликация по хешу)
        existing_hashes = self._get_existing_hashes()

        # Фильтрация дубликатов
        new_docs = []
        for doc in documents:
            content_hash = hashlib.sha256(
                doc.get(text_field, "").encode()
            ).hexdigest()[:16]
            if content_hash not in existing_hashes:
                doc['_hash'] = content_hash
                new_docs.append(doc)
            else:
                skipped += 1

        # Батчевая обработка
        for i in range(0, len(new_docs), self.batch_size):
            batch = new_docs[i:i + self.batch_size]
            texts = [doc.get(text_field, "") for doc in batch]

            # Усечение слишком длинных текстов
            texts = [t[:8000] for t in texts]

            embeddings = await self.embed_batch(texts)

            points = []
            for j, (doc, embedding) in enumerate(zip(batch, embeddings)):
                point_id = int(hashlib.sha256(
                    doc.get('id', str(i + j)).encode()
                ).hexdigest()[:8], 16) % (2**63)

                payload = {k: v for k, v in doc.items()
                           if k != text_field and not k.startswith('_')}
                payload['text_preview'] = doc.get(text_field, "")[:200]
                payload['_hash'] = doc['_hash']

                points.append(PointStruct(
                    id=point_id,
                    vector=embedding,
                    payload=payload
                ))

            self.qdrant.upsert(
                collection_name=self.collection,
                points=points
            )
            indexed += len(batch)

        return {'indexed': indexed, 'skipped': skipped, 'total': total}

    def _get_existing_hashes(self) -> set:
        """Получение хешей уже проиндексированных документов"""
        try:
            scroll_result = self.qdrant.scroll(
                collection_name=self.collection,
                scroll_filter=None,
                with_payload=["_hash"],
                limit=10000
            )
            return {point.payload.get('_hash', '') for point in scroll_result[0]}
        except Exception:
            return set()

Продвинутый пайплайн с поддержкой разных типов данных

class MultimodalEmbeddingPipeline:
    """Индексация текстов, таблиц и изображений"""

    def __init__(self):
        self.text_pipeline = EmbeddingPipeline("text_collection")
        self.table_pipeline = EmbeddingPipeline("table_collection")

    async def process_table(self, df: pd.DataFrame,
                              table_name: str,
                              description: str = "") -> list[dict]:
        """Таблица → набор индексируемых записей"""
        documents = []

        # Описание схемы таблицы
        schema_text = f"Table: {table_name}. {description}\n"
        schema_text += f"Columns: {', '.join(df.columns.tolist())}\n"
        schema_text += f"Sample data:\n{df.head(3).to_string()}"

        documents.append({
            'id': f"{table_name}_schema",
            'content': schema_text,
            'type': 'table_schema',
            'table_name': table_name
        })

        # Каждая строка как отдельная запись (для маленьких таблиц)
        if len(df) <= 1000:
            for idx, row in df.iterrows():
                row_text = f"Row {idx} in {table_name}: " + \
                           ", ".join([f"{col}={val}" for col, val in row.items()])
                documents.append({
                    'id': f"{table_name}_row_{idx}",
                    'content': row_text,
                    'type': 'table_row',
                    'table_name': table_name,
                    'row_index': idx
                })

        return documents

    async def process_structured_content(self, content: dict,
                                          content_type: str) -> list[dict]:
        """JSON/structured data → эмбеддинги"""
        import json

        # Плоское текстовое представление
        flat_text = self._flatten_dict(content)

        return [{
            'id': content.get('id', hashlib.md5(flat_text.encode()).hexdigest()),
            'content': flat_text,
            'type': content_type,
            'raw': json.dumps(content, ensure_ascii=False)[:1000]
        }]

    def _flatten_dict(self, d: dict, prefix: str = "") -> str:
        """Рекурсивное преобразование словаря в текст"""
        parts = []
        for key, value in d.items():
            full_key = f"{prefix}.{key}" if prefix else key
            if isinstance(value, dict):
                parts.append(self._flatten_dict(value, full_key))
            elif isinstance(value, list):
                parts.append(f"{full_key}: {', '.join(str(v) for v in value[:10])}")
            else:
                parts.append(f"{full_key}: {value}")
        return ". ".join(parts)

Поиск по проиндексированным данным

    async def semantic_search(self, query: str,
                               limit: int = 5,
                               score_threshold: float = 0.7) -> list[dict]:
        query_embedding = (await self.text_pipeline.embed_batch([query]))[0]

        results = self.text_pipeline.qdrant.search(
            collection_name=self.text_pipeline.collection,
            query_vector=query_embedding,
            limit=limit,
            score_threshold=score_threshold,
            with_payload=True
        )

        return [
            {
                'score': hit.score,
                'text': hit.payload.get('text_preview', ''),
                'metadata': {k: v for k, v in hit.payload.items()
                             if k not in ['text_preview', '_hash']}
            }
            for hit in results
        ]

Производительность пайплайна

Инкрементальное обновление через хеширование позволяет экономить до 90% затрат при ежедневных обновлениях.

Типичные ошибки при построении пайплайна

• Использование модели эмбеддингов, не подходящей под язык данных — мультиязычные модели (multilingual-e5-large) нужны для неанглийского контента.
• Слишком большие чанки (>512 токенов) снижают точность поиска на 15–20%.
• Отсутствие дедупликации ведёт к дублированию векторов и раздуванию хранилища.
• Игнорирование метаданных — без фильтрации по атрибутам поиск становится медленнее и менее точным.

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

Мы разрабатываем пайплайн под ключ:

• Анализ данных – определение оптимального chunking и модели.
• Проектирование архитектуры – выбор векторного хранилища, конфигурация батчей.
• Реализация – написание production-кода с обработкой ошибок и мониторингом.
• Тестирование – нагрузочное тестирование (100K+ запросов) и проверка качества эмбеддингов.
• Документация и обучение – Model Card, инструкция по эксплуатации, обучение команды.

Сроки: от 2 недель до 2 месяцев в зависимости от сложности. Стоимость рассчитывается индивидуально после аудита ваших данных.

Свяжитесь с нами для консультации и оценки проекта. Закажите разработку пайплайна – получите готовую инфраструктуру для семантического поиска. Получите консультацию по вашему проекту.

Согласно документации Qdrant, выбор векторного хранилища критичен для задержек. Мы используем Qdrant как наиболее производительное решение для production.

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-пайплайна под ключ.