Project: Build a Semantic Search Engine

কী বানাবে

এই course এর জন্য একটা search service। তাকে জিজ্ঞেস করো "my chatbot bill is higher for Bengali users", সে Tokenization lesson এর সেই paragraph টা ফেরত দেবে যেটা কারণ ব্যাখ্যা করে, paragraph টা English এ হোক বা বাংলায়। যে lesson গুলো পড়েছ সেখান থেকে প্রায় 800 টা paragraph index করবে, ছোট একটা HTTP API এর পেছনে search চালাবে, reciprocal rank fusion দিয়ে keyword আর অর্থের search মেলাবে, আর মেপে দেখবে অন্যভাবে বলা প্রশ্নে কোন mode আসলে সবচেয়ে ভালো কাজ করে।

পরের project এর production RAG system এর retrieval অংশ এটাই, আর নিজে নিজেও একটা পুরো, কাজের service।

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

• Embedding Models and Vector Databases।
• Semantic Search and Hybrid Search: BM25 আর RRF।
• Project: Find Similar Sentences with NumPy: bag-of-words এর শুরুর জায়গা।
• NumPy আর transformers সহ Python, আর multilingual-e5-small model (প্রায় 470 MB)।

Architecture

Play

File	কাজ
docs.json	799 টা paragraph (401 English, 398 বাংলা), lesson, section আর ভাষা সহ
embed.py	বহুভাষিক embedding model (embeddings lesson থেকে)
search.py	SearchIndex: BM25, vector search, RRF; vector disk এ cache করে
server.py	GET /search: input validation, k এর সীমা, JSON response, শুধু standard library
main.py	index বানায়, server চালায়, HTTP দিয়ে ডাকে, মান আর latency মাপে
test_search.py	tokenizer, determinism আর API-validation এর test

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

Step 1: document। 80+ অক্ষরের প্রতিটা lesson paragraph, code block আর table ছাড়া, দুই ভাষাতেই, প্রতিটার সাথে lesson এর slug আর section id। এই label গুলোই result কে click করার মতো বানায় আর evaluation সম্ভব করে।

Step 2: index। BM25 এর এমন tokenizer লাগবে যেটা বাংলা বোঝে: আগের [a-z]+ চুপচাপ প্রতিটা বাংলা অক্ষর ফেলে দিত। এখানকার pattern বাংলা Unicode block রাখে, কার চিহ্ন সহ। CPU তে 799 টা paragraph embed করতে প্রায় এক মিনিট লাগে, তাই vector disk এ save করে আবার load করা হয়; production এ এটাই তোমার vector database।

"""A small hybrid search engine: BM25 + embeddings + reciprocal rank fusion, with a saved index."""
import json
import math
import os
import re
from collections import Counter
import numpy as np
from embed import embed

TOKEN = re.compile(r"[a-z0-9ঀ-৿]+")          # English words, numbers, and Bangla letters with their marks

def tokenize(text):
    return TOKEN.findall(text.lower())

class SearchIndex:
    def __init__(self, docs, cache="vectors.npy"):
        self.docs = docs
        self.tf = [Counter(tokenize(d["text"])) for d in docs]
        self.len = [sum(c.values()) for c in self.tf]
        self.avg = sum(self.len) / len(docs)
        df = Counter(w for c in self.tf for w in c)
        self.idf = {w: math.log(1 + (len(docs) - n + 0.5) / (n + 0.5)) for w, n in df.items()}
        if os.path.exists(cache):                       # embedding is the slow part: do it once, then reuse
            self.vectors = np.load(cache)
        else:
            self.vectors = embed([d["text"] for d in docs])
            np.save(cache, self.vectors)

    def bm25(self, query, k1=1.5, b=0.75):
        q = tokenize(query)
        scores = np.zeros(len(self.docs))
        for i, (c, n) in enumerate(zip(self.tf, self.len)):
            for w in q:
                if w in c:
                    scores[i] += self.idf[w] * c[w] * (k1 + 1) / (c[w] + k1 * (1 - b + b * n / self.avg))
        return scores

    def search(self, query, mode="hybrid", k=5, candidates=50):
        keyword = list(np.argsort(-self.bm25(query), kind="stable")[:candidates])
        semantic = list(np.argsort(-(self.vectors @ embed([query], kind="query")[0]), kind="stable")[:candidates])
        if mode == "keyword":
            ranked = keyword
        elif mode == "semantic":
            ranked = semantic
        else:
            fused = Counter()
            for lst in (keyword, semantic):
                for r, i in enumerate(lst):
                    fused[i] += 1 / (60 + r + 1)
            ranked = [i for i, _ in fused.most_common()]
        return [{k: self.docs[i][k] for k in ("id", "lang", "lesson", "section")} | {"text": self.docs[i]["text"][:90]}
                for i in ranked[:k]]

