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

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

Без правильно побудованого пайплайну ембедінгів RAG-система видає нерелевантні відповіді або втрачає сенс через некоректний chunking. Ми будуємо production-ready пайплайни, які забезпечують латентність p99 менше 50 мс при індексації 1 мільйона документів. Наш досвід — 5+ років у ML-інженерії та понад 15 впроваджених систем семантичного пошуку. Компанія працює на ринку ML-рішень понад 5 років, реалізувала 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. Для проєктів з конфіденційними даними це єдиний варіант. OpenAI text-embedding-3-small має пропускну здатність у 2.7 рази вищу, ніж text-embedding-3-large, при вдвічі меншій вартості.

Як забезпечити якість ембедінгів?

Ми комбінуємо кілька технік:

• Дедуплікація за хешем вмісту — виключаємо подвійну індексацію однакових документів.
• Усічення занадто довгих текстів — не більше 8000 токенів на вхід моделі.
• Додавання метаданих — кожне векторне представлення містить текстове прев'ю та атрибути для фільтрації.
• Моніторинг дрейфу — щотижня перераховуємо розподіл ембедінгів і при необхідності переіндексуємо.
• Для покращення якості ембедінгів ми використовуємо контрастивне навчання на парних даних, нормалізацію L2 вихідних векторів та Product Quantization для зменшення розміру індексу без значної втрати точності.

Як налаштувати 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
        ]

Продуктивність пайплайну

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

• Використання моделі ембедінгів, яка не підходить під мову даних — мультимовні моделі (multilingual-e5-large) потрібні для неанглійського контенту.
• Занадто великі чанки (>512 токенів) знижують точність пошуку на 15–20%.
• Відсутність дедуплікації веде до дублювання векторів та роздування сховища.
• Ігнорування метаданих — без фільтрації за атрибутами пошук стає повільнішим і менш точним.

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

Ми розробляємо пайплайн під ключ:

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

Ми гарантуємо якість пайплайну: кожен проєкт проходить ретельне тестування на 100K+ запитів. Наші фахівці мають сертифікації AWS та OpenAI.

Терміни: від 2 тижнів до 2 місяців залежно від складності. Вартість розраховується індивідуально після аудиту ваших даних.

Зв'яжіться з нами для консультації та оцінки проєкту. Замовте розробку пайплайну – отримайте готову інфраструктуру для семантичного пошуку. Отримайте консультацію щодо вашого проєкту.

Згідно з документацією Qdrant, вибір векторного сховища критичний для затримок. Ми використовуємо Qdrant як найбільш продуктивне рішення для production.

Чому дата-інжиніринг визначає успіх 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-пайплайну під ключ.