Build a retrieval service that layers three production techniques — dense vector search, sparse keyword search fused with Reciprocal Rank Fusion (RRF), and cross-encoder reranking — and prove with a measured gains table that each layer earns its place.
You will load the same corpus into two vector stores (Chroma and pgvector-in-Docker) so you can compare the embedded-library path against the Postgres path, then evaluate four retrieval configurations against a 20-query golden set. The numbers you produce reproduce the layered-gains pattern from Anthropic's contextual-retrieval research: dense-only misses exact part numbers, BM25 misses paraphrases, hybrid recovers both, and the reranker fixes the ordering inside the candidate set. Employers name "hybrid search" and "reranking" verbatim in data-engineering postings — this lab is the artifact that proves you have done it, not just read about it.
~150 minutes hands-on · Advanced tier (maps to module M6).
Everything runs on your laptop for $0 — no cloud account, no API keys. You need Python 3.11+, Docker (for the pgvector half; the lab still works without it), and roughly 500 MB of disk for models.
mkdir -p ~/labs/de-lab-09 && cd ~/labs/de-lab-09
python3 -m venv .venv && source .venv/bin/activate
pip install sentence-transformers chromadb rank_bm25 numpy "psycopg[binary]"
The first run downloads two small models from Hugging Face (~90 MB each): sentence-transformers/all-MiniLM-L6-v2 (bi-encoder, 384 dims) and cross-encoder/ms-marco-MiniLM-L-6-v2 (reranker). After that, everything is offline.
Create make_corpus.py — a deterministic generator for a 120-article industrial-maintenance knowledge base plus a 20-query golden set. Every learner gets byte-identical data, so your numbers should match the expected table within a point or two.
"""Generate the plantops corpus: 120 maintenance-KB articles + 20 golden queries.
Deterministic (seeded) so every learner gets identical data."""
import json, random, pathlib
random.seed(42)
OUT = pathlib.Path("corpus"); OUT.mkdir(exist_ok=True)
SYSTEMS = [
("hydraulic press", "HYD"), ("conveyor belt", "CNV"), ("industrial chiller", "CHL"),
("packaging robot", "PKR"), ("steam boiler", "BLR"), ("air compressor", "CMP"),
("CNC lathe", "CNC"), ("paint booth ventilator", "PBV"), ("forklift charger", "FLC"),
("wastewater pump", "WWP"),
]
SYMPTOMS = [
"overheats after thirty minutes of continuous operation",
"emits a high-pitched whine under load",
"trips the breaker during startup",
"shows pressure drop below the rated threshold",
"vibrates excessively at high duty cycle",
"leaks fluid around the main seal",
"fails to reach operating temperature",
"displays error code and halts mid-cycle",
"draws abnormally high current at idle",
"stalls intermittently when ambient humidity is high",
"produces inconsistent output past the calibration window",
"makes a grinding noise during the reverse stroke",
]
FIXES = [
"replace the worn bearing assembly and repack with high-temperature grease",
"flush the coolant loop and refill with the approved glycol mixture",
"recalibrate the pressure transducer against the master gauge",
"tighten the drive coupling to the torque spec and recheck alignment",
"swap the failing contactor and inspect the control relay wiring",
"clean the intake filter and verify unobstructed airflow",
"update the controller firmware and restore factory PID defaults",
"replace the degraded O-ring kit and pressure-test the manifold",
]
BOILERPLATE = (
"Safety notice: before servicing any equipment, isolate the power supply, apply "
"lock-out tag-out, wear the plant-mandated PPE, and confirm the work permit is "
"signed by the shift supervisor. All maintenance actions must be recorded in the "
"computerized maintenance management system on the same shift. Units still under "
"warranty must not be opened without written approval from the vendor. "
)
docs, queries = [], []
part_codes = []
for i in range(120):
system, prefix = SYSTEMS[i % len(SYSTEMS)]
part_codes.append(f"{prefix}-{random.randint(1000, 9999)}-{random.choice('ABCDEF')}")
for i in range(120):
system, prefix = SYSTEMS[i % len(SYSTEMS)]
symptom = SYMPTOMS[i % len(SYMPTOMS)]
fix = FIXES[i % len(FIXES)]
part_code = part_codes[i]
doc_id = f"kb-{i:03d}"
# cross-references: two OTHER part codes appear in every doc (realistic noise for BM25)
xref = random.sample([p for p in part_codes if p != part_code], 2)
title = f"Troubleshooting: {system} {symptom.split()[0]} issue (part {part_code})"
body = (
BOILERPLATE
+ f"Applies to {system} units, service part {part_code}. "
f"Reported symptom: unit {symptom}. "
f"Diagnosis: inspect the {system} drive train and log readings for ten minutes. "
f"Resolution: {fix}. "
f"After the repair, run the standard acceptance test and record the result "
f"in the maintenance ledger. "
f"See also related service bulletins for parts {xref[0]} and {xref[1]}. "
f"Escalate to level-2 support if the {system} still {symptom} after the fix."
)
docs.append({"id": doc_id, "title": title, "part_code": part_code, "text": body})
# Golden queries: 10 semantic paraphrases (dense-friendly), 10 exact part-code lookups (sparse-friendly)
PARAPHRASE = {
"overheats after thirty minutes of continuous operation": "machine gets too hot after running half an hour",
"emits a high-pitched whine under load": "makes a high-pitched whining noise when under heavy load",
"trips the breaker during startup": "electrical panel cuts power the moment we switch it on",
"shows pressure drop below the rated threshold": "PSI reading keeps falling under the minimum spec",
"vibrates excessively at high duty cycle": "unit shakes badly when we run it flat out",
"leaks fluid around the main seal": "oil seeping out near the central gasket",
"fails to reach operating temperature": "never warms up to the target temp",
"displays error code and halts mid-cycle": "throws a fault message and stops halfway through",
"draws abnormally high current at idle": "amperage way above normal even when doing nothing",
"stalls intermittently when ambient humidity is high": "keeps stalling on and off when humidity is high",
}
# Each (system, symptom) combo appears twice in the corpus (120 docs, 60 combos), so
# for paraphrase queries BOTH twins are acceptable answers; part-code lookups have exactly one.
by_combo = {}
for i, d in enumerate(docs):
by_combo.setdefault((i % len(SYSTEMS), i % len(SYMPTOMS)), []).append(d["id"])
sem_ids = random.sample(range(120), 60)
count = 0
used = set()
for i in sem_ids:
d = docs[i]
system = SYSTEMS[i % len(SYSTEMS)][0]
symptom = SYMPTOMS[i % len(SYMPTOMS)]
if symptom in PARAPHRASE and count < 10:
rel = by_combo[(i % len(SYSTEMS), i % len(SYMPTOMS))]
queries.append({"qid": f"q-{count:02d}", "query": f"{system} {PARAPHRASE[symptom]}",
"relevant_ids": rel, "kind": "semantic"})
used.update(rel)
count += 1
code_ids = random.sample([i for i in range(120) if docs[i]["id"] not in used], 10)
for j, i in enumerate(code_ids):
d = docs[i]
queries.append({"qid": f"q-{10+j:02d}", "query": f"repair steps for part {d['part_code']}",
"relevant_ids": [d["id"]], "kind": "keyword"})
(OUT / "docs.jsonl").write_text("\n".join(json.dumps(d) for d in docs))
(OUT / "golden.jsonl").write_text("\n".join(json.dumps(q) for q in queries))
print(f"wrote {len(docs)} docs, {len(queries)} golden queries")
Run it: python make_corpus.py — you should see wrote 120 docs, 20 golden queries.
The corpus is deliberately adversarial in two realistic ways: every document shares a 70-word safety boilerplate (dilutes both embeddings and BM25 statistics, exactly like legal headers in enterprise documents), and every document cross-references two other part codes (so a part-code query gets lexical hits in wrong documents). Real corpora do this to you constantly.
Build the retrieval library. Create retrieval.py with the embedding index (Chroma), the BM25 index, the RRF fusion function, and the cross-encoder reranker. This is the core of the lab — read it, don't just paste it.
import json, pathlib
import chromadb
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer, CrossEncoder
docs = [json.loads(l) for l in pathlib.Path("corpus/docs.jsonl").read_text().splitlines()]
golden = [json.loads(l) for l in pathlib.Path("corpus/golden.jsonl").read_text().splitlines()]
texts = [d["title"] + "\n" + d["text"] for d in docs]
ids = [d["id"] for d in docs]
text_of = dict(zip(ids, texts))
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
emb = model.encode(texts, normalize_embeddings=True, show_progress_bar=False)
client = chromadb.Client()
col = client.create_collection("plantops", metadata={"hnsw:space": "cosine"})
col.add(ids=ids, embeddings=emb.tolist(), documents=texts)
tokenized = [t.lower().replace("(", " ").replace(")", " ").replace(",", " ")
.replace(".", " ").split() for t in texts]
bm25 = BM25Okapi(tokenized)
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def dense_search(query, k=20):
qv = model.encode([query], normalize_embeddings=True)
return col.query(query_embeddings=qv.tolist(), n_results=k)["ids"][0]
def sparse_search(query, k=20):
scores = bm25.get_scores(query.lower().split())
order = sorted(range(len(scores)), key=lambda i: -scores[i])[:k]
return [ids[i] for i in order]
def rrf(list_a, list_b, k=60, top=20):
scores = {}
for rank, doc in enumerate(list_a):
scores[doc] = scores.get(doc, 0) + 1.0 / (k + rank + 1)
for rank, doc in enumerate(list_b):
scores[doc] = scores.get(doc, 0) + 1.0 / (k + rank + 1)
return [d for d, _ in sorted(scores.items(), key=lambda x: -x[1])][:top]
def rerank(query, cand_ids, top=5):
pairs = [(query, text_of[c]) for c in cand_ids]
scores = reranker.predict(pairs, show_progress_bar=False)
order = sorted(range(len(cand_ids)), key=lambda i: -scores[i])
return [cand_ids[i] for i in order[:top]]
Understand RRF before you run it. RRF ignores raw scores entirely — dense cosine similarities and BM25 scores live on incompatible scales, so fusing raw scores is a classic bug. RRF only uses each document's rank in each list: 1/(60+rank). A document ranked 3rd by dense and 5th by BM25 beats one ranked 1st in a single list only. The constant 60 damps the advantage of a single #1 rank. Write one sentence in your notes explaining why score normalization (min-max scaling both lists) would be fragile here — then check your answer against the boilerplate: BM25 scores on this corpus cluster tightly because every document shares 70 identical words.
Build the evaluation harness. Create evaluate.py. It measures recall@1, recall@5, MRR, and latency (median + p95) for each configuration against the golden set. A query counts as a hit if any acceptable relevant_ids entry is retrieved.
import time, statistics
from retrieval import *
def evaluate(fn, label):
hits5 = hits1 = 0; rr = []; latencies = []
for q in golden:
rel = set(q["relevant_ids"])
t0 = time.perf_counter()
top5 = fn(q["query"])[:5]
latencies.append((time.perf_counter() - t0) * 1000)
ranks = [i for i, d in enumerate(top5) if d in rel]
if ranks:
hits5 += 1; rr.append(1 / (ranks[0] + 1))
if ranks[0] == 0: hits1 += 1
else:
rr.append(0)
lat = sorted(latencies); p95 = lat[int(0.95 * len(lat)) - 1]
print(f"{label:16s} r@1={hits1}/20 r@5={hits5}/20 MRR={sum(rr)/len(rr):.2f} "
f"med={statistics.median(latencies):.0f}ms p95={p95:.0f}ms")
evaluate(lambda q: dense_search(q, 5), "dense-only")
evaluate(lambda q: sparse_search(q, 5), "bm25-only")
evaluate(lambda q: rrf(dense_search(q), sparse_search(q), top=5), "hybrid-rrf")
evaluate(lambda q: rerank(q, rrf(dense_search(q), sparse_search(q), top=20)), "hybrid+rerank")
Run it and record the gains table. python evaluate.py. Expected output (your latencies depend on hardware; retrieval numbers should match within one query):
| Configuration | recall@1 | recall@5 | MRR | median latency | p95 latency |
|---|---|---|---|---|---|
| dense-only | 13/20 | 15/20 | 0.68 | ~6 ms | ~13 ms |
| bm25-only | 11/20 | 19/20 | 0.69 | <1 ms | <1 ms |
| hybrid (RRF) | 15/20 | 19/20 | 0.83 | ~5 ms | ~6 ms |
| hybrid + rerank | 19/20 | 20/20 | 0.96 | ~90 ms | ~100 ms |
Read the table like a production engineer: top-5 failure rate drops 25% → 5% → 0% as layers stack; top-1 failure drops 35% → 25% → 5%. One query sits in hybrid's top-20 but not its top-5 — the reranker rescues it, which is precisely the rule of thumb: if the right passage is in the top 20–50 but not the top 5, add a reranker before you swap embedding models.
Diagnose the failure modes by kind. Modify the harness to print misses split by kind. You should see dense-only failing mostly on keyword queries (part codes are weak signals in embedding space, and the cross-reference noise hurts) and bm25-only failing on semantic paraphrases ("keeps stalling on and off when humidity is high" shares almost no tokens with "stalls intermittently when ambient humidity is high"). Write two sentences in your notes naming which layer covers which failure class.
Load pgvector and run the same dense query path. Start Postgres with pgvector (this step needs Docker; skip to step 7 if you cannot run it, and note the skip in your artifact):
docker run -d --name lab09-pg -e POSTGRES_PASSWORD=lab -p 5433:5432 pgvector/pgvector:pg16
Create load_pgvector.py:
import json, pathlib, psycopg
from sentence_transformers import SentenceTransformer
docs = [json.loads(l) for l in pathlib.Path("corpus/docs.jsonl").read_text().splitlines()]
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
emb = model.encode([d["title"] + "\n" + d["text"] for d in docs], normalize_embeddings=True)
with psycopg.connect("host=localhost port=5433 user=postgres password=lab") as conn:
conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
conn.execute("DROP TABLE IF EXISTS kb")
conn.execute("""CREATE TABLE kb (id text PRIMARY KEY, title text, body text,
embedding vector(384))""")
with conn.cursor() as cur:
for d, e in zip(docs, emb):
cur.execute("INSERT INTO kb VALUES (%s, %s, %s, %s)",
(d["id"], d["title"], d["text"], str(e.tolist())))
conn.execute("CREATE INDEX ON kb USING hnsw (embedding vector_cosine_ops)")
conn.commit()
q = model.encode(["conveyor belt oil seeping out near the central gasket"],
normalize_embeddings=True)[0]
with psycopg.connect("host=localhost port=5433 user=postgres password=lab") as conn:
rows = conn.execute("""SELECT id, 1 - (embedding <=> %s::vector) AS sim
FROM kb ORDER BY embedding <=> %s::vector LIMIT 5""",
(str(q.tolist()), str(q.tolist()))).fetchall()
for r in rows: print(r)
Run the golden-set dense evaluation against pgvector too and add a row to your latency table. On 120 rows the numbers are close; the decision framework you should write down anyway: pgvector wins when you already run Postgres and stay under ~10M vectors (one fewer system, SQL joins with metadata, row-level security for ACL filtering); a dedicated store wins on managed scaling, built-in hybrid search, and multi-tenancy features.
Add a cost column. Everything here is self-hosted, so cost-per-query is compute time. Estimate it: at p95 ≈ 100 ms of single-core CPU for hybrid+rerank and a $0.04/hour cloud vCPU, one query costs about $0.000001 — effectively free. Now contrast with a managed reranker API at roughly $1–2 per 1,000 queries: at 1M queries/month that is $1,000–2,000/month versus one small CPU node. Put both numbers in your gains table; the trade is latency and ops burden versus invoice.
Hybrid is the production default because dense and sparse fail on disjoint query classes; the reranker is a precision layer over a recall layer. Always measure each addition — a layer that does not move recall@k or MRR on your golden set is latency you are paying for nothing.
python make_corpus.py reports exactly 120 docs and 20 queries.python evaluate.py shows monotonically improving MRR across dense → hybrid → hybrid+rerank, with hybrid+rerank at recall@5 = 20/20.keyword queries and BM25 failures concentrated in semantic queries.kb- id matching a conveyor belt seal-leak article), and \d kb in psql shows an hnsw index.A de-lab-09/ folder containing make_corpus.py, retrieval.py, evaluate.py, load_pgvector.py, and gains-table.md — the measured table (recall@1, recall@5, MRR, median/p95 latency, cost-per-query note) for all four configurations plus a pgvector-vs-Chroma latency comparison and a 5-line store-selection recommendation. Self-grading rubric (pass = 8/10):
~/.cache/huggingface.n_results: you created the collection twice in one process — restart Python or client.delete_collection first.psycopg.OperationalError: the container needs ~3 seconds to accept connections after docker run; also confirm port 5433 is free.random.seed(42) ran before any random call and that you did not edit corpus text.rrf(..., top=20) argument.Every component in this lab is local open source: embeddings and the cross-encoder run in-process via sentence-transformers, BM25 is pure Python, and both stores (Chroma, pgvector) run on your machine — no document, query, or embedding ever leaves the laptop. That makes this exact stack the reference architecture for client engagements where the MSA prohibits sending data to external AI APIs: swap the corpus for the client's documents inside their VPC, keep the same eval harness, and present the gains table as the evidence for every architecture decision. If the client later approves a managed reranker or hosted embeddings, you already have the golden set to prove whether the upgrade is worth the data-residency conversation.