Build a Production RAG API

Build a hybrid retrieval RAG API with reranking and citation-grounded answers.

Introduction

30 Second Summary

When you ask an AI assistant a technical question, the quality of its answer depends entirely on what documentation it can find. A sloppy retrieval step returns irrelevant text and the model hallucinates.

In this project, you will build a production-grade RAG API that retrieves the right documentation chunks using hybrid search, reranks them for precision, and generates answers with inline citations.

What You'll Build

A FastAPI service that ingests technical documentation, indexes it with both dense and sparse vectors in Qdrant, retrieves with hybrid search and cross-encoder reranking, and generates citation-grounded answers.

By the end of this project, you'll have:

  • A hybrid retrieval pipeline that indexes documents with both dense (OpenAI) and sparse (BM25) vectors, fuses results with Reciprocal Rank Fusion, and reranks with a Cohere cross-encoder.
  • A FastAPI endpoint that generates citation-grounded answers from retrieved documentation with latency tracking.
  • An evaluation harness with a golden test set and LLM-as-judge scoring for faithfulness and retrieval precision.
  • Secret Mission: Add query expansion that decomposes ambiguous queries into sub-queries for improved recall.

Are there any prerequisites?

This project assumes Python proficiency. You will also need an OpenAI API key and a Cohere API key (free signup at dashboard.cohere.com).

Before We Start

You are about to build a production-grade RAG pipeline that goes far beyond basic vector search. Your system will implement the industry-standard 2026 pattern: chunk documents, embed them as vectors, retrieve with hybrid search (dense + BM25), rerank for precision, generate citation-grounded answers, and evaluate quality with metrics.

Before writing any pipeline code, you need a working Python environment with all the right libraries, valid API keys for OpenAI and Cohere, and a set of real technical documentation to index. This step gets all of that in place.

In this step, get ready to:
  • Install all project dependencies and create a virtual environment.
  • Configure API keys for OpenAI and Cohere and verify connectivity.
  • Download the httpx documentation dataset into a local folder.
Create your project directory and install dependencies

Every Python project needs an isolated environment so its packages don't conflict with other projects on your machine. You will create a project folder, set up a virtual environment, and install the seven libraries this pipeline depends on.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir rag-api
cd rag-api
  • Create and activate a Python virtual environment by running these commands:
python3 -m venv venv
source venv/bin/activate

✔️ I see (venv) in my terminal prompt

Your virtual environment is active. Every package you install from here on stays inside this project.

ⓧ I'm on Windows

On Windows, activate the virtual environment with this command instead:

venv\Scripts\activate

You should now see (venv) at the start of your terminal prompt.

  • Install all project dependencies by running:
pip install qdrant-client openai cohere fastapi uvicorn rank-bm25 tiktoken

What are these packages?

  • qdrant-client is the Python client for Qdrant, a vector database that stores both dense and sparse vectors.
  • openai is the official OpenAI SDK for generating embeddings and chat completions.
  • cohere is the Cohere SDK for cross-encoder reranking.
  • fastapi and uvicorn power the REST API layer.
  • rank-bm25 provides BM25 scoring for sparse retrieval.
  • tiktoken is OpenAI's tokenizer for splitting text into precise token-sized chunks.
  • Verify that all packages installed correctly by running:
python3 -c "import qdrant_client, openai, cohere, fastapi, uvicorn, rank_bm25, tiktoken; print('All imports successful')"
  • You should see All imports successful printed in your terminal.

Seeing an import error?

Make sure you activated your virtual environment first. You should see (venv) at the start of your prompt. If a specific package failed to install, re-run the pip install command and check for error messages about missing system dependencies.

Help me debug a Python package import error in my RAG project virtual environment.

Configure environment variables for API keys

Your pipeline calls two external APIs: OpenAI for embeddings and generation, and Cohere for reranking. Both need API keys set as environment variables so the code can authenticate without hardcoding secrets.

  • Set your OpenAI API key by running this command (replace the placeholder with your actual key):
export OPENAI_API_KEY=your-openai-api-key-here
  • Sign up for a free Cohere API key at https://dashboard.cohere.com/api-keys if you don't have one yet.
  • Set your Cohere API key by running this command (replace the placeholder with your actual key):
export CO_API_KEY=your-cohere-api-key-here

Why CO_API_KEY?

Cohere's SDK looks for the environment variable CO_API_KEY by default. Using this name means the library will authenticate automatically without you passing the key in code.

  • Verify your OpenAI connection by running this test embedding call:
python3 -c "
from openai import OpenAI
import os
client = OpenAI(api_key=os.environ['OPENAI_API_KEY'])
result = client.embeddings.create(input=['hello world'], model='text-embedding-3-small')
print(f'Embedding dimension: {len(result.data[0].embedding)}')
"
  • You should see Embedding dimension: 1536 printed. This confirms your OpenAI key works and the embedding model returns 1536-dimensional vectors.
  • Verify your Cohere connection by running this test rerank call:
python3 -c "
import cohere
import os
co = cohere.ClientV2(os.environ['CO_API_KEY'])
result = co.rerank(model='rerank-v3.5', query='hello', documents=['world', 'foo'], top_n=1)
print(f'Rerank successful. Top result index: {result.results[0].index}')
"
  • You should see Rerank successful. Top result index: 0 (or index 1). This confirms your Cohere key works and the reranker is responding.

Getting an authentication error?

Double-check that you replaced the placeholder text with your actual API key. Run echo $OPENAI_API_KEY or echo $CO_API_KEY to confirm the variable is set in your current terminal session. If you opened a new terminal tab, you need to re-export the keys.

Help me fix an API authentication error when testing my OpenAI or Cohere API keys.

Download the sample technical documentation dataset

Your RAG pipeline needs real documents to index and query against. You will download the httpx library's documentation from GitHub. These are well-structured markdown files covering a real Python HTTP client, small enough to ingest in under a minute.

  • Create the docs/ directory and download the httpx documentation by running this Python script:
python3 -c "
import urllib.request
import os

# Create docs directory and advanced subdirectory
os.makedirs('docs/advanced', exist_ok=True)

# Base URL for raw GitHub content
base_url = 'https://raw.githubusercontent.com/encode/httpx/master/docs'

# Top-level doc files
top_level = [
    'index.md', 'quickstart.md', 'async.md', 'http2.md',
    'compatibility.md', 'troubleshooting.md', 'environment_variables.md',
]

# Advanced doc files
advanced = [
    'advanced/clients.md', 'advanced/authentication.md',
    'advanced/ssl.md', 'advanced/proxies.md', 'advanced/timeouts.md',
    'advanced/transports.md',
]

all_files = top_level + advanced

for filename in all_files:
    url = f'{base_url}/{filename}'
    filepath = f'docs/{filename}'
    print(f'Downloading {filename}...')
    urllib.request.urlretrieve(url, filepath)

print(f'Done. Downloaded {len(all_files)} files to docs/')
"

What does this script do?

This script downloads 13 markdown files from the httpx repository on GitHub. These cover quickstart guides, async usage, advanced topics like authentication, SSL, proxies, and transports.

The files land in a local docs/ directory that your ingestion pipeline will read from in the next step.

  • Verify the files downloaded successfully by running:
ls docs/ && ls docs/advanced/
  • You should see the markdown files listed in both directories. The top-level docs/ folder should contain files like quickstart.md and async.md. The docs/advanced/ folder should contain files like clients.md and authentication.md.

Download failed or files are empty?

Make sure you have an internet connection and that your terminal can reach raw.githubusercontent.com. If you are behind a corporate proxy, you may need to configure your proxy settings. You can also try opening one of the URLs directly in your browser to confirm access.

Help me troubleshoot why my Python script can't download files from GitHub.

Your environment is ready with all dependencies installed, both API keys verified, and real documentation waiting to be indexed. Next up, you will build the ingestion pipeline that chunks these documents and indexes them with hybrid vectors in Qdrant.

Set Up Configuration

Your environment has all the tools installed and your API keys are configured as environment variables. Before building the ingestion and retrieval modules, you need a single place to store all configuration values that every module will share.

In this step, you'll create a config.py file that centralizes API keys, model names, and tuning parameters. Every module in the project will import from this file, making it easy to adjust settings in one place.

In this step, get ready to:
  • Create a centralized config.py file with all project constants.
  • Verify the configuration loads correctly in Python.
