How to Build a Semantic Search Engine with Qdrant and OpenAI Embeddings

Table of Contents

• How to Build a Semantic Search Engine with Qdrant and OpenAI Embeddings
  • Understanding the Architecture: Why Qdrant and text-embedding [1]-3
  • Prerequisites and Environment Setup
• Create a virtual environment
• Install dependencies
• Start Qdrant locally with Docker
  • Core Implementation: Building the Semantic Search Pipeline
    • Step 1: Document Chunking and Embedding
    • Step 2: Embedding with Retry Logic and Rate Limiting
    • Step 3: Qdrant Collection Setup and Indexing

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

Semantic search has transformed how we interact with unstructured data. Unlike keyword-based search that matches exact terms, semantic search understands the meaning behind queries, returning results that are conceptually similar even when they share no common words. In this tutorial, we'll build a production-grade semantic search engine using Qdrant as our vector database [3] and OpenAI's text-embedding-3 models for generating embeddings.

By the end, you'll have a fully functional search system that can handle millions of documents with sub-50ms query times, complete with proper error handling, batch processing, and performance optimization.

Understanding the Architecture: Why Qdrant and text-embedding-3

Before diving into code, let's understand why this combination is powerful for production systems.

Qdrant is a vector similarity search engine written in Rust, designed for high-performance ANN (Approximate Nearest Neighbor) search. As of June 2026, Qdrant supports multiple indexing algorithms including HNSW (Hierarchical Navigable Small World) and payload filtering, making it suitable for hybrid search scenarios. Its key advantages include:

• Built-in filtering on payload metadata alongside vector search
• Support for multiple distance metrics (Cosine, Dot, Euclidean)
• Horizontal scaling with distributed deployment
• gRPC and REST APIs for client communication

OpenAI's text-embedding-3 models, released in early 2024, represent a significant improvement over their predecessors. According to OpenAI's documentation, text-embedding-3-small costs $0.02 per 1M tokens while text-embedding-3-large costs $0.13 per 1M tokens. These models support dimensions of 256, 512, or 1536 (for small) and 256, 512, 1024, or 3072 (for large), allowing you to trade off between storag [2]e efficiency and accuracy.

The architecture works as follows:

1. Documents are chunked and embedded using OpenAI [7]'s API
2. Embeddings are stored in Qdrant with their original text and metadata
3. User queries are embedded using the same model
4. Qdrant performs ANN search to find the nearest neighbors
5. Results are returned with similarity scores and metadata

Prerequisites and Environment Setup

You'll need Python 3.10+ and a running Qdrant instance. We'll use Docker for local development, but Qdrant also offers a cloud service at qdrant.io.

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install qdrant-client==1.12.0 openai==1.55.0 python-dotenv==1.0.1 tiktoken==0.8.0

# Start Qdrant locally with Docker
docker run -d --name qdrant -p 6333:6333 -p 6334:6334 qdrant/qdrant:v1.12.0

The Qdrant client library version 1.12.0 is the latest stable release as of June 2026. The tiktoken library is optional but useful for managing token limits when chunking documents.

Create a .env file in your project root:

OPENAI_API_KEY=your_openai_api_key_here
QDRANT_HOST=localhost
QDRANT_PORT=6333

Core Implementation: Building the Semantic Search Pipeline

Step 1: Document Chunking and Embedding

The first challenge in semantic search is handling documents that exceed the model's token limit. OpenAI's text-embedding-3-small has a maximum context length of 8191 tokens. We need to chunk documents intelligently, preserving semantic boundaries where possible.

import os
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
import tiktoken
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.http import models
from dotenv import load_dotenv

load_dotenv()

@dataclass
class Document:
    """Represents a document with its metadata and content."""
    id: str
    content: str
    metadata: Dict
    source: str  # e.g., "pdf", "web", "api"

class DocumentChunker:
    """Handles intelligent document chunking with overlap."""

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoder = tiktoken.get_encoding("cl100k_base")

    def chunk_document(self, document: Document) -> List[Dict]:
        """
        Split a document into overlapping chunks.
        Returns list of dicts with chunk_id, content, and metadata.
        """
        tokens = self.encoder.encode(document.content)
        chunks = []

        # Handle documents shorter than chunk_size
        if len(tokens) <= self.chunk_size:
            return [{
                "chunk_id": f"{document.id}_0",
                "content": document.content,
                "metadata": document.metadata,
                "token_count": len(tokens)
            }]

        # Sliding window with overlap
        start = 0
        chunk_index = 0
        while start < len(tokens):
            end = min(start + self.chunk_size, len(tokens))
            chunk_tokens = tokens[start:end]
            chunk_text = self.encoder.decode(chunk_tokens)

            chunks.append({
                "chunk_id": f"{document.id}_{chunk_index}",
                "content": chunk_text,
                "metadata": document.metadata,
                "token_count": len(chunk_tokens)
            })

            start += self.chunk_size - self.chunk_overlap
            chunk_index += 1

        return chunks

