n4nAI
Open console →
GlossaryHow-toRAG Architecture & Pipeline Design

How embedding models fit into a RAG pipeline

A step-by-step guide to integrating embedding models into a RAG pipeline, from model selection and chunking to indexing, retrieval, and evaluation.

n4n Team4 min read862 words

Audio narration

Coming soon — every post will get a voice note here.

Embedding models in RAG are the bridge between raw text and semantic search. They transform documents and queries into vectors so your retriever can find relevant context for the LLM. This guide walks through building that pipeline end to end — model selection, chunking, indexing, retrieval, and verification — with runnable code at each step.

Step 1: Choose an embedding model that matches your constraints

Start by defining your requirements: latency budget, dimensionality, language support, and whether you need instruction tuning. For most English-first RAG systems, text-embedding-3-small (1536 dims, ~62 MTEB) or text-embedding-3-large (3072 dims, ~65 MTEB) are solid defaults. If you need multilingual support or want to run locally, consider intfloat/multilingual-e5-large (1024 dims) or BAAI/bge-m3 (1024 dims, supports dense + sparse + ColBERT).

# Quick comparison helper
EMBEDDING_MODELS = {
    "openai-small": {"dim": 1536, "max_tokens": 8191, "provider": "openai"},
    "openai-large": {"dim": 3072, "max_tokens": 8191, "provider": "openai"},
    "e5-large": {"dim": 1024, "max_tokens": 512, "provider": "huggingface"},
    "bge-m3": {"dim": 1024, "max_tokens": 8192, "provider": "huggingface"},
}

Verify: Run a quick MTEB benchmark on your domain data if possible. At minimum, embed 100 representative chunks and confirm cosine similarity ranks known-relevant pairs higher than random pairs.

Step 2: Implement a chunking strategy before embedding

Embedding models have token limits. Chunking determines retrieval granularity and directly affects recall. Start with a recursive character splitter (500–1000 tokens, 10–20% overlap) for general text. For code, use language-aware splitters. For long documents, consider hierarchical chunking: large parent chunks for context, small child chunks for retrieval.

from langchain_text_splitters import RecursiveCharacterTextSplitter

def get_splitter(chunk_size=800, chunk_overlap=100):
    return RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )

# For code
from langchain_text_splitters import Language
from langchain_text_splitters import RecursiveCharacterTextSplitter as CodeSplitter

code_splitter = CodeSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=500,
    chunk_overlap=50,
)

Verify: Sample 20 chunks and confirm: (a) no chunk exceeds the model’s max tokens, (b) semantic units (paragraphs, functions) stay intact, (c) overlap preserves boundary context.

Step 3: Build the embedding pipeline with batching and retries

Embedding is I/O-bound. Batch requests (64–256 texts per call), retry on transient errors, and respect rate limits. If you’re using OpenAI-compatible endpoints, the client handles retries; for self-hosted models, implement your own.

import asyncio
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

client = AsyncOpenAI()  # or point to your gateway

@retry(
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(3),
)
async def embed_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = await client.embeddings.create(
        model=model,
        input=texts,
        encoding_format="float",
    )
    return [d.embedding for d in response.data]

async def embed_all(chunks: list[str], batch_size: int = 128) -> list[list[float]]:
    embeddings = []
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        embeddings.extend(await embed_batch(batch))
        # Respect rate limits — adjust based on your tier
        await asyncio.sleep(0.1)
    return embeddings

Verify: Embed a known dataset (e.g., 1k chunks from your corpus). Confirm: (a) output dimensionality matches the model spec, (b) no NaN/Inf values, (c) latency per batch is within budget, (d) retry logic triggers correctly on simulated 429/5xx responses.

Step 4: Store vectors in a search index with metadata

Choose a vector store that supports your scale and filter requirements. For <1M vectors, faiss or chromadb work locally. For production, consider pgvector, pinecone, weaviate, or qdrant. Always store metadata (source_id, chunk_index, timestamps) alongside vectors for filtering and citation.

import faiss
import numpy as np
import json
from pathlib import Path

