GenAI Data and Knowledge Layer

This document covers the design, development and maintenance of data pipelines to efficiently feed your Large Language Models with your proprietary data.

Table of Contents

• 1. Embeddings and Vector Store Fundamentals
  • 1.1 From Vectors to Semantics
  • 1.2 The Mathematics of Semantic Similarity
  • 1.3 Understanding Vector Spaces Through Visualization
  • 1.4 Comparing Embedding Models
  • 1.5 Evaluating Embedding Quality: MTEB and Beyond
  • 1.6 Demo: Generating Embeddings at Scale
  • 1.7 Demo: Robust Embedding Pipelines
  • 1.8 Demo: Semantic Similarity Search in Practice
• 2. Vector Database Architecture and Operations
  • 2.1 Why Vector Stores?
  • 2.2 Internal Operation: HNSW
  • 2.3 Indexing Algorithms: IVF and Product Quantization
  • 2.4 Large-Scale Storage Strategies
  • 2.5 Comparing Vector Databases
  • 2.6 pgvector and Self-Hosted Solutions
  • 2.7 Schema Design for Vector Databases
  • 2.8 Advanced Query Patterns
  • 2.9 Production Operations
  • 2.10 Monitoring, Index Health and Recovery
• 3. Knowledge Graphs and Structured Knowledge
  • 3.1 Knowledge Graph Fundamentals
  • 3.2 Knowledge Graphs vs Databases vs Vector Stores
  • 3.3 Ontology and Schema Design
  • 3.4 Entity Recognition from Unstructured Text
  • 3.5 Relation Extraction and Knowledge Construction
  • 3.6 Building Knowledge Graphs with LLMs
  • 3.7 Graph Embedding Techniques
  • 3.8 Hybrid Retrieval: Vectors and Graph Traversal
  • 3.9 Querying Knowledge Graphs
  • 3.10 Maintaining and Evolving Knowledge Graphs
• 4. Production Pipelines, Advanced Preprocessing and Synthetic Data
  • 4.1 End-to-End Data Pipeline Architecture
  • 4.2 Chunking Strategies
  • 4.3 Handling Multimodal Content
  • 4.4 Metadata Enrichment and Context Injection
  • 4.5 Pipeline Orchestration
  • 4.6 Synthetic Data Generation and Testing

1. Embeddings and Vector Store Fundamentals

1.1 From Vectors to Semantics

What is an embedding?

At its most fundamental level, an embedding is a numerical translation. It is the process of taking something from our world — a word, a sentence, a concept — and converting it into a language that machines truly understand: a mathematical representation. Concretely, this becomes a list of numbers, a vector. Think of it as a dense, unique fingerprint where the pattern of numbers encapsulates the meaning and context of the original text.

Visualizing dimensions

To visualize this, imagine a simple coordinate on a 2D map. The pair of numbers (x, y) indicates an exact location:

y
↑
|        ●  (3, 4)
|
|
+----------------→ x
       (0,0)

Instead of two dimensions, imagine a space with hundreds or even thousands of dimensions. This is impossible for us to visualize, but a computer can navigate it perfectly. Each dimension in this vector can be seen as capturing a latent feature of the text’s meaning:

Vector arithmetic and analogies

The positions are not arbitrary. Through training on enormous amounts of text, the model learns to organize meanings geometrically. Similar concepts end up in similar regions of the space.

The classic example of vector arithmetic:

King   = [0.8, 0.6, 0.9]
Man    = [0.7, 0.2, 0.4]
Woman  = [0.7, 0.9, 0.4]

King - Man + Woman
= [0.8-0.7+0.7, 0.6-0.2+0.9, 0.9-0.4+0.4]
= [0.8, 1.3, 0.9]

Queen ≈ [0.8, 1.25, 0.88]   ← very close!

The model encoded not only words, but the conceptual relationships between them.

Why embeddings are indispensable

Embeddings form the indispensable data layer for generative AI because they solve a critical problem:

• Human language is symbolic, ambiguous and incredibly rich
• Computers need data in structured numerical form
• Embeddings act as the perfect translator

When you query a modern AI system:

1. It embeds your question (converts it into a vector)
2. It navigates the high-dimensional space
3. It finds the nearest document/concept vectors

This is semantic search — what allows an AI to retrieve a passage about canine veterinary care when you ask “how do I care for my sick dog”, even if no words directly match.

1.2 The Mathematics of Semantic Similarity

We need a reliable “ruler” for our vector space. In mathematics, we have three main candidates.

Euclidean Distance

This is the standard distance we know in everyday life. In high dimensions, the formula extends naturally:

distance = √[(a₁-b₁)² + (a₂-b₂)² + (a₃-b₃)² + ... + (aₙ-bₙ)²]

Characteristics:

• A smaller distance means closer points
• Critical flaw: very sensitive to the magnitude (length) of the vectors
• A long document produces a very long vector, a short query produces a short vector — their distance may be large even if their meaning is similar

Dot Product

We multiply the corresponding components and sum them:

A · B = a₁b₁ + a₂b₂ + a₃b₃ + ... + aₙbₙ

Characteristics:

• A larger positive dot product indicates a greater degree of alignment
• Also influenced by both angle and magnitude
• Two long vectors will have a large dot product even if they are only loosely related

Cosine Similarity — the gold standard

Cosine similarity is designed to solve the magnitude problem. It does not ask “how close are these points?” but “do these vectors point in the same direction?”

              A · B
cos(θ) = ─────────────
           |A| · |B|

This normalization effectively reduces both vectors to unit length, then measures the angle between them.

Why is direction so powerful for language? Because the meaning of text resides more in its orientation in conceptual space than in its distance from an arbitrary origin. Length often correlates with text length or word frequency, but direction encodes semantic content.

Summary of measures:

• Euclidean distance and dot product are misled by magnitude
• Cosine similarity provides the cleanest measure of semantic alignment

💡 Cosine distance = 1 - cosine similarity — transforms the similarity measure into a distance metric (lower = closer).

1.3 Understanding Vector Spaces Through Visualization

The dimensionality problem

Each dimension in an embedding vector represents a latent feature — a subtle axis of meaning such as formality, technicality or emotion. Models use 300, 768 or 1,536 dimensions with ease, but our cognitive limit is three dimensions.

Principal Component Analysis (PCA)

PCA is a classic dimensionality reduction technique. It is a way to find the most informative viewing angles for your complex data.

How it works:

1. It algorithmically identifies principal components — the axes along which data varies most
2. The first component is the direction that captures the greatest variance in the data
3. The second component captures the next greatest variance while being perpendicular to the first
4. By projecting vectors onto only the 2 or 3 first components, we create the best possible “2D snapshot”

What visualization reveals

When you take a diverse collection of texts, generate their embeddings and use PCA to project them into 2D:

• Distinct clusters appear (finance, biology, etc.)
• Sub-clusters form within them (investment vs. personal budgeting)
• This proves that the embedding space is not random — it has a distinct semantic topology
• This reveals relative positioning (e.g., “dog” appears between the “domestic animals” and “wild canids” clusters)

1.4 Comparing Embedding Models

Different models, trained on different data with different objectives, will produce different vector representations.

The ecosystem is dominated by three key approaches:

OpenAI — text-embedding-3

• Strong general-purpose performance
• Notable innovation: the concept of dimension scaling
• Allows shortening output vectors (e.g., from 1,536 to 256 dimensions) with minimal and graceful performance degradation
• Offers a powerful lever to trade a little precision for massive savings in storage and query speed