Why this matters in production: Without proper chunking, you'll either lose context from truncation or waste tokens on irrelevant content. The overlap ensures that concepts spanning chunk boundaries aren't lost. The tiktoken encoder matches OpenAI's tokenization, giving accurate token counts.

Step 2: Embedding with Retry Logic and Rate Limiting

OpenAI's API has rate limits that vary by tier. According to OpenAI's documentation, the default rate limit for text-embedding-3-small is 3,000 RPM (requests per minute) for Tier 1 users. We need robust retry logic.

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APIError

class EmbeddingService:
    """Handles embedding generation with retry logic and batching."""

    def __init__(self, model: str = "text-embedding-3-small", dimensions: int = 512):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.dimensions = dimensions
        self.max_batch_size = 100  # OpenAI's batch limit for embeddings

        # Validate dimensions based on model
        valid_dims = {
            "text-embedding-3-small": [256, 512, 1536],
            "text-embedding-3-large": [256, 512, 1024, 3072]
        }
        if model in valid_dims and dimensions not in valid_dims[model]:
            raise ValueError(f"Invalid dimensions {dimensions} for model {model}. "
                           f"Valid options: {valid_dims[model]}")

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30),
        retry=retry_if_exception_type((RateLimitError, APIError))
    )
    def embed_text(self, text: str) -> List[float]:
        """Embed a single text string with retry logic."""
        response = self.client.embeddings.create(
            model=self.model,
            input=text,
            dimensions=self.dimensions
        )
        return response.data[0].embedding

    def embed_batch(self, texts: List[str]) -> List[List[float]]:
        """
        Embed a batch of texts, handling rate limits and batch size limits.
        Returns list of embedding vectors.
        """
        all_embeddings = []

        # Process in batches to respect API limits
        for i in range(0, len(texts), self.max_batch_size):
            batch = texts[i:i + self.max_batch_size]

            try:
                response = self.client.embeddings.create(
                    model=self.model,
                    input=batch,
                    dimensions=self.dimensions
                )
                # Sort by index to maintain order
                sorted_embeddings = sorted(response.data, key=lambda x: x.index)
                all_embeddings.extend([e.embedding for e in sorted_embeddings])

                # Rate limiting: sleep between batches
                if i + self.max_batch_size < len(texts):
                    time.sleep(0.5)  # 500ms between batches

            except RateLimitError as e:
                print(f"Rate limit hit at batch {i//self.max_batch_size}. "
                      f"Waiting 10 seconds..")
                time.sleep(10)
                # Retry this batch
                response = self.client.embeddings.create(
                    model=self.model,
                    input=batch,
                    dimensions=self.dimensions
                )
                sorted_embeddings = sorted(response.data, key=lambda x: x.index)
                all_embeddings.extend([e.embedding for e in sorted_embeddings])

        return all_embeddings

Edge case handling:

• Empty inputs: The OpenAI API returns an error for empty strings. You should validate inputs before sending.
• Rate limits: The exponential backoff with jitter prevents thundering herd problems.
• Batch ordering: OpenAI doesn't guarantee response order, so we sort by index.

Step 3: Qdrant Collection Setup and Indexing

Qdrant requires explicit collection creation with defined schema. This is where we configure the vector size, distance metric, and indexing algorithm.

