Retrieval and Vector Stores in LangChain

Complete course on designing scalable retrieval systems for LLM applications with LangChain.

Table of Contents

1. Overview — RAG Pipeline
2. Module 1 — Document Loaders and Data Ingestion
   • The Document Object
   • Common File Loaders
   • Web and API Loaders
   • Batch Loading — DirectoryLoader
   • Lazy Loading
3. Module 2 — Text Splitting Strategies
   • Why Split Text
   • CharacterTextSplitter
   • RecursiveCharacterTextSplitter
   • Chunk Size and Overlap
   • Specialized Splitters — Markdown, HTML, Code
4. Module 3 — Embeddings and Vector Representations
   • Why Embeddings Are Necessary
   • Embeddings as Vectors
   • Embedding Models with LangChain
   • Similarity Measures
   • Dimensions and Performance Trade-offs
   • Evaluating Embedding Quality
5. Module 4 — Vector Stores and Scalable Similarity Search
   • Why Vector Stores Are Necessary
   • FAISS — Local Vector Store
   • Pinecone — Cloud Vector Store
   • Distance Metrics and Scores
   • Metadata Filtering
   • MMR — Maximal Marginal Relevance
6. Module 5 — Retrievers and Advanced Retrieval Strategies
   • The Retriever Abstraction in LangChain
   • Types of Retrievers
   • Multi-Query Retriever
   • Contextual Compression
   • Evaluating Retrieval Systems
   • Trade-offs in Retrieval Systems
7. Module 6 — Structured and Hybrid Queries with LangChain
   • Querying Structured Data
   • Natural Language to SQL — LCEL and Agents
   • SQL Query Reliability and Validation
   • Hybrid Systems — Structured and Unstructured Data
   • Choosing the Right Data Access Approach
8. Summary and Best Practices

1. Overview — RAG Pipeline

RAG (Retrieval Augmented Generation) is the central architectural pattern of this course. External data does not go directly into the LLM — it passes through several stages.

flowchart LR
    A[📄 Data Sources\nFiles · Web · API · DB] --> B[Document Loaders\nIngestion]
    B --> C[Text Splitters\nChunking]
    C --> D[Embedding Models\nConversion to vectors]
    D --> E[Vector Store\nStorage & indexing]
    E --> F{User Query}
    F --> G[Retriever\nSimilarity search]
    G --> H[LLM\nResponse generation]
    H --> I[Final Response]

    style A fill:#f0f4ff,stroke:#4a90d9
    style E fill:#fff0e0,stroke:#e07c00
    style H fill:#e8f5e9,stroke:#2e7d32
    style I fill:#e8f5e9,stroke:#2e7d32

Role of Each Stage

2. Module 1 — Document Loaders and Data Ingestion

Data Ingestion Concept

LLMs are powerful models but they have no direct access to your files, websites, or databases. Data ingestion is the process that brings real-world data into the LLM pipeline.

LangChain provides more than 100 document loaders organized into categories:

mindmap
  root((Document Loaders))
    Local Files
      TextLoader
      CSVLoader
      PyPDFLoader
    Web
      WebBaseLoader
      WikipediaLoader
      ArxivLoader
    External APIs
      Loaders with API key
      Cloud services
    Databases
      SQL
      NoSQL
      Cloud storage

The Document Object

All loaders return Document objects — a standardized format, independent of the source.

classDiagram
    class Document {
        +str page_content
        +dict metadata
    }
    class MetadataExamples {
        +source: "path/to/file.pdf"
        +page: 3
        +row: 5
        +title: "Page title"
        +language: "en"
    }
    Document --> MetadataExamples : metadata contains

Advantages of standardization:

• Each pipeline component works with the same structure
• Changing the source (PDF → web) only requires changing the loader
• Metadata allows filtering searches, tracing sources, and debugging

Common File Loaders

TextLoader

# Load a text file as a LangChain Document
from langchain_community.document_loaders import TextLoader

loader = TextLoader("../data/files/sports_news.txt", encoding="utf-8")
documents = loader.load()

print(documents[0].page_content)
print(documents[0].metadata)

• Granularity: 1 file → 1 Document
• Usage: logs, raw text exports, system files

CSVLoader

# Load a CSV file as LangChain Documents
from langchain_community.document_loaders import CSVLoader

loader = CSVLoader("../data/files/sales_data.csv")
documents = loader.load()

print("Total rows loaded:", len(documents))

print(f"\nCONTENT:\n{documents[0].page_content}")
print(f"\nMETADATA:\n{documents[0].metadata}")

• Granularity: 1 CSV row → 1 Document
• Usage: product catalogs, analytical data, inventories

PyPDFLoader

# Load a PDF as LangChain Documents
from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("../data/files/research_overview.pdf")
documents = loader.load()

print("Number of documents:", len(documents))
print(f"\nCONTENT:\n{documents[0].page_content}")
print(f"\nMETADATA:\n{documents[0].metadata}")

• Granularity: 1 PDF page → 1 Document
• Usage: reports, research articles, documentation
• LangChain offers several PDFLoaders; PyPDFLoader is the most common

Web and API Loaders

WebBaseLoader

Extracts readable text from a web page’s HTML (uses beautifulsoup).

# Load multiple web pages as LangChain Documents
from langchain_community.document_loaders import WebBaseLoader

urls = [
    "https://docs.python.org/3/tutorial/interpreter.html",
    "https://docs.python.org/3/tutorial/introduction.html"
]

loader = WebBaseLoader(urls)
documents = loader.load()

print("Number of documents loaded:", len(documents))

for i, doc in enumerate(documents, 1):
    print(f"\n--- Document {i} ---")
    print("\nMetadata:")
    for key, value in doc.metadata.items():
        print(f"{key}: {value}")

    # Clean and preview content
    clean_text = " ".join(doc.page_content.split())
    print(clean_text[:500])

WikipediaLoader

# Load Wikipedia articles as LangChain Documents
from langchain_community.document_loaders import WikipediaLoader

loader = WikipediaLoader(
    query="Artificial Intelligence",
    load_max_docs=2
)

documents = loader.load()
print("Number of documents loaded:", len(documents))

for i, doc in enumerate(documents, 1):
    print(f"\n--- Article {i} ---")
    for key, value in doc.metadata.items():
        print(f"{key}: {value}")
    print(doc.page_content[:400])

ArxivLoader

# Load arXiv article abstracts
from langchain_community.document_loaders import ArxivLoader

loader = ArxivLoader(query="large language models", load_max_docs=2)
documents = loader.load()

for i, doc in enumerate(documents, 1):
    print(f"\n--- Paper {i} ---")
    for key, value in doc.metadata.items():
        print(f"{key}: {value}")
    print(doc.page_content[:500])

Batch Loading — DirectoryLoader

For processing hundreds or thousands of files without writing manual loops.

# Load all .txt files from a directory
from langchain_community.document_loaders import DirectoryLoader, TextLoader

loader = DirectoryLoader(
    path="../data/files",
    glob="**/*.txt",          # ** = recursive search in subdirectories
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"}
)

