Skip to main content
EN · FR Start for free
Menu
Lesson 14 of 18 · Embeddings, RAG, and Semantic Search

Complete RAG Pipeline

Lesson 14 of 18 Updated November 2, 2026 Developer
On this page
  1. Putting It All Together
  2. The Complete Pipeline
  3. Running the Pipeline
  4. Five Directions for Further Improvement
  5. Key Takeaways

Putting It All Together

Previous lessons covered each link in the chain separately: loading, chunking, encoding, indexing, retrieval, and generation. Now it’s time to weld them into a single, complete, end-to-end program that you can run as-is and then adapt to your own documents. This script fits in one file and requires only four dependencies.

pip install mistralai faiss-cpu numpy requests

The Complete Pipeline

Two design choices structure this code. First, we group all configuration at the top of the file: models, chunk size, batch size, and TOP_K are named constants. This saves you from hunting down hard-coded values buried inside functions when you need to tune behavior. Second, we split the logic into six independent functions, each corresponding to one step and printing its progress as [n/6]. This trace isn’t decorative: when the pipeline fails or returns a strange answer, it tells you immediately whether the problem lies in loading, chunking, or retrieval. Each function is also replaceable in isolation—you can swap in a PDF loader for load_data without touching the rest.

The main function chains the steps together, then queries the corpus with three questions about the Paul Graham essay used as our demo dataset. Notice that retrieval and generation run inside the loop, while indexing happens just once beforehand: re-encoding the entire corpus for every question would be absurd in terms of both cost and latency.

import os
import json
import numpy as np
import faiss
import requests
from mistralai.client import Mistral

# ============================================================
# Configuration
# ============================================================
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
EMBED_MODEL = "mistral-embed"
CHAT_MODEL = "mistral-large-latest"
CHUNK_SIZE = 2048
BATCH_SIZE = 25
TOP_K = 3

client = Mistral(api_key=MISTRAL_API_KEY)


# ============================================================
# Step 1: Load Data
# ============================================================
def load_data(url):
    """Load a text document from a URL."""
    print("[1/6] Loading data...")
    response = requests.get(url)
    response.raise_for_status()
    text = response.text
    print(f"  -> {len(text)} characters loaded")
    return text


# ============================================================
# Step 2: Create Chunks
# ============================================================
def create_chunks(text, chunk_size=CHUNK_SIZE):
    """Split text into chunks."""
    print("[2/6] Creating chunks...")
    chunks = []
    paragraphs = text.split("\n\n")
    current = ""

    for para in paragraphs:
        para = para.strip()
        if not para:
            continue
        if len(current) + len(para) + 2 <= chunk_size:
            current = current + "\n\n" + para if current else para
        else:
            if current:
                chunks.append(current)
            current = para

    if current:
        chunks.append(current)

    print(f"  -> {len(chunks)} chunks created (avg size: {sum(len(c) for c in chunks) // len(chunks)} chars)")
    return chunks


# ============================================================
# Step 3: Generate Embeddings
# ============================================================
def create_embeddings(chunks, batch_size=BATCH_SIZE):
    """Encode chunks into embeddings via mistral-embed."""
    print("[3/6] Generating embeddings...")
    all_embeddings = []

    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        response = client.embeddings.create(
            model=EMBED_MODEL,
            inputs=batch,
        )
        all_embeddings.extend([d.embedding for d in response.data])
        print(f"  -> Batch {i // batch_size + 1}/{(len(chunks) - 1) // batch_size + 1} encoded")

    embeddings = np.array(all_embeddings, dtype=np.float32)
    print(f"  -> Matrix: {embeddings.shape}")
    return embeddings


# ============================================================
# Step 4: Create FAISS Index
# ============================================================
def create_index(embeddings):
    """Create and populate a FAISS index."""
    print("[4/6] Creating FAISS index...")
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)
    index.add(embeddings)
    print(f"  -> Index: {index.ntotal} vectors, dimension {dimension}")
    return index