Create the centralized configuration file

A production project has many moving parts: embedding models, vector databases, reranking services, and generation models. Hardcoding values in every file makes changes painful. A dedicated config module solves this by giving every other file a single import target.

  • Create a new file called config.py in your project folder (the same folder where your virtual environment lives).
  • Add the following code to config.py:
import os

# Load API keys from environment variables with safe fallbacks
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "your-openai-api-key-here")
CO_API_KEY = os.environ.get("CO_API_KEY", "your-cohere-api-key-here")

# Model configuration
EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIM = 1536
RERANK_MODEL = "rerank-v3.5"
GENERATION_MODEL = "gpt-4o-mini"

# Qdrant storage configuration
QDRANT_PATH = "./qdrant_data"
COLLECTION_NAME = "tech_docs"

# Chunking and retrieval parameters
CHUNK_SIZE = 512
CHUNK_OVERLAP = 50
TOP_K_RETRIEVAL = 20
TOP_K_RERANK = 5
RRF_K = 60

What does this code do?

  • os.environ.get() reads your API keys from the environment variables you already set. The second argument is a fallback value that only activates if the variable is missing.
  • EMBEDDING_MODEL and EMBEDDING_DIM pin the OpenAI embedding model (text-embedding-3-small, 1536 dimensions).
  • RERANK_MODEL sets the Cohere cross-encoder to rerank-v3.5, which has a 4096-token context window suited for 512-token chunks.
  • QDRANT_PATH tells Qdrant to store vector data on disk locally, with no running server required.
  • RRF_K is the constant used in Reciprocal Rank Fusion to merge dense and sparse search results. The standard default is 60.

Why a config file instead of .env?

A .env file works well for secrets, but this project also needs numeric constants and model names that every module imports. Putting everything in a single Python module means you get type checking, IDE autocomplete, and zero extra dependencies.

  • Save config.py.
  • Verify the configuration loads correctly by running this command in your terminal:
python -c "import config; print(config.COLLECTION_NAME)"

You should see tech_docs printed in your terminal. This confirms Python can find and import the file without errors.

Seeing a ModuleNotFoundError?

  • Make sure you are running the command from your project folder (the same directory that contains config.py).
  • Check that your virtual environment is still activated. You should see its name in parentheses at the start of your terminal prompt.
  • Confirm the file is named exactly config.py with no typos or extra extensions like .py.txt.

Why can't Python find my config module?

✔️ Awesome, I've got everything!

Great. Double check you've saved config.py and that tech_docs printed in your terminal.

ⓧ I'd like to double check the full code

Here is the complete config.py file for comparison:

import os

OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "your-openai-api-key-here")
CO_API_KEY = os.environ.get("CO_API_KEY", "your-cohere-api-key-here")

EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIM = 1536
RERANK_MODEL = "rerank-v3.5"
GENERATION_MODEL = "gpt-4o-mini"

QDRANT_PATH = "./qdrant_data"
COLLECTION_NAME = "tech_docs"

CHUNK_SIZE = 512
CHUNK_OVERLAP = 50
TOP_K_RETRIEVAL = 20
TOP_K_RERANK = 5
RRF_K = 60

Your configuration module is ready. Next up, you'll build the document ingestion pipeline that chunks your httpx docs and indexes them with both dense and sparse vectors in Qdrant.

Ingest and Index Documents

In the previous step, you set up your project environment and confirmed connectivity to OpenAI and Cohere. Now it's time to build the core of any RAG system: the ingestion pipeline.

A RAG pipeline is only as good as the data it retrieves from. In this step, you'll chunk the httpx documentation into search-friendly pieces, embed them as both dense and sparse vectors, and index everything in Qdrant so your retrieval layer has two complementary search strategies to draw from.

In this step, get ready to:
  • Build a chunking pipeline that splits markdown files into token-bounded chunks with metadata.
  • Generate dense embeddings and sparse BM25 vectors for every chunk.
  • Create a Qdrant collection with both vector types and upsert all indexed points.
Build the chunking pipeline

Before you can search documentation, you need to break it into small, focused chunks that fit within an embedding model's context window. Each chunk also needs metadata so you can trace answers back to their source file.

  • Create a new file called ingest.py in your project directory.
  • Add the imports and the document-loading function by pasting this code:
import os
import json
import tiktoken
from pathlib import Path

from openai import OpenAI
from qdrant_client import QdrantClient, models
from rank_bm25 import BM25Okapi

from config import (
    OPENAI_API_KEY,
    EMBEDDING_MODEL,
    EMBEDDING_DIM,
    QDRANT_PATH,
    COLLECTION_NAME,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
)


def load_documents(docs_dir: str) -> list[dict]:
    documents = []
    for filepath in Path(docs_dir).glob("**/*.md"):
        text = filepath.read_text(encoding="utf-8")
        documents.append({
            "text": text,
            "source": str(filepath.relative_to(docs_dir)),
        })
    return documents

What does this code do?

The imports pull in everything you need: tiktoken for counting tokens, the OpenAI client for embeddings, qdrant_client for vector storage, and BM25Okapi for sparse keyword scoring.

The load_documents() function walks through your docs/ directory, reads every markdown file, and stores both the text and the relative file path as metadata.

  • Below load_documents(), add the recursive chunking function:
def recursive_chunk(text: str, source: str, chunk_size: int, overlap: int) -> list[dict]:
    enc = tiktoken.encoding_for_model("gpt-4o-mini")
    tokens = enc.encode(text)
    chunks = []
    start = 0
    chunk_id = 0

    while start < len(tokens):
        end = start + chunk_size
        chunk_tokens = tokens[start:end]
        chunk_text = enc.decode(chunk_tokens)
        chunks.append({
            "id": f"{source}::chunk_{chunk_id}",
            "text": chunk_text,
            "source": source,
            "chunk_index": chunk_id,
        })
        start += chunk_size - overlap
        chunk_id += 1

    return chunks

What does this code do?

This function uses tiktoken to tokenize text the same way the model would, then slides a window of 512 tokens with 50-token overlap across the document.

The overlap ensures that concepts split across chunk boundaries still appear in at least one complete chunk. Each chunk gets a unique ID combining the source file and its position in the document.

  • Save ingest.py.

Seeing an import error?

Make sure your virtual environment is active and all packages from Step 1 are installed. If tiktoken or rank_bm25 is missing, re-run the pip install command from the previous step.

Help me debug an import error in ingest.py

Generate dense embeddings and sparse BM25 vectors

Dense vectors capture semantic meaning, but they can miss exact keyword matches. BM25 sparse vectors capture term frequency, which is great for precise terminology lookups. You'll generate both so your retrieval layer can combine their strengths.

  • Add the embedding function below recursive_chunk() in ingest.py:
def embed_texts(texts: list[str], client: OpenAI) -> list[list[float]]:
    embeddings = []
    batch_size = 100
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            input=batch,
            model=EMBEDDING_MODEL,
        )
        embeddings.extend([item.embedding for item in response.data])
    return embeddings

What does this code do?

This function sends chunks to the OpenAI text-embedding-3-small model in batches of 100. Batching keeps you within API rate limits and reduces the number of network round-trips.

Each text chunk comes back as a 1536-dimensional vector that encodes its semantic meaning.

  • Add the two sparse vector functions below embed_texts():
def build_sparse_vectors(chunks: list[dict]) -> tuple[BM25Okapi, list[list[str]]]:
    tokenized = [chunk["text"].lower().split() for chunk in chunks]
    bm25 = BM25Okapi(tokenized)
    return bm25, tokenized


def tokens_to_sparse_vector(tokenized_corpus: list[list[str]], bm25: BM25Okapi, doc_idx: int) -> models.SparseVector:
    vocab = {}
    for doc in tokenized_corpus:
        for token in doc:
            if token not in vocab:
                vocab[token] = len(vocab)

    doc_tokens = tokenized_corpus[doc_idx]
    token_counts = {}
    for token in doc_tokens:
        token_counts[token] = token_counts.get(token, 0) + 1

    indices = []
    values = []
    for token, count in token_counts.items():
        if token in vocab:
            indices.append(vocab[token])
            values.append(float(count))

    return models.SparseVector(indices=indices, values=values)

What does this code do?