documents = loader.load()
print("Total Documents Loaded:", len(documents))

for doc in documents:
    print("\nFile:", doc.metadata.get("source"))
    print("Preview:", doc.page_content[:150])
    print(50 * "-")

Common glob patterns:

Lazy Loading

Lazy loading — process documents one by one instead of loading everything into memory.

flowchart TD
    subgraph Eager Loading
        E1[.load()] --> E2[All Documents in memory\n📦📦📦📦📦]
        E2 --> E3[Processing]
    end
    subgraph Lazy Loading
        L1[.lazy_load()] --> L2[Document 1\n📦]
        L2 --> L3[Immediate processing]
        L3 --> L4[Document 2\n📦]
        L4 --> L5[Immediate processing]
        L5 --> L6[...]
    end
    style E2 fill:#ffe0e0
    style L2 fill:#e0ffe0
    style L4 fill:#e0ffe0

# Lazy loading of PDFs from a directory
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader

loader = DirectoryLoader(
    path="../data/files",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader
)

# Lazy loading — each document is generated one at a time
for doc in loader.lazy_load():
    print("\nFile:", doc.metadata.get("source"))
    print("Preview:", doc.page_content[:100])
    print("-" * 50)

    # In production: split, generate embeddings, store in vector DB
    # without ever loading all docs simultaneously

When to use lazy loading:

• Directories with hundreds of files
• Very large PDFs
• Limited system memory

3. Module 2 — Text Splitting Strategies

Why Split Text

flowchart TD
    A[Raw Document\n📄 Large size] --> B{Problems}
    B --> C[LLM context window\nlimit]
    B --> D[Performance degradation\nInfo lost in the middle]
    B --> E[Poor embedding quality\nOne single large vector]
    B --> F[Imprecise retrieval\nToo much noise]

    C --> G[Solution: Text Splitting\n✂️ Focused chunks]
    D --> G
    E --> G
    F --> G

    style A fill:#ffe0e0
    style G fill:#e0ffe0

Reasons to split:

1. Limited context window — every LLM has a maximum token input
2. Model attention — LLMs pay less attention to the middle of long texts
3. Embedding quality — each chunk should represent a focused idea
4. Retrieval accuracy — a targeted chunk is easier to retrieve

Introduction to Text Splitters

In the ingestion pipeline, splitters sit between loaders and embeddings.

flowchart LR
    A[Document\npage_content + metadata] --> B[Text Splitter]
    B --> C[Chunk 1\n🔵 partial content\n+ metadata]
    B --> D[Chunk 2\n🔵 partial content\n+ metadata]
    B --> E[Chunk 3\n🔵 partial content\n+ metadata]
    B --> F[...]

Types of strategies:

CharacterTextSplitter

The simplest splitter — divides based on a separator and a target size.

2-step process:

flowchart TD
    A["Raw text\nAAA\n\nBBBBBB\n\nCCC"]
    A --> B["Step 1: Separation\nSplit by \\n\\n"]
    B --> C[Piece 1: AAA\n100 chars]
    B --> D[Piece 2: BBB\n250 chars]
    B --> E[Piece 3: CCC\n600 chars]
    C --> F["Step 2: Merging\nup to chunk_size=400"]
    D --> F
    E --> F
    F --> G[Chunk 1: AAA+BBB\n350 chars ✅]
    F --> H[Chunk 2: CCC\n600 chars ⚠️ OVER LIMIT]

Note: CharacterTextSplitter does not strictly guarantee chunk_size compliance. If an individual piece exceeds the limit, it is still kept intact.

# Demonstrate splitting with CharacterTextSplitter
from langchain_text_splitters import CharacterTextSplitter

text = (
    "A" * 100 + "\n\n" +
    "B" * 250 + "\n\n" +
    "C" * 600 + "\n\n" +
    "D" * 300 + "\n\n" +
    "E" * 500 + "\n\n" +
    "F" * 100
)

splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=400,
    chunk_overlap=0
)

chunks = splitter.split_text(text)

for i, chunk in enumerate(chunks):
    visible_content = chunk.replace("\n", "\\n")
    length = len(chunk)
    status = "OVER LIMIT" if length > 400 else "OK"
    print(f"Chunk {i+1}: {length} characters {status}")
    print(f"Content: |{visible_content}|")
    print("-" * 40)

Key parameters:

RecursiveCharacterTextSplitter

The recommended default splitter — respects natural text boundaries.

Principle: try several separators in cascade until the target size is reached.

flowchart TD
    A[Text too large] --> B{Try \\n\\n}
    B -- Chunks still too large --> C{Try \\n}
    C -- Chunks still too large --> D{Try '.'}
    D -- Chunks still too large --> E{Try ' '}
    E -- Still too large --> F{Split character by character}
    F --> G[Chunks guaranteed\nwithin limit ✅]
    B -- Chunks OK --> G
    C -- Chunks OK --> G
    D -- Chunks OK --> G
    E -- Chunks OK --> G

Default separator list:

["\n\n", "\n", ".", " ", ""]
# Paragraphs → Lines → Sentences → Words → Characters

# Load and split with RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader("../data/files/sports_news.txt", encoding="utf-8")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ".", " ", ""],
    chunk_size=800,
    add_start_index=True,   # stores start position in metadata
    chunk_overlap=0
)

split_docs = splitter.split_documents(documents)
print("\nTotal chunks created:", len(split_docs))

for i, chunk in enumerate(split_docs):
    print(f"\n--- Chunk {i+1} ---")
    print(f"Size: {len(chunk.page_content)} characters")
    print(f"Metadata:", chunk.metadata)
    print(f"Content: |{chunk.page_content}|")

Advantages over CharacterTextSplitter:

• Guarantees each chunk is within the size limit
• Does not cut in the middle of a sentence or word unless absolutely necessary
• Better embedding and retrieval quality in RAG systems

Chunk Size and Overlap

flowchart LR
    subgraph "Chunks too small"
        A1[🔴 Little context\nFragmented info\nRetrieval may miss\nrelated information]
    end
    subgraph "Chunks too large"
        A2[🔴 Too much information\nLess precise retrieval\nHigh computational cost\nRisk exceeding context window]
    end
    subgraph "Ideal size"
        A3[🟢 Balance\nSufficient context\nPrecise retrieval\nGood embedding quality]
    end

Starting recommendation: ~1000 characters then evaluate performance.

Overlap Concept

Without overlap:
[====Chunk 1====][====Chunk 2====][====Chunk 3====]
                ↑                ↑
           Info lost         Info lost

With overlap (chunk_overlap=100):
[====Chunk 1====]
             [overlap][====Chunk 2====]
                              [overlap][====Chunk 3====]

# Splitting with overlap
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader("../data/files/sports_news.txt", encoding="utf-8")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ".", " ", ""],
    chunk_size=450,
    chunk_overlap=100      # 100 characters of overlap between chunks
)

split_docs = splitter.split_documents(documents)
print("\nTotal chunks created:", len(split_docs))