class QdrantVectorStore:
    """Manages Qdrant collection operations and vector search."""

    def __init__(self, collection_name: str = "documents"):
        self.client = QdrantClient(
            host=os.getenv("QDRANT_HOST", "localhost"),
            port=int(os.getenv("QDRANT_PORT", "6333"))
        )
        self.collection_name = collection_name
        self.embedding_service = EmbeddingService()

    def create_collection(self, vector_size: int = 512):
        """
        Create or recreate the collection with optimal settings.

        HNSW parameters explained:
        - m: Number of bi-directional links per node (higher = more memory, better recall)
        - ef_construct: Size of dynamic candidate list for construction (higher = better quality)
        - full_scan_threshold: Threshold for switching to exact search
        """
        # Delete existing collection if it exists
        try:
            self.client.delete_collection(self.collection_name)
        except Exception:
            pass  # Collection doesn't exist

        self.client.create_collection(
            collection_name=self.collection_name,
            vectors_config=models.VectorParams(
                size=vector_size,
                distance=models.Distance.COSINE,
                hnsw_config=models.HnswConfigDiff(
                    m=16,  # Default: 16, range: 4-64
                    ef_construct=100,  # Default: 100, range: 4-1000
                    full_scan_threshold=10000  # Use exact search for small collections
                )
            ),
            optimizers_config=models.OptimizersConfigDiff(
                default_segment_number=2,  # Number of segments for parallel processing
                memmap_threshold_kb=20000  # Use memmap for segments > 20MB
            )
        )
        print(f"Collection '{self.collection_name}' created with vector size {vector_size}")

    def index_documents(self, documents: List[Document], chunker: DocumentChunker):
        """
        Index a list of documents into Qdrant.
        Handles chunking, embedding, and upserting in batches.
        """
        all_points = []

        for doc in documents:
            chunks = chunker.chunk_document(doc)

            # Extract texts for batch embedding
            chunk_texts = [chunk["content"] for chunk in chunks]

            # Generate embeddings in batch
            embeddings = self.embedding_service.embed_batch(chunk_texts)

            # Create Qdrant points
            for chunk, embedding in zip(chunks, embeddings):
                point = models.PointStruct(
                    id=hash(chunk["chunk_id"]),  # Use hash for consistent IDs
                    vector=embedding,
                    payload={
                        "chunk_id": chunk["chunk_id"],
                        "content": chunk["content"],
                        "source": doc.source,
                        "metadata": json.dumps(chunk["metadata"]),
                        "token_count": chunk["token_count"]
                    }
                )
                all_points.append(point)

        # Upsert in batches of 100 (Qdrant's recommended batch size)
        batch_size = 100
        for i in range(0, len(all_points), batch_size):
            batch = all_points[i:i + batch_size]
            self.client.upsert(
                collection_name=self.collection_name,
                points=batch,
                wait=True  # Wait for indexing to complete
            )
            print(f"Indexed {min(i + batch_size, len(all_points))}/{len(all_points)} points")

        print(f"Successfully indexed {len(all_points)} chunks")
        return len(all_points)

Important architectural decisions:

• Hashing for point IDs: Using Python's built-in hash() is deterministic but not cryptographically secure. For production, consider using hashlib.sha256 or UUIDs to avoid collisions.
• Cosine distance: This is the standard for semantic search with OpenAI embeddings, as they are normalized to unit length.
• Wait parameter: Setting wait=True ensures consistency but increases latency. For high-throughput systems, use wait=False and implement eventual consistency checks.

Step 4: Semantic Search with Payload Filtering

The real power of Qdrant comes from combining vector similarity with payload filtering. This enables hybrid search scenarios like "find documents similar to X that were created after date Y."

class SemanticSearchEngine:
    """High-level search engine combining embedding and Qdrant search."""

    def __init__(self, vector_store: QdrantVectorStore):
        self.vector_store = vector_store
        self.embedding_service = vector_store.embedding_service

    def search(
        self,
        query: str,
        limit: int = 10,
        score_threshold: Optional[float] = None,
        filter_conditions: Optional[Dict] = None
    ) -> List[Dict]:
        """
        Perform semantic search with optional filtering.

        Args:
            query: Natural language query string
            limit: Maximum number of results
            score_threshold: Minimum similarity score (0-1)
            filter_conditions: Qdrant filter conditions for payload filtering

        Returns:
            List of search results with content, metadata, and scores
        """
        # Generate query embedding
        query_vector = self.embedding_service.embed_text(query)

        # Build search request
        search_params = {
            "collection_name": self.vector_store.collection_name,
            "query_vector": query_vector,
            "limit": limit,
            "with_payload": True,
            "with_vectors": False  # Don't return vectors to save bandwidth
        }

        # Add optional filters
        if score_threshold is not None:
            search_params["score_threshold"] = score_threshold

        if filter_conditions:
            search_params["query_filter"] = models.Filter(**filter_conditions)

        # Execute search
        results = self.vector_store.client.search(**search_params)

        # Format results
        formatted_results = []
        for result in results:
            formatted_results.append({
                "chunk_id": result.payload["chunk_id"],
                "content": result.payload["content"],
                "source": result.payload["source"],
                "metadata": json.loads(result.payload["metadata"]),
                "score": result.score,
                "token_count": result.payload["token_count"]
            })

        return formatted_results

    def search_with_filters(
        self,
        query: str,
        source: Optional[str] = None,
        date_range: Optional[tuple] = None,
        limit: int = 10
    ) -> List[Dict]:
        """
        Convenience method for common filter patterns.

        Example: search_with_filters("machine learning", source="arxiv", limit=5)
        """
        must_conditions = []

        if source:
            must_conditions.append(
                models.FieldCondition(
                    key="source",
                    match=models.MatchValue(value=source)
                )
            )

        if date_range:
            start_date, end_date = date_range
            must_conditions.append(
                models.FieldCondition(
                    key="metadata.created_at",
                    range=models.Range(
                        gte=start_date,
                        lte=end_date
                    )
                )
            )

        filter_conditions = None
        if must_conditions:
            filter_conditions = {
                "must": must_conditions
            }

        return self.search(query, limit=limit, filter_conditions=filter_conditions)