# ============================================================
# Step 5: Retrieve Relevant Chunks
# ============================================================
def retrieve(question, index, chunks, k=TOP_K):
    """Retrieve the most relevant chunks."""
    print(f"[5/6] Searching for: '{question[:50]}...'")
    response = client.embeddings.create(
        model=EMBED_MODEL,
        inputs=[question],
    )
    query_vector = np.array([response.data[0].embedding], dtype=np.float32)

    distances, indices = index.search(query_vector, k)

    results = []
    for dist, idx in zip(distances[0], indices[0]):
        if idx >= 0:
            results.append({
                "text": chunks[idx],
                "distance": float(dist),
            })

    print(f"  -> {len(results)} chunks retrieved")
    return results


# ============================================================
# Step 6: Generate Answer
# ============================================================
def generate(question, context_chunks):
    """Generate the answer using retrieved context."""
    print("[6/6] Generating answer...")

    context = "\n\n---\n\n".join([c["text"] for c in context_chunks])

    prompt = f"""Informative context:

{context}

---

Based ONLY on the context above, answer the following question precisely and completely.

Question: {question}"""

    response = client.chat.complete(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": "You are an assistant that answers based on the provided context."},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )

    answer = response.choices[0].message.content
    print(f"  -> Answer generated ({len(answer)} characters)")
    return answer


# ============================================================
# Main Pipeline
# ============================================================
def main():
    """Run the complete RAG pipeline."""
    # 1. Load
    url = "https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt"
    text = load_data(url)

    # 2. Chunk
    chunks = create_chunks(text)

    # 3. Encode
    embeddings = create_embeddings(chunks)

    # 4. Index
    index = create_index(embeddings)

    # 5+6. Question -> Answer
    questions = [
        "What were the two main things the author worked on before college?",
        "What did the author do during their time at Y Combinator?",
        "What programming language did the author learn first?",
    ]

    for question in questions:
        print(f"\n{'='*60}")
        context = retrieve(question, index, chunks)
        answer = generate(question, context)
        print(f"\nQ: {question}")
        print(f"A: {answer}")


if __name__ == "__main__":
    main()

Running the Pipeline

Save this code as rag_pipeline.py, export your API key in the environment—never hard-code it in the file, which will eventually end up in a repository—and then run the script. The output walks through the six steps, then processes the three questions with their answers. If any answer seems off-topic, compare it against the chunks displayed in step 5 before blaming the generation model.

export MISTRAL_API_KEY="your-api-key"
python rag_pipeline.py

Five Directions for Further Improvement

This pipeline is a foundation, not a finished product, and each of its limitations points to a next step. The most immediate is persistence: in its current form, the script re-encodes the entire corpus on every run, whereas saving the FAISS index and chunks would make subsequent executions nearly instant. Re-ranking is the next level up, passing FAISS results through a cross-encoder that judges relevance far more finely than vector distance alone. Metadata filtering addresses another need: constraining search by date, category, or author in addition to similarity. Two final improvements focus on user experience: streaming, which displays the answer as it generates rather than after several seconds of silence, and conversation history, which lets users follow up with “what about version 2?” without restating their entire question.

# 1. Save the index to avoid recalculating
faiss.write_index(index, "my_rag.index")
np.save("my_chunks.npy", np.array(chunks, dtype=object))

# 2. Add re-ranking
# After FAISS retrieval, use a cross-encoder model
# to re-order results by fine-grained relevance

# 3. Metadata filtering
# Add filters (date, category, author) in addition
# to vector similarity

# 4. Response streaming
# Use client.chat.stream() to display the answer
# as it generates

# 5. Conversation history
# Add previous exchanges to the prompt
# for a contextual RAG chatbot

Key Takeaways

  • The complete RAG pipeline breaks down into 6 steps: load, chunk, encode, index, retrieve, generate
  • Each step is an independent function, easy to test and replace
  • Use temperature=0.1 for factual answers
  • Save the FAISS index and embeddings to avoid recalculating on every run
  • This pipeline is a solid starting point to enrich based on your needs
← Previous lessonRetrieval and GenerationNext lesson →Corrective RAG