class FaissStore:
    def __init__(self, dim: int, index_path: str = "faiss.index", meta_path: str = "meta.json"):
        self.dim = dim
        self.index_path = index_path
        self.meta_path = meta_path
        self.index = faiss.IndexFlatIP(dim)  # Inner product = cosine on normalized vectors
        self.metadata = []

    def add(self, embeddings: list[list[float]], metadatas: list[dict]):
        vecs = np.array(embeddings, dtype=np.float32)
        faiss.normalize_L2(vecs)
        self.index.add(vecs)
        self.metadata.extend(metadatas)

    def search(self, query_vec: np.ndarray, k: int = 10) -> list[dict]:
        faiss.normalize_L2(query_vec)
        scores, indices = self.index.search(query_vec, k)
        return [
            {"score": float(scores[0][i]), **self.metadata[indices[0][i]]}
            for i in range(len(indices[0]))
            if indices[0][i] != -1
        ]

    def persist(self):
        faiss.write_index(self.index, self.index_path)
        with open(self.meta_path, "w") as f:
            json.dump(self.metadata, f)

    @classmethod
    def load(cls, dim: int, index_path: str, meta_path: str):
        store = cls(dim, index_path, meta_path)
        store.index = faiss.read_index(index_path)
        with open(meta_path) as f:
            store.metadata = json.load(f)
        return store

Verify: After indexing, run 10 test queries. Confirm: (a) top-k results include expected documents, (b) metadata fields are queryable (e.g., filter by source_id), (c) index persists and reloads correctly.

Step 5: Implement retrieval with query preprocessing

Raw user queries often underperform. Preprocess: rewrite with an LLM (HyDE), expand with sub-queries, or prepend instructions for instruction-tuned models (e.g., "query: " for E5, "Represent this sentence for searching relevant passages: " for BGE). Then embed and search.

QUERY_INSTRUCTIONS = {
    "e5-large": "query: ",
    "bge-m3": "Represent this sentence for searching relevant passages: ",
    "openai-small": "",  # No instruction needed
    "openai-large": "",
}

async def retrieve(
    query: str,
    store: FaissStore,
    embed_model: str,
    k: int = 10,
    filter_fn=None,
) -> list[dict]:
    instruction = QUERY_INSTRUCTIONS.get(embed_model, "")
    query_text = f"{instruction}{query}"
    query_emb = (await embed_batch([query_text], model=embed_model))[0]
    query_vec = np.array([query_emb], dtype=np.float32)
    results = store.search(query_vec, k=k * 2)  # Over-fetch for filtering
    if filter_fn:
        results = [r for r in results if filter_fn(r)]
    return results[:k]

Verify: Create a small eval set (20–50 queries with known relevant doc IDs). Measure recall@k and MRR. Target: recall@10 > 0.85 on your domain. If low, revisit chunking (Step 2) or query preprocessing.

Bi-encoder retrieval (embedding model) optimizes for recall. A cross-encoder reranker scores (query, doc) pairs directly, improving precision. Use cross-encoder/ms-marco-MiniLM-L-6-v2 (fast) or BAAI/bge-reranker-large (stronger). Rerank top-50 to top-5–10.

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2", max_length=512)

def rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
    pairs = [(query, c.get("text", "")) for c in candidates]
    scores = reranker.predict(pairs)
    for c, s in zip(candidates, scores):
        c["rerank_score"] = float(s)
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:top_k]

Verify: Compare retrieval-only vs. retrieval+rerank on your eval set. Expect 5–15% relative improvement in nDCG@10. Confirm latency overhead is acceptable (<100ms for top-50).

Step 7: Assemble the full RAG pipeline

Wire retrieval into your LLM call. Pass retrieved chunks as context with citations. Enforce token budgets: reserve space for system prompt, user query, and completion.

from openai import AsyncOpenAI

llm_client = AsyncOpenAI()

SYSTEM_PROMPT = """You are a precise assistant. Answer using ONLY the provided context.
Cite sources inline like [1], [2] using the doc_id from each chunk.
If the context is insufficient, say so."""