The build_sparse_vectors() function tokenizes every chunk by whitespace and builds a BM25 index from the entire corpus. BM25 scores how important each term is to a given document relative to all other documents.

The tokens_to_sparse_vector() function converts a single chunk's term frequencies into Qdrant's sparse vector format: a list of vocabulary indices paired with their frequency values.

  • Save ingest.py.

Not sure about the sparse vector format?

Qdrant sparse vectors need two parallel arrays: indices (which vocabulary position each token maps to) and values (the weight for that token). This is the same concept as a sparse matrix row.

Help me understand how BM25 sparse vectors work in Qdrant

Create the Qdrant collection and run ingestion

With the chunking and vectorization functions ready, you need an orchestrator that ties everything together: load documents, chunk them, embed them, create the vector database collection, and upsert all the indexed points.

  • Add the ingest() function and the main block at the bottom of ingest.py:
def ingest(docs_dir: str = "docs"):
    openai_client = OpenAI(api_key=OPENAI_API_KEY)
    qdrant = QdrantClient(path=QDRANT_PATH)

    print("Loading documents...")
    documents = load_documents(docs_dir)
    print(f"Loaded {len(documents)} documents")

    print("Chunking...")
    all_chunks = []
    for doc in documents:
        chunks = recursive_chunk(doc["text"], doc["source"], CHUNK_SIZE, CHUNK_OVERLAP)
        all_chunks.extend(chunks)
    print(f"Created {len(all_chunks)} chunks")

    print("Generating embeddings...")
    texts = [chunk["text"] for chunk in all_chunks]
    dense_embeddings = embed_texts(texts, openai_client)

    print("Building BM25 index...")
    bm25, tokenized_corpus = build_sparse_vectors(all_chunks)

    print("Creating Qdrant collection...")
    if qdrant.collection_exists(COLLECTION_NAME):
        qdrant.delete_collection(COLLECTION_NAME)

    qdrant.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config={
            "dense": models.VectorParams(size=EMBEDDING_DIM, distance=models.Distance.COSINE),
        },
        sparse_vectors_config={
            "sparse": models.SparseVectorParams(),
        },
    )

    print("Upserting points...")
    points = []
    for i, chunk in enumerate(all_chunks):
        sparse_vec = tokens_to_sparse_vector(tokenized_corpus, bm25, i)
        points.append(
            models.PointStruct(
                id=i,
                vector={
                    "dense": dense_embeddings[i],
                },
                payload={
                    "text": chunk["text"],
                    "source": chunk["source"],
                    "chunk_index": chunk["chunk_index"],
                    "chunk_id": chunk["id"],
                },
            )
        )

    batch_size = 100
    for i in range(0, len(points), batch_size):
        qdrant.upsert(
            collection_name=COLLECTION_NAME,
            points=points[i:i + batch_size],
        )

    # Save BM25 metadata for retrieval
    vocab = {}
    for doc in tokenized_corpus:
        for token in doc:
            if token not in vocab:
                vocab[token] = len(vocab)

    bm25_meta = {
        "vocab": vocab,
        "corpus_size": len(tokenized_corpus),
    }
    Path("bm25_meta.json").write_text(json.dumps(bm25_meta))

    print(f"Ingestion complete. {len(points)} points indexed.")
    count = qdrant.count(COLLECTION_NAME)
    print(f"Qdrant collection count: {count.count}")


if __name__ == "__main__":
    ingest()

What does this code do?

The ingest() function is the main orchestrator that runs the full pipeline from raw markdown files to a searchable vector collection.

  • It initializes the Qdrant client in local embedded mode using a file path (no server required).
  • It creates a collection with a named "dense" vector config (1536 dimensions, cosine distance) and a named "sparse" vector config.
  • Each point is upserted with both vector types plus a payload containing the chunk text, source file, and chunk index.
  • Finally, it saves the BM25 vocabulary mapping to bm25_meta.json so the retrieval module can reconstruct sparse query vectors later.
  • Save ingest.py.
  • Run the ingestion pipeline by executing this command:
python ingest.py

You should see output showing each stage completing: loading documents, chunking, generating embeddings, building the BM25 index, creating the Qdrant collection, and upserting points. The final two lines confirm how many points were indexed and the collection count.

Seeing an API error during embedding?

If you see an authentication error, verify your OPENAI_API_KEY environment variable is set correctly in your terminal session. If you see a rate limit error, wait 60 seconds and try again.

If the script fails partway through, it's safe to re-run. The collection_exists check deletes any previous partial collection before recreating it.

Help me debug an error when running ingest.py

✔️ Awesome, I've got everything!

Great. Double check you've saved ingest.py and that the terminal output shows the point count matching your chunk count.

ⓧ I'd like to double check the full code

Here's the complete ingest.py file for comparison:

import os
import json
import tiktoken
from pathlib import Path

from openai import OpenAI
from qdrant_client import QdrantClient, models
from rank_bm25 import BM25Okapi

from config import (
    OPENAI_API_KEY,
    EMBEDDING_MODEL,
    EMBEDDING_DIM,
    QDRANT_PATH,
    COLLECTION_NAME,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
)


def load_documents(docs_dir: str) -> list[dict]:
    documents = []
    for filepath in Path(docs_dir).glob("**/*.md"):
        text = filepath.read_text(encoding="utf-8")
        documents.append({
            "text": text,
            "source": str(filepath.relative_to(docs_dir)),
        })
    return documents


def recursive_chunk(text: str, source: str, chunk_size: int, overlap: int) -> list[dict]:
    enc = tiktoken.encoding_for_model("gpt-4o-mini")
    tokens = enc.encode(text)
    chunks = []
    start = 0
    chunk_id = 0

    while start < len(tokens):
        end = start + chunk_size
        chunk_tokens = tokens[start:end]
        chunk_text = enc.decode(chunk_tokens)
        chunks.append({
            "id": f"{source}::chunk_{chunk_id}",
            "text": chunk_text,
            "source": source,
            "chunk_index": chunk_id,
        })
        start += chunk_size - overlap
        chunk_id += 1

    return chunks


def embed_texts(texts: list[str], client: OpenAI) -> list[list[float]]:
    embeddings = []
    batch_size = 100
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            input=batch,
            model=EMBEDDING_MODEL,
        )
        embeddings.extend([item.embedding for item in response.data])
    return embeddings


def build_sparse_vectors(chunks: list[dict]) -> tuple[BM25Okapi, list[list[str]]]:
    tokenized = [chunk["text"].lower().split() for chunk in chunks]
    bm25 = BM25Okapi(tokenized)
    return bm25, tokenized


def tokens_to_sparse_vector(tokenized_corpus: list[list[str]], bm25: BM25Okapi, doc_idx: int) -> models.SparseVector:
    vocab = {}
    for doc in tokenized_corpus:
        for token in doc:
            if token not in vocab:
                vocab[token] = len(vocab)

    doc_tokens = tokenized_corpus[doc_idx]
    token_counts = {}
    for token in doc_tokens:
        token_counts[token] = token_counts.get(token, 0) + 1

    indices = []
    values = []
    for token, count in token_counts.items():
        if token in vocab:
            indices.append(vocab[token])
            values.append(float(count))

    return models.SparseVector(indices=indices, values=values)