for i, chunk in enumerate(split_docs):
    print(f"\n--- Chunk {i+1} ---")
    print(chunk.page_content)

Overlap rules:

Specialized Splitters — Markdown, HTML, Code

MarkdownHeaderTextSplitter

# Split a Markdown document based on heading hierarchy
from langchain_text_splitters import MarkdownHeaderTextSplitter

markdown_text = """
# Machine Learning Guide

## Introduction
Machine learning is a field of artificial intelligence that focuses on building systems that learn from data.
These systems improve automatically as they are exposed to more data over time.

## Types of Learning

### Supervised Learning
In supervised learning, models are trained using labeled data.

### Unsupervised Learning
In unsupervised learning, the algorithm discovers patterns without labeled data.

## Applications
Machine learning is used in healthcare, finance, recommendation systems, and many other domains.
"""

headers = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers)
docs = markdown_splitter.split_text(markdown_text)

for i, doc in enumerate(docs, 1):
    print(f"\nChunk {i}")
    print("Metadata:", doc.metadata)
    print(doc.page_content)

Markdown + Recursive combination (for large sections):

from langchain_text_splitters import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter

# Step 1: Split by structure
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers)
docs = markdown_splitter.split_text(markdown_text)

# Step 2: Re-split sections that are too large
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=20
)

final_docs = recursive_splitter.split_documents(docs)
# Metadata (heading hierarchy) is preserved in each final chunk

Source Code Splitting

# Split Python and C code with language structure awareness
from langchain_text_splitters import RecursiveCharacterTextSplitter, Language

python_code = """
def calculate_average(numbers):
    total = 0
    count = 0
    for n in numbers:
        total += n
        count += 1
    return total / count

def find_max(numbers):
    max_value = numbers[0]
    for n in numbers:
        if n > max_value:
            max_value = n
    return max_value

def normalize_data(numbers):
    avg = calculate_average(numbers)
    return [n - avg for n in numbers]
"""

# Use the Python-language-aware splitter
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=300,
    chunk_overlap=0
)

python_docs = python_splitter.create_documents([python_code])
for i, doc in enumerate(python_docs, 1):
    print(f"\n--- Python Chunk {i} ---")
    print(doc.page_content)

4. Module 3 — Embeddings and Vector Representations

Why Embeddings Are Necessary

Keyword search looks for exact matches — it misses meaning:

Query: "How to improve sleep quality at night?"

Keyword results: documents containing exactly "sleep quality"
Embedding results: semantically similar documents such as
  ✅ "Tips for better bedtime habits"
  ✅ "Common causes of insomnia"
  ✅ "Why sleep schedule is important"

Embeddings as Vectors

An embedding is a list of numbers (vector) representing the meaning of text in a high-dimensional space.

flowchart LR
    A["Text\n'Machine learning\nis a branch of AI'"] --> B[Embedding Model]
    B --> C["Vector\n[0.12, -0.45, 0.87,\n 0.23, -0.11, ...]"]
    C --> D["Point in\nN-dimensional space"]

Simplified geometric representation:

         ↑ Dimension 2 (happiness?)
         |
  🟢 "He is very happy"
  🟢 "She feels joyful"    ← similar texts = close together
         |
  -------+-------→ Dimension 1 (emotion?)
         |
  🔴 "The stock fell"      ← different texts = far apart
  🔴 "Cooking pasta recipe"

Properties:

• Dense vectors: most values contain useful information (vs. sparse like bag-of-words)
• Fixed dimension: all vectors have the same number of dimensions, regardless of text length
• Captured semantics: similar-meaning texts → nearby vectors in space

Embedding Models with LangChain

LangChain provides a unified interface for all embedding providers.

Two main methods:

OpenAI Embeddings

# Generate query and document embeddings with OpenAI
from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv

load_dotenv()  # Load OPENAI_API_KEY from .env

# Initialize the model
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Query embedding
query_embedding = embeddings.embed_query("What is machine learning?")
print("\nEmbedding dimension:", len(query_embedding))
print("First 10 values:")
print(query_embedding[:10])

# Embeddings for multiple documents
documents = [
    "Machine learning is a branch of artificial intelligence.",
    "Deep learning uses neural networks with many layers.",
    "Cooking recipes require ingredients and instructions."
]

doc_embeddings = embeddings.embed_documents(documents)
print("\nNumber of embeddings:", len(doc_embeddings))
print("Dimension:", len(doc_embeddings[0]))

Hugging Face Embeddings (Open Source)

# Generate embeddings with Hugging Face
from langchain_huggingface import HuggingFaceEndpointEmbeddings
from dotenv import load_dotenv

load_dotenv()

embeddings = HuggingFaceEndpointEmbeddings(
    model="sentence-transformers/all-MiniLM-L6-v2"   # 384 dimensions
)

query = "What is machine learning?"
query_embedding = embeddings.embed_query(query)
print("\nDimension:", len(query_embedding))  # 384

documents = [
    "Machine learning is a branch of artificial intelligence.",
    "Deep learning uses neural networks with many layers.",
    "Cooking recipes require ingredients and instructions."
]

doc_embeddings = embeddings.embed_documents(documents)
print("Number of embeddings:", len(doc_embeddings))

Similarity Measures

Once texts are converted to vectors, we compare them with similarity measures:

flowchart TD
    subgraph "Euclidean Distance (L2)"
        E["d = √(Σ(a_i - b_i)²)\nSmaller = more similar\nSensitive to magnitude"]
    end
    subgraph "Cosine Similarity"
        C["sim = (A·B) / (‖A‖·‖B‖)\nValue between -1 and 1\nMeasures the angle\n⭐ Most used in NLP"]
    end
    subgraph "Dot Product"
        D["sim = Σ(a_i · b_i)\nLarger = more similar\nSensitive to direction AND magnitude"]
    end

Manual implementation (cosine similarity):

# Rank documents by similarity with a query
from langchain_openai import OpenAIEmbeddings
import numpy as np
from dotenv import load_dotenv

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

query = "How can I improve my sleep quality at night?"

documents = [
    "Penguins live in Antarctica",
    "Tips for better sleep and healthy bedtime habits",
    "The solar system has eight planets",
    "Ways to reduce stress before sleep",
    "How to cook pasta",
    "Meditation helps with relaxation",
    "Why sleep schedule is important",
    "Common causes of insomnia and sleep disorders",
]

# Convert to vectors
query_vector = embeddings.embed_query(query)
doc_vectors = embeddings.embed_documents(documents)

# Cosine similarity function
def cosine_similarity(vec1, vec2):
    dot = np.dot(vec1, vec2)
    norm = np.linalg.norm(vec1) * np.linalg.norm(vec2)
    return dot / norm

# Calculate and sort
results = []
for doc, vec in zip(documents, doc_vectors):
    score = cosine_similarity(query_vector, vec)
    results.append((score, doc))

results.sort(key=lambda x: x[0], reverse=True)