Cohere — Embed models (embed-english-v3)

• Exceptionally robust and well-suited to complex enterprise-level retrieval
• Strong focus on Retrieval-Augmented Generation (RAG)
• Optimized not only for similarity, but for finding passages that genuinely help an LLM generate a better response
• Native multilingual support

BAAI — BGE (open-source)

• Open-source model from the Beijing Academy of Artificial Intelligence
• Has consistently dominated open-source benchmarks such as MTEB
• Performance rivaling the best proprietary models
• Full data control and no API latency
• Requires infrastructure management (GPU resources, engineering overhead)

How to choose?

📌 The best platform for finding open-source models is Hugging Face.

1.5 Evaluating Embedding Quality: MTEB and Beyond

The Massive Text Embedding Benchmark (MTEB)

MTEB is a comprehensive standardized test for embedding models. It evaluates a model’s ability across several task categories:

Beyond MTEB: domain-specific evaluation

⚠️ MTEB is a starting point. It measures general capability on public datasets. Your application is likely domain-specific.

Domain-specific evaluation is non-negotiable for production systems.

How to proceed:

1. Create a golden dataset — a collection of representative queries with relevant documents identified manually
2. Run your own mini-MTEB — generate embeddings with different candidate models
3. Measure performance with metrics:
   • Normalized Discounted Cumulative Gain (NDCG)
   • Precision@K
   • Recall@K
4. LLM-as-a-judge — use a powerful model to evaluate the relevance of retrieved passages

The model that performs best on your data is the winner, regardless of its MTEB rank.

1.6 Demo: Generating Embeddings at Scale

This demo shows the practical reality of generating embeddings for a real document collection, with a Python pipeline using LangChain.

What a real-world embedding pipeline must do

1. Chunking — Documents are too long for embedding models; split them at paragraph/sentence boundaries
2. Batching — Group chunks into batches and send multiple chunks at once (dramatically faster)
3. Automatic retry — APIs can fail (network issues, rate limits); implement automatic retries with exponential backoff
4. Fail gracefully — If a batch fails, log the error and continue with the rest
5. Metadata tracking — Attach each chunk to its source document

Complete pipeline code

import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from tenacity import retry, stop_after_attempt, wait_exponential
import logging
from typing import List

# 1. CONFIGURATION
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize the embedding model
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=256  # Reduced dimension for efficiency
)

