Project: Build a Production RAG System

কী বানাবে

এই course এর উপর একটা প্রশ্নোত্তর service, demo না, production system এর মতো করে বানানো: hybrid retrieval, reranking, একটা relevance threshold যেটা "জানি না" বলে, নম্বর দেওয়া citation সহ answer, প্রতিটা answer এ নিজে থেকে check, check fail করলে নিরাপদ একটা fallback, token streaming, আর বারবার আসা প্রশ্নের জন্য একটা cache।

ইচ্ছা করেই ছোট একটা model দিয়ে চালাবে (CPU তে Qwen2.5-0.5B)। এটা সীমাবদ্ধতা না, সুবিধা: দুর্বল model প্রায়ই ফেল করে, তাই প্রতিটা নিরাপত্তার স্তর কাজ করতে দেখবে। শক্তিশালী model এ একই code কম ফেল করে, তবু প্রতিটা স্তরই লাগে।

যা আগে জানা লাগবে

• RAG Architecture আর Reranking।
• Structured Outputs: model এর output validate করা।
• Project: Build a Semantic Search Engine: এই project তার search.py আর docs.json আবার ব্যবহার করে।
• Local model গুলো: multilingual-e5-small, mmarco-mMiniLMv2-L12-H384-v1 আর Qwen2.5-0.5B-Instruct।

Architecture

Play

File	কাজ
search.py, docs.json	799 টা paragraph এর উপর hybrid retrieval (আগের project)
checks.py	grounding check আর উদ্ধৃতি দিয়ে fallback, কোনো model ছাড়াই test করা যায়
rag.py	pipeline: cache, retrieve, rerank, threshold, stream, check, fallback
main.py	চারটা প্রশ্ন: দুইটার answer আছে, একটা বিষয়ের বাইরে, একটা বারবার
test_checks.py	check গুলোর unit test

ধাপে ধাপে বানাও

Step 1: চওড়া করে আনো, সরু করে rerank করো। Hybrid search 20 টা candidate দেয় (recall); cross-encoder সেরা 3 টা রাখে (precision)। শুধু 3 টা ছোট chunk prompt এ যায়, তাতে prompt সস্তা আর মনোযোগী থাকে।

Step 2: প্রাসঙ্গিক কিছু না থাকলে না বলো। সেরা rerank score একটা threshold এর নিচে হলে (এখানে −2, বিষয়ের ভেতরের আর বাইরের প্রশ্নের score দেখে বাছা) LLM না ডেকেই "জানি না" ফেরত দাও। Model কে বানিয়ে বলতে দেওয়ার চেয়ে সস্তা আর বেশি সৎ।

Step 3: citation সহ generate, streaming। Source গুলো prompt এ যায় [1] থেকে [3] নম্বর দিয়ে। TextIteratorStreamer token তৈরি হওয়ার সাথে সাথে দিয়ে দেয়, তাই UI সাথে সাথে text দেখাতে পারে।

Step 4: প্রতিটা answer check করো। সাধারণ code, কোনো model না: citation আছে কি, সেগুলো এমন source এ দেখাচ্ছে কি যেটা আছে, আসলেই কোনো answer আছে কি, শুধু "[1] [2] [3]" না তো? এই check গুলো checks.py তে থাকে যাতে unit test করা যায়।

Step 5: নিরাপদে fallback। কোনো check fail করলে খসড়া দেখাবে না। সেরা source এর প্রথম দুইটা বাক্য দেখাও, cite করে। কম সাবলীল, সবসময় ভিত্তিসহ।

Step 6: cache। Normalise করা প্রশ্নকে key বানাও (ছোট হাতের, ছাঁটা); বারবার এলে খরচ শূন্য।

"""Grounding checks and the extractive fallback: plain Python, testable without any model."""
import re

def check(answer, n_sources):
    """Cheap grounding checks on the model's answer. Each returns a problem description."""
    cited = {int(n) for n in re.findall(r"\[(\d+)\]", answer)}
    words = re.sub(r"\[\d+\]", "", answer).split()
    problems = [f"cites [{n}] which doesn't exist" for n in sorted(cited) if not 1 <= n <= n_sources]
    if "don't know" in answer.lower():
        return problems
    if not cited:
        problems.append("no citation")
    if len(words) < 5:
        problems.append("no real answer, only citations")
    return problems

