How to Build a RAG Chatbot That Actually Retrieves

Almost every RAG chatbot project fails in the same place, and it is not the place students expect. The model call is four lines of code. The prompt is a paragraph. What actually decides whether your document chatbot answers correctly is whether the right passage was in the context window at all — and for most projects, it wasn't. The model then did what models do: it wrote a fluent, confident, wrong answer from whatever it was handed.

So if you are building a RAG chatbot as a mini project or a final-year capstone, understand the shape of the work before you start: the retrieval is the project. The LLM call is the easy part. This post walks through the parts that actually take engineering — structure-aware chunking, embedding and querying with pgvector, an evaluation set you build by hand, a citation contract, and a refusal path for when retrieval comes up empty.

The Python below is real. The chunker and the scoring functions are tested and their behaviour is described accurately. The database and API code is real code, but you will need to wire up your own DSN, schema, and API key before it runs — I am not going to pretend a blog post is a working repository. The embedding model runs locally and costs nothing, so you can build the whole retrieval half before you spend a rupee on API calls. That ordering is deliberate.

Why does my RAG chatbot answer confidently and wrongly?

Because a language model given irrelevant context does not say "this context is irrelevant." It pattern-matches. Hand it five chunks about leave policy when the question was about gratuity eligibility, and it will assemble something gratuity-shaped out of leave-policy vocabulary. The failure is silent, and it looks exactly like a correct answer.

That means the interesting question in your project is never "which model did you use." It is: for a question whose answer exists in your corpus, how often is the passage containing that answer actually in the top k results? That number has a name — recall@k — and if you cannot state yours, you do not know whether your system works.

An LLM cannot fix a retrieval failure. It can only make one harder to see.

What is wrong with fixed-size chunking?

Here is the chunker almost every tutorial starts with, and it is the single biggest destroyer of retrieval quality in student RAG projects:

# The wrong way. Do not ship this.
def chunk(text, size=1000, overlap=200):
    out = []
    for i in range(0, len(text), size - overlap):
        out.append(text[i:i + size])
    return out

It slices on character count, which has nothing to do with meaning. A definition gets severed from the term it defines. A table header lands in one chunk and its rows in the next. A clause that begins "This shall not apply where..." ends up in a different chunk from the rule it negates — so retrieval returns the rule, the model reads the rule, and the model tells the user the opposite of the truth. The overlap parameter does not save you; it just means you now have two chunks that are each half-wrong.

Chunk on structure instead. Documents already carry it: headings, paragraphs, list items, fenced code, table boundaries. Split on those, pack the pieces up to a token budget, and carry the heading into the chunk so the embedding knows what the text is about. Three details in the code below are load-bearing, and every one of them is a bug I have watched a structure-aware chunker ship with. It flushes the chunk whenever the heading changes, so a chunk is never labelled with a heading that only applies to its first block. It tracks fenced-code state, so the comments inside a Python listing are not mistaken for Markdown headings. And it requires whitespace after the hashes, so a table row beginning with a hash is not a section break.

import re
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer

# 384 dimensions, 512-token input limit. Runs on a laptop CPU. Free.
model = SentenceTransformer("BAAI/bge-small-en-v1.5")
tok = model.tokenizer

def n_tokens(text: str) -> int:
    return len(tok.encode(text, add_special_tokens=False))

@dataclass
class Chunk:
    doc_id: str
    ordinal: int
    heading: str
    text: str

HEADING_RE = re.compile(r"#{1,6}\s")     # a hash with no space is not a heading
FENCE_RE   = re.compile(r"^\s*(```|~~~)")

def split_blocks(markdown: str):
    """Yield (heading, block) pairs. A fenced code block is one block."""
    heading, buf, fence = "", [], None

    def flush():
        nonlocal buf
        block = "\n".join(buf).strip()
        buf = []
        return block or None

    for line in markdown.splitlines():
        m = FENCE_RE.match(line)
        if fence is not None:                 # inside a code fence: never parse headings
            buf.append(line)
            if m and m.group(1) == fence:
                fence = None
                if (b := flush()):
                    yield heading, b
            continue
        if m:                                 # opening a fence
            if (b := flush()):
                yield heading, b
            fence = m.group(1)
            buf.append(line)
        elif HEADING_RE.match(line):
            if (b := flush()):
                yield heading, b
            heading = line.lstrip("#").strip()
        elif not line.strip():
            if (b := flush()):
                yield heading, b
        else:
            buf.append(line)
    if (b := flush()):
        yield heading, b

