Разработка 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.