print("\nSimilarity scores:\n")
for score, doc in results:
    print(f"{score:.3f} -> {doc}")

print("\nTop 3 most relevant documents:\n")
for score, doc in results[:3]:
    print(f"{score:.3f} -> {doc}")

Important point: query and document embeddings must use the same model. Each model creates its own vector space — mixing models makes comparisons incoherent.

Dimensions and Performance Trade-offs

quadrantChart
    title Trade-offs: Embedding model size
    x-axis Small --> Large
    y-axis Low semantic quality --> High semantic quality
    quadrant-1 Large model\nHigh quality
    quadrant-2 Ideal not found
    quadrant-3 Small model\nLow quality
    quadrant-4 Small model\nGood quality/cost
    all-MiniLM-L6-v2 (384d): [0.15, 0.45]
    text-embedding-3-small (1536d): [0.55, 0.70]
    text-embedding-3-large (3072d): [0.85, 0.92]

Model comparison:

Evaluating Embedding Quality

Evaluation metrics:

$$\text{Recall@K} = \frac{\text{Relevant docs in top K}}{\text{Total relevant docs}}$$

$$\text{Precision@K} = \frac{\text{Relevant docs in top K}}{K}$$

# Evaluate embedding-based search with Recall@K and Precision@K
from langchain_openai import OpenAIEmbeddings
import numpy as np
from dotenv import load_dotenv

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

query = "How can I improve my sleep quality?"

documents = [
    "Penguins live in Antarctica",
    "Tips for better sleep and healthy bedtime habits",
    "The solar system has eight planets",
    "Ways to reduce stress before sleep",
    "How to cook pasta",
    "Meditation helps with relaxation",
    "Why sleep schedule is important",
    "Exercise improves overall health",
    "How to stay productive during the day",
    "Common causes of insomnia and sleep disorders",
]

# Ground truth (actually relevant documents)
relevant_docs = {
    "Tips for better sleep and healthy bedtime habits",
    "Ways to reduce stress before sleep",
    "Why sleep schedule is important",
    "Common causes of insomnia and sleep disorders"
}

query_vec = embeddings.embed_query(query)
doc_vecs = embeddings.embed_documents(documents)

def cosine_similarity(vec1, vec2):
    dot = np.dot(vec1, vec2)
    norm = np.linalg.norm(vec1) * np.linalg.norm(vec2)
    return dot / norm

results = sorted(
    [(cosine_similarity(query_vec, vec), doc) for doc, vec in zip(documents, doc_vecs)],
    reverse=True
)

k = 3
top_k_docs = [doc for _, doc in results[:k]]
relevant_found = sum(1 for doc in top_k_docs if doc in relevant_docs)

recall_at_k = relevant_found / len(relevant_docs)
precision_at_k = relevant_found / k

print(f"\nRecall@{k}: {recall_at_k:.2f}")
print(f"Precision@{k}: {precision_at_k:.2f}")

5. Module 4 — Vector Stores and Scalable Similarity Search

Why Vector Stores Are Necessary

Problem with the naive approach:

• 1 million documents × 768 dimensions = 768 million values to compare for every query
• Storing in a Python list is impractical at scale

flowchart TD
    subgraph "KNN - Exact Nearest Neighbor"
        K1[Query] --> K2[Compare with ALL vectors]
        K2 --> K3[100% accurate ✅\nSlow for large volume ❌\nO(n) per query]
    end
    subgraph "ANN - Approximate Nearest Neighbor"
        A1[Query] --> A2[Index filters relevant regions]
        A2 --> A3[Compare only a subset]
        A3 --> A4[~95-99% accurate ✅\nFast O(log n) ✅\nIdeal for production]
    end
    style K3 fill:#fff0e0
    style A4 fill:#e0ffe0

Vector store responsibilities:

1. Store vectors
2. Index for fast search (ANN)
3. Perform similarity search

LangChain Unified Interface for Vector Stores

flowchart TD
    App[Application] --> LC[LangChain Interface]
    LC --> FAISS[FAISS\nLocal]
    LC --> Pine[Pinecone\nCloud]
    LC --> Chroma[Chroma\nLocal]
    LC --> Weav[Weaviate\nCloud]

    subgraph "Common Methods"
        M1[add_documents()]
        M2[similarity_search()]
        M3[delete()]
    end

FAISS — Local Vector Store

FAISS (Facebook AI Similarity Search) — ideal for development and small systems.

# FAISS vector store: add, search, delete, update, save
import faiss
from langchain_community.docstore.in_memory import InMemoryDocstore
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from dotenv import load_dotenv

load_dotenv()

# Initialize the embedding model
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Create the FAISS index (dimension must match the model)
dimension = 1536   # text-embedding-3-small = 1536 dimensions
index = faiss.IndexFlatL2(dimension)

# Create the vector store
vector_store = FAISS(
    embedding_function=embeddings,
    index=index,
    docstore=InMemoryDocstore(),   # store original text
    index_to_docstore_id={},       # map vector ↔ document
)

# Create documents
documents = [
    Document(page_content="Penguins live in Antarctica"),
    Document(page_content="Tips for better sleep and healthy bedtime habits"),
    Document(page_content="The solar system has eight planets"),
    Document(page_content="Ways to reduce stress before sleep"),
    Document(page_content="Cook pasta in 5 minutes"),
    Document(page_content="Meditation helps with relaxation"),
]

ids = [f"d{i:03}" for i in range(1, len(documents) + 1)]

# Add documents (automatically generates embeddings)
vector_store.add_documents(documents=documents, ids=ids)

# Similarity search
query = "How can I improve my sleep quality?"
results = vector_store.similarity_search(query, k=3)

print("\nTop results:")
for doc in results:
    print(doc.page_content)

# Delete a document
vector_store.delete(ids=["d004"])

# Update (delete + re-insert)
vector_store.delete(ids=["d005"])
updated_doc = Document(page_content="Good sleep and regular exercise are essential.")
vector_store.add_documents(documents=[updated_doc], ids=["d005"])

# Save to disk
vector_store.save_local("faiss_index")

Load from disk:

# Load FAISS from disk and perform a search
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from dotenv import load_dotenv

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = FAISS.load_local(
    "faiss_index",
    embeddings,
    allow_dangerous_deserialization=True  # Safe if we created the file ourselves
)

query = "How can I improve my sleep quality?"
results = vector_store.similarity_search(query, k=3)

for doc in results:
    print(doc.page_content)

Pinecone — Cloud Vector Store

Pinecone — for production systems with automatic scalability and persistence.

# Pinecone vector store: create index, add, search, delete, update
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from pinecone import Pinecone, ServerlessSpec
from dotenv import load_dotenv
import os

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Initialize Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index_name = "langchain-demo"

# Create the index only if it doesn't exist
if not pc.has_index(index_name):
    pc.create_index(
        name=index_name,
        dimension=1536,       # must match the embedding model
        metric="cosine",      # similarity metric
        spec=ServerlessSpec(
            cloud="aws",
            region="us-east-1"
        )
    )