def chunk_document(doc_id: str, markdown: str, max_tokens: int = 350,
                   overlap_blocks: int = 1) -> list[Chunk]:
    chunks, cur, cur_tokens, ordinal = [], [], 0, 0

    def flush():
        nonlocal cur, cur_tokens, ordinal
        if not cur:
            return
        chunks.append(Chunk(doc_id, ordinal, cur[0][0],
                            "\n\n".join(b for _, b in cur)))
        ordinal += 1
        # carry overlap only when it will not make the next chunk a superset of this one
        cur = cur[-overlap_blocks:] if overlap_blocks and len(cur) > overlap_blocks else []
        cur_tokens = sum(n_tokens(b) for _, b in cur)

    for heading, block in split_blocks(markdown):
        if cur and heading != cur[-1][0]:     # heading changed: close the chunk
            flush()
            cur, cur_tokens = [], 0           # and never carry overlap across a heading
        if cur and cur_tokens + n_tokens(block) > max_tokens:
            flush()
        cur.append((heading, block))
        cur_tokens += n_tokens(block)
    flush()
    return chunks

The overlap is now a whole block, not 200 arbitrary characters, so the boundary between two chunks always falls where the document itself had a boundary. Without the heading-change flush, a chunk containing the tail of "Casual Leave" and the start of "Gratuity" gets stored under the heading "Casual Leave" and embedded as if it were about leave — which is precisely the definition-severed-from-its-term failure the structure-aware chunker exists to prevent.

Embedding and storing with pgvector

You do not need a dedicated vector database for a project of this size. Postgres with the pgvector extension gives you vector search, keyword search, and your application tables in one place, with one backup story and one connection pool. Start here; graduate later if you ever actually need to.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
    id           bigserial PRIMARY KEY,
    doc_id       text        NOT NULL,
    ordinal      int         NOT NULL,
    heading      text        NOT NULL DEFAULT '',
    content      text        NOT NULL,
    content_hash text        NOT NULL,      -- sha256(doc_id || ordinal || content)
    embedding    vector(384) NOT NULL,
    tsv          tsvector GENERATED ALWAYS AS (
                     to_tsvector('english', heading || ' ' || content)
                 ) STORED,
    UNIQUE (doc_id, ordinal)                -- makes re-ingest idempotent
);

-- Approximate nearest neighbour index for cosine distance.
CREATE INDEX chunks_embedding_hnsw ON chunks
    USING hnsw (embedding vector_cosine_ops);

-- Keyword index. You will need it; see below.
CREATE INDEX chunks_tsv ON chunks USING gin (tsv);

Now embed and insert. Three details matter. Normalize the vectors, so cosine distance behaves. BGE models are trained asymmetrically — passages are embedded as-is, but queries get a specific instruction prefix, and skipping it quietly costs you retrieval quality for no reason. And make re-ingest idempotent: the UNIQUE constraint plus ON CONFLICT DO UPDATE means running the ingest twice on the same document updates rows in place instead of inserting a second copy under fresh ids. Be clear-eyed about what that does and does not buy you, though. Chunk ids are a bigserial surrogate key. They are stable across a re-ingest of the same chunking, and they are meaningless across a different one — change the chunk size and every boundary moves, so the rows are genuinely different rows. That is why the eval set in the next section is labelled against text, not against ids.

import hashlib
import psycopg
from pgvector.psycopg import register_vector

QUERY_PREFIX = "Represent this sentence for searching relevant passages: "

def embed_passages(texts: list[str]):
    return model.encode(texts, batch_size=32, normalize_embeddings=True)

def embed_query(question: str):
    return model.encode(QUERY_PREFIX + question, normalize_embeddings=True)

def content_hash(c: Chunk) -> str:
    payload = f"{c.doc_id}\x00{c.ordinal}\x00{c.text}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()

def ingest(conn, doc_id: str, chunks: list[Chunk]) -> None:
    payload = [f"{c.heading}\n\n{c.text}".strip() for c in chunks]
    vecs = embed_passages(payload)
    with conn.cursor() as cur:
        for c, v in zip(chunks, vecs):
            cur.execute(
                """INSERT INTO chunks
                       (doc_id, ordinal, heading, content, content_hash, embedding)
                   VALUES (%s, %s, %s, %s, %s, %s)
                   ON CONFLICT (doc_id, ordinal) DO UPDATE SET
                       heading      = EXCLUDED.heading,
                       content      = EXCLUDED.content,
                       content_hash = EXCLUDED.content_hash,
                       embedding    = EXCLUDED.embedding""",
                (c.doc_id, c.ordinal, c.heading, c.text, content_hash(c), v),
            )
        # a re-chunk can produce fewer chunks than last time; drop the stale tail
        cur.execute("DELETE FROM chunks WHERE doc_id = %s AND ordinal >= %s",
                    (doc_id, len(chunks)))
    conn.commit()

