How to Build a RAG System for Document Search in Python: A Step-by-Step Guide 2026

In 2026, Retrieval-Augmented Generation (RAG) has become the de facto standard for building enterprise AI assistants. According to Gartner, over 70% of LLM deployments in the enterprise segment use RAG architecture. The reason is simple: language models hallucinate, and RAG anchors responses to facts from documents.

In this guide, I'll show you how to build a production-ready RAG system in Python from scratch. We'll go through the full pipeline: text extraction from PDF and DOCX, chunking, embeddings, vector indexing in Qdrant, and integration with an LLM for answer generation. By the end of the article, you'll have a working prototype ready for scaling.

RAG Architecture: What's Inside the Black Box

The classic RAG pipeline consists of two phases:

1. Indexing (offline): documents → chunks → embeddings → vector DB
2. Inference (online): query → query embedding → top-k chunk retrieval → context + prompt → LLM → answer

The key advantage of RAG over fine-tuning is that you don't retrain the model. The LLM remains frozen, and knowledge is updated through the vector DB. This provides:
- Relevance: update documents, and the system immediately answers with new information
- Transparency: you can show the user where the answer came from
- Cost control: indexing is cheap, and inference via LLM is only on queries

Tools of 2026: What We Use

Why this stack? Voyage-3-lite offers embedding quality on par with OpenAI text-embedding-3-small but at 3x lower cost. Qdrant is the best open-source vector DB with a graph-based HNSW index. DeepSeek-V3 is the cheapest LLM with a 128K token context, ideal for RAG.

Step 1. Extracting Text from Documents

The first problem: documents are rarely "plain text." PDFs can be scanned images, and DOCX can have complex layouts. We use the unstructured library, which can detect layout and extract text considering structure.

import hashlib
from unstructured.partition.auto import partition

def extract_text(file_path: str) -> list[dict]:
    """Extracts document elements with metadata"""
    elements = partition(
        filename=file_path,
        strategy="auto",          # auto-detect document type
        pdf_infer_table_structure=True,  # recognize tables in PDF
        languages=["rus", "eng"]  # support Russian and English
    )

    docs = []
    for el in elements:
        if el.text.strip():
            doc_id = hashlib.md5(el.text.encode()).hexdigest()[:12]
            docs.append({
                "doc_id": doc_id,
                "text": el.text,
                "type": str(type(el).__name__),  # Title, NarrativeText, Table, etc.
                "page_number": el.metadata.page_number if el.metadata else None
            })
    return docs

# Example: parse a PDF report
docs = extract_text("annual_report_2025.pdf")
print(f"Extracted {len(docs)} elements")
# Output: Extracted 347 elements

unstructured automatically determines what to do: for text PDFs it uses PyPDF2, for scans — OCR via Tesseract (if installed), for DOCX — python-docx. Important: on production systems, I recommend running this in a separate microservice, as OCR consumes a lot of CPU.

Step 2. Chunking: Cutting Correctly

LLMs have a context limit (DeepSeek-V3 has 128K tokens, but that doesn't mean you should cram everything). Research shows: the optimal chunk size for RAG is 512-1024 tokens with 10-20% overlap. Why? Too small chunks lose context, too large chunks dilute semantics.

We use the recursive splitter from LangChain, which respects paragraph and sentence boundaries:

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1024,          # in characters, roughly 250-300 tokens
    chunk_overlap=200,        # overlap to preserve context
    separators=["\n\n", "\n", ". ", " ", ""],  # separator priority
    length_function=len,
    is_separator_regex=False
)

def chunk_documents(docs: list[dict]) -> list[dict]:
    chunks = []
    for doc in docs:
        texts = text_splitter.split_text(doc["text"])
        for i, chunk_text in enumerate(texts):
            chunks.append({
                "chunk_id": f"{doc['doc_id']}_chunk_{i}",
                "text": chunk_text,
                "metadata": {
                    "source": doc.get("source", "unknown"),
                    "page": doc.get("page_number"),
                    "type": doc.get("type"),
                    "chunk_index": i
                }
            })
    return chunks