Step 3: API। সাধারণ http.server, কারণ logic টাই আসল। খেয়াল করো সে কী কী ফিরিয়ে দেয়: খালি বা 300+ অক্ষরের query, অজানা mode, আর 1 থেকে 20 এর বাইরের যেকোনো k কে সীমায় আনে। Client কে কখনো ঠিক করতে দিও না তোমার server কতটা কাজ করবে।

"""GET /search?q=...&mode=hybrid|keyword|semantic&k=5  ->  JSON results. Standard library only."""
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

def make_server(index, port=8099):
    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            url = urlparse(self.path)
            params = {k: v[0] for k, v in parse_qs(url.query).items()}
            q = params.get("q", "").strip()
            mode = params.get("mode", "hybrid")
            if url.path != "/search" or not q or len(q) > 300 or mode not in ("hybrid", "keyword", "semantic"):
                return self.reply(400, {"error": "use /search?q=<1-300 chars>&mode=hybrid|keyword|semantic"})
            k = min(max(int(params.get("k", 5)), 1), 20)          # clamp: never trust client limits
            self.reply(200, {"query": q, "mode": mode, "results": index.search(q, mode, k)})

        def reply(self, status, body):
            data = json.dumps(body, ensure_ascii=False).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def log_message(self, *args):                             # keep the demo output quiet
            pass

    return ThreadingHTTPServer(("127.0.0.1", port), Handler)

Step 4: চালাও আর মাপো। Label দেওয়া বারোটা প্রশ্ন, ইচ্ছা করেই অন্যভাবে বলা যাতে lesson এর শব্দের সাথে কম মেলে (overfitting এর জন্য "aces the practice papers but fails the real exam")। দুইটা metric: hit@1 (সঠিক lesson প্রথম result) আর MRR@5 (প্রথম সঠিক result এর 1/rank এর গড়; 1.0 মানে নিখুঁত)।

import json
import threading
import time
import urllib.parse
import urllib.request
from search import SearchIndex
from server import make_server

docs = json.load(open("docs.json"))
t = time.time()
index = SearchIndex(docs, cache="/tmp/vectors.npy")            # fresh index every run of this demo
print(f"indexed {len(docs)} paragraphs ({sum(d['lang'] == 'bn' for d in docs)} Bangla) "
      f"into {index.vectors.shape} vectors")

server = make_server(index)
threading.Thread(target=server.serve_forever, daemon=True).start()

def get(params):
    url = "http://127.0.0.1:8099/search?" + urllib.parse.urlencode(params)
    try:
        with urllib.request.urlopen(url) as r:
            return r.status, json.load(r)
    except urllib.error.HTTPError as e:
        return e.code, json.load(e)

status, body = get({"q": "why does bangla cost more tokens", "k": 3})
print(f"\nGET /search?q=why does bangla cost more tokens&k=3 -> {status}")
for r in body["results"]:
    print(f"  [{r['lang']}] {r['lesson']}#{r['section']}: {r['text'][:70]}...")
print("GET /search?q= ->", get({"q": ""}))

