Post 16 introduced Atlas Vector Search. This post builds an end-to-end production-shaped RAG (Retrieval-Augmented Generation) app on top of it: chunking + embedding the corpus, indexing into Atlas, querying with hybrid (lexical + semantic) retrieval, optional reranking, prompt assembly, streaming the LLM response, observability + cost tracking, and the production guardrails (rate limits, citation, evaluation). The framework is LangChain (Python); the embedder is OpenAI text-embedding-3-large; the LLM is GPT-4o (any equivalent works).

Series map: 19 Beanie+FastAPI → 20 Outbox → 21 CQRS + ES → 22 RAG Production Build (this post) → 23 SQL→Mongo migration → … → 29 K8s + mongosync + cost.

Learning Outcome

Chunk + embed + index a document corpus into Atlas Vector Search.
Retrieve with vector + lexical + filter (hybrid).
Rerank with a cross-encoder when needed.
Assemble prompts with citations.
Stream the LLM response back to clients.
Track cost + latency + relevance per query.
Evaluate retrieval quality with a small offline eval set.

Chapter 1 — Why MongoDB for RAG

The naïve RAG stack is:

1
2 documents → embedder → vector DB → retrieval → LLM

Mongo + Atlas Vector Search replaces a separate vector DB:

Documents + their embeddings + their metadata live in one collection in one database.
Hybrid search (lexical + vector) is one pipeline (Post 16 Ch 2).
Filter on metadata (tenant, recency, category) alongside vector search.
Same backup / monitoring / connection-pool patterns as the rest of your Mongo workload.
No data sync between document store + vector store.

Trade-off vs Pinecone / Weaviate / pgvector: Atlas Vector Search is HNSW only (no IVF or hybrid index types in 2026); cluster cost scales with index size. For < a few hundred million vectors, Atlas wins on operational simplicity.

Chapter 2 — Ingest pipeline — chunk + embed + insert

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30 # ingest.py
from typing import Iterable
from openai import OpenAI
from pymongo import MongoClient
from langchain_text_splitters import RecursiveCharacterTextSplitter
 
oa = OpenAI()
db = MongoClient(os.environ["MONGO_URI"])["docs"]
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
 
def embed(texts: list[str]) -> list[list[float]]:
    resp = oa.embeddings.create(model="text-embedding-3-large", input=texts)
    return [d.embedding for d in resp.data]
 
def ingest(source_id: str, text: str, metadata: dict):
    chunks = splitter.split_text(text)
    vectors = embed(chunks)
    docs = [
        {
            "source_id":   source_id,
            "chunk_index": i,
            "text":        chunk,
            "embedding":   vec,
            "metadata":    metadata,
            "indexed_at":  datetime.utcnow(),
        }
        for i, (chunk, vec) in enumerate(zip(chunks, vectors))
    ]
    db.chunks.insert_many(docs)

Chunking guidelines:

800-1200 tokens per chunk for general documents. Smaller for FAQs (200-400); larger for narrative text (1500-2000).
Overlap of 10-15 % to preserve context across boundaries.
For code / structured docs, split at structural boundaries (functions, sections).

Embedding batch size: OpenAI accepts up to ~2048 inputs per call; tune to your rate limit. Batch dramatically reduces cost + latency.

Index the embedding (Atlas UI or API)

1
2
3
4
5
6
7
8
9
10
11
12
13
14 // once, on the chunks collection
db.chunks.createSearchIndex({
  name: "chunks_vec",
  type: "vectorSearch",
  definition: {
    fields: [
      { type: "vector", path: "embedding", numDimensions: 3072, similarity: "cosine" },
      { type: "filter", path: "metadata.tenant" },
      { type: "filter", path: "metadata.category" },
      { type: "filter", path: "indexed_at" }
    ]
  }
})

Also create an Atlas Search lexical index on text for hybrid (Chapter 4).