def first_sentences(text, n=2):
    return " ".join(re.split(r"(?<=[.।?!])\s+", text)[:n])

"""A RAG service: hybrid retrieve -> rerank -> threshold -> grounded answer with citations -> checks -> cache."""
import hashlib
import time
import numpy as np
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer, TextIteratorStreamer
from threading import Thread
from llm import model, tokenizer
from checks import check, first_sentences
from search import SearchIndex

CE = "cross-encoder/mmarco-mMiniLMv2-L12-H384-v1"
_ce_tok = AutoTokenizer.from_pretrained(CE)
_ce = AutoModelForSequenceClassification.from_pretrained(CE).eval()

SYSTEM = ("You answer questions about an AI engineering course using ONLY the numbered sources. "
          "Cite every fact like [1]. If the sources don't answer the question, reply exactly: I don't know.")

class RAG:
    def __init__(self, docs, min_rerank_score=0.0, k=3):
        self.index = SearchIndex(docs, cache="/tmp/rag_vectors.npy")
        self.docs = {d["id"]: d for d in docs}
        self.min_score, self.k = min_rerank_score, k
        self.cache = {}

    def rerank(self, question, hits):
        texts = [self.docs[h["id"]]["text"] for h in hits]
        enc = _ce_tok([question] * len(texts), texts, padding=True, truncation=True, max_length=256, return_tensors="pt")
        with torch.no_grad():
            scores = _ce(**enc).logits[:, 0].numpy()
        order = np.argsort(-scores)
        return [(hits[i], float(scores[i])) for i in order]

    def ask(self, question, stream=None):
        key = hashlib.sha256(question.strip().lower().encode()).hexdigest()
        if key in self.cache:
            return self.cache[key] | {"cached": True}
        t0 = time.perf_counter()
        hits = self.index.search(question, mode="hybrid", k=20)          # 1. retrieve wide
        ranked = self.rerank(question, hits)[: self.k]                    # 2. rerank, keep the best few
        if ranked[0][1] < self.min_score:                                 # 3. nothing relevant: don't guess
            result = {"answer": "I don't know: this isn't covered in the course material I can search.",
                      "sources": [], "top_score": ranked[0][1]}
        else:
            sources = [self.docs[h["id"]] for h, _ in ranked]
            ctx = "\n".join(f"[{i + 1}] ({s['lesson']}) {s['text']}" for i, s in enumerate(sources))
            answer = self.generate([{"role": "system", "content": SYSTEM},
                                    {"role": "user", "content": f"Sources:\n{ctx}\n\nQuestion: {question}"}], stream)
            problems = check(answer, len(sources))                        # 4. grounding checks on the output
            draft = answer
            if problems:                                                  # 5. fallback: quote the best source
                answer = f"{first_sentences(sources[0]['text'])} [1]"
            result = {"answer": answer, "draft": draft, "top_score": ranked[0][1], "problems": problems,
                      "fallback": bool(problems), "sources": [f"{s['lesson']}#{s['section']}" for s in sources]}
        result["ms"] = round((time.perf_counter() - t0) * 1000)
        self.cache[key] = result                                          # 6. cache by normalised question
        return result | {"cached": False}

    def generate(self, messages, stream=None, max_new_tokens=90):
        ids = tokenizer.apply_chat_template(messages, add_generation_prompt=True, return_tensors="pt")
        streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
        args = dict(input_ids=ids, attention_mask=torch.ones_like(ids), max_new_tokens=max_new_tokens,
                    do_sample=False, pad_token_id=tokenizer.eos_token_id, streamer=streamer)
        Thread(target=model.generate, kwargs=args).start()
        parts = []
        for piece in streamer:                                            # tokens arrive while the model works
            parts.append(piece)
            if stream:
                stream(piece)
        return "".join(parts).strip()

import json
from rag import RAG

rag = RAG(json.load(open("docs.json")), min_rerank_score=-2.0)
chunks = []

