Embedding Pipeline Development for Data Indexing

Without a properly built embedding pipeline, a RAG system returns irrelevant answers or loses context due to incorrect chunking. We build production-ready pipelines that ensure p99 latency under 50 ms when indexing 1 million documents. Indexing via the OpenAI API text-embedding-3-small costs around $2 per 1M documents, while self-hosted solutions on an A10G deliver nearly double the throughput and cost about $0.5 per hour of GPU. With 5+ years of experience in ML engineering and over 15 deployed semantic search systems, we guarantee reliable infrastructure.

How to Choose the Right Embedding Model?

Text chunking is a critical step. Too short chunks lose context; too long ones blur semantics. The optimal size is 256–512 tokens for English and 128–256 for Russian. We use recursive splitting with 10–20% overlap to preserve boundary meanings. Without this, even the best embedding models produce up to 30% misclicks. The choice of embedding model depends on budget, latency, and data sensitivity. The table below compares popular options:

Self-hosted models (BGE-M3) are 1.5–2 times faster than the OpenAI API on large volumes but require a dedicated GPU. For sensitive data, this is the only option.

How to Ensure Embedding Quality?

We combine several techniques: content hash deduplication to prevent double indexing, truncation of texts longer than 8000 tokens, addition of metadata (text preview and attributes) to each vector, and drift monitoring with weekly recalculation of the embedding distribution.

Pipeline Implementation

Steps to Set Up

1. Choose a model and storage: API (OpenAI, Cohere) or self-hosted (BGE-M3). Select a vector database: Qdrant, Pinecone, Weaviate.
2. Design chunking: set chunk size and overlap for your language and content type. Test on a sample of 1000 documents.
3. Implement deduplication: use SHA-256 hashing to skip duplicates during incremental indexing.
4. Batch processing: configure batch size (100 for API, 500–1000 for self-hosted) with retries using exponential backoff. According to OpenAI Embedding API best practices, a batch size of 100 is recommended.
5. Monitor and test: run load testing on 100K+ queries, measure p99 latency and error rate.

Basic Code Example

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]]: 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: 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() 

Advanced: Multimodal and Structured Data

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]: 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) 

Search Integration

 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 ] 

Performance Benchmarks

Incremental updating via hashing saves up to 90% of costs during daily updates. For example, total cost for 1M documents: ~$2 (API) or ~$0.5 GPU per hour for self-hosted.

Common Pitfalls

• Using an embedding model unsuitable for the data language – multilingual models (multilingual-e5-large) are needed for non-English content.
• Chunks too large (>512 tokens) reduce search accuracy by 15–20%.
• Lack of deduplication leads to duplicate vectors and storage bloat.
• Ignoring metadata – without attribute filtering, search becomes slower and less precise.

Our Expertise and Deliverables

With over 5 years of experience in ML engineering and more than 15 deployed semantic search systems, we guarantee reliable pipelines. We provide a written guarantee for p99 latency under 50 ms and cost savings up to 90% with incremental updates. Our turnkey service includes:

• Data analysis – determining optimal chunking and model.
• Architecture design – choosing a vector store, configuring batches.
• Implementation – writing production code with error handling and monitoring.
• Testing – load testing (100K+ requests) and embedding quality checks.
• Documentation and training – Model Card, operation manual, and team training.

Timeline: from 2 weeks to 2 months depending on complexity. Cost is calculated individually after an audit of your data. Contact us for a consultation and project evaluation.