def ingest(docs_dir: str = "docs"):
    openai_client = OpenAI(api_key=OPENAI_API_KEY)
    qdrant = QdrantClient(path=QDRANT_PATH)

    print("Loading documents...")
    documents = load_documents(docs_dir)
    print(f"Loaded {len(documents)} documents")

    print("Chunking...")
    all_chunks = []
    for doc in documents:
        chunks = recursive_chunk(doc["text"], doc["source"], CHUNK_SIZE, CHUNK_OVERLAP)
        all_chunks.extend(chunks)
    print(f"Created {len(all_chunks)} chunks")

    print("Generating embeddings...")
    texts = [chunk["text"] for chunk in all_chunks]
    dense_embeddings = embed_texts(texts, openai_client)

    print("Building BM25 index...")
    bm25, tokenized_corpus = build_sparse_vectors(all_chunks)

    print("Creating Qdrant collection...")
    if qdrant.collection_exists(COLLECTION_NAME):
        qdrant.delete_collection(COLLECTION_NAME)

    qdrant.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config={
            "dense": models.VectorParams(size=EMBEDDING_DIM, distance=models.Distance.COSINE),
        },
        sparse_vectors_config={
            "sparse": models.SparseVectorParams(),
        },
    )

    print("Upserting points...")
    points = []
    for i, chunk in enumerate(all_chunks):
        sparse_vec = tokens_to_sparse_vector(tokenized_corpus, bm25, i)
        points.append(
            models.PointStruct(
                id=i,
                vector={
                    "dense": dense_embeddings[i],
                },
                payload={
                    "text": chunk["text"],
                    "source": chunk["source"],
                    "chunk_index": chunk["chunk_index"],
                    "chunk_id": chunk["id"],
                },
            )
        )

    batch_size = 100
    for i in range(0, len(points), batch_size):
        qdrant.upsert(
            collection_name=COLLECTION_NAME,
            points=points[i:i + batch_size],
        )

    # Save BM25 metadata for retrieval
    vocab = {}
    for doc in tokenized_corpus:
        for token in doc:
            if token not in vocab:
                vocab[token] = len(vocab)

    bm25_meta = {
        "vocab": vocab,
        "corpus_size": len(tokenized_corpus),
    }
    Path("bm25_meta.json").write_text(json.dumps(bm25_meta))

    print(f"Ingestion complete. {len(points)} points indexed.")
    count = qdrant.count(COLLECTION_NAME)
    print(f"Qdrant collection count: {count.count}")


if __name__ == "__main__":
    ingest()

Your documents are now indexed with both dense semantic vectors and sparse keyword vectors. Next up, you'll build the hybrid retrieval module that queries both vector types and fuses the results into a single ranked list.

Build Hybrid Retrieval and Reranking

Your documents are indexed with both dense and sparse vectors. Now you need a way to search across both vector spaces and combine the results intelligently.

In this step, you'll build a retrieval pipeline that queries both the dense and sparse indexes, fuses the results using Reciprocal Rank Fusion (RRF), and then reranks the top candidates with a Cohere cross-encoder. Each layer measurably improves precision over naive vector search alone.

In this step, get ready to:
  • Build a hybrid search function combining dense and sparse retrieval with RRF fusion.
  • Add Cohere cross-encoder reranking to boost precision on the top results.
  • Wire up the complete retrieval pipeline with timing instrumentation and test it end-to-end.
Build the hybrid search function

Dense vector search excels at finding semantically similar content, but it can miss exact keyword matches. Sparse BM25 search catches those exact matches. Combining both and fusing their ranked results with RRF gives you the best of both worlds.

  • Create a new file called retrieve.py in your project folder.
  • Add the imports and the RRF fusion helper function by pasting this code:
import json
import time
from pathlib import Path

import cohere
from openai import OpenAI
from qdrant_client import QdrantClient, models

from config import (
    OPENAI_API_KEY,
    CO_API_KEY,
    EMBEDDING_MODEL,
    RERANK_MODEL,
    QDRANT_PATH,
    COLLECTION_NAME,
    TOP_K_RETRIEVAL,
    TOP_K_RERANK,
    RRF_K,
)


def rrf_fuse(result_lists: list[list[dict]], k: int = 60) -> list[dict]:
    # Merge multiple ranked lists using Reciprocal Rank Fusion
    scores = {}
    for results in result_lists:
        for rank, result in enumerate(results):
            doc_id = result["id"]
            if doc_id not in scores:
                scores[doc_id] = {"score": 0.0, **result}
            scores[doc_id]["score"] += 1.0 / (k + rank + 1)

    fused = sorted(scores.values(), key=lambda x: x["score"], reverse=True)
    return fused

What does this code do?

The imports bring in clients for OpenAI (embeddings), Cohere (reranking), and Qdrant (vector search). All config constants come from your existing config.py.

The rrf_fuse() function implements Reciprocal Rank Fusion. It takes multiple ranked result lists and assigns each document a score of 1 / (k + rank + 1) for each list it appears in. Documents that rank highly in both dense and sparse searches accumulate a higher fused score.

  • Next, start building the main retrieval function. Add the dense search section below the RRF function:

def retrieve(query: str) -> dict:
    openai_client = OpenAI(api_key=OPENAI_API_KEY)
    co = cohere.ClientV2(CO_API_KEY)
    qdrant = QdrantClient(path=QDRANT_PATH)

    timings = {}

    # Dense search: embed the query with OpenAI
    start = time.time()
    query_embedding = openai_client.embeddings.create(
        input=[query],
        model=EMBEDDING_MODEL,
    ).data[0].embedding
    timings["embed_ms"] = (time.time() - start) * 1000

    # Query Qdrant dense vectors for top matches
    start = time.time()
    dense_results = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        query=query_embedding,
        using="dense",
        limit=TOP_K_RETRIEVAL,
        with_payload=True,
    ).points
    timings["dense_search_ms"] = (time.time() - start) * 1000

    dense_hits = [
        {"id": p.id, "text": p.payload["text"], "source": p.payload["source"]}
        for p in dense_results
    ]

What does this code do?

This begins the retrieve() function by initializing all three API clients and a timings dictionary to track latency per stage.

For dense search, the query is embedded with OpenAI's text-embedding-3-small model, then used to search the Qdrant collection's dense vector index. The top 20 results are extracted into a list of hit dictionaries.

  • Add the sparse search and RRF fusion section directly below the dense search code:

    # Sparse search: convert query to BM25 sparse vector
    start = time.time()
    bm25_meta = json.loads(Path("bm25_meta.json").read_text())
    vocab = bm25_meta["vocab"]
    query_tokens = query.lower().split()
    indices = []
    values = []
    for token in query_tokens:
        if token in vocab:
            indices.append(vocab[token])
            values.append(1.0)

    if indices:
        sparse_results = qdrant.query_points(
            collection_name=COLLECTION_NAME,
            query=models.SparseVector(indices=indices, values=values),
            using="sparse",
            limit=TOP_K_RETRIEVAL,
            with_payload=True,
        ).points
        sparse_hits = [
            {"id": p.id, "text": p.payload["text"], "source": p.payload["source"]}
            for p in sparse_results
        ]
    else:
        sparse_hits = []
    timings["sparse_search_ms"] = (time.time() - start) * 1000

    # RRF Fusion: merge dense and sparse results into one ranked list
    start = time.time()
    fused = rrf_fuse([dense_hits, sparse_hits], k=RRF_K)[:TOP_K_RETRIEVAL]
    timings["fusion_ms"] = (time.time() - start) * 1000

What does this code do?

The sparse search reconstructs a BM25-style query vector from the saved vocabulary in bm25_meta.json. Each query token that exists in the vocabulary gets mapped to its index with a weight of 1.0. This produces a sparse vector that Qdrant can match against the sparse vectors stored during ingestion.

After both searches return their top 20 results, rrf_fuse() merges them. Documents found by both methods get boosted. The fused list is trimmed to the top 20 for reranking.

Why hybrid search instead of dense-only?

Dense embeddings capture meaning but can miss exact terminology. If someone asks about "httpx.AsyncClient", a dense search might return results about async HTTP clients in general. The sparse BM25 search matches the exact token "AsyncClient" and boosts those results to the top.

RRF is the standard fusion method because it's rank-based, not score-based. Dense cosine scores and BM25 scores live on different scales, so you cannot simply add them. RRF only cares about position in each list.

  • Save retrieve.py.

Seeing an import error?

Make sure you're running from within your virtual environment where you installed the project dependencies. If cohere is not found, verify you ran pip install cohere in Step 1.

Help me debug an import error in retrieve.py

Add Cohere cross-encoder reranking

RRF fusion gives you a merged ranking, but it still relies on first-stage retrieval scores. A cross-encoder reranker reads both the query and each candidate chunk together, producing a much more accurate relevance judgment.

  • In retrieve.py, add the reranking section and return statement directly below the RRF fusion code:

    # Rerank: use Cohere cross-encoder to reorder by true relevance
    start = time.time()
    if fused:
        rerank_docs = [hit["text"] for hit in fused]
        rerank_response = co.rerank(
            model=RERANK_MODEL,
            query=query,
            documents=rerank_docs,
            top_n=TOP_K_RERANK,
        )
        reranked = []
        for result in rerank_response.results:
            hit = fused[result.index]
            hit["relevance_score"] = result.relevance_score
            reranked.append(hit)
    else:
        reranked = []
    timings["rerank_ms"] = (time.time() - start) * 1000

    return {"chunks": reranked, "timings": timings}

