Build a RAG prototype with pgvector retrieval

Introduction

Engineers often spend days wiring together document loaders, chunkers, and vector stores before they can run a single end-to-end retrieval query. That setup cost slows prototype feedback loops in sprint environments where course-corrections need to happen every few days, not weeks. This lesson walks through a minimal but complete RAG pipeline — from document ingestion and recursive chunking through embedding and retrieval — using Pydantic models to enforce data integrity at every stage. By the end, you'll be able to stand up a working RAG prototype that loads documents, creates semantically coherent chunks, stores embeddings, and returns grounded answers.

Key Terminology

• Recursive Chunking — A splitting strategy that walks a ranked separator list ("\n\n", "\n", " ", "") in order, stopping at the first separator that keeps every part within chunk_size, so each Chunk ends at a natural language boundary rather than mid-sentence.
• Chunk Overlap — The number of characters shared between adjacent chunks, set by ChunkingConfig.chunk_overlap; prevents retrieval from returning a fragment that straddles a boundary, at the cost of a larger token budget.
• Content Checksum — A SHA-256 digest of raw document bytes stored in Document.checksum; used as the deduplication key so re-ingesting the same file does not create a duplicate Document record in the pipeline.
• Character Offset — The start_char and end_char fields on each Chunk that record where in the parent document's text the chunk begins and ends, making retrieved passages traceable back to their exact source location.
• ChunkingConfig — A Pydantic BaseModel that validates the three key chunking knobs — chunk_size, chunk_overlap, and separators — centralising all tunable parameters into a single typed object rather than scattered constants.
• Data Contract — The guarantee, enforced by Pydantic at construction time, that every Document and Chunk carries the required fields in the correct types before data advances to the next pipeline stage.

Concepts

Pydantic Models as Pipeline Data Contracts

A RAG pipeline is a chain of stages — ingestion, chunking, embedding, retrieval — where each stage consumes the output of the one before it. Without explicit data contracts, a missing field or mistyped value discovered at query time requires tracing backwards through multiple functions to find where bad data entered. Pydantic flips that dynamic: Document and Chunk are validated at construction, so a malformed doc_id or a token_count of the wrong type raises immediately at the point of creation rather than silently corrupting retrieval results later.

In sprint-paced prototyping this matters more, not less. When a course-correction is expected every few days, a silent data-shape bug that only surfaces during retrieval wastes a full evaluation cycle. By making Document and Chunk the typed hand-off objects between every stage, you get Python-native documentation of what the pipeline promises to carry — and an instant failure signal when that promise is broken (see Code Walkthrough).

Recursive Chunking vs. Fixed-Size Splitting

Fixed-size splitting is straightforward: slice every N characters and advance a cursor. The problem is that meaningful content rarely aligns with a fixed cursor position — a paragraph can end at character 318 and the next idea starts at 320, but a hard split at 512 throws the opening words of that new idea onto the tail of one chunk and the rest onto the head of the next. Retrieval returns one fragment and the answer is incomplete.

Recursive chunking avoids this by trying separators in ranked order — paragraph breaks first, then line breaks, then spaces, then individual characters as a last resort — and stopping at the first separator that keeps all parts within chunk_size. In practice this is almost always a paragraph or line boundary, not a mid-word cut. The resulting chunks are semantically coherent because they terminate where the author signaled a boundary, not where a counter happened to expire.

Chunk Overlap and Sprint Trade-offs

chunk_overlap exists because an answer can straddle a chunk boundary: the first half of the relevant sentence sits at the tail of chunk N and the second half at the head of chunk N+1. Without overlap, retrieval surfaces only one chunk and the answer is truncated. With overlap, both chunks carry enough shared context that either one alone is useful.

The cost is real: overlap duplicates tokens across chunks, inflating embedding storage and the context window consumed when retrieved chunks are assembled into a prompt. The right sprint-cycle move is to start with the default (chunk_overlap: 64 characters in ChunkingConfig) and adjust based on observed retrieval misses — not to pre-optimise before retrieval data exists. ChunkingConfig is designed for exactly this kind of fast iteration: swap one validated field value, re-ingest, re-query (see Code Walkthrough).

Source Traceability via Character Offsets

Embedding and retrieval discard document structure — the model sees token sequences, not filenames or page numbers. When a retrieved chunk looks wrong or suspiciously fabricated, there is no built-in way to confirm which document it came from or where inside that document it lived.

The Chunk model addresses this directly by persisting start_char and end_char alongside content. These offsets record exactly where the chunk sits inside the parent Document.content string. Combined with doc_id linking back to Document.filename and Document.checksum, every chunk carries a full provenance chain from raw bytes through to the retrieved passage. In a sprint evaluation where the question "why did the model return this?" needs an answer in minutes, that traceability is the difference between a fast fix and a blind re-ingest.