Vector search alone will embarrass you

Embeddings are good at meaning and bad at literal tokens. Ask for "section 12(b)" or a part number, and the nearest-neighbour search returns passages that are semantically about the same topic while missing the one that literally contains the string. Postgres full-text search is good at exactly the thing embeddings are weak at, and you already created the index. Be precise about what it is, though: to_tsvector('english', ...) does lexeme matching, not exact-string matching. It stems and splits, so "section 12(b)" becomes the lexemes section, 12 and b — which is enough to find the right clause. It is not enough for an identifier like ERR_4021-B, which gets fragmented on the punctuation. For identifiers, add a second index using the simple config, or a GIN trigram index via pg_trgm for real substring matching. Then fuse the two ranked lists with Reciprocal Rank Fusion — no score calibration, no tuning, just ranks.

-- hybrid.sql
WITH vec AS (
    SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> %(q)s) AS rank
    FROM chunks
    ORDER BY embedding <=> %(q)s
    LIMIT 50
),
kw AS (
    SELECT id, ROW_NUMBER() OVER (ORDER BY ts_rank_cd(tsv, query) DESC) AS rank
    FROM chunks, websearch_to_tsquery('english', %(text)s) query
    WHERE tsv @@ query
    ORDER BY ts_rank_cd(tsv, query) DESC
    LIMIT 50
)
SELECT c.id, c.doc_id, c.heading, c.content,
       kw.rank AS kw_rank,
       COALESCE(1.0 / (60 + vec.rank), 0)
     + COALESCE(1.0 / (60 + kw.rank),  0) AS score,
       1 - (c.embedding <=> %(q)s)         AS cosine_similarity
FROM vec
FULL OUTER JOIN kw USING (id)
JOIN chunks c ON c.id = COALESCE(vec.id, kw.id)
ORDER BY score DESC
LIMIT %(k)s;

The 60 in the denominator is the standard RRF constant. Do not agonise over it. Do, however, notice the LIMIT 50 in the vec CTE, because there is a trap right there: pgvector's hnsw.ef_search defaults to 40, and asking an HNSW index for more rows than its candidate list size quietly degrades recall. The query above asks for 50. So ef_search is a precondition of this query, not a remedy you reach for later — set it before you run it. Joining from the two CTEs rather than scanning chunks and filtering matters too: the outer query then touches about a hundred candidate rows instead of the whole table.

import os
from pathlib import Path
from psycopg.rows import dict_row

DSN = os.environ["DATABASE_URL"]
HYBRID_SQL = Path("hybrid.sql").read_text(encoding="utf-8")

def retrieve(conn, question: str, k: int = 6) -> list[dict]:
    qv = embed_query(question)
    with conn.cursor(row_factory=dict_row) as cur:
        # Must be >= the LIMIT in the vec CTE, or HNSW silently loses recall.
        # SET LOCAL scopes it to the current transaction (psycopg opens one for you).
        cur.execute("SET LOCAL hnsw.ef_search = 100")
        cur.execute(HYBRID_SQL, {"q": qv, "text": question, "k": k})
        return cur.fetchall()

Build the eval set before you touch the prompt

This is the step that separates a project that works from a project that demos. Open your corpus. Write 20 to 30 questions a real user would ask. For each one, copy out the exact span of text that answers it. That is your eval set. It takes an afternoon and it is the most valuable artifact in the repository — more valuable than the code, because the code is replaceable and the labels are not.

Label against the text, not against chunk ids. This is the part people get wrong, and it is worth being explicit about why. The moment you change the chunk size — the very first experiment this post tells you to run — every chunk boundary moves and every id is meaningless. If your labels point at chunk 412, they now point at a chunk that no longer contains what you meant. You would silently invalidate the entire eval set with the first thing you tried, and the numbers would still come out looking plausible. A gold answer span survives re-chunking because it is a property of the document, not of your chunking strategy. So a retrieval counts as a hit when the retrieved chunk's content contains the gold span.

import json, re, statistics

# evalset.jsonl — one object per line, written by hand:
# {"question": "How many days of casual leave am I entitled to?",
#  "gold_texts": ["twelve days of casual leave"]}

