SUBMISSION: Case 4 — Enterprise Hybrid RAG with Cross-Encoder Reranking

Industry Context: A technical documentation assistant for a Kubernetes cloud platform must answer complex engineer inquiries across 500 internal markdown guides. Pure dense semantic vector search fails on exact CLI flags (--kubeconfig, --dry-run=client) and error codes, while pure BM25 keyword search misses conceptual queries. Build an enterprise hybrid search pipeline.

---

Deliverables

1. ingest_pipeline.py: Document loader and recursive chunker (chunk size 512, overlap 64) embedding documents into Chroma/pgvector while building an inverted BM25 index.
2. hybrid_retriever.py: Hybrid search engine combining dense vector cosine similarity and BM25 scores via Reciprocal Rank Fusion (RRF, k=60), followed by a cross-encoder reranker to yield the top-5 passages.
3. rag_generator.py: Prompt engine generating answers with grounded source citations ([Doc X, Section Y]) and an explicit "I don't have enough information" fallback when relevance falls below threshold.
4. rag_eval.py: Evaluation script measuring context retrieval recall and answer faithfulness.

---

Starter Code Scaffold

from typing import List, Dict, Any

def reciprocal_rank_fusion(dense_results: List[str], bm25_results: List[str], k: int = 60) -> List[tuple]:
    """Combine dense and sparse ranked lists using Reciprocal Rank Fusion (RRF)."""
    rrf_score = {}
    for rank, doc_id in enumerate(dense_results):
        rrf_score[doc_id] = rrf_score.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
    for rank, doc_id in enumerate(bm25_results):
        rrf_score[doc_id] = rrf_score.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
    return sorted(rrf_score.items(), key=lambda x: x[1], reverse=True)

---

100-Point Evaluation Rubric

• Criterion 1: Ingestion & Chunking Strategy (25 pts) — Document chunking with clean sentence boundary preservation, metadata indexing, dimension consistency.
• Criterion 2: Hybrid Retrieval & RRF Fusion (25 pts) — Dual retrieval pipeline marrying dense vector embeddings and BM25 sparse index via Reciprocal Rank Fusion.
• Criterion 3: Cross-Encoder Reranking (25 pts) — Second-stage scoring with cross-encoder reranker to compress top 20 candidates into top 5 high-density chunks.
• Criterion 4: Faithfulness, Citations & Fallback (25 pts) — Traceable source citations ([Doc, Section]) and strict "I don't know" fallback on out-of-domain queries.

---

Implementation Guide & Case Steps

"Build a chatbot that answers questions from a corpus of 500 support documents. Production-ready: hybrid search, reranking, citations, 'I don't know' fallback, evaluation."

---

Step 1 — Architecture overview

User question
    ↓
[Embed query]
    ↓
[Hybrid search (dense + BM25)] → top 20 candidates
    ↓
[Cross-encoder rerank] → top 5
    ↓
[LLM with retrieved context + citations + "I don't know" fallback]
    ↓
Answer with citations

5 components. Let's build each.

Step 2 — Document ingestion

from langchain.text_splitter import RecursiveCharacterTextSplitter

def ingest_documents(docs):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

    chunks = []
    for doc in docs:
        for chunk_text in splitter.split_text(doc.text):
            chunks.append({
                "text": chunk_text,
                "metadata": {
                    "source_id": doc.id,
                    "source_title": doc.title,
                    "source_url": doc.url,
                    "ingested_at": datetime.utcnow().isoformat(),
                }
            })
    return chunks

# Embed and index
from openai import OpenAI
client = OpenAI()

def embed_chunks(chunks):
    texts = [c["text"] for c in chunks]
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=texts,
    )
    for chunk, embedding_data in zip(chunks, response.data):
        chunk["embedding"] = embedding_data.embedding
    return chunks

# Store in Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

qdrant = QdrantClient(host="localhost", port=6333)
qdrant.recreate_collection(
    collection_name="support_docs",
    vectors_config=VectorParams(size=3072, distance=Distance.COSINE),
)

def index_chunks(chunks):
    points = [
        PointStruct(
            id=i,
            vector=chunk["embedding"],
            payload={"text": chunk["text"], **chunk["metadata"]}
        )
        for i, chunk in enumerate(chunks)
    ]
    qdrant.upsert(collection_name="support_docs", points=points)

Step 3 — Hybrid search

from rank_bm25 import BM25Okapi
import re

# Build BM25 index
tokenized_corpus = [re.findall(r"\w+", c["text"].lower()) for c in chunks]
bm25 = BM25Okapi(tokenized_corpus)