Performance considerations:

• Score threshold: Setting a minimum score (e.g., 0.7) can dramatically improve result quality by filtering out noise.
• Payload indexing: For efficient filtering, ensure you create payload indexes in Qdrant for frequently filtered fields. This is done during collection creation.
• Vector storage: Setting with_vectors=False reduces response size by ~90% since vectors are large (512-3072 floats).

Step 5: Production Deployment and Monitoring

For production use, you'll want to add monitoring, caching, and error handling.

import logging
from functools import lru_cache
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class ProductionSearchEngine(SemanticSearchEngine):
    """Extended search engine with caching and monitoring."""

    def __init__(self, vector_store: QdrantVectorStore, cache_size: int = 1000):
        super().__init__(vector_store)
        self.cache_size = cache_size
        self.search_count = 0
        self.total_latency = 0.0

    @lru_cache(maxsize=1000)
    def _cached_embedding(self, query: str) -> tuple:
        """Cache embeddings for frequently searched queries."""
        return tuple(self.embedding_service.embed_text(query))

    def search(self, query: str, **kwargs) -> List[Dict]:
        """Monitored search with caching."""
        start_time = datetime.now()

        # Use cached embedding if available
        query_vector = list(self._cached_embedding(query))

        # Override query vector in search params
        search_params = {
            "collection_name": self.vector_store.collection_name,
            "query_vector": query_vector,
            "limit": kwargs.get("limit", 10),
            "with_payload": True,
            "with_vectors": False
        }

        # Add optional filters
        if "score_threshold" in kwargs:
            search_params["score_threshold"] = kwargs["score_threshold"]
        if "filter_conditions" in kwargs:
            search_params["query_filter"] = models.Filter(**kwargs["filter_conditions"])

        # Execute search with error handling
        try:
            results = self.vector_store.client.search(**search_params)
            latency = (datetime.now() - start_time).total_seconds()

            # Update metrics
            self.search_count += 1
            self.total_latency += latency

            # Log slow queries
            if latency > 1.0:
                logger.warning(f"Slow query detected: {latency:.2f}s for query: {query[:50]}..")

            # Format results
            formatted_results = []
            for result in results:
                formatted_results.append({
                    "chunk_id": result.payload["chunk_id"],
                    "content": result.payload["content"],
                    "source": result.payload["source"],
                    "metadata": json.loads(result.payload["metadata"]),
                    "score": result.score,
                    "token_count": result.payload["token_count"]
                })

            return formatted_results

        except Exception as e:
            logger.error(f"Search failed for query '{query[:50]}..': {str(e)}")
            raise

    def get_metrics(self) -> Dict:
        """Return performance metrics."""
        avg_latency = self.total_latency / max(self.search_count, 1)
        return {
            "total_searches": self.search_count,
            "average_latency_ms": avg_latency * 1000,
            "cache_size": self._cached_embedding.cache_info().currsize
        }

Complete Example: Putting It All Together

Here's how to use the entire system end-to-end:

def main():
    # Initialize components
    chunker = DocumentChunker(chunk_size=500, chunk_overlap=50)
    vector_store = QdrantVectorStore(collection_name="tech_docs")

    # Create collection with 512-dimensional vectors
    vector_store.create_collection(vector_size=512)

    # Sample documents
    documents = [
        Document(
            id="doc1",
            content="""Machine learning is a subset of artificial intelligence that enables 
            systems to learn and improve from experience without being explicitly programmed. 
            Deep learning, a subset of machine learning, uses neural networks with multiple 
            layers to progressively extract higher-level features from raw input.""",
            metadata={"created_at": "2024-01-15", "author": "John Doe", "category": "AI"},
            source="blog"
        ),
        Document(
            id="doc2",
            content="""Vector databases like Qdrant are optimized for storing and querying 
            high-dimensional vectors. They use approximate nearest neighbor (ANN) algorithms 
            like HNSW to perform similarity search at scale, making them ideal for semantic 
            search applications.""",
            metadata={"created_at": "2024-03-20", "author": "Jane Smith", "category": "Databases"},
            source="documentation"
        ),
        Document(
            id="doc3",
            content="""Python's asyncio library provides a framework for writing concurrent 
            code using the async/await syntax. It's particularly useful for I/O-bound 
            operations like API calls and database queries, allowing you to handle thousands 
            of connections simultaneously.""",
            metadata={"created_at": "2024-06-10", "author": "Bob Johnson", "category": "Programming"},
            source="tutorial"
        )
    ]

    # Index documents
    vector_store.index_documents(documents, chunker)

    # Create search engine
    search_engine = ProductionSearchEngine(vector_store)

    # Perform searches
    print("\n=== Semantic Search Results ===")
    results = search_engine.search("neural networks and deep learning", limit=3)
    for r in results:
        print(f"Score: {r['score']:.4f} | Source: {r['source']}")
        print(f"Content: {r['content'][:100]}..")
        print()

    # Search with filters
    print("=== Filtered Search ===")
    filtered_results = search_engine.search_with_filters(
        "database optimization",
        source="documentation",
        limit=2
    )
    for r in filtered_results:
        print(f"Score: {r['score']:.4f} | Source: {r['source']}")
        print(f"Content: {r['content'][:100]}..")
        print()

    # Print metrics
    print("=== Performance Metrics ===")
    print(json.dumps(search_engine.get_metrics(), indent=2))

if __name__ == "__main__":
    main()

Edge Cases and Production Considerations

1. Handling Empty or Malformed Inputs

Always validate inputs before processing:

def validate_query(query: str) -> str:
    """Validate and sanitize search queries."""
    if not query or not query.strip():
        raise ValueError("Query cannot be empty")
    if len(query) > 1000:
        raise ValueError("Query exceeds maximum length of 1000 characters")
    # Remove potentially dangerous characters
    sanitized = query.strip()[:1000]
    return sanitized

2. Managing Token Limits

OpenAI's embedding models have a maximum input length. For text-embedding-3-small, this is 8191 tokens. Always truncate or chunk inputs:

def truncate_to_token_limit(text: str, max_tokens: int = 8000) -> str:
    """Truncate text to fit within token limits, leaving room for overhead."""
    encoder = tiktoken.get_encoding("cl100k_base")
    tokens = encoder.encode(text)
    if len(tokens) > max_tokens:
        tokens = tokens[:max_tokens]
        text = encoder.decode(tokens)
    return text

3. Qdrant Connection Pooling

For high-throughput applications, use connection pooling:

from qdrant_client import QdrantClient
from qdrant_client.http import ApiClient

# Create a client with connection pooling
client = QdrantClient(
    host="localhost",
    port=6333,
    grpc_port=6334,
    prefer_grpc=True,  # gRPC is faster for bulk operations
    https=False,
    api_key=None,
    timeout=30,
    limits=ApiClient(
        pool_size=10,  # Connection pool size
        pool_connections=10,
        pool_maxsize=20
    )
)

What's Next

You now have a production-ready semantic search engine. Here are some natural extensions:

1. Hybrid search: Combine semantic search with keyword-based BM25 scoring using Qdrant's built-in sparse vectors or an external search engine like Elasticsearch
2. Multi-modal search: Extend to image or audio search by using appropriate embedding models
3. Real-time indexing: Implement streaming ingestion with Apache Kafka or RabbitMQ
4. A/B testing framework: Compare different embedding models or chunking strategies

For further reading, check out our guides on vector database optimization and embedding model selection.

The complete code for this tutorial is available on GitHub. Remember to monitor your OpenAI API usage and Qdrant resource consumption as you scale. With proper configuration, this system can handle millions of documents with sub-50ms query latency.

---