for q in ["Why does Bangla text cost more tokens than English?",
          "What does the KV cache store?",
          "What is the capital of Australia?",
          "why does bangla text cost more tokens than english?"]:
    chunks.clear()
    r = rag.ask(q, stream=chunks.append)
    print(f"\nQ: {q}")
    print(f"  answer:  {r['answer']}")
    print(f"  sources: {r['sources']}   top rerank score {r['top_score']:.2f}")
    if r.get("problems"):
        print(f"  rejected draft: {r['draft'][:110]}...")
        print(f"  checks failed: {r['problems']} -> answered by quoting source [1]")
    print(f"  cached: {r['cached']}" + (f", streamed in {len(chunks)} pieces" if chunks else ""))

Output:

Q: Why does Bangla text cost more tokens than English?
  answer:  Why Bangla costs more. In UTF-8, an English letter is 1 byte, a Bangla letter is 3. [1]
  sources: ['tokenization#how-it-works', 'tokenization#takeaway', 'tokenization#concept']   top rerank score 10.68
  rejected draft: Bengali text costs more tokens than English because of the way BPE (Byte Pair Encoding) is used to merge words...
  checks failed: ['no citation'] -> answered by quoting source [1]
  cached: False, streamed in 91 pieces

Q: What does the KV cache store?
  answer:  KV cache. In self-attention, each new token needs the keys and values of every earlier token. [1]
  sources: ['inference-decoding#how-it-works', 'inference-decoding#takeaway', 'inference-decoding#how-it-works']   top rerank score 6.66
  rejected draft: The KV cache stores past keys and values so that each step only processes the new token, and the context windo...
  checks failed: ['no citation'] -> answered by quoting source [1]
  cached: False, streamed in 37 pieces

Q: What is the capital of Australia?
  answer:  I don't know: this isn't covered in the course material I can search.
  sources: []   top rerank score -4.58
  cached: False

Q: why does bangla text cost more tokens than english?
  answer:  Why Bangla costs more. In UTF-8, an English letter is 1 byte, a Bangla letter is 3. [1]
  sources: ['tokenization#how-it-works', 'tokenization#takeaway', 'tokenization#concept']   top rerank score 10.68
  rejected draft: Bengali text costs more tokens than English because of the way BPE (Byte Pair Encoding) is used to merge words...
  checks failed: ['no citation'] -> answered by quoting source [1]
  cached: True

কী দেখলাম:

• বিষয়ের বাইরের প্রশ্ন LLM পর্যন্ত পৌঁছায়ই না। "Capital of Australia" এর সেরা rerank score −4.58, −2 এর threshold এর নিচে, তাই service সাথে সাথে "I don't know" বলে। কোনো বানানো answer নেই, LLM এর খরচ নেই।
• দুইটা আসল প্রশ্নেই retrieval আর reranking ঠিক: বাংলার প্রশ্নে তিনটা Tokenization chunk (সর্বোচ্চ score 10.68), KV cache এর প্রশ্নে তিনটা Inference chunk।
• Model এর খসড়া check এ ফেল করে। বাংলার খসড়ায় কোনো citation নেই আর আংশিক ভুল ("BPE tends to create fewer tokens")। Check গুলো সেটা বাতিল করে, আর user পায় সেরা source এর প্রথম বাক্যগুলো: "In UTF-8, an English letter is 1 byte, a Bangla letter is 3. [1]"। ছোট, ঠিক, cite করা। KV cache এর ক্ষেত্রেও একই।
• বানানোর সময় একটা "ভালো" prompt জিনিস আরো খারাপ করেছিল। System prompt এ একটা citation এর উদাহরণ দেওয়ায় model শুধু "[1] [2] [3]" দিয়ে reply দিচ্ছিল, আর check এর প্রথম version সেটা পার হতে দিয়েছিল। এজন্যই একটা no real answer, only citations check আর তার একটা test আছে।
• বারবার আসা প্রশ্ন (বড় ছোট হাতের অক্ষর আলাদা) cache থেকে আসে।
• Streaming এ একটা ফাঁদ আছে: বাংলার answer এমন একটা খসড়ার 91 টা টুকরো stream করেছে যেটা পরে বাতিল হয়েছে। Production এ হয় stream করে পরে সংশোধন দেখাও, নয়তো দেখানোর আগে check করো। এই সমঝোতা নিচে আছে।

Test করো

from checks import check, first_sentences

def test_good_answer_passes():
    assert check("A Bangla letter is 3 bytes in UTF-8 [1], so it needs more tokens [2].", 3) == []