index = pc.Index(index_name)
vector_store = PineconeVectorStore(index=index, embedding=embeddings)

# Add documents
documents = [
    Document(page_content="Penguins live in Antarctica"),
    Document(page_content="Tips for better sleep and healthy bedtime habits"),
    Document(page_content="Ways to reduce stress before sleep"),
]
ids = [f"d{i:03}" for i in range(1, len(documents) + 1)]

vector_store.add_documents(documents=documents, ids=ids)

# similarity_search, delete, etc. are identical to FAISS
results = vector_store.similarity_search("improve sleep quality", k=3)
for doc in results:
    print(doc.page_content)

FAISS vs Pinecone comparison:

Distance Metrics and Scores

# Search with similarity scores
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from dotenv import load_dotenv

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.load_local("faiss_index", embeddings,
                                allow_dangerous_deserialization=True)

query = "How can I improve my sleep quality?"

# Raw scores (depend on metric)
results_with_score = vector_store.similarity_search_with_score(query, k=3)
for doc, score in results_with_score:
    print(f"Raw L2 score: {score:.4f} | {doc.page_content}")
    # FAISS L2: LOWER score = more similar

# Normalized scores (0 to 1, metric-independent)
results_normalized = vector_store.similarity_search_with_relevance_scores(query, k=3)
for doc, score in results_normalized:
    print(f"Relevance: {score:.4f} | {doc.page_content}")
    # Always: HIGHER score = more relevant

Score interpretation:

Metadata Filtering

Semantic search alone is not sufficient when structural constraints are needed.

flowchart TD
    A[Query: 'leave policy 2026 in India'] --> B{Without filtering}
    B --> C[Semantic results\nmay include:\n❌ India policy 2022\n❌ US policy 2026\n❌ General information]

    A --> D{With metadata filtering\ncountry=India, year=2026}
    D --> E[Filtered results:\n✅ India 2026 policy only]

    style C fill:#ffe0e0
    style E fill:#e0ffe0

# Store documents with metadata and perform filtered search
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from pinecone import Pinecone, ServerlessSpec
from dotenv import load_dotenv
import os

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

# Documents with structured metadata
docs = [
    Document(
        page_content="2026 leave policy for employees in India with updated maternity benefits",
        metadata={"country": "India", "dept": "HR", "year": 2026, "type": "policy"}
    ),
    Document(
        page_content="2023 leave policy for employees in India",
        metadata={"country": "India", "dept": "HR", "year": 2023, "type": "policy"}
    ),
    Document(
        page_content="2026 leave policy for employees in US",
        metadata={"country": "US", "dept": "HR", "year": 2026, "type": "policy"}
    ),
]

# Search without filtering
query = "What is the leave policy for employees?"
results = vector_store.similarity_search(query, k=3)
print("Without filtering:", [doc.page_content[:50] for doc in results])

# Search with metadata filtering
results_filtered = vector_store.similarity_search(
    query,
    k=3,
    filter={"country": "India", "year": 2026}   # only India + 2026
)
print("With filtering:", [doc.page_content[:50] for doc in results_filtered])

MMR — Maximal Marginal Relevance

Problem: similarity search often returns redundant results.

Pure similarity (k=4) for "Applications of vector databases":
1. "Vector databases are used in search engines"
2. "Vector databases improve semantic search"
3. "Vector databases power modern search systems"
4. "Vector databases are used in recommendation systems"
→ All about search! Lack of diversity

MMR (k=4, fetch_k=10):
1. "Vector databases are used in search engines"       ← relevant
2. "Vector databases are used in recommendation systems" ← diversified
3. "Vector databases are applied in fraud detection"   ← diversified
4. "Vector databases are used in image search"         ← diversified
→ Covers multiple applications!

flowchart LR
    A[Query] --> B[ANN Search\nfetch_k=10 candidates]
    B --> C[MMR Re-ranking]
    C --> D[Result 1\nmost relevant]
    C --> E[Result 2\nrelevant AND different from 1]
    C --> F[Result 3\nrelevant AND different from 1 and 2]
    C --> G[Result k\nrelevant AND diverse]
    
    style C fill:#fff0e0

# Compare similarity search and MMR search
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from pinecone import Pinecone, ServerlessSpec
from dotenv import load_dotenv
import os

load_dotenv()

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
# ... index and vector_store initialization

query = "Applications of vector databases"

# Pure similarity search
results_sim = vector_store.similarity_search(query, k=4)
print("\n--- Similarity Search Results ---")
for i, doc in enumerate(results_sim, 1):
    print(f"{i}. {doc.page_content}")

# MMR search (balance relevance + diversity)
results_mmr = vector_store.max_marginal_relevance_search(
    query,
    k=4,          # final number of results
    fetch_k=10    # initial pool before re-ranking
)
print("\n--- MMR Search Results ---")
for i, doc in enumerate(results_mmr, 1):
    print(f"{i}. {doc.page_content}")

When to use MMR:

• In RAG systems to give the LLM more diverse context
• When documents are very similar to each other
• To avoid the LLM receiving repetitive information

6. Module 5 — Retrievers and Advanced Retrieval Strategies

The Retriever Abstraction in LangChain

A retriever takes a query and returns a list of relevant Documents. It hides the complexity of data access.

classDiagram
    class BaseRetriever {
        +invoke(query: str) list[Document]
    }
    class VectorStoreRetriever {
        +vectorstore
        +search_type
        +search_kwargs
        +invoke(query)
    }
    class BM25Retriever {
        +invoke(query)
    }
    class MultiQueryRetriever {
        +retriever: BaseRetriever
        +llm: LLM
        +invoke(query)
    }
    class ContextualCompressionRetriever {
        +base_retriever: BaseRetriever
        +base_compressor
        +invoke(query)
    }
    
    BaseRetriever <|-- VectorStoreRetriever
    BaseRetriever <|-- BM25Retriever
    BaseRetriever <|-- MultiQueryRetriever
    BaseRetriever <|-- ContextualCompressionRetriever

Key point: all retrievers are Runnable in LangChain — they can be chained with .invoke() and integrated into LCEL pipelines.

Types of Retrievers

flowchart TD
    R[LangChain Retrievers]
    R --> S[Source Retrievers\nDirect data access]
    R --> W[Strategy/Wrapper Retrievers\nImprove results]

    S --> S1[VectorStoreRetriever\nFAISS, Pinecone, etc.]
    S --> S2[BM25Retriever\nKeyword search]
    S --> S3[WikipediaRetriever\nWikipedia API]
    S --> S4[Custom retrievers]

    W --> W1[MultiQueryRetriever\nMultiple query versions]
    W --> W2[ContextualCompressionRetriever\nNoise reduction]

BM25 Retriever (keyword search)

# Retrieve best documents with BM25Retriever
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document

docs = [
    Document(page_content="The football world cup is held every four years"),
    Document(page_content="Python is a programming language"),
    Document(page_content="Argentina won the football world cup in 2022"),
    Document(page_content="Cricket is a bat and ball game"),
    Document(page_content="Basketball is played on a court with a hoop"),
    Document(page_content="The weather is nice today"),
]