chunks = chunk_documents(docs)
print(f"Obtained {len(chunks)} chunks from {len(docs)} elements")
# Output: Obtained 892 chunks from 347 elements

An important nuance: for tables, you should use a separate splitter that preserves rows as a whole. unstructured returns tables as separate elements with type Table — it's better not to split them but index them entirely.

Step 3. Embeddings: Turning Text into Vectors

Embeddings are the "bridge" between text and the vector DB. In 2026, market leaders are: OpenAI text-embedding-3-small (1536 dim), Voyage-3-lite (1024 dim), and Cohere Embed v3 (1024 dim). I choose Voyage AI for price/quality ratio.

import voyageai
import numpy as np

VOYAGE_API_KEY = "your-key"  # store in .env, not in code!
vo = voyageai.Client(api_key=VOYAGE_API_KEY)

def embed_chunks(chunks: list[dict], batch_size: int = 128) -> np.ndarray:
    """Gets embeddings for all chunks in batches"""
    texts = [chunk["text"] for chunk in chunks]
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        result = vo.embed(
            texts=batch,
            model="voyage-3-lite",
            input_type="document",  # optimization for long texts
            truncation=True
        )
        all_embeddings.extend(result.embeddings)
        print(f"Processed {min(i+batch_size, len(texts))}/{len(texts)} chunks")

    return np.array(all_embeddings, dtype=np.float32)

embeddings = embed_chunks(chunks)
print(f"Embedding dimension: {embeddings.shape}")
# Output: Embedding dimension: (892, 1024)

Tip: always use input_type="document" for indexing and input_type="query" for search. Voyage AI trains different projection heads for these modes, improving retrieval accuracy by 5-8% on Recall@10 metric.

Step 4. Vector DB: Qdrant in Docker

Qdrant is a Rust-based vector DB with HNSW index, providing latency <10ms on 1M vectors. Run locally via Docker:

docker run -d --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage \
  qdrant/qdrant:latest

Now create a collection and upload chunks:

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

client = QdrantClient(host="localhost", port=6333)

COLLECTION_NAME = "documents_rag"

# Create collection if it doesn't exist
client.recreate_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=VectorParams(
        size=1024,  # Voyage-3-lite embedding dimension
        distance=Distance.COSINE  # cosine similarity
    )
)

# Prepare points for insertion
points = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
    points.append(PointStruct(
        id=i,  # unique integer ID
        vector=embedding.tolist(),
        payload={
            "chunk_id": chunk["chunk_id"],
            "text": chunk["text"],
            "source": chunk["metadata"].get("source", ""),
            "page": chunk["metadata"].get("page", 0)
        }
    ))

# Insert in batches of 256 points
BATCH_SIZE = 256
for i in range(0, len(points), BATCH_SIZE):
    batch = points[i:i+BATCH_SIZE]
    client.upsert(
        collection_name=COLLECTION_NAME,
        points=batch
    )
    print(f"Uploaded {min(i+BATCH_SIZE, len(points))}/{len(points)} points")

print("Indexing complete!")
# Output: Uploaded 892/892 points
#        Indexing complete!

Why Qdrant, not Pinecone? In 2026, Pinecone is still popular, but for self-hosted solutions, Qdrant gives more control: you manage the data, no lock-in, and for 1M vectors you only pay for the server ($30/month on Hetzner vs $70/month on Pinecone).

Step 5. Retriever: Finding Relevant Chunks

Now let's write a search function that finds top-k chunks for a query:

def search_documents(query: str, top_k: int = 5) -> list[dict]:
    """Searches for relevant chunks by query"""
    # Get query embedding (query mode!)
    query_embedding = vo.embed(
        texts=[query],
        model="voyage-3-lite",
        input_type="query"
    ).embeddings[0]

    # Search in Qdrant
    search_result = client.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_embedding,
        limit=top_k,
        with_payload=True,
        score_threshold=0.65  # filter out irrelevant results
    )

    results = []
    for scored_point in search_result:
        results.append({
            "text": scored_point.payload["text"],
            "score": scored_point.score,
            "source": scored_point.payload.get("source", ""),
            "page": scored_point.payload.get("page", 0)
        })

    return results

