From your first text chunk and your first embedding, to a production RAG system with re-ranking, hybrid search, and faithfulness evaluation.
0 / 10 completed
Level 1 · Fundamentals
What RAG is and why it exists
Goal: understand the problem RAG solves and set up your working environment.
Retrieval-Augmented Generation (RAG) solves a very concrete problem: an LLM only "knows" what it saw during training, with a fixed cutoff date, and has no idea about your internal documents, your knowledge base, or information that changed yesterday. RAG connects the model to an external, up-to-date source of information at query time, instead of relying only on what the model memorized.
The basic architecture has two parts: a retriever that finds the text fragments most relevant to a question, and a generator (the LLM) that writes the answer using those fragments as context.
Install the dependencies
TERMINAL
pip install openai psycopg[binary] pgvector
export OPENAI_API_KEY="sk-your-api-key-here"
# Postgres with the pgvector extension (Docker is the fastest way to try it out)
docker run -d --name pg-rag -e POSTGRES_PASSWORD=postgres -p 5432:5432 pgvector/pgvector:pg17
setup.sql
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
id SERIAL PRIMARY KEY,
content TEXT NOT NULL,
embedding VECTOR(1536), -- dimension of text-embedding-3-small
source TEXT
);
Challenge
Spin up the Postgres container with pgvector and confirm the extension installed correctly by running SELECT * FROM pg_extension WHERE extname = 'vector';
Level 2 · Chunking
Splitting documents into fragments
Goal: split long documents into chunks that an embedding model can process well.
You can't feed a 50-page document straight into an embedding call and expect good results — it needs to be split into manageable fragments. Chunking is one of the decisions that most affects the final quality of a RAG system, and yet it's often treated as a minor detail.
Recursive character splitting with overlap
chunking.py
def split_into_chunks(text: str, size: int = 800, overlap: int = 150) -> list[str]:
"""Splits text into overlapping chunks, respecting paragraph breaks when possible."""
separators = ["\n\n", "\n", ". ", " "]
chunks = []
start = 0
while start < len(text):
end = min(start + size, len(text))
fragment = text[start:end]
# If we haven't reached the end, cut at the closest separator going backward
if end < len(text):
for sep in separators:
pos = fragment.rfind(sep)
if pos > size * 0.5: # avoid cutting too short
fragment = fragment[:pos + len(sep)]
break
chunks.append(fragment.strip())
start += len(fragment) - overlap # the overlap avoids losing context at the edge
return [c for c in chunks if c]
Why the overlap matters
Without overlap, an idea that crosses the boundary between two chunks gets cut in half in both — neither chunk A nor chunk B has the full context. An overlap of 15-20% of the chunk size is usually a good starting point; for highly technical documents with long definitions, it's worth pushing it higher.
Challenge
Take a real document (an article from this blog, for example) and try splitting it with size=400 and size=1200. Compare how many chunks come out and whether any idea gets awkwardly cut off.
Level 3 · Embeddings
Turning text into vectors
Goal: generate embeddings for the chunks using the OpenAI API.
An embedding is a numeric representation (a vector) of the meaning of a piece of text — texts with similar meaning produce vectors that are close to each other in that space. That mathematical closeness is what lets you "search by meaning" instead of by exact word matching.
embeddings.py
from openai import OpenAI
client = OpenAI()
def generate_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small", # 1536 dimensions, a solid default for production
input=text,
)
return response.data[0].embedding
vector = generate_embedding("RAG connects an LLM to up-to-date external information")
print(len(vector)) # 1536
text-embedding-3-small costs $0.02 per million tokens and is enough for 80% of production RAG use cases. text-embedding-3-large (3072 dimensions) gives you slightly higher quality at a higher cost — only worth it once you've actually measured that small falls short for your case.
Challenge
Generate embeddings for 3 sentences: two with similar meaning but different words, and one completely different. Compute cosine similarity between pairs (level 5 shows you how) and confirm the two similar ones end up closer together.
Level 4 · Vector store
Storing embeddings in pgVector
Goal: insert chunks and their embeddings into Postgres using the pgvector extension.
pgVector adds a vector data type and distance operators to Postgres — so you can store your embeddings in the same database you already use for everything else, without adding a separate vector store to your infrastructure.
ingestion.py
import psycopg
from pgvector.psycopg import register_vector
conn = psycopg.connect("postgresql://postgres:postgres@localhost:5432/postgres")
register_vector(conn)
def save_chunk(content: str, embedding: list[float], source: str):
with conn.cursor() as cur:
cur.execute(
"INSERT INTO documents (content, embedding, source) VALUES (%s, %s, %s)",
(content, embedding, source),
)
conn.commit()
# Full pipeline: document -> chunks -> embeddings -> storage
text = open("product_manual.txt").read()
for chunk in split_into_chunks(text):
emb = generate_embedding(chunk)
save_chunk(chunk, emb, source="product_manual.txt")
Challenge
Ingest 2-3 different text documents and confirm with SELECT count(*) FROM documents; that all the chunks were saved with the correct source.
Level 5 · Retrieval
Cosine similarity search
Goal: find the most relevant chunks for a question using pgvector's cosine distance operator.
pgvector exposes the <=> operator for cosine distance — the smaller the distance, the more similar the vectors. Since cosine distance ranges from 0 to 2, 1 - distance gives you a similarity score between -1 and 1, which is more intuitive to read.
retrieval.py
def search_relevant(question: str, top_k: int = 4) -> list[dict]:
question_embedding = generate_embedding(question)
with conn.cursor() as cur:
cur.execute(
"""
SELECT content, source, 1 - (embedding <=> %s) AS similarity
FROM documents
ORDER BY embedding <=> %s
LIMIT %s
""",
(question_embedding, question_embedding, top_k),
)
return [
{"content": r[0], "source": r[1], "similarity": r[2]}
for r in cur.fetchall()
]
results = search_relevant("How do I configure notifications?")
for r in results:
print(f"[{r['similarity']:.3f}] {r['source']}: {r['content'][:80]}...")
For large datasets (hundreds of thousands of vectors), add an HNSW index to the embedding column — without an index, pgvector does a full sequential scan on every search, which doesn't scale.
Challenge
Run CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops); and compare search time before and after with EXPLAIN ANALYZE.
Level 6 · Augmented generation
From chunks to a sourced answer
Goal: combine the retrieved chunks with the original question in a prompt, and generate an answer that cites where the information came from.
rag.py
def answer_with_rag(question: str) -> str:
chunks = search_relevant(question, top_k=4)
context = "\n\n".join(
f"[Source: {c['source']}]\n{c['content']}" for c in chunks
)
prompt = f"""Answer the question using ONLY the information in the context.
If the context doesn't have the answer, explicitly say you don't have that information.
Cite the source in brackets at the end of each claim.
Context:
{context}
Question: {question}"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
return response.choices[0].message.content
print(answer_with_rag("How do I configure notifications?"))
The instruction "if the context doesn't have the answer, say so explicitly" is the single most important line in the prompt — without it, the model will fill in the gaps with information it invented instead of admitting it doesn't know.
Challenge
Ask a question you know is NOT covered in your ingested documents, and confirm the model admits it doesn't have that information instead of hallucinating an answer.
Level 7 · Re-ranking
Improving retrieval quality
Goal: reorder the vector search results with a specialized model before passing them to the LLM.
Cosine similarity search is fast but approximate — it brings back reasonable candidates, not necessarily the best ones in the right order. A re-ranker (typically a cross-encoder) evaluates the question-chunk pair with more precision, at the cost of being slower — which is why it's only applied to the top-N candidates the vector search already returned, not to the whole database.
reranking.py
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def search_with_reranking(question: str, initial_top_k: int = 15, final_top_k: int = 4) -> list[dict]:
candidates = search_relevant(question, top_k=initial_top_k) # broad vector search
pairs = [[question, c["content"]] for c in candidates]
scores = reranker.predict(pairs) # the cross-encoder scores each pair
for c, score in zip(candidates, scores):
c["rerank_score"] = float(score)
candidates.sort(key=lambda c: c["rerank_score"], reverse=True)
return candidates[:final_top_k]
This pattern of "broad, cheap search + precise re-ranking over a few candidates" is the same principle behind nearly every modern search system, from search engines to recommendation systems.
Challenge
Compare the order of the top-4 results with and without re-ranking for an ambiguous question — you'll notice the re-ranker often moves to first place a chunk the vector search had ranked lower.
Level 8 · Hybrid search
Combining semantic and lexical search
Goal: combine vector search with full-text search so you don't miss exact matches.
Vector search is excellent for meaning, but weak on exact terms: error codes, product names, part numbers, specific acronyms. Postgres already ships with full-text search (tsvector), which does find exact matches — combining the two gives better results than either one alone.
hybrid_search.py
def hybrid_search(question: str, top_k: int = 4) -> list[dict]:
question_embedding = generate_embedding(question)
with conn.cursor() as cur:
cur.execute(
"""
WITH vector_search AS (
SELECT id, content, source,
ROW_NUMBER() OVER (ORDER BY embedding <=> %s) AS vector_rank
FROM documents
ORDER BY embedding <=> %s LIMIT 20
),
text_search AS (
SELECT id, content, source,
ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', %s)) DESC) AS text_rank
FROM documents
WHERE to_tsvector('english', content) @@ plainto_tsquery('english', %s)
LIMIT 20
)
-- Reciprocal Rank Fusion: combines both rankings into a single score
SELECT COALESCE(v.content, t.content) AS content,
COALESCE(v.source, t.source) AS source,
COALESCE(1.0 / (60 + v.vector_rank), 0) + COALESCE(1.0 / (60 + t.text_rank), 0) AS rrf_score
FROM vector_search v
FULL OUTER JOIN text_search t ON v.id = t.id
ORDER BY rrf_score DESC
LIMIT %s
""",
(question_embedding, question_embedding, question, question, top_k),
)
return [{"content": r[0], "source": r[1]} for r in cur.fetchall()]
Reciprocal Rank Fusion (RRF) is the standard technique for combining two different rankings without having to normalize scores across different scales (cosine similarity vs. text rank) — it simply sums 1/(k + position) from each list, giving more weight to results that rank well in both searches.
Challenge
Try a question that includes an exact code or term (e.g. "error E404" or a specific product name) and compare the results of search_relevant vs. hybrid_search.
Level 9 · Evaluation
Measuring whether your RAG actually works
Goal: evaluate the faithfulness of answers and detect when the system should admit it doesn't know.
A RAG system that "looks fine" in a handful of manual tests can be silently failing in production. Two key metrics: faithfulness (is the answer fully supported by the retrieved context, or did the model add information from its training?) and relevancy (was the retrieved context actually related to the question?).
evaluation.py
def evaluate_faithfulness(question: str, context: str, answer: str) -> dict:
"""Uses a second LLM as a judge to verify the answer is supported by the context."""
prompt = f"""Evaluate whether the ANSWER is fully supported by the CONTEXT.
Respond in JSON: {{"faithful": true/false, "reason": "..."}}
CONTEXT:
{context}
ANSWER:
{answer}"""
result = client.chat.completions.create(
model="gpt-4o-mini", # a cheap model is enough for this verification task
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
import json
return json.loads(result.choices[0].message.content)
Handling the "I don't have enough information" case
If the similarity of the best retrieved chunk is well below your other typical results (for example, under 0.3 on your dataset), that's a signal there's probably no relevant context — it's worth answering "I don't have information about this" instead of trying to generate an answer from weak context.
Challenge
Put together a set of 10 test questions (some with a clear answer in your documents, others without one) and run evaluate_faithfulness on each — count how many fail.
Level 10 · Final architecture
Production RAG: continuous ingestion and scaling
Goal: assemble everything above into an architecture that stays up to date on its own and scales with document and query volume.
1. Incremental ingestion pipeline
In production, documents change — you don't want to reprocess the entire corpus every time a file gets updated. The standard solution: store a hash of each document's content, and only re-ingest (re-chunk + re-embed) the ones that changed since the last run.
incremental_ingestion.py
import hashlib
def needs_reingestion(doc_id: str, content: str) -> bool:
current_hash = hashlib.sha256(content.encode()).hexdigest()
with conn.cursor() as cur:
cur.execute("SELECT hash FROM documents_meta WHERE doc_id = %s", (doc_id,))
row = cur.fetchone()
if row is None or row[0] != current_hash:
return True
return False
# In the ingestion pipeline: only re-process what changed
for doc_id, content in source_documents.items():
if needs_reingestion(doc_id, content):
delete_old_chunks(doc_id)
for chunk in split_into_chunks(content):
save_chunk(chunk, generate_embedding(chunk), source=doc_id)
2. Caching embeddings for frequent queries
If your RAG receives repeated or similar questions (common in customer support), caching the question embedding and the retrieval results avoids redundant calls to both the embeddings API and the database.
3. Scaling the vector store
final architecture
Source documents (S3, CMS, database)
│
▼
Incremental ingestion pipeline (only re-processes what changed)
│
▼
Chunking + Embeddings (batch, with caching of already-generated embeddings)
│
▼
pgVector (with HNSW index, read replicas if query volume grows)
│
▼
User query → Embedding → Hybrid search (vector + text) → Re-ranking
│
▼
LLM generates answer with cited sources
│
▼
Faithfulness evaluation (continuous sampling, not on every request)
With this, you have the full path: from a loose text chunk in Level 2, to a production RAG system with self-maintaining ingestion, hybrid search, re-ranking, and continuous evaluation of whether the answers are actually grounded in your documents.
Final challenge
Take the full pipeline from levels 1-9 and wrap it in a FastAPI POST /ask endpoint, with the incremental ingestion pipeline running separately as a scheduled job.