def test_missing_citation_fails():
    assert check("Bangla letters take more bytes, so more tokens.", 3) == ["no citation"]

def test_citations_without_an_answer_fail():
    assert check("[1] [2] [3]", 3) == ["no real answer, only citations"]

def test_citing_a_source_that_does_not_exist_fails():
    assert check("The cache stores keys and values [7].", 3) == ["cites [7] which doesn't exist"]

def test_honest_dont_know_passes():
    assert check("I don't know.", 3) == []

def test_fallback_quotes_whole_sentences_in_both_languages():
    assert first_sentences("One. Two. Three.") == "One. Two."
    assert first_sentences("প্রথম বাক্য। দ্বিতীয়। তৃতীয়।") == "প্রথম বাক্য। দ্বিতীয়।"

if __name__ == "__main__":
    for name, fn in list(globals().items()):
        if name.startswith("test_"):
            fn()
            print("pass", name)

pass test_good_answer_passes
pass test_missing_citation_fails
pass test_citations_without_an_answer_fail
pass test_citing_a_source_that_does_not_exist_fails
pass test_honest_dont_know_passes
pass test_fallback_quotes_whole_sentences_in_both_languages

প্রতিটা test বানানোর সময় দেখা একটা ব্যর্থতার ধরন: citation নেই, শুধু citation, এমন source এর citation যেটার অস্তিত্ব নেই, আর সৎ "I don't know" যেটা বাতিল হওয়া চলবে না। Fallback কে English আর বাংলা দুই ভাষাতেই বাক্যের শেষে কাটতে হবে (। হলো বাংলার দাঁড়ি)। Unit test এর পরের ধাপ হলো প্রত্যাশিত source সহ আসল প্রশ্নের একটা evaluation set, CI তে চালানো: Part 8 ঠিক এই system এর জন্যই সেটা বানায়।

Trade-offs

Choice	ভালো দিক	খরচ
Rerank threshold দিয়ে না বলা	বিষয়ের বাইরে বানানো answer নেই; LLM call বাঁচে	অস্বাভাবিকভাবে বলা কিছু answer দেওয়ার মতো প্রশ্নও বাদ পড়ে; label দেওয়া প্রশ্নে threshold tune করো
উদ্ধৃতি দিয়ে fallback	সবসময় ভিত্তিসহ আর cite করা	ভালো generate করা answer এর চেয়ে কম কাজের; উদ্ধৃতি হুবহু প্রশ্নের answer নাও হতে পারে
Streaming	user এক second এর মধ্যে text দেখে	বাতিল খসড়া আগেই দেখানো হয়ে গেছে; উপায়: প্রথম বাক্য জমিয়ে check করো, stream করে পরে বদলাও, বা ঝুঁকির বিষয়ে দেখানোর আগে check করো
Normalise করা প্রশ্নে cache	বারবার এলে বিনা খরচে	document বদলালে পুরনো হয়ে যায়: re-index এ মুছে দাও, বা key তে index এর version রাখো; ভাগ করা cache এ কখনো user এর ব্যক্তিগত answer cache কোরো না
ছোট local model	সস্তা, ব্যক্তিগত, প্রতিটা ব্যর্থতা দেখায়	বেশিরভাগ answer fallback এ যায়; শক্তিশালী model অনেক বেশি বার check পার হতো (Part 8 এ মাপো)
String ভিত্তিক check	দ্রুত, test করা যায়, model লাগে না	আকার check করে (citation আছে), সত্য না (cite করা text দাবিটা সমর্থন করে কিনা); তার জন্য LLM judge বা NLI লাগে (Part 8)

আরো এগিয়ে নাও

1. একটা check যোগ করো যে answer এর প্রতিটা সংখ্যা (দিন, টাকা, শতাংশ) cite করা source এর text এ আছে।
2. Prompt এ প্রতি user এর কথোপকথনের ইতিহাস (শেষ 3 টা turn) যোগ করো আর cache আবার test করো: এখন cache key তে ইতিহাস রাখতে হবে কেন?
3. আরো শক্তিশালী hosted বা local model বসাও, আবার চালাও, আর গোনো কতগুলো answer এখন fallback ছাড়াই check পার হয়।
4. আগের project এর search server এ একটা /ask endpoint যোগ করো যেটা server-sent events দিয়ে token stream করে।