# Test
results = search_documents("What was the company's revenue in 2025?", top_k=3)
for r in results:
    print(f"Score: {r['score']:.3f} | Source: {r['source']} p.{r['page']}")
    print(f"Text: {r['text'][:100]}...")
    print("---")

Note the score_threshold=0.65. This protects against garbage results: if the query is not about the documents at all, the retriever returns an empty list, and the LLM honestly says "information not found" instead of hallucinating.

Step 6. Answer Generation: LLM + Context

The final step is to pass the found chunks to the LLM with the right prompt. We use DeepSeek-V3 via API:

from openai import OpenAI  # DeepSeek is compatible with OpenAI SDK

DEEPSEEK_API_KEY = "your-key"
client_llm = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com/v1"
)

def generate_answer(query: str, context_chunks: list[dict]) -> str:
    """Generates an answer based on context"""
    # Build context from found chunks
    context = "\n\n---\n\n".join([
        f"[Source: {chunk['source']}, p. {chunk['page']}]\n{chunk['text']}"
        for chunk in context_chunks
    ])

    system_prompt = """You are an AI assistant for searching corporate documents.
    Answer in Russian, using only the provided context.
    If the context does not contain the answer, write: "Information not found in documents."
    Always indicate the source of information in the format [Source: name, p. X]."""

    user_prompt = f"Context:\n{context}\n\nQuestion: {query}"

    response = client_llm.chat.completions.create(
        model="deepseek-chat",  # DeepSeek-V3
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.3,  # low temperature for factuality
        max_tokens=1024
    )

    return response.choices[0].message.content

# Full pipeline
def rag_pipeline(query: str) -> str:
    chunks = search_documents(query, top_k=5)
    if not chunks:
        return "Information not found in documents."
    return generate_answer(query, chunks)

# Test
answer = rag_pipeline("What was the company's revenue in 2025?")
print(answer)
# Output: According to the annual report, the company's revenue in 2025 amounted to 12.4 billion rubles, which is 18% higher than in 2024. [Source: annual_report_2025.pdf, p. 7]

Optimization: What Really Matters in Production

Once the prototype works, you need to think about production quality. Here are three critical improvements:

1. Hybrid search (keyword + vector)

Pure vector search works poorly with abbreviations and exact names (e.g., "TIN 7701234567"). The solution is to add BM25 search via Tantivy or Elasticsearch and combine results with embeddings. Qdrant supports hybrid search natively since version 1.10.

2. Reranking

After retrieving the first 20-50 chunks, use a cross-encoder for reranking. The model BAAI/bge-reranker-v2-m3 gives +10-15% improvement in answer accuracy. Run it on GPU or via API (e.g., Cohere Rerank).

3. Meta-filtering

Add filters in Qdrant by metadata: document date, type, author. This allows queries like "find in contracts from 2025 the clause on penalties." Filtering is done at the index level before ANN search, so it doesn't slow down.

Cost Comparison

For comparison, using GPT-4o-mini would cost $0.15/M input tokens, but answer quality for RAG is lower than DeepSeek-V3. GPT-4o is $2.50/M, 10x more expensive.

Conclusion

We built a full-fledged RAG system in Python that can index PDF/DOCX, search them, and answer questions with source citations. The entire code fits in ~150 lines, and the operating cost is $41 per month for 10K queries.

What's next? Add image processing (diagrams, charts) via multimodal CLIP embeddings, connect an agent layer with memory for multi-turn dialogues, and deploy it all on FastAPI for integration with corporate systems.

RAG is not just a trendy buzzword. It's a practical tool that solves a real problem: how to make LLMs useful for working with documents without losing control over facts. If you work with data — master this pipeline. It will pay off from the very first project.

Want to dive deeper into Data Science and building AI systems? Check out the course "Data Science from Scratch" — we cover Python for data analysis, working with Pandas and NumPy, visualization, statistics, and machine learning on real projects. RAG is just one of many tools you'll master.