# Quality: 12 questions whose right lesson we know. hit@1 = it's the first result; MRR averages 1/rank.
LABELLED = [  # paraphrases on purpose: few of the lessons' own words
    ("my chatbot bill is higher for Bengali users", "tokenization"),
    ("answers get slower as the chat grows longer", "inference-decoding"),
    ("training error shot up to infinity after a few steps", "optimization"),
    ("blocking honest customers vs letting cheaters through", "training-evaluation"),
    ("aces the practice papers but fails the real exam", "overfitting-regularization"),
    ("softmax becomes too peaky when vectors are wide", "self-attention"),
    ("the bot made up a number that looked perfectly valid", "structured-outputs"),
    ("combine exact word matching with meaning matching", "semantic-hybrid-search"),
    ("rotating vectors so the model knows word order", "embeddings-positions"),
    ("a column that only gets filled after the customer left", "feature-engineering"),
    ("one weird big spender ruined my customer groups", "clustering"),
    ("early layers of a deep net stop learning", "training-networks"),
]
print(f"\n{'mode':<9} hit@1   MRR@5   p50 latency")
for mode in ("keyword", "semantic", "hybrid"):
    hit1, rr, times = 0, 0.0, []
    for q, lesson in LABELLED:
        t = time.perf_counter()
        _, body = get({"q": q, "mode": mode})
        times.append((time.perf_counter() - t) * 1000)
        ranks = [i + 1 for i, r in enumerate(body["results"]) if r["lesson"] == lesson]
        hit1 += bool(ranks) and ranks[0] == 1
        rr += 1 / ranks[0] if ranks else 0               # reciprocal rank of the first right result
    print(f"{mode:<9} {hit1:>2}/{len(LABELLED)}  {rr / len(LABELLED):5.2f}   {sorted(times)[len(times) // 2]:5.0f} ms")
server.shutdown()

Output:

indexed 799 paragraphs (398 Bangla) into (799, 384) vectors

GET /search?q=why does bangla cost more tokens&k=3 -> 200
  [en] tokenization#how-it-works: Why Bangla costs more. In UTF-8, an English letter is 1 byte, a Bangla...
  [en] tokenization#concept: This matters far more than it sounds. You pay per token, the context w...
  [en] tokenization#practice: Change the Bangla part of the training corpus from  3 to  120, so Bang...
GET /search?q= -> (400, {'error': 'use /search?q=<1-300 chars>&mode=hybrid|keyword|semantic'})

mode      hit@1   MRR@5   p50 latency
keyword    6/12   0.59      22 ms
semantic   8/12   0.78      21 ms
hybrid    10/12   0.86      20 ms

কী দেখলাম:

• একটা আসল query তিনটা Tokenization paragraph ফেরত দেয়, "Why Bangla costs more" paragraph টা প্রথমে, lesson আর section সহ, যাতে একটা UI সরাসরি সেখানে link দিতে পারে।
• খালি query পায় 400, সাথে কী অনুমোদিত সেটা বলা একটা message, crash বা অকারণে দামি একটা search এর বদলে।
• অন্যভাবে বলা প্রশ্নে মান: keyword search 12 এর মাত্র 6 বার সঠিক lesson প্রথমে পায় (MRR 0.59): অন্যভাবে বলা প্রশ্নের সাথে তার শব্দ মেলে না। Semantic search পায় 8 (0.78)। Hybrid পায় 10 (0.86), দুইটার যেকোনোটার চেয়ে ভালো, কারণ একটা যে প্রশ্ন মিস করে অন্যটা সেটা ধরে।
• এই মাপের প্রথম version অকাজের ছিল। Lesson এর নিজের শব্দ থেকে কপি করা প্রশ্ন প্রতিটা mode এ 12/12 পেয়েছিল। সহজ test প্রশ্ন পার্থক্য লুকিয়ে ফেলে; আসল user রা যেভাবে লিখবে সেভাবে লেখো।
• Latency: CPU তে 799 টা document এ প্রতি request এ প্রায় 20 ms, প্রায় পুরোটাই query embed করায়। এখানকার BM25 প্রতিটা document এর উপর একটা Python loop, এই সাইজে ঠিক আছে, আর 1 লাখ+ এ প্রথমেই বদলাতে হবে (একটা inverted index দিয়ে)।