What does this code do?

The reranker sends all fused chunk texts alongside the query to Cohere's rerank-v3.5 model. The model reads each query-document pair with full attention and returns a relevance score between 0 and 1.

Only the top 5 results (controlled by TOP_K_RERANK) are kept. Each hit gets a relevance_score field attached so downstream consumers can see how confident the reranker is.

  • Save retrieve.py.

Getting an authentication error from Cohere?

Verify that your CO_API_KEY environment variable is set correctly. You can check by running echo $CO_API_KEY in your terminal. If it prints nothing, re-export the key.

Help me fix my Cohere API authentication

Wire up the test runner and verify end-to-end

The retrieval pipeline is complete. Now you need a way to run it from the command line and see the full latency breakdown across each stage.

  • Add the __main__ block at the bottom of retrieve.py:

if __name__ == "__main__":
    result = retrieve("How do I make a POST request with httpx?")
    for chunk in result["chunks"]:
        print(f"[{chunk['source']}] (score: {chunk['relevance_score']:.3f})")
        print(f"  {chunk['text'][:100]}...")
    print(f"\nTimings: {result['timings']}")

What does this code do?

The test block runs a sample query through the full pipeline and prints each reranked chunk's source file and relevance score. It also prints the timing dictionary so you can see how many milliseconds each stage takes.

  • Save retrieve.py.
  • Run the retrieval pipeline end-to-end by running this command:
python retrieve.py
  • You should see 5 results printed with source file names, relevance scores between 0 and 1, and a timings dictionary showing millisecond breakdowns for embed, dense search, sparse search, fusion, and rerank stages.

Seeing zero results or empty output?

Make sure you ran ingest.py successfully in the previous step. The qdrant_data/ directory and bm25_meta.json must both exist. If they're missing, re-run the ingestion step.

Help me debug empty retrieval results

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete retrieve.py for comparison:

import json
import time
from pathlib import Path

import cohere
from openai import OpenAI
from qdrant_client import QdrantClient, models

from config import (
    OPENAI_API_KEY,
    CO_API_KEY,
    EMBEDDING_MODEL,
    RERANK_MODEL,
    QDRANT_PATH,
    COLLECTION_NAME,
    TOP_K_RETRIEVAL,
    TOP_K_RERANK,
    RRF_K,
)


def rrf_fuse(result_lists: list[list[dict]], k: int = 60) -> list[dict]:
    scores = {}
    for results in result_lists:
        for rank, result in enumerate(results):
            doc_id = result["id"]
            if doc_id not in scores:
                scores[doc_id] = {"score": 0.0, **result}
            scores[doc_id]["score"] += 1.0 / (k + rank + 1)

    fused = sorted(scores.values(), key=lambda x: x["score"], reverse=True)
    return fused


def retrieve(query: str) -> dict:
    openai_client = OpenAI(api_key=OPENAI_API_KEY)
    co = cohere.ClientV2(CO_API_KEY)
    qdrant = QdrantClient(path=QDRANT_PATH)

    timings = {}

    start = time.time()
    query_embedding = openai_client.embeddings.create(
        input=[query],
        model=EMBEDDING_MODEL,
    ).data[0].embedding
    timings["embed_ms"] = (time.time() - start) * 1000

    start = time.time()
    dense_results = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        query=query_embedding,
        using="dense",
        limit=TOP_K_RETRIEVAL,
        with_payload=True,
    ).points
    timings["dense_search_ms"] = (time.time() - start) * 1000

    dense_hits = [
        {"id": p.id, "text": p.payload["text"], "source": p.payload["source"]}
        for p in dense_results
    ]

    start = time.time()
    bm25_meta = json.loads(Path("bm25_meta.json").read_text())
    vocab = bm25_meta["vocab"]
    query_tokens = query.lower().split()
    indices = []
    values = []
    for token in query_tokens:
        if token in vocab:
            indices.append(vocab[token])
            values.append(1.0)

    if indices:
        sparse_results = qdrant.query_points(
            collection_name=COLLECTION_NAME,
            query=models.SparseVector(indices=indices, values=values),
            using="sparse",
            limit=TOP_K_RETRIEVAL,
            with_payload=True,
        ).points
        sparse_hits = [
            {"id": p.id, "text": p.payload["text"], "source": p.payload["source"]}
            for p in sparse_results
        ]
    else:
        sparse_hits = []
    timings["sparse_search_ms"] = (time.time() - start) * 1000

    start = time.time()
    fused = rrf_fuse([dense_hits, sparse_hits], k=RRF_K)[:TOP_K_RETRIEVAL]
    timings["fusion_ms"] = (time.time() - start) * 1000

    start = time.time()
    if fused:
        rerank_docs = [hit["text"] for hit in fused]
        rerank_response = co.rerank(
            model=RERANK_MODEL,
            query=query,
            documents=rerank_docs,
            top_n=TOP_K_RERANK,
        )
        reranked = []
        for result in rerank_response.results:
            hit = fused[result.index]
            hit["relevance_score"] = result.relevance_score
            reranked.append(hit)
    else:
        reranked = []
    timings["rerank_ms"] = (time.time() - start) * 1000

    return {"chunks": reranked, "timings": timings}


if __name__ == "__main__":
    result = retrieve("How do I make a POST request with httpx?")
    for chunk in result["chunks"]:
        print(f"[{chunk['source']}] (score: {chunk['relevance_score']:.3f})")
        print(f"  {chunk['text'][:100]}...")
    print(f"\nTimings: {result['timings']}")

Your retrieval pipeline now searches both vector spaces, fuses the results, and reranks with a cross-encoder. Next up, you'll build the generation layer that turns these retrieved chunks into cited answers through a FastAPI endpoint.

Build the Generation Module and FastAPI App

Your retrieval pipeline finds the most relevant chunks. Now you need a layer that takes those chunks and produces a human-readable answer with cited sources.

In this step, you will build the generation module that enforces citation discipline and wrap the full pipeline in a FastAPI application with a queryable endpoint.

In this step, get ready to:
  • Build a generation module that assembles context and calls OpenAI with citation rules.
  • Create a FastAPI app with a health check and a query endpoint.
  • Start the server and verify end-to-end question answering with curl.
Build the generation module with citation enforcement

A RAG system that retrieves well but generates poorly is still a bad system. The generation module must force the model to ground every claim in the retrieved context. You will use a system prompt that requires inline citations like [1], [2] so users can verify every answer.

  • Create a new file called generate.py in your project folder.
  • Add the imports and system prompt by pasting this code:
from openai import OpenAI

from config import OPENAI_API_KEY, GENERATION_MODEL


SYSTEM_PROMPT = """You are a technical documentation assistant. Answer the user's question based ONLY on the provided context chunks. 

Rules:
1. Cite your sources using [1], [2], etc. notation corresponding to the chunk numbers.
2. Every factual claim must have a citation.
3. If the context does not contain enough information to answer, say "I don't have enough information to answer this question."
4. Be concise and direct."""

What does this code do?

The SYSTEM_PROMPT is the contract that controls the LLM's behaviour. It forces the model to only use information from the provided chunks. It requires numbered citations so every claim is traceable.

Rule 3 is critical for production systems. Without it, the model will hallucinate answers when the context is insufficient.

You won't see a change yet. The next chunk adds the function that uses this prompt.

  • Below the SYSTEM_PROMPT constant, add the generate_answer function:
def generate_answer(query: str, chunks: list[dict]) -> dict:
    openai_client = OpenAI(api_key=OPENAI_API_KEY)

    context_parts = []
    sources = []
    for i, chunk in enumerate(chunks, 1):
        context_parts.append(f"[{i}] Source: {chunk['source']}\n{chunk['text']}")
        sources.append({"index": i, "source": chunk["source"], "text": chunk["text"][:200]})

    context = "\n\n---\n\n".join(context_parts)

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
    ]

    response = openai_client.chat.completions.create(
        model=GENERATION_MODEL,
        messages=messages,
        temperature=0.1,
        max_tokens=500,
    )

    answer = response.choices[0].message.content
    return {"answer": answer, "sources": sources}

What does this function do?

The function formats retrieved chunks into numbered context blocks ([1] Source: quickstart.md\n...) so the model knows which citation number maps to which source.