Code Walkthrough

Now that you have the Document, Chunk, and ChunkingConfig models grounding the pipeline's data contracts, the code below shows how they fit together in a working ingest path.

Document ingestion starts with two Pydantic models that enforce data integrity at every stage. Document tracks ingested files by checksum to prevent duplicate processing; Chunk preserves character offsets so retrieved passages can be traced back to their exact source location. RAGPrototype wires both models into a single ingest method that extracts text, hashes content for deduplication, and returns a typed Document ready for the chunking stage:

Code snippetpython
1import hashlib
2import openai
3from pydantic import BaseModel
4from typing import List, Optional
5
6class Document(BaseModel):
7    doc_id: str
8    filename: str
9    content: str
10    content_type: str  # pdf, markdown, text
11    page_count: Optional[int] = None
12    char_count: int
13    checksum: str
14
15class Chunk(BaseModel):
16    chunk_id: str
17    doc_id: str
18    content: str
19    chunk_index: int
20    start_char: int
21    end_char: int
22    embedding: Optional[List[float]] = None
23    token_count: int
24
25class RAGPrototype:
26    def __init__(self, proxy_url: str, db_url: str):
27        self.client = openai.OpenAI(
28            api_key="student-token",
29            base_url=proxy_url
30        )
31        self.db_url = db_url
32
33    async def ingest(self, filename: str, content: bytes) -> Document:
34        text = self._extract_text(filename, content)
35        return Document(
36            doc_id=hashlib.sha256(content).hexdigest()[:16],
37            filename=filename,
38            content=text,
39            content_type=self._detect_type(filename),
40            char_count=len(text),
41            checksum=hashlib.sha256(content).hexdigest()
42        )

Once a Document is ingested, chunking strategy determines retrieval quality. Fixed-size splits often cut sentences mid-thought; recursive chunking walks a ranked separator list — paragraph breaks first, then newlines, then spaces — so each chunk lands at a natural language boundary. ChunkingConfig exposes the key knobs as validated Pydantic fields, and recursive_chunk applies them:

Code snippetpython
1from pydantic import BaseModel
2from typing import List
3
4class ChunkingConfig(BaseModel):
5    """Configuration for recursive text chunking."""
6    chunk_size: int = 512
7    chunk_overlap: int = 64
8    separators: List[str] = ["\n\n", "\n", " ", ""]
9
10def recursive_chunk(text: str, config: ChunkingConfig) -> List[str]:
11    """Split text at natural boundaries using the separator hierarchy."""
12    for sep in config.separators:
13        parts = text.split(sep) if sep else list(text)
14        if all(len(p) <= config.chunk_size for p in parts):
15            return parts
16    # Fallback: hard split at chunk_size with overlap
17    return [
18        text[i : i + config.chunk_size]
19        for i in range(0, len(text), config.chunk_size - config.chunk_overlap)
20    ]

The chunk_overlap field is especially important during rapid prototyping: too little overlap fragments context at boundaries; too much inflates your token budget and slows retrieval during sprint evaluations.

Confirm that calling recursive_chunk with a multi-paragraph string and the default ChunkingConfig returns chunks whose length does not exceed chunk_size and that adjacent chunks share the expected overlap text.

Do's and Don'ts

Having walked through RAG prototype above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do use checksum (SHA-256 of raw bytes) as the deduplication key in Document.doc_id — hashing the original content bytes before text extraction catches duplicate uploads regardless of filename, preventing redundant embedding calls that bloat your vector store during sprint iteration.
2. ✓Do prefer recursive chunking over fixed-size splits by configuring ChunkingConfig.separators from coarsest to finest ("\n\n" → "\n" → " " → "") — this keeps sentence and paragraph boundaries intact, which directly improves retrieval precision because embeddings for semantically whole units outperform embeddings that cut mid-sentence.
3. ✓Do preserve start_char and end_char on every Chunk — these character offsets let you trace any retrieved passage back to its exact location in the source Document, which is essential for validating grounding and debugging retrieval quality without re-reading the entire corpus.

Don'ts

1. ✗Don't set chunk_overlap to zero in ChunkingConfig — without overlap, the hard-split fallback severs context exactly at chunk_size boundaries, so a sentence straddling two chunks becomes unembeddable as a unit and retrieval silently misses queries that span that boundary.
2. ✗Don't bypass the Document and Chunk Pydantic models by passing raw dicts through the pipeline — skipping model validation removes the type guarantee that embedding is either a valid List[float] or None, allowing half-embedded chunks to enter the vector store and cause silent retrieval failures at query time.
3. ✗Don't call recursive_chunk before ingest has computed and stored the checksum — chunking a duplicate document that was never deduplicated via doc_id results in redundant embedding work and inflated retrieval noise, precisely the sprint-cycle slowdown the ingest-first contract is designed to prevent.