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

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