def normalize(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip().lower()

def is_hit(content: str, gold_texts: list[str]) -> bool:
    hay = normalize(content)
    return any(normalize(g) in hay for g in gold_texts)

def load_evalset(path: str) -> list[dict]:
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]

def evaluate(conn, evalset: list[dict], ks=(1, 3, 5, 10)) -> dict:
    kmax = max(ks)
    hits = {k: 0 for k in ks}
    reciprocal_ranks = []

    for row in evalset:
        retrieved = retrieve(conn, row["question"], k=kmax)
        flags = [is_hit(r["content"], row["gold_texts"]) for r in retrieved]

        for k in ks:
            if any(flags[:k]):
                hits[k] += 1

        rank = next((i + 1 for i, f in enumerate(flags) if f), None)
        reciprocal_ranks.append(1.0 / rank if rank else 0.0)

    n = len(evalset)
    return {**{f"recall@{k}": hits[k] / n for k in ks},
            "mrr": statistics.fmean(reciprocal_ranks)}

if __name__ == "__main__":
    with psycopg.connect(DSN) as conn:
        register_vector(conn)
        print(evaluate(conn, load_evalset("evalset.jsonl")))

Now you have a number. Change the chunk size and run it again — and because the labels are text spans, the eval set still means what it meant before. Turn off the keyword arm and run it again. Drop the query prefix and run it again. Every one of those is a five-minute experiment with an honest answer at the end, and you will be surprised at least twice. Put the resulting table in your report — a chart of recall@5 across four chunking strategies is worth more to an examiner than another screenshot of a chat bubble.

And be honest with yourself about what counts as evidence. "It works on my three test questions" is not evidence — three questions you invented after building the system will happen to be the three the system handles. Twenty-five questions written from the documents, before you tuned anything, with a measured recall@5 and every single failure listed and explained: that is evidence. The failures are the most interesting part. Go read them.

Citations are not a feature, they are the contract

A grounded answer must be checkable. That means every claim carries the id of the chunk it came from, the UI renders that id as a link to the source passage, and — critically — you verify after generation that the cited ids are actually in the set you retrieved. A model that cites a chunk you never gave it has hallucinated the citation too, and you should reject the whole answer rather than show it.

The refusal path is the anti-hallucination guardrail

Your system needs a way to say "I don't know." Two independent gates, because either one alone leaks. Gate one: if retrieval did not turn up anything that looks like a real match, do not call the model at all. Gate two: force the model into a structured response with an explicit sufficiency flag, and let it decline. Gate one has a subtlety that will bite you. The obvious implementation checks the similarity of the top result — but the top result is the top RRF-fused result, and it may have been surfaced entirely by the keyword arm. Someone asks for "section 12(b)", the keyword arm correctly ranks the exact-match clause first, its cosine similarity is mediocre because embeddings are bad at literal tokens, and your refusal gate throws away the exact case you added hybrid search to fix. So check the best similarity across all hits, and let a strong keyword hit override the floor.

The threshold itself is the other trap, and it is why the code below refuses to ship with a default. BGE embeddings are strongly anisotropic: their similarities are compressed into the upper part of the range, so two completely unrelated short texts typically score somewhere around 0.6 to 0.75, not near zero. A floor of 0.45 would therefore never fire — you would believe you had two gates while shipping one, and you would also conclude your retrieval is excellent because everything scores 0.7. For bge-small the useful threshold usually lands somewhere around 0.7 to 0.8, and the only way to find yours is to measure it. Add 10 questions to your eval set whose answers are deliberately not in the corpus, and pick the threshold that refuses those without refusing the answerable ones. Your refusal rate becomes a measured number too.

import json
import anthropic

client = anthropic.Anthropic()   # reads ANTHROPIC_API_KEY

ANSWER_SCHEMA = {
    "type": "object",
    "properties": {
        "sufficient": {"type": "boolean"},
        "answer":     {"type": "string"},
        "citations":  {"type": "array", "items": {"type": "integer"}},
    },
    "required": ["sufficient", "answer", "citations"],
    "additionalProperties": False,
}

SYSTEM = (
    "Answer only from the numbered sources provided. Every factual sentence "
    "must cite the id of the source it came from. If the sources do not "
    "contain the answer, set sufficient to false, leave answer empty, and "
    "cite nothing. Never use prior knowledge. Respond with the final answer only."
)

REFUSAL = {"sufficient": False, "answer": "", "citations": []}

MIN_SIMILARITY = None   # You MUST calibrate this on unanswerable questions.