retriever = BM25Retriever.from_documents(docs, k=2)
results = retriever.invoke("football world cup")

for i, doc in enumerate(results, 1):
    print(f"{i}. {doc.page_content}")

WikipediaRetriever (simple RAG)

# Retrieve Wikipedia context and use an LLM to generate a response
from langchain_community.retrievers import WikipediaRetriever
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

retriever = WikipediaRetriever(top_k_results=3)
llm = ChatOpenAI(model="gpt-4o-mini")

query = "Tell me about the recent Football World Cup"

# Retrieve documents
results = retriever.invoke(query)

# Combine context
context = "\n\n".join([doc.page_content for doc in results])

# Ask the LLM
response = llm.invoke(f"Answer based on this context:\n{context}\n\nQuestion: {query}")
print(response.content)

VectorStore Retriever (most common)

# Demonstrate different retriever strategies on a vector store
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from dotenv import load_dotenv

load_dotenv()

docs = [
    Document(page_content="Sleep improves brain function and memory",
             metadata={"topic": "sleep", "category": "health"}),
    Document(page_content="Good sleep habits improve sleep quality",
             metadata={"topic": "sleep", "category": "health"}),
    Document(page_content="How to cook pasta perfectly",
             metadata={"topic": "cooking", "category": "food"}),
    Document(page_content="Deep sleep is important for recovery",
             metadata={"topic": "sleep", "category": "health"}),
]

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(docs, embeddings)

# Default retriever (k=4)
retriever = vectorstore.as_retriever()
results = retriever.invoke("How to improve sleep quality?")

# Top K retriever
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
results = retriever.invoke("How to improve sleep quality?")

# Retriever with metadata filter
retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 2,
        "filter": {"category": "health"}
    }
)
results = retriever.invoke("How to improve sleep quality?")

# MMR retriever
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 2, "fetch_k": 6}
)
results = retriever.invoke("How to improve sleep quality?")

Multi-Query Retriever

Generates multiple query variants to improve recall.

sequenceDiagram
    participant U as User
    participant MQ as MultiQueryRetriever
    participant LLM as LLM (generation)
    participant BR as Base Retriever
    participant VS as Vector Store

    U->>MQ: "WiFi not working on laptop"
    MQ->>LLM: Generate 3 variants
    LLM-->>MQ: "Laptop WiFi connectivity issues"\n"How to fix wireless adapter"\n"Network not connecting on PC"
    MQ->>BR: Query 1
    BR->>VS: Search
    VS-->>BR: Docs 1,2,3
    MQ->>BR: Query 2
    BR->>VS: Search
    VS-->>BR: Docs 2,4,5
    MQ->>BR: Query 3
    BR->>VS: Search
    VS-->>BR: Docs 3,6,7
    MQ-->>U: Union without duplicates: {1,2,3,4,5,6,7}

# MultiQueryRetriever using LangChain's built-in class
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_classic.retrievers.multi_query import MultiQueryRetriever
from dotenv import load_dotenv

load_dotenv()

docs = [
    Document(page_content="If your laptop is not turning on, check the power supply and battery."),
    Document(page_content="Slow laptop performance can be improved by closing background applications."),
    Document(page_content="WiFi issues on laptops can be caused by outdated drivers."),
    Document(page_content="Restarting the router can fix internet connection problems."),
    Document(page_content="Check network adapter settings if WiFi is not connecting."),
    Document(page_content="Check if airplane mode is enabled if WiFi is not working."),
]

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(docs, embeddings)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

llm = ChatOpenAI(temperature=0)

multi_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=llm
)

query = "WiFi not working on laptop"
results = multi_retriever.invoke(query)

print("Retrieved documents:")
for doc in results:
    print("-", doc.page_content)

Custom LCEL implementation:

# Custom multi-query retriever with LCEL
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from dotenv import load_dotenv

load_dotenv()

# ... (same docs and vectorstore as above)

llm = ChatOpenAI(temperature=0)

# Prompt to generate variants
prompt = ChatPromptTemplate.from_template("""
Generate 3 different versions of the following query.
Return each query on a new line.

Query: {query}
""")

def clean_queries(text: str):
    return [q.strip().lstrip("1234567890.- ") for q in text.split("\n") if q.strip()]

generate_queries_chain = (
    prompt
    | llm
    | StrOutputParser()
    | RunnableLambda(clean_queries)
)

def get_unique_documents(queries, retriever):
    all_docs = []
    for query in queries:
        all_docs.extend(retriever.invoke(query))
    unique = {doc.page_content: doc for doc in all_docs}
    return list(unique.values())

retrieval_chain = RunnableLambda(
    lambda queries: get_unique_documents(queries, base_retriever)
)

multi_query_chain = generate_queries_chain | retrieval_chain

results = multi_query_chain.invoke({"query": "WiFi not working on laptop"})
for doc in results:
    print("-", doc.page_content)

When to use Multi-Query:

• Large data volumes with varied phrasings
• Ambiguous or complex queries
• Simple retriever misses results

When to avoid:

• Fast responses needed
• Small dataset
• Minimizing LLM costs

Contextual Compression

Retrievers often return documents with off-topic information. Contextual compression filters out the noise.

flowchart LR
    A[Query\n"Capital of France?"] --> B[Base Retriever\nTop 2 docs]
    B --> C["Doc 1: 'The capital of France is Paris.\nIt is known for the Eiffel Tower.\nParis has many museums...'"]
    B --> D["Doc 2: 'The capital of India is New Delhi.\nIt is known for its rich history.'"]
    C --> E[Document Compressor]
    D --> E
    E --> F["Compressed Doc 1:\n'The capital of France is Paris.'"]
    E --> G[Doc 2 eliminated\nNot relevant]
    F --> H[LLM\nFocused response]

    style C fill:#fff0e0
    style D fill:#fff0e0
    style F fill:#e0ffe0

# ContextualCompressionRetriever to filter and reduce retrieved content
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_classic.retrievers import ContextualCompressionRetriever
from langchain_classic.retrievers.document_compressors import LLMChainExtractor
from langchain_core.documents import Document
from dotenv import load_dotenv

load_dotenv()

docs = [
    Document(page_content="The capital of Japan is Tokyo. It is famous for its technology and culture."),
    Document(page_content="The capital of India is New Delhi. It is known for its rich history."),
    Document(page_content="The capital of France is Paris. It is known for the Eiffel Tower."),
]

vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# Create the LLM compressor
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
compressor = LLMChainExtractor.from_llm(llm)

# Retriever with compression
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

query = "What is the capital of France?"

# Without compression: 2 full docs
results = base_retriever.invoke(query)
print("Base retriever results:")
for doc in results:
    print("-", doc.page_content.strip())

# With compression: only relevant parts
compressed_results = compression_retriever.invoke(query)
print("\nResults after compression:")
for doc in compressed_results:
    print("-", doc.page_content.strip())