# 2. CHUNKING STRATEGY
def chunk_document(text: str, chunk_size: int = 500, chunk_overlap: int = 50) -> List[str]:
    """Split the document into semantically coherent chunks."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    return text_splitter.split_text(text)

# 3. BATCH PROCESSING WITH RETRY
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def embed_batch_with_retry(texts: List[str]) -> List[List[float]]:
    """Embed a batch of texts with automatic retry logic."""
    try:
        return embeddings.embed_documents(texts)
    except Exception as e:
        logger.error(f"Embedding batch failed: {str(e)}")
        raise

# 4. MAIN PIPELINE
def generate_document_embeddings(
    documents: List[str], batch_size: int = 100
) -> List[tuple]:
    """
    End-to-end pipeline: chunk -> batch -> embed -> store.
    Returns a list of tuples (chunk_text, embedding_vector, metadata).
    """
    all_chunks = []
    metadata = []
    
    # Step 1: Preprocess and chunk all documents
    logger.info(f"Processing {len(documents)} documents...")
    for i, doc in enumerate(documents):
        chunks = chunk_document(doc)
        all_chunks.extend(chunks)
        metadata.extend([
            {"doc_id": i, "chunk_index": j} for j in range(len(chunks))
        ])
    
    logger.info(f"Created {len(all_chunks)} total chunks.")
    
    # Step 2: Process in batches
    results = []
    for batch_start in range(0, len(all_chunks), batch_size):
        batch_end = batch_start + batch_size
        batch_texts = all_chunks[batch_start:batch_end]
        batch_meta = metadata[batch_start:batch_end]
        
        logger.info(
            f"Embedding batch "
            f"{batch_start//batch_size + 1}/"
            f"{(len(all_chunks)-1)//batch_size + 1}"
        )
        
        try:
            batch_embeddings = embed_batch_with_retry(batch_texts)
            
            for text, embedding, meta in zip(
                batch_texts, batch_embeddings, batch_meta
            ):
                results.append((text, embedding, meta))
                
        except Exception as e:
            logger.error(
                f"Failed to process batch starting at index "
                f"{batch_start}: {str(e)}"
            )
    
    logger.info(f"Successfully embedded {len(results)} chunks.")
    return results

# 5. EXECUTION
if __name__ == "__main__":
    sample_documents = [
        "Quantum mechanics fundamentally changed how we understand matter "
        "and energy at the atomic scale. Its principles underpin modern "
        "computing and materials science.",
        "Natural language processing enables machines to interpret human text. "
        "Transformer-based architectures have dramatically improved the "
        "accuracy of text understanding tasks.",
        "Renewable energy sources are essential for reducing carbon emissions. "
        "Solar and wind power costs have fallen sharply over the past decade, "
        "accelerating global adoption."
    ]
    
    embedded_chunks = generate_document_embeddings(
        sample_documents, batch_size=2
    )
    
    if embedded_chunks:
        sample_text, sample_embedding, sample_meta = embedded_chunks[0]
        print(f"\nSample chunk text: {sample_text[:100]}...")
        print(f"Embedding dimension: {len(sample_embedding)}")
        print(f"First 5 values: {sample_embedding[:5]}")
        print(f"Metadata: {sample_meta}")

Typical output: 3 documents processed into 3 chunks, embeddings in 2 batches via the OpenAI API. Each result contains the truncated text, a 256-dimensional embedding vector and metadata linking it back to its source document.

1.7 Demo: Robust Embedding Pipelines

This improved version adds four critical production concepts to the basic pipeline.

The four pillars of a robust pipeline

1. Adaptive Chunking — Different strategies depending on document type (academic articles, code, general text)
2. Intelligent Caching — Storing embeddings in Redis with 30-day expiration, avoiding re-embedding identical content
3. Rate Limiting with backoff — Respects API constraints by estimating token usage
4. Cost Optimization — Real-time token tracking and cost calculation; dimension reduction to 256 instead of 1,536 (≈ 85% saving in storage)

Complete robust pipeline code

import hashlib
import time
from typing import List, Dict
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    SentenceTransformersTokenTextSplitter
)
from langchain_openai import OpenAIEmbeddings
import redis
import tiktoken

class RobustEmbeddingPipeline:
    def __init__(self, model_name="text-embedding-3-small", dimensions=256):
        self.embeddings = OpenAIEmbeddings(
            model=model_name,
            dimensions=dimensions
        )
        self.cache = redis.Redis(
            host='localhost', port=6379, decode_responses=False
        )
        self.tokenizer = tiktoken.get_encoding("cl100k_base")
        
    def _cost_estimator(self, tokens: int) -> float:
        """Estimate the cost in USD for embedding operations."""
        # text-embedding-3-small: $0.02 per 1M tokens
        cost_per_million = 0.02
        return (tokens / 1_000_000) * cost_per_million
    
    def smart_chunker(
        self, text: str, document_type: str = "general"
    ) -> List[str]:
        """Adaptive chunking based on content type."""
        if document_type == "academic":
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=800, chunk_overlap=100,
                separators=["\n## ", "\n# ", "\n\n", "\n", ". ", " "]
            )
        elif document_type == "code":
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=600, chunk_overlap=50,
                separators=["\ndef ", "\nclass ", "\n\n", "\n", " "]
            )
        else:
            splitter = SentenceTransformersTokenTextSplitter(
                chunk_size=512, chunk_overlap=50
            )
        
        chunks = splitter.split_text(text)
        
        # Enrich each chunk with position context
        enriched_chunks = []
        for i, chunk in enumerate(chunks):
            header = f"[Chunk {i+1} of {len(chunks)}] "
            enriched_chunks.append(header + chunk)
            
        return enriched_chunks
    
    def _rate_limiter(self, batch_size: int):
        """Respect API rate limits."""
        tokens_in_batch = batch_size * 512
        if tokens_in_batch > 800000:
            time.sleep(60)
        elif tokens_in_batch > 400000:
            time.sleep(10)
    
    def get_cached_embedding(self, text: str) -> List[float] | None:
        """Check the cache for an existing embedding."""
        text_hash = hashlib.sha256(text.encode()).hexdigest()
        cached = self.cache.get(f"embedding:{text_hash}")
        if cached:
            return eval(cached)  # In production, use proper serialization
        return None
    
    def set_cached_embedding(self, text: str, embedding: List[float]):
        """Store the embedding in cache with 30-day expiration."""
        text_hash = hashlib.sha256(text.encode()).hexdigest()
        self.cache.setex(
            f"embedding:{text_hash}", 2592000, str(embedding)
        )
    
    def process_documents(self, documents: List[Dict]) -> List[Dict]:
        """
        Robust document processing.
        Each dict must have: 'text', 'id', 'type'
        """
        results = []
        total_cost = 0
        total_tokens = 0
        
        for doc in documents:
            chunks = self.smart_chunker(
                doc['text'], doc.get('type', 'general')
            )
            
            for chunk_num, chunk in enumerate(chunks):
                # Check cache first
                cached_embedding = self.get_cached_embedding(chunk)
                if cached_embedding:
                    results.append({
                        'doc_id': doc['id'],
                        'chunk': chunk_num,
                        'embedding': cached_embedding,
                        'source': 'cache'
                    })
                    continue
                
                self._rate_limiter(len(chunks) - chunk_num)
                
                chunk_tokens = len(self.tokenizer.encode(chunk))
                chunk_cost = self._cost_estimator(chunk_tokens)
                total_tokens += chunk_tokens
                total_cost += chunk_cost
                
                try:
                    embedding = self.embeddings.embed_query(chunk)
                    self.set_cached_embedding(chunk, embedding)
                    
                    results.append({
                        'doc_id': doc['id'],
                        'chunk': chunk_num,
                        'embedding': embedding,
                        'source': 'api',
                        'tokens': chunk_tokens
                    })
                except Exception as e:
                    print(
                        f"Failed on document {doc['id']}, "
                        f"chunk {chunk_num}: {str(e)}"
                    )
        
        print(f"Total tokens processed: {total_tokens:,}")
        print(f"Estimated cost: ${total_cost:.4f}")
        print(
            f"Cache hits: "
            f"{len([r for r in results if r['source']=='cache'])}"
        )
        
        return results

# Usage
if __name__ == "__main__":
    pipeline = RobustEmbeddingPipeline(dimensions=256)
    
    sample_docs = [
        {
            'id': 'article_1',
            'text': "Climate change impacts include rising sea levels...",
            'type': 'general'
        },
        {
            'id': 'article_2',
            'text': "def compute_embedding(text):\n    vector = ...",
            'type': 'code'
        }
    ]
    
    results = pipeline.process_documents(sample_docs)

1.8 Demo: Semantic Similarity Search in Practice

Semantic similarity search transforms a natural language question into precise, relevant results. Here is the complete flow, from query formulation to ranked retrieval.

The four steps of search

1. Query Formulation — Transform the question into an embedding with optional expansion (synonyms, reformulations)
2. Similarity Calculation — Cosine similarity between the query embedding and each document embedding, with metadata filter support
3. Ranking and Re-ranking — Initial ranking by semantic similarity, then hybrid re-ranking (boost chunks containing query terms)
4. Relevance Tuning — Iterative parameter adjustment (score threshold) based on feedback

Semantic search engine code

import numpy as np
from typing import List, Dict
from sklearn.preprocessing import normalize

class SemanticSearchEngine:
    def __init__(self, embedding_model, vector_store):
        self.model = embedding_model
        self.store = vector_store
        
    def formulate_query(
        self, user_query: str, query_expansion: bool = False
    ) -> List[float]:
        """Transform the query into an embedding, with optional expansion."""
        base_embedding = self.model.embed_query(user_query)
        
        if query_expansion:
            expanded_queries = [
                user_query,
                f"details about {user_query}",
                f"explanation of {user_query}"
            ]
            embeddings = [
                self.model.embed_query(q) for q in expanded_queries
            ]
            base_embedding = np.mean(embeddings, axis=0)
        
        return base_embedding
    
    def calculate_similarity(
        self, query_embedding, doc_embedding, method="cosine"
    ) -> float:
        """Calculate similarity between query and document."""
        query_vec = np.array(query_embedding)
        doc_vec = np.array(doc_embedding)
        
        if method == "cosine":
            query_norm = normalize(query_vec.reshape(1, -1))[0]
            doc_norm = normalize(doc_vec.reshape(1, -1))[0]
            return np.dot(query_norm, doc_norm)
        elif method == "dot":
            return np.dot(query_vec, doc_vec)
    
    def search(
        self, user_query: str, top_k: int = 5,
        score_threshold: float = 0.7,
        use_metadata_filter: Dict = None
    ) -> List[Dict]:
        """Main search pipeline."""
        query_embedding = self.formulate_query(
            user_query, query_expansion=True
        )
        
        results = []
        for doc_id, doc_data in self.store.items():
            # Apply metadata filter
            if use_metadata_filter:
                if not all(
                    doc_data["metadata"].get(k) == v
                    for k, v in use_metadata_filter.items()
                ):
                    continue
            
            similarity = self.calculate_similarity(
                query_embedding, doc_data["embedding"]
            )
            
            if similarity >= score_threshold:
                results.append({
                    "doc_id": doc_id,
                    "text": doc_data["text"],
                    "score": similarity,
                    "metadata": doc_data["metadata"]
                })
        
        results.sort(key=lambda x: x["score"], reverse=True)
        return self._rerank_results(results[:top_k], user_query)
    
    def _rerank_results(
        self, initial_results: List[Dict], query: str
    ) -> List[Dict]:
        """Hybrid re-ranking: 70% semantic, 30% keyword."""
        for result in initial_results:
            query_terms = set(query.lower().split())
            chunk_terms = set(result["text"].lower().split())
            overlap = (
                len(query_terms.intersection(chunk_terms))
                / len(query_terms)
            )
            result["adjusted_score"] = (
                (0.7 * result["score"]) + (0.3 * overlap)
            )
        
        initial_results.sort(
            key=lambda x: x["adjusted_score"], reverse=True
        )
        return initial_results
    
    def tune_relevance(
        self, query: str, expected_results: List[str],
        tuning_param: str = "score_threshold"
    ) -> Dict:
        """Iterative relevance tuning based on feedback."""
        best_params = {"score_threshold": 0.7, "top_k": 5}
        
        for threshold in [0.6, 0.7, 0.8, 0.85]:
            results = self.search(query, score_threshold=threshold)
            retrieved_ids = [r["doc_id"] for r in results]
            relevant_retrieved = len(
                set(retrieved_ids) & set(expected_results)
            )
            precision = (
                relevant_retrieved / len(retrieved_ids)
                if retrieved_ids else 0
            )
            print(f"Threshold {threshold}: Precision = {precision:.2f}")
            
            if precision > 0.8:
                best_params["score_threshold"] = threshold
                break
        
        return best_params

💡 In production, with real embeddings (OpenAI, Cohere), optimal precision scores typically fall between 0.7 and 0.85.

2. Vector Database Architecture and Operations

2.1 Why Vector Stores?

The limits of traditional databases

Relational or NoSQL databases are designed for a different world. They excel at:

• Exact matches (finding a specific user ID)
• Range queries (transactions within a time period)
• Indexing structures such as B-trees optimized for one-dimensional searches

Vector similarity search is a fundamentally different operation: we are not looking for an exact match, but for the nearest vectors in a space of hundreds of dimensions. Computing cosine similarity against every vector (brute-force approach) is prohibitive at scale — this is the curse of dimensionality.

What Vector Databases do

They are designed from the ground up to:

• Store dense vectors
• Perform fast Approximate Nearest Neighbor (ANN) search
• Use specialized indexing algorithms that create intelligent shortcuts

The core idea is to trade a tiny percentage of precision for massive speed gains — thousands of times faster than a brute-force scan.

2.2 Internal Operation: HNSW

Hierarchical Navigable Small World (HNSW) has become the de facto standard for modern vector databases (Pinecone, Weaviate, Qdrant).

The founding concept: the navigable small world graph

Imagine each vector as a node connected to a small set of other nodes (its neighbors). From any node, you can walk toward its neighbors, then toward their neighbors, progressively approaching a target. This is the “six degrees of separation” property.

HNSW’s innovation: the hierarchy

Instead of a single flat graph, HNSW builds multiple layers:

How a search unfolds

1. Top layer — Start from an entry point and navigate toward the nearest neighbor of the query. Repeat until no closer neighbor is found → we are in the right region of space
2. Descent — Drop to the next layer using the current position as the starting point. More nodes, finer connections
3. Bottom layer — We are in the precise neighborhood where the true nearest neighbors live. Final, more exhaustive search to return the top K results

Performance and trade-offs

• Search complexity: ~logarithmic relative to dataset size
• For 1 million vectors, HNSW examines only a few thousand nodes (99.9% reduction)
• High RAM consumption: often 30 to 50 GB per billion vectors
• Parameters to tune:
  • M — maximum number of connections per node
  • efConstruction — rigor of graph construction during indexing

✅ HNSW won because it offers exceptional recall at low latency without requiring special hardware acceleration.

2.3 Indexing Algorithms: IVF and Product Quantization

Inverted File Indexing (IVF)

The idea is simple: partition the dataset into K regions (Voronoi cells) via an algorithm like k-means.

How it works:

1. Each cell is represented by its centroid
2. On insertion, a vector is assigned to its nearest centroid
3. On query, IVF finds the n_probe nearest centroids and searches only within those cells

Product Quantization (PQ)

PQ is a compression technique. Instead of storing each 768-dimensional vector in float32:

1. Split each vector into M equal sub-vectors
2. Apply k-means to each sub-space independently, generating a codebook of ~256 centroids per sub-space
3. Replace each sub-vector with the ID of its nearest centroid
4. The original vector becomes just M small integers

Memory savings: factor of 10 to 20× and substantial acceleration.

IVF vs PQ — Comparison

📌 In production, IVF and PQ are frequently combined (e.g., FAISS — Facebook AI Similarity Search).

2.4 Large-Scale Storage Strategies

Memory vs. disk

Compression

A 768-dimensional float32 vector occupies 3 KB. At one billion vectors, that is 3 TB just for the vectors.

💡 Many systems use multi-tier compression: coarse compressed index for initial filtering, then re-ranking with full-precision vectors.

Sharding

Replication

Replication serves two masters: availability and read scalability. With 3 replicas, you can handle 3× the query volume and survive a hardware failure.

A production deployment might use memory-mapped disk storage with PQ compression, sharded across 16 nodes, each with 2 replicas for high availability.

2.5 Comparing Vector Databases

Pinecone

• ✅ Fully managed, closed-source
• ✅ Setup in minutes (API key → indexing)
• ✅ Automatic scaling, replication and index optimization
• ❌ Vendor lock-in, linear costs, minimal control
• 🎯 Ideal for: startups and teams prioritizing speed to market

Weaviate

• ✅ Open-source with cloud offering
• ✅ Designed for hybrid search from the ground up
• ✅ Built-in vectorization modules (plug in OpenAI or Cohere)
• ✅ GraphQL interface, rich native metadata filtering
• ❌ More resource-intensive, monolithic “all-in-one” architecture
• 🎯 Ideal for: sophisticated filtering and hybrid search

Qdrant

• ✅ Written in Rust, exceptional speed and memory efficiency
• ✅ Fine-grained control over indexing parameters (HNSW, quantization)
• ✅ Available as open-source, cloud, or self-hosted binary
• ❌ More complexity to manage
• 🎯 Ideal for: cost and performance optimization at scale

Chroma

• ✅ Open-source, deeply integrated into the Python ecosystem
• ✅ Native integration with LangChain and LlamaIndex
• ✅ Lightweight, runs embedded within your application
• ❌ Designed for local development and small-to-medium workloads
• ❌ Not designed for petabyte-scale or distributed deployments
• 🎯 Ideal for: prototyping and local experimentation (the “SQLite” of vector databases)

Summary: how to choose?

2.6 pgvector and Self-Hosted Solutions

pgvector — the PostgreSQL extension

If you already use PostgreSQL, pgvector is remarkably compelling:

• It is an extension that adds vector data types and similarity search operators directly into PostgreSQL
• Define a column as vector(1536), create an index, and query with <-> for Euclidean distance or <=> for cosine similarity
• Vectors live alongside relational data, joinable with customer tables, history, etc.

Advantages:

• No new query language to learn
• Backups, replication, high availability managed by PostgreSQL
• Supports both IVFFlat and HNSW

Limitations:

• Single-node by nature
• Index build times slow with dataset size
• Complex filters can degrade performance

Other self-hosted solutions

• Elasticsearch with vector plugins
• Redis with RedisVL
• SQLite with sqlite-vss

Self-hosted vs. Managed Services

2.7 Schema Design for Vector Databases

Three key concepts determine whether your retrieval will be fast, accurate and maintainable.

1. Metadata Filtering

Most vector databases allow attaching key-value pairs to each vector (timestamps, sources, categories). The key design decision:

2. Namespace Organization

Namespaces are logical partitions within a single vector database instance. Use cases:

• Multi-tenancy: separate spaces per client
• A/B testing: separate indexes for different embedding model versions
• Deleting one client’s data without affecting others

3. Hybrid Search Configuration

Hybrid search merges semantic similarity with traditional keyword matching. The schema must:

• Store both embeddings and full text
• Configure the weight of each signal
• Specify which fields are keyword-searchable
• Define how the fusion algorithm combines scores (e.g., Reciprocal Rank Fusion)

2.8 Advanced Query Patterns

Pattern 1: Metadata-filtered Search

Scenario: a support bot for a SaaS, the user asks about “login issues” but is on version 2.5.

# Qdrant filter predicate
client.search(
    collection_name="docs",
    query_vector=embedding,
    query_filter=models.Filter(
        must=[
            models.FieldCondition(
                key="version",
                match=models.MatchValue(value="2.5")
            ),
            models.FieldCondition(
                key="date",
                range=models.Range(gte="2024-01-01")
            )
        ]
    ),
    limit=10
)

This is not post-filtering — it is filter-aware traversal, dramatically faster.

Pattern 2: Multi-vector Retrieval for rich documents

PDFs contain text, tables and graphics. A single vector per document loses everything.

# At indexing time: one document, multiple vectors
chunks = [
    {"text": "Introduction to transformers",
     "doc_id": "paper123", "type": "text"},
    {"text": "Table: Model accuracy comparison",
     "doc_id": "paper123", "type": "table"},
    {"text": "Figure: Attention mechanism",
     "doc_id": "paper123", "type": "image_caption"}
]

# At query time: multi-type retrieval, then aggregation
results = vector_db.search(query, limit=5)
doc_groups = group_by_doc_id(results)

Pattern 3: Late Interaction with ColBERT

Instead of a single vector per query, store embeddings at the token level. At search time, compute the maximum similarity between tokens. This captures matches like “deep learning” in one chunk and “neural networks” in another.

2.9 Production Operations

Incremental updates

Data is not static. Rebuilding the full index every night does not scale beyond a few million vectors. Vector databases support real-time upserts. HNSW graphs allow dynamic insertion.

⚠️ Deletions are more complex — many databases mark vectors as deleted but leave them in place until a background compaction.

Batch operations

Golden rule: batch sizes between 100 and 1,000 depending on dimension and database limits. Monitor memory during large batches.

Index rebuild strategies

Indexes degrade over time (deleted vectors leaving holes, updates fragmenting the graph). Approaches:

• Maintenance windows: nightly or weekly rebuilds
• Blue-Green Index Deployment: build a new index in the background while the old one serves queries, then atomically switch over

Minimizing downtime

To target five 9s (99.999% availability):

• Replication — node failures cause no interruption
• Rolling upgrades — updates without full shutdown
• Blue-green rebuilds — index maintenance without blocking queries
• Health checks and automatic failover

2.10 Monitoring, Index Health and Recovery

What to monitor?

Backup strategies

1. Full exports — Dump vectors and metadata to object storage (S3). Slow but exhaustive. Weekly with daily incremental exports
2. Snapshots — If supported (Qdrant, Weaviate), filesystem snapshots copyable while the database is running

⚠️ A backup you cannot restore is worthless. Test your restores!

Disaster Recovery

📌 Document your recovery procedure and practice it quarterly.

Performance debugging

1. Index parameters — HNSW efSearch too low? Increase it. Too high? Wasted resources
2. Metadata filters — Highly selective filters can accelerate; poorly designed ones kill performance
3. Resource contention — CPU saturation, disk I/O, memory pressure

3. Knowledge Graphs and Structured Knowledge

3.1 Knowledge Graph Fundamentals

What is a Knowledge Graph?

Forget tables and vector spaces for a moment. Think about how you store information in your own mind. You do not have a spreadsheet of facts — you have a network of interconnected ideas. This is the essence of a knowledge graph: a data model that represents information as a network of entities and the relationships between them.

Components of the graph data model

Two main implementations

Property Graphs (e.g., Neo4j):

• Very developer-friendly
• Nodes with properties, relationships with a name and their own properties
• Flexible, ideal for recommendation engines or fraud detection

RDF (Resource Description Framework):

• Standard from the Semantic Web community
• Each piece of information is decomposed into triples: Subject → Predicate → Object
• Example: Einstein → hasBirthDate → 1879-03-14
• Interoperable data, queryable via SPARQL

The Semantic Web

The original vision of the Semantic Web was to create a web of linked data that computers could understand and reason over, not just display. RDF is its foundation, enabling data from different sources to be connected and interpreted through shared definitions called ontologies.

3.2 Knowledge Graphs vs Databases vs Vector Stores

The most powerful hybrid architecture

The most powerful GenAI architectures combine all three:

1. The user query is sent to the vector store → semantically relevant chunks
2. Simultaneously, named entities extracted from the query are used to traverse the knowledge graph → connected facts and relationships
3. The results are merged → responses that are both conceptually relevant and enriched with precise structured context

3.3 Ontology and Schema Design

“If the graph data model is like the alphabet, the ontology is the grammar.”

Classes and properties

Ontology languages

RDFS (RDF Schema):

• Allows defining classes, properties and establishing simple hierarchies
• Example: Manager is a subclass of Person

OWL (Web Ontology Language):

• Complex logic and constraints
• Disjoint classes (a thing cannot be both a Person and an Organization)
• Restrictions (a “mother” must be a person having at least one child)
• sameAs relation to unify terms from different datasets

Reasoning

A reasoner can examine your data and OWL ontology and infer new implicit facts:

• If “having a child makes someone a parent” and “Alice has a child Bob” → Alice is automatically classified as a parent
• If acquired is a sub-property of owns → acquisitions are returned in queries about owns
• Transitive relations → automatically infer ancestors
• Equivalent entities → unify identifiers

The expressiveness vs. performance trade-off

The more detailed and logical the ontology, the smarter the graph. However, complex OWL reasoning is computationally expensive and can dramatically slow queries.

In production:

• Use a simpler property graph schema rather than heavy OWL
• Perform complex reasoning offline and store results as simple facts
• Design an ontology that is “just expressive enough”

3.4 Entity Recognition from Unstructured Text

Named Entity Recognition (NER) is the task of locating and classifying segments of text into predefined categories.

Example: “Apple acquired Darwin AI in 2024” → Apple (Organization), Darwin AI (Organization), 2024 (Date)

Approaches

💡 The most modern approach: use an LLM with few-shot prompting — provide a few examples of domain-specific entities, and the model extracts them with surprising accuracy without fine-tuning.

3.5 Relation Extraction and Knowledge Construction

“Entity recognition answers what, relation extraction answers how.”

Approaches

Relation Classification (classic approach):

• Take text surrounding two entities and predict the relationship
• Fine-tune a transformer model on datasets like TACRED or FewRel
• Choose from a fixed set of relationships: acquired, headquartered_in, founded_by, etc.

Zero-shot Extraction with LLMs (modern approach):

• Prompt an LLM: “Extract all entity relationships as subject-predicate-object triples”
• The LLM understands semantics and can generate relationships never seen during training
• No fine-tuning required

Example:

Input: “The research team at Google Brain published the Transformer architecture”

Output:

• (Google Brain, employs, research team)
• (research team, published, Transformer architecture)

3.6 Building Knowledge Graphs with LLMs

Instead of training custom models, you can prompt an LLM to return structured data directly loadable into a graph database.

Extraction code with LangChain and OpenAI

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List

# Initialize the LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Define the desired structure with Pydantic
class Entity(BaseModel):
    name: str = Field(description="The entity name")
    type: str = Field(
        description="Entity type: Person, Organization, "
                     "Product, Technology"
    )
    
class Relationship(BaseModel):
    source: str = Field(description="Source entity name")
    target: str = Field(description="Target entity name")
    type: str = Field(description="Relationship type")

class KnowledgeGraph(BaseModel):
    entities: List[Entity] = Field(description="List of unique entities")
    relationships: List[Relationship] = Field(
        description="List of relationships"
    )

# Create a parser
parser = JsonOutputParser(pydantic_object=KnowledgeGraph)

# Build the prompt
prompt = ChatPromptTemplate.from_template("""
    Extract a knowledge graph from the text below.
    
    Rules:
    - Identify all relevant entities
    - Extract relationships between them
    - Use consistent entity names
    - Only include explicit information from the text
    
    {format_instructions}
    
    Text: {text}