def answer(conn, question: str) -> dict:
    if MIN_SIMILARITY is None:
        raise RuntimeError("Calibrate MIN_SIMILARITY on your unanswerable set first.")

    hits = retrieve(conn, question, k=6)
    if not hits:
        return dict(REFUSAL)

    # Gate 1: retrieval confidence. Cheapest refusal there is — no API call.
    # Best similarity across ALL hits, not hits[0]: the top RRF result may have
    # come from the keyword arm and carry a mediocre cosine score.
    best_sim = max(h["cosine_similarity"] for h in hits)
    exact_match = any(h["kw_rank"] is not None and h["kw_rank"] <= 3 for h in hits)
    if best_sim < MIN_SIMILARITY and not exact_match:
        return dict(REFUSAL)

    sources = "\n\n".join(f"[{h['id']}] {h['heading']}\n{h['content']}" for h in hits)

    msg = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        system=SYSTEM,
        messages=[{"role": "user",
                   "content": f"Sources:\n\n{sources}\n\nQuestion: {question}"}],
        output_config={"format": {"type": "json_schema", "schema": ANSWER_SCHEMA}},
    )

    # The model has a refusal path of its own. Do not crash on it.
    # stop_reason "refusal" can arrive with an empty content list (IndexError),
    # and "max_tokens" with truncated JSON (JSONDecodeError). Both are refusals.
    if msg.stop_reason in ("refusal", "max_tokens") or not msg.content:
        return dict(REFUSAL)
    try:
        result = json.loads(msg.content[0].text)
    except json.JSONDecodeError:
        return dict(REFUSAL)

    # Gate 2: citations must point at chunks we actually retrieved.
    allowed = {h["id"] for h in hits}
    result["citations"] = [c for c in result["citations"] if c in allowed]
    if result["sufficient"] and not result["citations"]:
        return dict(REFUSAL)   # an answer with no surviving evidence is a refusal

    return result

Note the last gate carefully: when the citation filter strips every id the model produced, the answer text is discarded along with them. Demoting the sufficiency flag but keeping the prose is a bug — a caller that renders result["answer"] would happily display an answer whose evidence you just proved was invented. If you would rather not hand-roll the JSON parsing at all, the SDK's client.messages.parse() validates the response against the same schema for you; you still need the stop_reason check.

The cost and latency budget

Do the arithmetic before you demo, because someone will ask. Six chunks at roughly 350 tokens is about 2,100 tokens of sources, plus the system prompt and the question — call it 2,400 input tokens. A grounded answer with citations is maybe 250 output tokens. At Claude Opus 4.8's list price of $5 per million input tokens and $25 per million output, that is roughly $0.018 per question, about 55 questions per dollar. The same call on Claude Haiku 4.5 ($1 and $5 per million) is roughly $0.004, about 270 questions per dollar. Those are the two numbers to put on the slide; which tradeoff you make is a product decision, not a technical one. And note where the money is not: retrieval costs you nothing in API spend, because the query embedding runs locally and the search is a database query. It is not free — the encoder burns CPU and the search burns I/O — but almost all of your money and almost all of your latency lives in the generation call. Optimising your vector index for speed before you have measured end-to-end p95 is premature by a wide margin.

Prompt caching will not save you here either, and you should know why before you try it. Caching is a prefix match — it only pays off when a large chunk of the beginning of the prompt is byte-identical across requests. In a RAG loop the retrieved sources change with every question, so the only stable prefix is the system prompt, and on Opus 4.8 the minimum cacheable prefix is 4,096 tokens. The system prompt above is under a hundred. It would never cache, silently, with no error. The lever that actually reduces cost is retrieving fewer, better chunks — which brings you right back to recall@k.

What to actually build, in order

1. 1Ingest and chunk on structure. Assert that no chunk exceeds the encoder's token limit.
2. 2Embed locally with a small open model. Store in pgvector with an HNSW index and a tsvector column.
3. 3Hand-write 20 to 30 question/gold-span pairs. Measure recall@k and MRR. Write the number down.
4. 4Iterate on retrieval only — chunk size, overlap, hybrid fusion, ef_search, query prefix — until recall@5 stops improving.
5. 5Add 10 unanswerable questions and calibrate the similarity floor against them.
6. 6Only now add the generation call, with a citation contract and both refusal gates.

Notice that the model does not appear until step six. That ordering is the whole argument of this post, and it is the thing that will make your project defensible when someone asks the hard question — not "does it work," but "how do you know."

This is how we work on final-year and mini projects at Tenzok: eval set first, retrieval measured, guardrails in the code rather than in the report. If you are building a RAG chatbot and your recall number does not exist yet, that is the place to start — with us or without us.