It calls gpt-4o-mini with low temperature (0.1) to keep answers factual. It returns the answer plus a sources list with truncated text for the API response.

  • Save generate.py.
  • Test the module by running:
python -c "from generate import generate_answer; print('generate module OK')"

You should see generate module OK printed with no import errors.

Seeing an import error?

Make sure config.py is in the same directory as generate.py. Confirm your virtual environment is activated and openai is installed.

Help me debug my generate.py import error

✔️ Awesome, I've got everything!

Great. Double check you have saved generate.py before moving on.

ⓧ I'd like to double check the full code

from openai import OpenAI

from config import OPENAI_API_KEY, GENERATION_MODEL


SYSTEM_PROMPT = """You are a technical documentation assistant. Answer the user's question based ONLY on the provided context chunks. 

Rules:
1. Cite your sources using [1], [2], etc. notation corresponding to the chunk numbers.
2. Every factual claim must have a citation.
3. If the context does not contain enough information to answer, say "I don't have enough information to answer this question."
4. Be concise and direct."""


def generate_answer(query: str, chunks: list[dict]) -> dict:
    openai_client = OpenAI(api_key=OPENAI_API_KEY)

    context_parts = []
    sources = []
    for i, chunk in enumerate(chunks, 1):
        context_parts.append(f"[{i}] Source: {chunk['source']}\n{chunk['text']}")
        sources.append({"index": i, "source": chunk["source"], "text": chunk["text"][:200]})

    context = "\n\n---\n\n".join(context_parts)

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
    ]

    response = openai_client.chat.completions.create(
        model=GENERATION_MODEL,
        messages=messages,
        temperature=0.1,
        max_tokens=500,
    )

    answer = response.choices[0].message.content
    return {"answer": answer, "sources": sources}
Create the FastAPI application

You now have all the pipeline components: ingestion, retrieval, and generation. The final piece is an HTTP API that ties them together so any client can send a question and receive a structured answer.

  • Create a new file called app.py in your project folder.
  • Add the imports, app instance, and Pydantic response models:
import time

from fastapi import FastAPI
from pydantic import BaseModel

from retrieve import retrieve
from generate import generate_answer

# Create the FastAPI application instance
app = FastAPI(title="Production RAG API")


# Define request and response schemas
class QueryRequest(BaseModel):
    question: str


class SourceItem(BaseModel):
    index: int
    source: str
    text: str


class QueryResponse(BaseModel):
    answer: str
    sources: list[SourceItem]
    latency_ms: float

What do these models do?

Pydantic models define the exact shape of your API's input and output. QueryRequest ensures every request has a question field.

QueryResponse guarantees every response includes the answer text, a list of cited sources, and the total latency. FastAPI auto-generates documentation from these models.

You won't see a change yet. The next chunk adds the actual endpoints.

  • Below the QueryResponse class, add the two API endpoints:
@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/query", response_model=QueryResponse)
def query_endpoint(request: QueryRequest):
    start = time.time()

    # Run the full retrieval pipeline
    retrieval_result = retrieve(request.question)
    chunks = retrieval_result["chunks"]

    # Generate a citation-grounded answer
    generation_result = generate_answer(request.question, chunks)

    latency_ms = (time.time() - start) * 1000

    return QueryResponse(
        answer=generation_result["answer"],
        sources=generation_result["sources"],
        latency_ms=round(latency_ms, 1),
    )

What do these endpoints do?

The GET /health endpoint is a liveness check. Load balancers and monitoring tools hit this to verify the service is up.

The POST /query endpoint accepts a JSON body with a question, runs the full hybrid retrieval + reranking + generation pipeline, and returns a structured JSON response. The latency_ms field lets you monitor performance per request.

  • Save app.py.

Seeing a syntax or import error?

Confirm that retrieve.py and generate.py are both saved in the same directory as app.py. Check that fastapi and pydantic are installed in your virtual environment.

Help me fix my app.py import or syntax error

✔️ Awesome, I've got everything!

Great. Double check you have saved app.py before starting the server.

ⓧ I'd like to double check the full code

import time

from fastapi import FastAPI
from pydantic import BaseModel

from retrieve import retrieve
from generate import generate_answer

app = FastAPI(title="Production RAG API")


class QueryRequest(BaseModel):
    question: str


class SourceItem(BaseModel):
    index: int
    source: str
    text: str


class QueryResponse(BaseModel):
    answer: str
    sources: list[SourceItem]
    latency_ms: float


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/query", response_model=QueryResponse)
def query_endpoint(request: QueryRequest):
    start = time.time()

    retrieval_result = retrieve(request.question)
    chunks = retrieval_result["chunks"]

    generation_result = generate_answer(request.question, chunks)

    latency_ms = (time.time() - start) * 1000

    return QueryResponse(
        answer=generation_result["answer"],
        sources=generation_result["sources"],
        latency_ms=round(latency_ms, 1),
    )
Start the server and test the API

With both files saved, you can start the FastAPI server and send a real question through the full pipeline.

  • Start the development server by running:
uvicorn app:app --reload

What does this command do?

The uvicorn command starts an ASGI server. app:app tells it to import the app object from the app.py module. The --reload flag auto-restarts the server when you edit files.

You should see output indicating Uvicorn is running on http://127.0.0.1:8000.

  • Open a second terminal tab (keep the server running in the first).
  • Test the health endpoint by running:
curl http://127.0.0.1:8000/health

You should see {"status":"ok"} as the response.

  • Send a real question to the query endpoint by running:
curl -X POST http://127.0.0.1:8000/query -H "Content-Type: application/json" -d '{"question": "How do I make a POST request with httpx?"}'

You should see a JSON response with three fields: answer (containing inline citations like [1], [2]), sources (a list of source references), and latency_ms (the total processing time).

Getting a connection refused error?

Make sure the uvicorn server is still running in your other terminal tab. Check that you are using port 8000 in the curl command. If the server crashed on startup, check its terminal output for the specific error.

Help me debug why my FastAPI server won't start or respond

Your RAG pipeline is now a fully queryable API. Next up, you will build an evaluation harness to measure how well it actually performs.

Evaluate the RAG Pipeline

Your RAG API is live and answering questions with citations. But how do you know the answers are actually correct?

Without measurement, you are guessing. In this step, you will build an evaluation harness that scores your pipeline on three dimensions: faithfulness (is the answer grounded in context?), relevance (does it address the question?), and context precision (did retrieval surface the right chunks?). This gives you concrete numbers to prove your pipeline works.

In this step, get ready to:
  • Create a golden test set of 10 questions with expected sources.
  • Build an LLM-as-judge evaluation script that scores faithfulness, relevance, and context precision.
  • Run the evaluation and interpret the results table.
Build a golden test set

A golden test set is a curated list of questions with known correct sources. You will use these to automatically score how well your retrieval and generation perform across different query types.

  • Create the eval/ directory by running this command:
mkdir eval
  • Create the golden test set file by running:
touch eval/golden_set.json
  • Confirm the file exists by running:
ls eval/

You should see golden_set.json in the output.

  • Open eval/golden_set.json in your editor and paste the following test set:
[
  {"question": "How do I make a POST request with JSON body using httpx?", "expected_source": "quickstart.md"},
  {"question": "How do I set custom headers in httpx?", "expected_source": "quickstart.md"},
  {"question": "What is the difference between httpx.Client and httpx.AsyncClient?", "expected_source": "async.md"},
  {"question": "How do I handle timeouts in httpx?", "expected_source": "quickstart.md"},
  {"question": "How do I upload a file with httpx?", "expected_source": "quickstart.md"},
  {"question": "What authentication methods does httpx support?", "expected_source": "advanced.md"},
  {"question": "How do I use connection pooling in httpx?", "expected_source": "advanced.md"},
  {"question": "How do I stream a response body in httpx?", "expected_source": "quickstart.md"},
  {"question": "What is the recommended way to deploy a Kubernetes cluster?", "expected_source": "NONE"},
  {"question": "How do I configure SSL certificates in httpx?", "expected_source": "advanced.md"}
]

What does this test set cover?

The 10 questions span three categories that stress-test different parts of your pipeline.

  • Standard retrieval questions (questions 1-8) verify that your hybrid search finds the right documentation chunks.
  • The out-of-scope question (question 9, about Kubernetes) tests whether your system correctly responds with "I don't know" when the answer is not in the indexed docs.
  • Multi-topic questions (like Client vs AsyncClient) test whether retrieval can surface information spread across multiple sections.
  • Save eval/golden_set.json.