Types of compressors:

Trade-off: compression improves accuracy but adds latency and cost (additional LLM call).

Evaluating Retrieval Systems

flowchart TD
    subgraph "Evaluation Metrics"
        R["Recall@K\n= Relevant docs found / Total relevant docs\nMeasures COVERAGE"]
        P["Precision@K\n= Relevant docs found / K\nMeasures ACCURACY"]
        F["F1@K\n= 2 × (P×R) / (P+R)\nBalance between P and R"]
        N["NDCG\nAccounts for RANK\n(position in results)"]
    end

    style R fill:#e0f0ff
    style P fill:#e0f0ff
    style F fill:#e0ffe0
    style N fill:#fff0e0

Practical example:

Trade-offs in Retrieval Systems

              Precision
                 ▲
                /|\
               / | \
              /  |  \
             /   |   \
            /    |    \
           /     |     \
          ▼------+------▼
        Speed          Cost

Trade-off summary:

7. Module 6 — Structured and Hybrid Queries with LangChain

Querying Structured Data

Fundamental difference:

flowchart LR
    subgraph "Structured Data"
        S1[SQL Tables\nCSV / Excel]
        S2["Analytical queries\nSELECT SUM(sales)\nWHERE region='West'"]
    end
    subgraph "Unstructured Data"
        U1[Texts / PDFs\nEmails / Documents]
        U2["Semantic queries\n'Summarize customer\ncomplaints for Q2'"]
    end
    S1 --> S2
    U1 --> U2

Processing flow — SQL with LangChain:

sequenceDiagram
    participant U as User
    participant LC as LangChain/LLM
    participant DB as Database

    U->>LC: "What is total revenue for 2025?"
    LC->>LC: Understand DB schema
    LC->>LC: Generate SQL: SELECT SUM(sales) FROM orders WHERE year=2025
    LC->>DB: Execute SQL query
    DB-->>LC: [(1234567,)]
    LC->>LC: Formulate response
    LC-->>U: "The total revenue for 2025 was $1,234,567"

Create the test database:

# Create and populate an SQLite database with sales data
import sqlite3

conn = sqlite3.connect("sales.db")
cursor = conn.cursor()

cursor.execute("""
CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY,
    region TEXT,
    sales INTEGER
)
""")

cursor.execute("DELETE FROM orders")

cursor.executemany(
    "INSERT INTO orders (region, sales) VALUES (?, ?)",
    [
        ("West", 100), ("East", 150), ("South", 130),
        ("West", 200), ("North", 180), ("East", 300),
        ("South", 170), ("West", 150), ("North", 220),
        ("East", 250)
    ]
)

conn.commit()
conn.close()
print("Database created.")

Natural Language to SQL — LCEL and Agents

LCEL Pipeline

# LCEL pipeline: SQL generation, execution and response synthesis
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.tools.sql_database.tool import QuerySQLDatabaseTool
from langchain_core.runnables import RunnablePassthrough
from dotenv import load_dotenv

load_dotenv()

db = SQLDatabase.from_uri("sqlite:///sales.db")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
sql_executor = QuerySQLDatabaseTool(db=db)
schema = db.get_table_info()

# SQL generation prompt
sql_prompt = ChatPromptTemplate.from_template("""
You are an SQL expert.

Given this database schema:
{schema}

Write a SQL query to answer:
{question}

Return ONLY the SQL query. Do not include markdown formatting.
""")

sql_generator = (
    {"schema": lambda _: schema, "question": lambda x: x["question"]}
    | sql_prompt
    | llm
    | StrOutputParser()
)

def clean_sql(query: str) -> str:
    query = query.strip()
    if query.startswith("```"):
        query = query.replace("```sql", "").replace("```", "")
    return query.strip()

# Final response prompt
answer_prompt = ChatPromptTemplate.from_template("""
Answer the user's question using the SQL result.

User Question: {question}
SQL Query Used: {query}
SQL Result: {result}
Final Answer: Provide a clear natural language answer based only on the SQL result.
""")

# Full pipeline
full_chain = (
    RunnablePassthrough.assign(query=sql_generator)
    | RunnablePassthrough.assign(query=lambda x: clean_sql(x["query"]))
    | RunnablePassthrough.assign(result=lambda x: sql_executor.invoke(x["query"]))
    | answer_prompt
    | llm
    | StrOutputParser()
)

# Test with multiple questions
questions = [
    "What is total sales in West region?",
    "What is the average sales in East region?",
    "Show total sales grouped by region.",
    "Which region has the highest total sales?"
]

for q in questions:
    print("====================================")
    response = full_chain.invoke({"question": q})
    print(response)

SQL Agent with Security

# Improve agent security by validating SQL queries
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain.agents import create_agent
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

db = SQLDatabase.from_uri("sqlite:///sales.db")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()

base_sql_tool = [t for t in tools if t.name == "sql_db_query"][0]

@tool
def safe_sql_query(query: str) -> str:
    """Executes ONLY secure SELECT queries on the database."""
    q = query.lower()

    if not q.strip().startswith("select"):
        return "Error: Only SELECT queries are allowed."

    forbidden = ["drop", "delete", "update", "insert", "alter"]
    if any(word in q for word in forbidden):
        return "Error: Unsafe SQL detected."

    return base_sql_tool.invoke(query)

# Replace the execution tool with the secure version
safe_tools = [safe_sql_query if t.name == "sql_db_query" else t for t in tools]

agent = create_agent(
    model=llm,
    tools=safe_tools,
    system_prompt="You are a data analyst. Use tools to answer SQL questions. Only SELECT queries allowed."
)

questions = [
    "What is total sales in West region?",
    "Delete all records from orders table",   # will be blocked
    "What is the average sales in East region?"
]

for q in questions:
    print("====================================")
    print("Question:", q)
    response = agent.invoke({"messages": [{"role": "user", "content": q}]})
    print("Response:", response["messages"][-1].content)

SQL Query Reliability and Validation

Two levels of validation:

flowchart TD
    A[User question] --> B[LLM generates SQL]
    B --> C{Manual validation\nsecurity rules}
    C -- SELECT only --> D{LLM Checker\nSQL correct?}
    C -- DROP/DELETE/UPDATE --> E[❌ Blocked]
    D -- Correct --> F[Execute SQL]
    D -- Incorrect --> G[LLM corrects SQL]
    G --> F
    F --> H{Result validation}
    H -- Valid result --> I[LLM generates response]
    H -- Empty/invalid result --> J[Error message]
    I --> K[Final response]

    style E fill:#ffe0e0
    style K fill:#e0ffe0

# LLM-based SQL checker to correct queries before execution
checker_prompt = ChatPromptTemplate.from_template("""
You are a SQL expert.

Given the database schema, the user's question, and the SQL query,
check if the query correctly answers the question.

Database Schema: {schema}
User Question: {question}
SQL Query: {query}

If the query is correct, return it unchanged.
If incorrect, return a corrected SQL query.
Return ONLY the SQL query.
""")