Chapter 3 — Retrieval — vector

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22 def retrieve_vector(query: str, k: int = 10, tenant: str | None = None) -> list[dict]:
    q_vec = embed([query])[0]
    pipeline = [
        {
            "$vectorSearch": {
                "index": "chunks_vec",
                "path": "embedding",
                "queryVector": q_vec,
                "numCandidates": 200,   # search width
                "limit": k,
                **({"filter": {"metadata.tenant": tenant}} if tenant else {}),
            }
        },
        {
            "$project": {
                "_id": 1, "text": 1, "metadata": 1,
                "score": {"$meta": "vectorSearchScore"},
            }
        },
    ]
    return list(db.chunks.aggregate(pipeline))

Knobs:

numCandidates: HNSW search width. Higher = better recall + slower. Start 200; raise to 500-1000 if recall is bad.
limit: top-k to return.
filter: pre-filters narrow the search space. Big wins for multi-tenant or category-scoped queries.

Chapter 4 — Hybrid retrieval (lexical + vector)

Vector search misses exact-keyword hits; lexical misses semantic paraphrases. Combine for best recall.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49 def retrieve_hybrid(query: str, k: int = 10, tenant: str | None = None) -> list[dict]:
    q_vec = embed([query])[0]
 
    pipeline = [
        {
            "$vectorSearch": {
                "index": "chunks_vec",
                "path": "embedding",
                "queryVector": q_vec,
                "numCandidates": 200,
                "limit": 20,
                **({"filter": {"metadata.tenant": tenant}} if tenant else {}),
            }
        },
        {"$addFields": {"vec_score": {"$meta": "vectorSearchScore"}, "lex_score": 0}},
        {"$unionWith": {
            "coll": "chunks",
            "pipeline": [
                {
                    "$search": {
                        "index": "chunks_text",
                        "compound": {
                            "must": [{"text": {"query": query, "path": "text"}}],
                            **({"filter": [{"text": {"query": tenant, "path": "metadata.tenant"}}]} if tenant else {}),
                        }
                    }
                },
                {"$limit": 20},
                {"$addFields": {"lex_score": {"$meta": "searchScore"}, "vec_score": 0}},
            ]
        }},
        # Reciprocal Rank Fusion
        {"$group": {
            "_id": "$_id",
            "text": {"$first": "$text"},
            "metadata": {"$first": "$metadata"},
            "vec_score": {"$max": "$vec_score"},
            "lex_score": {"$max": "$lex_score"},
        }},
        # weighted combination — tune per workload
        {"$addFields": {"combined": {"$add": [
            {"$multiply": ["$vec_score", 0.6]},
            {"$multiply": ["$lex_score", 0.4]}
        ]}}},
        {"$sort": {"combined": -1}},
        {"$limit": k},
    ]
    return list(db.chunks.aggregate(pipeline))

The weights (0.6 / 0.4) are workload-dependent — tune with an eval set (Chapter 8).

A better scoring approach is Reciprocal Rank Fusion (RRF) — combine ranks, not raw scores:

1
2 rrf_score = sum_over_rankings(1 / (k + rank))

Atlas + the LangChain integration provide hybrid retriever helpers that do this for you.

Chapter 5 — Optional reranking

Hybrid retrieval typically returns 20-50 candidates. A cross-encoder reranker scores each candidate against the query more accurately (slow per pair, fast on small batches).

1
2
3
4
5
6
7
8
9 # pseudo — using cohere reranker
from cohere import Client as Cohere
co = Cohere()
 
def rerank(query: str, candidates: list[dict], top_n: int = 5) -> list[dict]:
    texts = [c["text"] for c in candidates]
    r = co.rerank(model="rerank-english-v3.0", query=query, documents=texts, top_n=top_n)
    return [candidates[ri.index] for ri in r.results]

When to add reranking:

Quality matters more than 50-100 ms latency.
You're already calling Cohere / OpenAI / Jina for embeddings.
Hybrid alone isn't returning the right doc top-1.

Skip when latency matters more than relevance lift.

Chapter 6 — Prompt assembly with citations

1
2
3
4
5
6
7
8
9
10
11
12
13 def assemble_prompt(query: str, chunks: list[dict]) -> list[dict]:
    context = "\n\n".join(
        f"[{i+1}] {c['text']} (source: {c['metadata'].get('source_id', '?')})"
        for i, c in enumerate(chunks)
    )
    system = (
        "You answer questions using ONLY the provided context. "
        "Cite sources by their bracketed numbers, e.g. [1][3]. "
        "If the context doesn't contain the answer, say so."
    )
    user = f"Context:\n{context}\n\nQuestion: {query}"
    return [{"role": "system", "content": system}, {"role": "user", "content": user}]