""")

# Chain it together
chain = prompt | llm | parser

# Sample text
text = """
Apple acquired DarwinAI in 2024 to boost its AI capabilities. 
DarwinAI was founded by Alexander Wong in Waterloo, Canada. 
The company specialized in efficient computer vision models.
"""

# Run the extraction
result = chain.invoke({
    "text": text,
    "format_instructions": parser.get_format_instructions()
})

# Display results
print("Entities found:")
for entity in result["entities"]:
    print(f"  - {entity['name']} ({entity['type']})")

print("\nRelationships found:")
for rel in result["relationships"]:
    print(f"  - {rel['source']} --[{rel['type']}]--> {rel['target']}")

Graph validation

Before loading data into a graph database, validate the LLM output:

def validate_graph(data):
    """Quality validation before loading into the graph."""
    entity_names = [e["name"] for e in data["entities"]]
    
    for rel in data["relationships"]:
        if rel["source"] not in entity_names:
            print(f"Warning: {rel['source']} not in entities")
        if rel["target"] not in entity_names:
            print(f"Warning: {rel['target']} not in entities")
    
    # Remove entity duplicates
    unique = {e["name"]: e for e in data["entities"]}.values()
    data["entities"] = list(unique)
    return data

validated = validate_graph(result)

Improving with Few-shot Prompting

For production, adding examples to the prompt dramatically improves consistency:

examples = """
Example 1:
Text: "Satya Nadella is CEO of Microsoft."
Output: {{"entities": [
    {{"name": "Satya Nadella", "type": "Person"}},
    {{"name": "Microsoft", "type": "Organization"}}
  ], "relationships": [
    {{"source": "Satya Nadella", "target": "Microsoft",
      "type": "CEO_OF"}}
  ]}}