sql_checker = checker_prompt | llm | StrOutputParser()

def check_sql(query: str, question: str) -> str:
    checked_query = sql_checker.invoke({
        "query": query,
        "question": question,
        "schema": schema
    })
    print("\n--- SQL Checker ---")
    print("Original SQL:", query)
    print("Verified SQL:", checked_query)
    return checked_query

Common challenges:

Hybrid Systems — Structured and Unstructured Data

A hybrid system combines SQL (numerical data) and vector search (semantic understanding).

flowchart TD
    A[User question] --> B[Router LLM]
    B -- structured --> C[SQL Pipeline\ndatabase]
    B -- unstructured --> D[Vector Store Retriever\ndocuments]
    B -- hybrid --> E[SQL + Vector\ncombined]
    C --> F[Structured response]
    D --> G[Semantic response]
    E --> H[Complete response\n= numbers + context]

    style C fill:#e0f0ff
    style D fill:#fff0e0
    style E fill:#e0ffe0
    style H fill:#e0ffe0

# Hybrid pipeline: SQL + vector store with intelligent routing
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.utilities import SQLDatabase
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.tools.sql_database.tool import QuerySQLDatabaseTool
from langchain_core.runnables import RunnablePassthrough, RunnableBranch, RunnableLambda
from langchain_community.vectorstores import FAISS
from dotenv import load_dotenv

load_dotenv()

db = SQLDatabase.from_uri("sqlite:///sales.db")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
sql_executor = QuerySQLDatabaseTool(db=db)
schema = db.get_table_info()

# Routing prompt
router_prompt = ChatPromptTemplate.from_template("""
You are routing questions to data sources.
- Requires numeric calculation from database → structured
- Requires document summarization or text → unstructured
- Requires BOTH numeric data AND document insights → hybrid

Return ONLY one word in lowercase: structured, unstructured, or hybrid
Question: {question}
""")

router_chain = (
    router_prompt
    | llm
    | StrOutputParser()
    | RunnableLambda(lambda x: x.strip().lower())
)

# Vector store for unstructured data
vectorstore = FAISS.load_local(
    "faiss_index",
    OpenAIEmbeddings(model="text-embedding-3-small"),
    allow_dangerous_deserialization=True
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# Routing branching is handled by RunnableBranch
# structured  → sql_chain
# unstructured → vector_chain
# hybrid      → both combined

Example questions and their routing:

Choosing the Right Data Access Approach

flowchart TD
    A[Data type?]
    A -- Structured\ntables/DB --> B[Question type?]
    A -- Unstructured\ntexts/docs --> C[Semantic understanding\n→ Document Retrieval]
    A -- Both --> D[Requires both?\n→ Hybrid System]

    B -- Analytical\ntotals/averages --> E[SQL]
    B -- Descriptive\nexplanations --> C

    style E fill:#e0f0ff
    style C fill:#fff0e0
    style D fill:#e0ffe0

Approach comparison table:

Golden rule: Always choose the approach based on the problem, not on the most advanced tool. Simplicity is a virtue.

8. Summary and Best Practices

Overall Architecture of a Complete RAG System

flowchart TD
    subgraph "Ingestion Phase (offline)"
        A1[Data sources\nPDF, Web, CSV, API] -->|Document Loaders| A2[Documents]
        A2 -->|Text Splitters| A3[Chunks]
        A3 -->|Embedding Model| A4[Vectors]
        A4 -->|Vector Store| A5[(Vector index\nFAISS / Pinecone)]
    end

    subgraph "Retrieval Phase (online)"
        B1[User question] -->|Embedding Model| B2[Query vector]
        B2 -->|Similarity Search / MMR| A5
        A5 -->|Top K Documents| B3[Relevant context]
        B3 -->|LLM + Prompt| B4[Final response]
    end

    style A5 fill:#fff0e0,stroke:#e07c00
    style B4 fill:#e8f5e9,stroke:#2e7d32

RAG System Design Checklist

Data Loading

• Identify sources (PDF, web, CSV, DB)
• Choose the appropriate loader for each format
• Use lazy_load for large volumes

Splitting

• Use RecursiveCharacterTextSplitter by default
• Start with chunk_size=1000, chunk_overlap=100-200
• Adapt based on the embedding model (don’t exceed its context window)
• Use specialized splitters for Markdown/HTML/code

Embeddings

• Use the same model for documents and queries
• Consider the cost/quality/speed trade-off
• For prototypes: all-MiniLM-L6-v2 (free, 384d)
• For production: text-embedding-3-small or text-embedding-3-large

Vector Store

• FAISS for development/small volumes
• Pinecone/Weaviate for production
• Add metadata for filtering
• Use MMR to avoid redundancy

Retrieval

• Start with simple VectorStoreRetriever
• Add MultiQueryRetriever if results lack coverage
• Add ContextualCompressionRetriever if too much noise
• Measure with Recall@K, Precision@K, F1@K

SQL Security

• Validate that generated queries are SELECT only
• Block DROP, DELETE, UPDATE, INSERT, ALTER
• Add an LLM checker to correct errors
• Validate empty or invalid results

Decision Flow: Which Retriever to Choose?

flowchart TD
    Q[Is my retriever missing relevant results?]
    Q -- Yes --> MQ[Use MultiQueryRetriever\nMore coverage]
    Q -- No --> R[Do the results contain\ntoo much noise?]
    R -- Yes --> CC[Use ContextualCompressionRetriever\nLess noise]
    R -- No --> D[Are the results too similar?]
    D -- Yes --> MMR[Use MMR\nMore diversity]
    D -- No --> OK[Simple VectorStoreRetriever\nSuffices for this case]

    style OK fill:#e0ffe0
    style MQ fill:#e0f0ff
    style CC fill:#fff0e0
    style MMR fill:#f0e0ff

Summary of Python Packages Used

# Install dependencies
pip install langchain
pip install langchain-community
pip install langchain-openai
pip install langchain-huggingface
pip install langchain-pinecone
pip install langchain-classic      # MultiQueryRetriever, ContextualCompressionRetriever
pip install langchain-text-splitters
pip install faiss-cpu              # or faiss-gpu
pip install pinecone-client
pip install pypdf                  # PyPDFLoader
pip install beautifulsoup4         # WebBaseLoader
pip install wikipedia              # WikipediaLoader
pip install arxiv                  # ArxivLoader
pip install rank-bm25              # BM25Retriever
pip install python-dotenv          # .env loading
pip install numpy                  # cosine_similarity

Environment Variables (.env)

OPENAI_API_KEY=sk-...
PINECONE_API_KEY=pcsk_...
HUGGINGFACEHUB_API_TOKEN=hf_...

Final note: The goal is not to use the most advanced approach, but the simplest one that reliably answers the user’s question. Always choose the approach based on the problem, not the tool.

Search Terms

retrieval · vector · stores · langchain · rag · search · embeddings · artificial · intelligence · generative · ai · data · retriever · sql · loaders · loading · splitting · store · structured · systems · text · concept · document · embedding