Key rules baked into the prompt:

Use only context.
Cite numerically (cheap to render in UI).
Refuse when context is insufficient (mitigates hallucination).

The UI then maps [1][3] to the actual source_id URLs from chunks.

Chapter 7 — LLM call + streaming

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 def answer_stream(query: str, tenant: str | None = None):
    candidates = retrieve_hybrid(query, k=20, tenant=tenant)
    top = rerank(query, candidates, top_n=5)
    messages = assemble_prompt(query, top)
 
    stream = oa.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=0.2,
        stream=True,
    )
    for delta in stream:
        token = delta.choices[0].delta.content
        if token:
            yield token
 
    # after streaming finishes, return citations
    yield {"citations": [{"i": i+1, **c["metadata"], "preview": c["text"][:200]} for i, c in enumerate(top)]}

In FastAPI:

1
2
3
4
5
6 from fastapi.responses import StreamingResponse
 
@app.post("/ask")
async def ask(q: AskBody):
    return StreamingResponse(answer_stream(q.query, q.tenant), media_type="text/event-stream")

Stream as Server-Sent Events (SSE) for a chatty UI feel.

The RAG query path as a funnel from 200 vector candidates down to 5 reranked chunks and a cited answer

Chapter 8 — Evaluation

A small eval set is the difference between "I think it's good" and "I know it's improving".

Build 50-200 queries with the expected source(s). Run retrieval; measure:

Recall@k — was the right doc in the top-k?
MRR — mean reciprocal rank of the first correct doc.
End-to-end accuracy — manually + via LLM-judge on a sample.

1
2
3
4
5
6
7
8
9
10
11
12
13 def evaluate(eval_set):
    metrics = {"recall@1": 0, "recall@5": 0, "mrr": 0}
    for ex in eval_set:
        results = retrieve_hybrid(ex["query"], k=10, tenant=ex.get("tenant"))
        gold = set(ex["expected_chunk_ids"])
        ranks = [i for i, r in enumerate(results) if r["_id"] in gold]
        if ranks:
            metrics["recall@1"]  += int(ranks[0] == 0)
            metrics["recall@5"]  += int(ranks[0] < 5)
            metrics["mrr"]       += 1 / (ranks[0] + 1)
    for k in metrics: metrics[k] /= len(eval_set)
    return metrics

Run on every PR that touches retrieval. CI gate on regression > X %.

LangChain ships an eval module; LangSmith provides hosted eval + traces; ragas is a popular OSS framework for RAG-specific metrics.

Chapter 9 — Observability + cost

Track per request:

Query string + tenant + filters.
Embedding tokens (input length).
Retrieved chunk IDs + scores (for offline replay).
Reranker latency.
LLM input + output tokens.
End-to-end latency.

Persist to a traces collection or use LangSmith / Helicone / Langfuse. Use this data to:

Identify slow queries.
Estimate per-query cost (embedding tokens × embedding price + LLM tokens × LLM price).
Find queries with low retrieval quality (no high-score matches).

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 @app.post("/ask")
async def ask(q: AskBody):
    t0 = time.time()
    candidates = retrieve_hybrid(q.query)
    t1 = time.time()
    top = rerank(q.query, candidates)
    answer, in_tokens, out_tokens = llm_call(top, q.query)
    t2 = time.time()
 
    await db.traces.insert_one({
        "query": q.query, "tenant": q.tenant,
        "retrieved_ids": [str(c["_id"]) for c in candidates],
        "selected_ids":  [str(c["_id"]) for c in top],
        "in_tokens": in_tokens, "out_tokens": out_tokens,
        "lat_retrieve_ms": int((t1-t0)*1000),
        "lat_llm_ms":      int((t2-t1)*1000),
        "at": datetime.utcnow(),
    })
    return {"answer": answer, "citations": [...]}