Test করো

import json
import threading
import urllib.error
import urllib.request
from search import SearchIndex, tokenize
from server import make_server

docs = json.load(open("docs.json"))[:40]
index = SearchIndex(docs, cache="/tmp/test_vectors.npy")

def test_bangla_tokens_keep_their_vowel_signs():
    assert tokenize("টাকা ফেরত, ERR-517!") == ["টাকা", "ফেরত", "err", "517"]

def test_results_are_deterministic_and_limited():
    a = index.search("token cost", k=3)
    assert a == index.search("token cost", k=3) and len(a) == 3

def test_api_rejects_bad_input_and_clamps_k():
    server = make_server(index, port=8098)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    def get(path):
        try:
            with urllib.request.urlopen("http://127.0.0.1:8098" + path) as r:
                return r.status, json.load(r)
        except urllib.error.HTTPError as e:
            return e.code, json.load(e)
    assert get("/search?q=")[0] == 400
    assert get("/search?q=hi&mode=drop+table")[0] == 400
    assert get("/search?q=" + "a" * 301)[0] == 400
    status, body = get("/search?q=vectors&k=1000")
    assert status == 200 and len(body["results"]) == 20       # k=1000 was clamped to 20
    server.shutdown()

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

pass test_bangla_tokens_keep_their_vowel_signs
pass test_results_are_deterministic_and_limited
pass test_api_rejects_bad_input_and_clamps_k

• বাংলা token কার চিহ্ন রাখে: "টাকা" কে একটা token ই থাকতে হবে, "া" তে ভাঙা যাবে না। এই একটা test ই [a-z] এর bug ধরে ফেলত।
• Deterministic, সীমিত result: একই query একই list দেয়, চাওয়া দৈর্ঘ্যের।
• API validation: খালি, বেশি লম্বা আর অজানা mode এর request পায় 400; k=1000 সীমায় এসে 20 হয়।

Trade-offs

Choice	ভালো দিক	খরচ
Paragraph ধরে chunk	নির্দিষ্ট result যেটা একটা section এ link দেয়	কয়েক paragraph এ ছড়ানো একটা তথ্যের জন্য দুইটা result লাগতে পারে
.npy file এ cache করা vector	সহজ, সাথে সাথে restart	আবার না বানিয়ে update করা যায় না; vector DB (pgvector, Qdrant) যোগ/মোছা/ছাঁকা সামলায়
Brute-force vector search	একদম ঠিক, 800 vector এ ঠিক আছে	লাখ লাখে ANN index (HNSW) এ যাও
Python loop এ BM25	পড়া যায়	প্রতি query তে O(document); inverted index বা search engine নাও
http.server	কোনো dependency নেই	auth, rate limiting বা metric নেই; production এ gateway এর পেছনে FastAPI/Go
Label দেওয়া 12 টা প্রশ্ন	বড় পার্থক্য ধরে	ছোট পার্থক্যের জন্য খুব কম (Part 1 এর statistics); Part 8 ঠিকঠাক একটা set বানায়

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

1. Hybrid এর প্রথম 20 টা result এ reranking lesson এর reranker যোগ করো, দেখো hit@1 10/12 এর উপরে ওঠে কিনা।
2. একটা lang filter যোগ করো (/search?q=...&lang=bn), search এর ভেতরে, ranking এর আগে লাগানো, পরে না।
3. BM25 loop কে একটা inverted index (শব্দ → document id এর list) দিয়ে বদলাও আর 10 গুণ document এ সময় মাপো।
4. 10 টা Banglish প্রশ্ন লেখো ("token keno beshi lage bangla te") আর প্রতিটা mode মাপো। কোনটা আগে ভাঙে?