async def rag_answer(query: str, store: FaissStore, embed_model: str, k: int = 5) -> dict:
    # Retrieve
    candidates = await retrieve(query, store, embed_model, k=20)
    # Rerank
    reranked = rerank(query, candidates, top_k=k)
    # Build context
    context_blocks = []
    for i, c in enumerate(reranked, 1):
        context_blocks.append(f"[{i}] (doc_id: {c['doc_id']}) {c['text']}")
    context = "\n\n".join(context_blocks)
    # Generate
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
    ]
    response = await llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0,
        max_tokens=512,
    )
    return {
        "answer": response.choices[0].message.content,
        "citations": [c["doc_id"] for c in reranked],
        "context_used": len(reranked),
    }

Verify: Run 10 end-to-end queries. Check: (a) answers cite actual retrieved chunks, (b) no hallucinated citations, (c) context fits within model’s context window, (d) latency breakdown (embed + retrieve + rerank + generate) meets SLA.

Step 8: Monitor and iterate on embedding quality

Embedding drift happens — corpus changes, query distribution shifts. Log every query with: query text, retrieved chunk IDs, rerank scores, user feedback (thumbs up/down), and LLM response. Periodically re-evaluate recall@k on a held-out set. If recall drops >5%, re-embed with a newer model or adjust chunking.

import logging
import uuid

query_logger = logging.getLogger("rag.queries")
query_logger.setLevel(logging.INFO)

def log_query(query: str, results: list[dict], answer: str, latency_ms: int):
    query_logger.info(json.dumps({
        "query_id": str(uuid.uuid4()),
        "query": query,
        "retrieved": [{"doc_id": r["doc_id"], "score": r.get("score"), "rerank": r.get("rerank_score")} for r in results],
        "answer_preview": answer[:200],
        "latency_ms": latency_ms,
    }))

Verify: Build a dashboard (Grafana, Datadog, or even a notebook) showing: p50/p95 latency per stage, recall@k over time, citation accuracy (sampled), and error rates. Set alerts on recall regression.

Common pitfalls to avoid

  • Skipping normalization: FAISS IndexFlatIP requires L2-normalized vectors for cosine similarity. Forgetting this silently breaks ranking.
  • Chunking too large: 2000-token chunks dilute signal. The embedding becomes an average of unrelated concepts.
  • Ignoring instruction prefixes: E5 and BGE models require their query prefixes. Omitting them degrades performance 10–20%.
  • No metadata filtering: Without source_id or timestamp filters, you can’t implement “search only recent docs” or “exclude deprecated versions.”
  • Embedding once, never again: Model improvements (e.g., text-embedding-3-small → text-embedding-3-large) are free recall gains if you re-index.

When to upgrade the embedding model

Move to a larger model when: (a) recall@10 plateaus below 0.8 on your eval set, (b) you add multilingual requirements, (c) you need stronger instruction following (e.g., “find code that handles auth”). The dimension increase (1536 → 3072) doubles index size and search latency — budget for it.

You now have a production-ready embedding pipeline: chunk → embed → index → retrieve → rerank → generate, with verification at each stage. The embedding model is the foundation — choose deliberately, monitor continuously, and re-embed when the data or the models improve.

Tagsembedding-modelsrag-architecturepipelinellm

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
GlossaryHow to choose an embedding model for your RAG pipelineA step-by-step guide to selecting the right embedding model for your RAG pipeline, with benchmarks, code, and validation strategies.GlossaryAnatomy of a production RAG architectureA senior engineer dissects the components, tradeoffs, and failure modes of a production RAG architecture — from ingestion to retrieval to generation — with code patterns and hard-won lessons.GlossaryAgentic RAG: adding reasoning steps to retrievalAgentic RAG adds multi-step reasoning to retrieval, letting agents plan, decompose queries, and iterate on results instead of single-pass lookup.GlossaryCommon failure points in RAG pipeline designSenior engineer breaks down the real-world failure points in RAG pipelines — chunking, retrieval, generation, and evaluation — with code examples and honest tradeoffs.GlossaryHow to design a RAG pipeline for enterprise documentsA step-by-step guide to building a production RAG pipeline for enterprise documents, covering ingestion, chunking, embedding, retrieval, and evaluation with runnable code.GlossaryNaive RAG vs advanced RAG: architectural differencesA practitioner's comparison of naive vs advanced RAG architectures across retrieval quality, latency, cost, and operational complexity with a clear verdict by use case.

All rag architecture & pipeline design posts →