Atlas Charts on the traces collection gives quick dashboards.

Chapter 10 — Production guardrails

Guardrail	Why
Rate limit per user / tenant	LLM calls are slow + expensive
Max input length validation	refuses crafted huge inputs
PII / secret scrubbing in inputs + outputs	privacy + compliance
Citation enforcement	hallucination mitigation; user trust
Refusal on low-confidence retrieval	"I don't have the information" beats wrong answers
Logging with masking	debug without leaking
Eval-gated deploys	CI on RAG quality
A/B test retrieval strategies	data-driven tuning
Cost alerts per tenant	catch runaway usage
Streaming	better UX, lower perceived latency
Cache identical question answers per tenant (with TTL)	cost saver

Chapter 11 — Common pitfalls

Pitfall	Cause	Fix
Embeddings dimension mismatch	swapped model didn't rebuild index	rebuild index when changing embedder
Low recall with high `limit`	`numCandidates` too small	raise to 500-1000
Mixed tenants in results	no filter on metadata	add `filter` to `$vectorSearch`
LLM ignores context, hallucinates	weak system prompt	strengthen "USE ONLY context"; refusal instruction; eval
Latency spikes	reranker on every query	rerank top-20 only; or skip when not needed
Cost spike	one user spamming	rate-limit + cache identical Qs
Stale answers after document update	embeddings not updated	change-stream-driven re-embed (Post 12)
Long context blowing token limit	too many chunks in prompt	smaller `top_n`; smaller chunks
Reranker different lang than docs	language mismatch	use multilingual rerank model
Eval set drift	new feature areas not covered	grow eval set quarterly

Practice Exercises

Exercise 1 — Ingest a small corpus

Download ~50 markdown docs (your repo's docs / Wikipedia subset). Chunk + embed + insert. Build the vector index.

Exercise 2 — Vector-only retrieval

Endpoint that takes a query, runs $vectorSearch, returns top-5 chunks + scores. Test 10 questions; rate manually.

Exercise 3 — Hybrid retrieval

Add lexical ($search text) to the pipeline. Compare top-5 across vector / lexical / hybrid for the same 10 questions.

Exercise 4 — Reranker

Add Cohere rerank step after hybrid. Confirm top-1 quality improves on questions where exact keywords matter.

Exercise 5 — Streaming endpoint

Wire the streaming pipeline into FastAPI SSE. Render in a small HTML page that progressively displays tokens.

Exercise 6 — Eval harness

Build a 20-question eval set with known gold chunks. Run recall@1 / recall@5 / MRR. Tune numCandidates, chunk size, reranker on/off; track which combination wins.

Cheat-sheet

What	How
Embedder	OpenAI `text-embedding-3-large` (3072 dim) or `-small` (1536)
LLM	GPT-4o / Claude / Gemini — interchangeable
Chunk size	800-1200 tokens, 10-15 % overlap
Index	Atlas Vector Search HNSW + filter fields
Pre-filter	metadata.tenant / category / date inside `$vectorSearch.filter`
Hybrid	`$vectorSearch` + `$unionWith` lexical + RRF combination
Rerank	cross-encoder (Cohere, Jina, etc.) on top-20
Prompt	system="use only context"; cite with [n]
Stream	SSE / chunked HTTP
Eval	Recall@k / MRR / end-to-end accuracy
Trace	persist per request → `traces` collection
Cost	track input + output tokens; rate-limit; cache
Guardrails	refusal on low-confidence; citations; PII scrub
Re-embed on doc change	change stream → re-embed → upsert chunks

What's next

Post 23 covers migrating from SQL / Postgres to MongoDB — schema-mapping principles, data-load tooling (mongoimport, AWS DMS, custom ETL), dual-write strategies, and the staged cut-over playbook.

References

Atlas Vector Search docs
LangChain MongoDB integration
LlamaIndex MongoDB integration
LangChain × MongoDB partnership announcement
ragas — RAG evaluation framework
Anthropic + OpenAI prompt engineering guides — system prompt patterns.

Note: this post is in progress. Planned additions: full chat UI in Next.js; GraphRAG with $graphLookup; agentic loop (tool-use) example; cost dashboards.