JSON parse error when running the eval later?

Make sure there are no trailing commas after the last object in the array. JSON does not allow trailing commas like Python does.

help me fix a JSON parse error in my golden_set.json file

✔️ Awesome, I've got everything!

Great. Make sure you have saved eval/golden_set.json before moving on.

ⓧ I'd like to double check the full code

[
  {"question": "How do I make a POST request with JSON body using httpx?", "expected_source": "quickstart.md"},
  {"question": "How do I set custom headers in httpx?", "expected_source": "quickstart.md"},
  {"question": "What is the difference between httpx.Client and httpx.AsyncClient?", "expected_source": "async.md"},
  {"question": "How do I handle timeouts in httpx?", "expected_source": "quickstart.md"},
  {"question": "How do I upload a file with httpx?", "expected_source": "quickstart.md"},
  {"question": "What authentication methods does httpx support?", "expected_source": "advanced.md"},
  {"question": "How do I use connection pooling in httpx?", "expected_source": "advanced.md"},
  {"question": "How do I stream a response body in httpx?", "expected_source": "quickstart.md"},
  {"question": "What is the recommended way to deploy a Kubernetes cluster?", "expected_source": "NONE"},
  {"question": "How do I configure SSL certificates in httpx?", "expected_source": "advanced.md"}
]
Implement LLM-as-judge evaluation

Instead of manually reading every answer, you will use an LLM to automatically judge quality. This approach (called LLM-as-judge) sends the question, context, and answer to a model with specific scoring criteria and gets back a numeric score.

Why LLM-as-judge instead of a library like RAGAS?

A custom LLM-as-judge implementation teaches the same concepts (faithfulness, relevance, precision) while being more transparent and debuggable. You can see exactly what prompts produce each score.

Once you understand the mechanics, adopting a framework like RAGAS for CI integration becomes straightforward because you already know what it does under the hood.

  • Create a new file called evaluate.py in your project root.
  • Add the imports and evaluation prompt constants by pasting this code:
import json
from pathlib import Path

from openai import OpenAI

from config import OPENAI_API_KEY, GENERATION_MODEL
from retrieve import retrieve
from generate import generate_answer


FAITHFULNESS_PROMPT = """You are evaluating a RAG system's answer for faithfulness.

Given the retrieved context chunks and the generated answer, determine if every claim in the answer is supported by the context.

Context:
{context}

Answer:
{answer}

Score the faithfulness from 0.0 to 1.0:
- 1.0: Every claim is fully supported by the context
- 0.5: Some claims are supported, others are not
- 0.0: The answer contains claims not found in context

Respond with ONLY a JSON object: {{"score": <float>, "reason": "<brief explanation>"}}"""


RELEVANCE_PROMPT = """You are evaluating a RAG system's answer for relevance.

Given the original question and the generated answer, determine if the answer addresses the question.

Question:
{question}

Answer:
{answer}

Score the relevance from 0.0 to 1.0:
- 1.0: The answer directly and completely addresses the question
- 0.5: The answer partially addresses the question
- 0.0: The answer does not address the question at all

Respond with ONLY a JSON object: {{"score": <float>, "reason": "<brief explanation>"}}"""


CONTEXT_PRECISION_PROMPT = """You are evaluating retrieval quality for a RAG system.

Given the question and the retrieved chunks, determine what fraction of the chunks are actually relevant to answering the question.

Question:
{question}

Retrieved chunks:
{chunks}

Score the context precision from 0.0 to 1.0:
- 1.0: All retrieved chunks are relevant to the question
- 0.5: About half the chunks are relevant
- 0.0: None of the chunks are relevant

Respond with ONLY a JSON object: {{"score": <float>, "reason": "<brief explanation>"}}"""

What do these three prompts measure?

  • FAITHFULNESS_PROMPT checks if the generated answer only contains claims that appear in the retrieved context. A low score here means hallucination.
  • RELEVANCE_PROMPT checks if the answer actually addresses the user's question. A low score means the answer went off-topic.
  • CONTEXT_PRECISION_PROMPT checks if the retrieved chunks are relevant to the query. A low score means retrieval is surfacing noise.
  • Add the llm_judge() helper function below the prompts:
def llm_judge(prompt: str) -> dict:
    # Send the scoring prompt to the LLM and parse the JSON response
    client = OpenAI(api_key=OPENAI_API_KEY)
    response = client.chat.completions.create(
        model=GENERATION_MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
        max_tokens=200,
    )
    text = response.choices[0].message.content.strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"score": 0.0, "reason": "Failed to parse judge response"}

What does this function do?

This is a reusable wrapper that sends any scoring prompt to gpt-4o-mini with temperature=0.0 for deterministic output. It parses the JSON response and gracefully handles parse failures by returning a zero score.

  • Add the evaluate_query() function that scores a single question:
def evaluate_query(question: str) -> dict:
    # Run the full pipeline for one question and score it on all three metrics
    retrieval_result = retrieve(question)
    chunks = retrieval_result["chunks"]

    generation_result = generate_answer(question, chunks)
    answer = generation_result["answer"]

    # Format context for the faithfulness judge
    context_text = "\n\n".join([f"[{i+1}] {c['text']}" for i, c in enumerate(chunks)])
    # Format chunks for the precision judge (truncated for token efficiency)
    chunks_text = "\n\n".join([f"Chunk {i+1}: {c['text'][:300]}" for i, c in enumerate(chunks)])

    faithfulness = llm_judge(
        FAITHFULNESS_PROMPT.format(context=context_text, answer=answer)
    )
    relevance = llm_judge(
        RELEVANCE_PROMPT.format(question=question, answer=answer)
    )
    precision = llm_judge(
        CONTEXT_PRECISION_PROMPT.format(question=question, chunks=chunks_text)
    )

    return {
        "question": question,
        "answer": answer,
        "faithfulness": faithfulness["score"],
        "relevance": relevance["score"],
        "context_precision": precision["score"],
    }

What does this function do?

For each question, evaluate_query() runs the full RAG pipeline (retrieve + generate) and then calls the LLM judge three times to score faithfulness, relevance, and context precision independently.

Each judge call is independent. This means a question can score high on relevance but low on faithfulness if the answer addresses the right topic but hallucinates details.

  • Add the run_evaluation() function and the main block at the bottom of the file:
def run_evaluation(golden_set_path: str = "eval/golden_set.json"):
    # Load the golden set and evaluate every question
    golden_set = json.loads(Path(golden_set_path).read_text())

    results = []
    for item in golden_set:
        print(f"Evaluating: {item['question'][:60]}...")
        result = evaluate_query(item["question"])
        results.append(result)

    # Calculate aggregate scores across all questions
    avg_faithfulness = sum(r["faithfulness"] for r in results) / len(results)
    avg_relevance = sum(r["relevance"] for r in results) / len(results)
    avg_precision = sum(r["context_precision"] for r in results) / len(results)

    # Print results as a formatted table
    print("\n" + "=" * 60)
    print("EVALUATION RESULTS")
    print("=" * 60)
    print(f"{'Question':<40} {'Faith':>6} {'Relev':>6} {'Prec':>6}")
    print("-" * 60)
    for r in results:
        print(f"{r['question'][:38]:<40} {r['faithfulness']:>6.2f} {r['relevance']:>6.2f} {r['context_precision']:>6.2f}")
    print("-" * 60)
    print(f"{'AVERAGE':<40} {avg_faithfulness:>6.2f} {avg_relevance:>6.2f} {avg_precision:>6.2f}")
    print("=" * 60)

    return results


if __name__ == "__main__":
    run_evaluation()

What does this function do?

The run_evaluation() function loads all 10 questions from the golden set, evaluates each one, and prints a summary table with per-question scores and aggregate averages. This gives you a single command to benchmark your entire pipeline.

  • Save evaluate.py.

Import error when running the script?

Make sure evaluate.py is in the same directory as retrieve.py, generate.py, and config.py. The imports reference these files as sibling modules.

help me fix import errors in my evaluate.py file

✔️ Awesome, I've got everything!

Great. Double check you have saved evaluate.py in your project root directory.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