Example 2:
Text: "Python was created by Guido van Rossum."
Output: {{"entities": [
    {{"name": "Python", "type": "Language"}},
    {{"name": "Guido van Rossum", "type": "Person"}}
  ], "relationships": [
    {{"source": "Guido van Rossum", "target": "Python",
      "type": "CREATED"}}
  ]}}
"""

3.7 Graph Embedding Techniques

Graph embeddings translate the graph structure (nodes and connections) into dense vector representations, making the graph usable for machine learning and hybrid search.

Technique families

Code: Node Embeddings with Node2Vec

import networkx as nx
from node2vec import Node2Vec
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# Create a graph (Zachary's karate club — classic social network)
G = nx.karate_club_graph()

# Generate walks and train Node2Vec
node2vec = Node2Vec(
    G, dimensions=64, walk_length=30, num_walks=200, workers=4
)
model = node2vec.fit(window=10, min_count=1, batch_words=4)

# Get the embedding for node 0
embedding_0 = model.wv['0']

# Embeddings for all nodes
embeddings = np.array([model.wv[str(node)] for node in G.nodes()])

# Cosine similarity between node 0 and node 33
sim = cosine_similarity([embeddings[0]], [embeddings[33]])[0][0]
print(f"Similarity between node 0 and node 33: {sim:.3f}")

# Top 5 nodes most similar to node 0
similar_nodes = model.wv.most_similar('0', topn=5)
print("Nodes most similar to node 0:")
for node, score in similar_nodes:
    print(f"  - node {node}: similarity {score:.3f}")

Code: Graph Neural Networks (GCN) with PyTorch Geometric

import torch
import torch.nn.functional as F
from torch_geometric.nn import GCNConv

class GCN(torch.nn.Module):
    def __init__(self, in_channels, hidden_channels, out_channels):
        super().__init__()
        self.conv1 = GCNConv(in_channels, hidden_channels)
        self.conv2 = GCNConv(hidden_channels, out_channels)

    def forward(self, x, edge_index):
        x = self.conv1(x, edge_index)
        x = F.relu(x)
        x = self.conv2(x, edge_index)
        return x  # node embeddings

# x: node feature matrix (e.g., text embeddings)
# edge_index: graph connections
# model = GCN(in_channels=384, hidden_channels=128, out_channels=64)
# embeddings = model(x, edge_index)

This network passes messages along the edges, so the final embedding of a node incorporates both its own features and those of its neighbors.

The bidirectional translation

Once node embeddings are obtained, you can:

• Store them in a vector database alongside document chunks
• Search for similar nodes via vector similarity (graph traversal via vectors)
• Take an arbitrary query vector and find the nearest node embeddings, then traverse the graph from those nodes

This is the heart of hybrid retrieval — combining vector similarity with graph relationships.

3.8 Hybrid Retrieval: Vectors and Graph Traversal

“Vector search finds things that look like your query. Graph traversal finds things that are connected to your query. Together, they capture what each would miss alone.”

The multi-stage flow

Reciprocal Rank Fusion

The simplest and most effective fusion algorithm for combining two ranked lists:

def reciprocal_rank_fusion(
    vector_results, graph_results, k=60
):
    """Merges two ranked lists into a single re-ranked list."""
    fusion_scores = {}
    
    for rank, result in enumerate(vector_results):
        doc_id = result['chunk_id']
        fusion_scores[doc_id] = (
            fusion_scores.get(doc_id, 0) + 1 / (rank + k)
        )
    
    for rank, result in enumerate(graph_results):
        doc_id = result['chunk_id']
        fusion_scores[doc_id] = (
            fusion_scores.get(doc_id, 0) + 1 / (rank + k)
        )
    
    ranked = sorted(
        fusion_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )
    return ranked

# Usage
vector_list = vector_search(query, vector_db, top_k=10)
graph_list = graph_traversal(entity_ids, graph_connection)

vector_results = [{'chunk_id': r.id} for r in vector_list]
graph_results = [{'chunk_id': r['chunk_id']} for r in graph_list]

final_ranking = reciprocal_rank_fusion(vector_results, graph_results)

When to use Hybrid Retrieval?

• When your knowledge graph captures non-obvious relationships from text alone
• When users need to discover things across connections (e.g., “find suppliers of companies similar to X”)

3.9 Querying Knowledge Graphs

SPARQL — the SQL of graphs

SPARQL is the standard query language for RDF graphs. You write patterns that match sub-graphs, and the database returns the satisfying bindings.

Basic example — find what Apple has acquired:

PREFIX : <http://example.org/>
SELECT ?company ?year
WHERE {
    :Apple :acquired ?company .
    ?company :foundedIn ?year .
}

Advanced example — find all indirect acquisitions (acquisition chain):

PREFIX : <http://example.org/>
SELECT ?company
WHERE {
    :Apple :acquired+ ?company .
}

The + means one or more repetitions of the acquired predicate.

Cypher — Neo4j’s language

Cypher is more visual, using ASCII arrows to represent patterns:

MATCH (apple:Company {name: 'Apple'})-[:ACQUIRED*]->(company:Company)
RETURN company.name

The * after the relationship type means one or more hops. You can also specify [:ACQUIRED*1..3] for between 1 and 3 hops.

Reasoning in queries

• If acquired is a sub-property of owns, queries on owns also return acquisitions
• Transitive properties (like isParent) allow automatically inferring grandparents and ancestors
• Identifier unification (Apple = Apple Inc.) treats entities as a single node

Python example with RDFLib

from rdflib import Graph, Namespace, Literal

g = Graph()
ns = Namespace("http://example.org/")

g.add((ns.Apple, ns.acquired, ns.DarwinAI))
g.add((ns.DarwinAI, ns.foundedIn, Literal(2017)))

# SPARQL query
query = """
PREFIX : <http://example.org/>
SELECT ?company ?year
WHERE {
    :Apple :acquired ?company .
    ?company :foundedIn ?year .
}
"""

for row in g.query(query):
    print(f"Company: {row.company}, Founded: {row.year}")

💡 For full OWL reasoning, use a dedicated triple store like Stardog or GraphDB with built-in reasoners.

3.10 Maintaining and Evolving Knowledge Graphs

Managing updates

• Incremental updates: when new documents arrive, run the extraction pipeline and add new triples
• Provenance tracking: store which source document generated each triple. When a document is updated, delete all its triples before re-extracting

Duplicate detection

Common problem: the same entity appears in different forms (“Apple Incorporated”, “Apple”, “Apple Computer”).

from fuzzywuzzy import fuzz
import itertools

def find_duplicate_entities(entities, threshold=85):
    """Find probable duplicates based on name similarity."""
    duplicates = []
    
    for e1, e2 in itertools.combinations(entities, 2):
        name1, name2 = e1['name'], e2['name']
        similarity = fuzz.token_sort_ratio(name1, name2)
        
        if similarity > threshold:
            duplicates.append({
                'source': name1,
                'target': name2,
                'score': similarity
            })
    
    return duplicates

# Example
entities = [
    {'name': 'Apple Inc.', 'type': 'Organization'},
    {'name': 'Apple', 'type': 'Organization'},
    {'name': 'Apple Computer', 'type': 'Organization'},
    {'name': 'Microsoft', 'type': 'Organization'}
]

dupes = find_duplicate_entities(entities)
for d in dupes:
    print(f"'{d['source']}' might be same as '{d['target']}' "
          f"({d['score']}%)")

In production, combine multiple signals: name similarity, same relationship patterns, same document contexts, and similar embeddings.

Schema evolution

The ontology is not static. The safe approach is additive changes:

1. Add new types and properties
2. Mark old ones as deprecated
3. Migrate data progressively

Versioning

Production best practices

• Run duplicate detection jobs nightly
• Monitor schema usage to identify newly needed types
• Keep at least one week of snapshots for rollback
• Always test schema changes on a copy before deploying

4. Production Pipelines, Advanced Preprocessing and Synthetic Data

4.1 End-to-End Data Pipeline Architecture

The pipeline can be broken down into five key stages, orchestrated by a tool such as Airflow or Prefect.

flowchart LR
    A[Data Ingestion] --> B[Cleaning and Normalization]
    B --> C[Chunking]
    C --> D[Embedding Generation]
    D --> E[Indexing]
    
    style A fill:#4A90D9,stroke:#333,color:#fff
    style B fill:#7B68EE,stroke:#333,color:#fff
    style C fill:#E67E22,stroke:#333,color:#fff
    style D fill:#27AE60,stroke:#333,color:#fff
    style E fill:#E74C3C,stroke:#333,color:#fff

Orchestration (Airflow, Prefect) ensures each file is cleaned before being chunked, and chunked before being embedded, with retry handling and alerts.

4.2 Chunking Strategies

Chunking is a trade-off between context and precision:

• Chunks too large → information is diluted, details get lost
• Chunks too small → precision gain but context loss

Strategy 1: Fixed-size Chunking

The simplest approach: split every N characters with an optional overlap.

def fixed_size_chunking(text, chunk_size=500, overlap=50):
    chunks = []
    start = 0
    text_length = len(text)
    
    while start < text_length:
        end = min(start + chunk_size, text_length)
        if start > 0:
            start = start - overlap
        chunk = text[start:end]
        chunks.append(chunk)
        start = end
    
    return chunks

# Example
sample_text = "Your long document text goes here. " * 50
chunks = fixed_size_chunking(sample_text, chunk_size=200, overlap=20)
print(f"Generated {len(chunks)} chunks")
print(f"First chunk: {chunks[0][:100]}...")

Trade-off: Simplicity vs. coherence — fast and predictable, but often cuts sentences in the middle.

Strategy 2: Sentence-based Chunking

Split at sentence boundaries, preserving linguistic units.

import nltk
nltk.download('punkt_tab')
from nltk.tokenize import sent_tokenize

def sentence_based_chunking(text, max_sentences=3):
    sentences = sent_tokenize(text)
    chunks = []
    current_chunk = []
    
    for sentence in sentences:
        current_chunk.append(sentence)
        if len(current_chunk) >= max_sentences:
            chunk_text = " ".join(current_chunk)
            chunks.append(chunk_text)
            current_chunk = []
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

# Example
sample = ("This is the first sentence. This is the second. "
          "This is the third. Here is a fourth. And a fifth.")
chunks = sentence_based_chunking(sample, max_sentences=2)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1}: {chunk}")

Trade-off: Respects grammatical boundaries, but may create conceptually disconnected chunks.

Strategy 3: Recursive Chunking (often the most effective)

Respects the document’s natural hierarchy: attempts to split by paragraphs first, then sentences, then words.

def recursive_chunking(text, separators=None, chunk_size=500):
    if separators is None:
        separators = ["\n\n", "\n", ". ", " ", ""]
    
    if len(text) <= chunk_size:
        return [text]
    
    separator = separators[0]
    if separator == "":
        return [
            text[i:i+chunk_size]
            for i in range(0, len(text), chunk_size)
        ]
    
    splits = text.split(separator)
    
    if len(splits) == 1:
        return recursive_chunking(text, separators[1:], chunk_size)
    
    chunks = []
    current_chunk = ""
    for split in splits:
        piece = split + (
            separator if split != splits[-1] else ""
        )
        if len(current_chunk) + len(piece) <= chunk_size:
            current_chunk += piece
        else:
            if current_chunk:
                chunks.append(current_chunk)
            if len(piece) > chunk_size:
                sub_chunks = recursive_chunking(
                    piece, separators[1:], chunk_size
                )
                chunks.extend(sub_chunks)
                current_chunk = ""
            else:
                current_chunk = piece
    
    if current_chunk:
        chunks.append(current_chunk)
    
    return chunks

# Example
text = ("Section 1\n\nThis is a paragraph with multiple sentences. "
        "It talks about topic A. Still topic A.\n\n"
        "Section 2\n\nThis paragraph covers topic B. "
        "It has details about B.")
chunks = recursive_chunking(text, chunk_size=50)
for i, chunk in enumerate(chunks):
    print(f"Recursive Chunk {i+1}: {chunk}")

How to choose?

🔑 The key is experimentation — run retrievals with different strategies, examine what is returned, and adjust size and overlap based on user query types.

4.3 Handling Multimodal Content

In the real world, documents are not just text — there are PDFs with images, PowerPoint slides with charts, web pages with tables.

Extraction strategies by content type

Code: processing a multimodal PDF

import pdfplumber
import pytesseract
from PIL import Image
import io

def extract_content_from_pdf(pdf_path):
    content_chunks = []
    
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            # Extract text
            text = page.extract_text()
            if text:
                content_chunks.append({
                    "type": "text",
                    "page": page_num,
                    "content": text
                })
            
            # Extract tables
            tables = page.extract_tables()
            for table in tables:
                table_str = "\n".join([
                    "| " + " | ".join(row) + " |"
                    for row in table if any(row)
                ])
                content_chunks.append({
                    "type": "table",
                    "page": page_num,
                    "content": table_str
                })
            
            # Extract images (OCR)
            for img in page.images:
                img_data = img["stream"].get_data()
                image = Image.open(io.BytesIO(img_data))
                ocr_text = pytesseract.image_to_string(image)
                if ocr_text.strip():
                    content_chunks.append({
                        "type": "image_text",
                        "page": page_num,
                        "content": ocr_text
                    })
    
    return content_chunks

# Usage
chunks = extract_content_from_pdf("sample_document.pdf")
for chunk in chunks:
    print(f"Type: {chunk['type']}, Page: {chunk['page']}")
    print(f"Content preview: {chunk['content'][:100]}...\n")

Indexing strategies

4.4 Metadata Enrichment and Context Injection

The problem

A paragraph extracted from the middle of a document may reference “the company” or “the results” without specifying what they refer to. Without context, the LLM will not know how to interpret the chunk.

Essential metadata

Two complementary approaches

1. Injection — Modify the chunk text:

[Title: Q4 Report] [Section: Revenue]
This quarter saw a 15% increase...

→ The info becomes part of the vector, improves retrieval for queries mentioning the title

2. Metadata Filtering — Store separately in the vector database: → Allows restricting searches by filter without polluting the vector

Metadata enrichment code

import re
from typing import List, Dict

def enrich_chunks_with_metadata(
    document_title: str, content: str
) -> List[Dict]:
    """
    Split content by headers and create enriched chunks.
    Each chunk receives the document title and its section header.
    """
    header_pattern = r'^(#{1,3})\s+(.+)$'
    lines = content.split('\n')
    
    chunks = []
    current_section = "Introduction"
    current_text = []
    
    for line in lines:
        header_match = re.match(header_pattern, line)
        if header_match:
            if current_text:
                chunk_text = " ".join(current_text)
                enriched = (
                    f"[Title: {document_title}] "
                    f"[Section: {current_section}] "
                    f"{chunk_text}"
                )
                chunks.append({
                    "text": enriched,
                    "metadata": {
                        "title": document_title,
                        "section": current_section,
                        "original_text": chunk_text
                    }
                })
                current_text = []
            current_section = header_match.group(2).strip()
        else:
            if line.strip():
                current_text.append(line.strip())
    
    # Don't forget the last chunk
    if current_text:
        chunk_text = " ".join(current_text)
        enriched = (
            f"[Title: {document_title}] "
            f"[Section: {current_section}] "
            f"{chunk_text}"
        )
        chunks.append({
            "text": enriched,
            "metadata": {
                "title": document_title,
                "section": current_section,
                "original_text": chunk_text
            }
        })
    
    return chunks

# Example
doc_title = "2023 Annual Report"
doc_content = """
## Executive Summary
The company achieved record revenue in 2023.
Growth was driven by international expansion.