def hybrid_search(query, top_k=20):
    # Dense
    query_emb = client.embeddings.create(
        model="text-embedding-3-large",
        input=[query]
    ).data[0].embedding
    dense_results = qdrant.search(
        collection_name="support_docs",
        query_vector=query_emb,
        limit=top_k * 2,
    )

    # Sparse (BM25)
    tokens = re.findall(r"\w+", query.lower())
    bm25_scores = bm25.get_scores(tokens)
    sparse_top_idx = bm25_scores.argsort()[::-1][:top_k * 2]

    # Reciprocal rank fusion
    rrf_scores = {}
    k = 60  # standard RRF constant
    for rank, hit in enumerate(dense_results):
        rrf_scores[hit.id] = rrf_scores.get(hit.id, 0) + 1 / (k + rank + 1)
    for rank, idx in enumerate(sparse_top_idx):
        rrf_scores[int(idx)] = rrf_scores.get(int(idx), 0) + 1 / (k + rank + 1)

    # Top by combined score
    top_ids = sorted(rrf_scores.keys(), key=lambda i: -rrf_scores[i])[:top_k]
    return [chunks[i] for i in top_ids]

Step 4 — Reranking

import cohere
co = cohere.Client(api_key=os.environ["COHERE_API_KEY"])

def rerank(query, candidates, top_k=5):
    docs = [c["text"] for c in candidates]
    response = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=docs,
        top_n=top_k,
    )
    return [candidates[r.index] for r in response.results]

Step 5 — Generation with citations

import anthropic

claude = anthropic.Anthropic()

def answer_with_rag(user_question):
    candidates = hybrid_search(user_question, top_k=20)
    top_chunks = rerank(user_question, candidates, top_k=5)

    context_block = "\n\n".join([
        f"[Source {i+1}: {c['source_title']}]\n{c['text']}"
        for i, c in enumerate(top_chunks)
    ])

    response = claude.messages.create(
        model="claude-sonnet-4-6",
        system=[{
            "type": "text",
            "text": '''You are a support assistant. Answer the user's question using ONLY the provided context.

Rules:
- Cite sources for each claim using [Source X] format.
- If the context doesn't contain the answer, say "I don't have information about that in our docs. Could you rephrase or contact support?"
- Keep responses under 200 words.
- Don't make up information not in the context.''',
            "cache_control": {"type": "ephemeral"}
        }],
        messages=[{
            "role": "user",
            "content": f"Context:\n{context_block}\n\nQuestion: {user_question}"
        }],
        max_tokens=400,
    )
    return {
        "answer": response.content[0].text,
        "sources": [{"title": c["source_title"], "url": c["source_url"]} for c in top_chunks],
    }

Step 6 — Eval

eval_set = [
    {
        "question": "How do I return a defective product?",
        "expected_in_answer": ["return", "30 days", "shipping label"],
    },
    {
        "question": "What's your refund policy for digital products?",
        "expected_in_answer": ["digital", "no refund", "exception"],
    },
    {
        "question": "Tell me about your company's history.",
        "expected_in_answer": ["I don't have information"],  # Should refuse
    },
    # ...30-50 cases
]

def evaluate(rag_fn, eval_set):
    results = []
    for case in eval_set:
        output = rag_fn(case["question"])["answer"]
        passed = all(phrase.lower() in output.lower() for phrase in case["expected_in_answer"])
        results.append({"question": case["question"], "passed": passed, "output": output})
    pass_rate = sum(r["passed"] for r in results) / len(results)
    return {"pass_rate": pass_rate, "results": results}

For open-ended quality, add LLM-as-judge evaluation:

def judge_answer(question, answer, expected_qualities):
    prompt = f'''Evaluate this answer to a customer question on these criteria:
{expected_qualities}

Question: {question}
Answer: {answer}

Output JSON: {{"correctness": 1-5, "groundedness": 1-5, "tone": 1-5, "concerns": "string or null"}}'''
    response = claude.messages.create(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=200,
    )
    return json.loads(response.content[0].text)

Step 7 — Production monitoring

Log every request:

def answer_with_rag_logged(user_question, user_id):
    start = time.time()
    result = answer_with_rag(user_question)
    latency_ms = (time.time() - start) * 1000

    log_to_warehouse({
        "request_id": str(uuid.uuid4()),
        "user_id": user_id,
        "question": user_question,
        "answer": result["answer"],
        "sources": result["sources"],
        "latency_ms": latency_ms,
        "served_at": datetime.utcnow().isoformat(),
        "model": "claude-sonnet-4-6",
    })

    return result

Monitor:

• Latency (p50, p95, p99).
• "I don't know" rate (too high = poor coverage; too low = possible hallucination).
• Cost per request.
• Sample 5-10 random requests/day for manual quality review.

Self-check MCQs