from openai import OpenAI

from config import OPENAI_API_KEY, GENERATION_MODEL
from retrieve import retrieve
from generate import generate_answer


FAITHFULNESS_PROMPT = """You are evaluating a RAG system's answer for faithfulness.

Given the retrieved context chunks and the generated answer, determine if every claim in the answer is supported by the context.

Context:
{context}

Answer:
{answer}

Score the faithfulness from 0.0 to 1.0:
- 1.0: Every claim is fully supported by the context
- 0.5: Some claims are supported, others are not
- 0.0: The answer contains claims not found in context

Respond with ONLY a JSON object: {{"score": <float>, "reason": "<brief explanation>"}}"""


RELEVANCE_PROMPT = """You are evaluating a RAG system's answer for relevance.

Given the original question and the generated answer, determine if the answer addresses the question.

Question:
{question}

Answer:
{answer}

Score the relevance from 0.0 to 1.0:
- 1.0: The answer directly and completely addresses the question
- 0.5: The answer partially addresses the question
- 0.0: The answer does not address the question at all

Respond with ONLY a JSON object: {{"score": <float>, "reason": "<brief explanation>"}}"""


CONTEXT_PRECISION_PROMPT = """You are evaluating retrieval quality for a RAG system.

Given the question and the retrieved chunks, determine what fraction of the chunks are actually relevant to answering the question.

Question:
{question}

Retrieved chunks:
{chunks}

Score the context precision from 0.0 to 1.0:
- 1.0: All retrieved chunks are relevant to the question
- 0.5: About half the chunks are relevant
- 0.0: None of the chunks are relevant

Respond with ONLY a JSON object: {{"score": <float>, "reason": "<brief explanation>"}}"""


def llm_judge(prompt: str) -> dict:
    client = OpenAI(api_key=OPENAI_API_KEY)
    response = client.chat.completions.create(
        model=GENERATION_MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
        max_tokens=200,
    )
    text = response.choices[0].message.content.strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"score": 0.0, "reason": "Failed to parse judge response"}


def evaluate_query(question: str) -> dict:
    retrieval_result = retrieve(question)
    chunks = retrieval_result["chunks"]

    generation_result = generate_answer(question, chunks)
    answer = generation_result["answer"]

    context_text = "\n\n".join([f"[{i+1}] {c['text']}" for i, c in enumerate(chunks)])
    chunks_text = "\n\n".join([f"Chunk {i+1}: {c['text'][:300]}" for i, c in enumerate(chunks)])

    faithfulness = llm_judge(
        FAITHFULNESS_PROMPT.format(context=context_text, answer=answer)
    )
    relevance = llm_judge(
        RELEVANCE_PROMPT.format(question=question, answer=answer)
    )
    precision = llm_judge(
        CONTEXT_PRECISION_PROMPT.format(question=question, chunks=chunks_text)
    )

    return {
        "question": question,
        "answer": answer,
        "faithfulness": faithfulness["score"],
        "relevance": relevance["score"],
        "context_precision": precision["score"],
    }


def run_evaluation(golden_set_path: str = "eval/golden_set.json"):
    golden_set = json.loads(Path(golden_set_path).read_text())

    results = []
    for item in golden_set:
        print(f"Evaluating: {item['question'][:60]}...")
        result = evaluate_query(item["question"])
        results.append(result)

    avg_faithfulness = sum(r["faithfulness"] for r in results) / len(results)
    avg_relevance = sum(r["relevance"] for r in results) / len(results)
    avg_precision = sum(r["context_precision"] for r in results) / len(results)

    print("\n" + "=" * 60)
    print("EVALUATION RESULTS")
    print("=" * 60)
    print(f"{'Question':<40} {'Faith':>6} {'Relev':>6} {'Prec':>6}")
    print("-" * 60)
    for r in results:
        print(f"{r['question'][:38]:<40} {r['faithfulness']:>6.2f} {r['relevance']:>6.2f} {r['context_precision']:>6.2f}")
    print("-" * 60)
    print(f"{'AVERAGE':<40} {avg_faithfulness:>6.2f} {avg_relevance:>6.2f} {avg_precision:>6.2f}")
    print("=" * 60)

    return results


if __name__ == "__main__":
    run_evaluation()
Run evaluation and interpret results

Now you will run the evaluation against all 10 questions in your golden set. This takes a few minutes because each question triggers the full pipeline (retrieve, generate) plus three judge calls.

  • Run the evaluation script by executing:
python evaluate.py

How long does this take?

Expect 2-4 minutes total. Each question makes 4 API calls (1 embedding, 1 rerank, 1 generation, 3 judge calls). The script prints progress as it evaluates each question so you can see it working.

When it finishes, you should see a formatted results table with per-question scores and an AVERAGE row at the bottom.

How to interpret the results

  • Aggregate faithfulness above 0.8 means your system rarely hallucinates. It generates answers grounded in the retrieved context.
  • Context precision above 0.7 means your hybrid retrieval pipeline is surfacing relevant chunks most of the time.
  • A low faithfulness score on a specific question signals hallucination. Your model made claims not in the context.
  • A low context precision score signals a retrieval miss. Your search returned irrelevant chunks for that query.

Use weak queries to improve your pipeline

Look for questions with low scores. If context precision is low but faithfulness is high, your retrieval needs tuning (try adjusting chunk size or overlap). If faithfulness is low but precision is high, your generation prompt needs tightening.

Scores are all 0.0?

This usually means the LLM judge is not returning valid JSON. Check that your OPENAI_API_KEY environment variable is set and that gpt-4o-mini is accessible from your account.

help me debug why my evaluation scores are all zero

Secret mission

Add Query Expansion for Ambiguous Queries

Your hybrid retrieval pipeline works well for clear, specific questions. But what about vague queries like "how do I use httpx"? Query expansion uses an LLM to decompose ambiguous questions into targeted sub-queries, dramatically improving recall on the hardest cases in your golden set.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. The OpenAI and Cohere API charges (~$0.30-$0.50 total) were incurred during development and require no further action.

Resources you used:

  • Local Qdrant data stored in the ./qdrant_data/ directory (on-disk, no running service).
  • Project files: config.py, ingest.py, retrieve.py, generate.py, app.py, evaluate.py, query_expand.py, and bm25_meta.json.
  • Downloaded httpx documentation in the docs/ directory.
  • Golden test set in eval/golden_set.json.
  • Python virtual environment created for this project.

Keep everything running

No action needed. Choose this if you plan to keep iterating on your RAG pipeline, adding more documents, or tuning your retrieval parameters.

  • The qdrant_data/ directory persists between runs with no ongoing cost.
  • Your virtual environment stays active for future development sessions.
  • No cloud services are running. You only incur API costs when you execute the scripts.

Pause - I'll come back to this later

Stop any running processes to free up system resources. Your files and data stay on disk for when you return.

  • If the FastAPI server is still running in your terminal, press Ctrl+C to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • All project files and the qdrant_data/ directory remain on disk for your next session.

Delete - I don't want to use this again

Remove all project files to free up disk space. This deletes your vector database, downloaded docs, and all Python scripts.

  • If the FastAPI server is still running, press Ctrl+C to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the entire project directory by running:
cd ~/Desktop
rm -rf production-rag-api

On Windows, use this command instead:

cd ~\Desktop
rmdir /s /q production-rag-api
  • This removes all Python files, the qdrant_data/ directory, the docs/ folder, the eval/ folder, and your virtual environment.
  • Confirm the directory is gone by running:
ls

You should no longer see the project folder in the output.

Nice Work!

Nice Work!

You just built a production-grade RAG API from scratch. Your system ingests technical documentation, retrieves with hybrid search, reranks for precision, and generates citation-grounded answers with measurable quality scores.

You've learned how to:

  • Build a hybrid retrieval pipeline combining dense vectors (OpenAI text-embedding-3-small) and sparse vectors (BM25) with Reciprocal Rank Fusion and Cohere cross-encoder reranking in Qdrant embedded mode.
  • Create a FastAPI endpoint that generates citation-grounded answers with inline source references and latency tracking.
  • Implement an evaluation harness with a golden test set and LLM-as-judge scoring for faithfulness, relevance, and context precision.
  • Secret Mission: Added query expansion that decomposes ambiguous queries into sub-queries for improved recall.

Ready to quiz yourself?