## Financial Highlights
Net income increased by 25% compared to 2022.
Operating expenses were well controlled.
"""

enriched_chunks = enrich_chunks_with_metadata(doc_title, doc_content)
for chunk in enriched_chunks:
    print("Enriched Text:")
    print(chunk["text"])
    print("Metadata:")
    print(chunk["metadata"])
    print("-" * 50)

Output:

Enriched Text:
[Title: 2023 Annual Report] [Section: Executive Summary] The company achieved record revenue in 2023. Growth was driven by international expansion.
Metadata:
{'title': '2023 Annual Report', 'section': 'Executive Summary', ...}

💡 In practice, both approaches are often combined: inject the short title and section header into the text, and store complete metadata (paths, timestamps, type) as filterable fields.

4.5 Pipeline Orchestration

Airflow ensures each task runs at the right time, in the right order, and that failures are handled gracefully. At the heart of Airflow is the concept of a DAG (Directed Acyclic Graph).

Airflow DAG code

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

from your_pipeline import (
    fetch_documents, clean_and_chunk,
    generate_embeddings, index_vectors, cleanup
)

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'embedding_pipeline',
    default_args=default_args,
    description='End-to-end embedding generation pipeline',
    schedule_interval='@hourly',
    catchup=False
)

fetch_task = PythonOperator(
    task_id='fetch_documents',
    python_callable=fetch_documents, dag=dag
)

chunk_task = PythonOperator(
    task_id='clean_and_chunk',
    python_callable=clean_and_chunk, dag=dag
)

embed_task = PythonOperator(
    task_id='generate_embeddings',
    python_callable=generate_embeddings, dag=dag
)

index_task = PythonOperator(
    task_id='index_vectors',
    python_callable=index_vectors, dag=dag
)

cleanup_task = PythonOperator(
    task_id='cleanup',
    python_callable=cleanup, dag=dag
)

# Define dependencies
fetch_task >> chunk_task >> embed_task >> index_task >> cleanup_task

Key orchestration points

4.6 Synthetic Data Generation and Testing

The problem

Real user data is scarce, especially at the launch of a new system. You do not have thousands of real queries to test with.

The solution: LLMs to generate test cases

Take documents and ask an LLM to generate question-answer pairs. You can also generate:

• Edge cases requiring multi-hop reasoning
• Questions with synonyms absent from the text
• Adversarial examples (semantically similar but on a different subject)

Synthetic generation code

import openai
from typing import List, Dict
import json

def generate_synthetic_qa(
    document_text: str, num_pairs: int = 5
) -> List[Dict]:
    prompt = f"""
    Based on the following document, generate {num_pairs} 
    question-answer pairs.
    Make the questions diverse: some should be factual, 
    some should require inference, and some should use 
    synonyms not found in the text.
    
    Document:
    {document_text[:2000]}
    
    Return the pairs as a JSON list with 'question' and 
    'answer' keys.
    """
    
    response = openai.ChatCompletion.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )
    
    qa_pairs = json.loads(
        response.choices[0].message.content
    )
    return qa_pairs

# Usage
document = (
    "The Eiffel Tower was completed in 1889. "
    "It is located in Paris, France. "
    "It was designed by Gustave Eiffel."
)
qa_pairs = generate_synthetic_qa(document, num_pairs=3)

for qa in qa_pairs:
    print(f"Q: {qa['question']}")
    print(f"A: {qa['answer']}\n")

Continuous validation

Integrate synthetic data generation into the pipeline:

1. Each addition of new documents → generate synthetic QA pairs
2. Run an automatic evaluation job
3. Track metrics like hit rate or Mean Reciprocal Rank over time
4. If metrics drop → alert and investigate

⚠️ Synthetic data carries the biases of the LLM that generated it. However, combined with progressive deployment and real query monitoring, it forms a robust quality assurance strategy.

Summary

This course covers the complete journey, from the fundamental mathematics of embeddings to architectural decisions for vector databases, through building knowledge graphs and production-ready pipelines.

Search Terms

genai · data · knowledge · layer · rag · vector · search · embeddings · artificial · intelligence · generative · ai · graph · embedding · graphs · pipeline · strategies · databases · chunking · hybrid · metadata · semantic · approaches · architecture