Building a Local RAG Pipeline without External APIs

Building a Local RAG Pipeline without External APIs

Retrieval-Augmented Generation (RAG) has become the go-to pattern for building AI applications that can answer questions about private or domain-specific data. While most tutorials rely on OpenAI, Anthropic, or other cloud APIs, running a fully local RAG pipeline gives you complete data privacy, zero per-query costs, and full control over every component. This tutorial walks you through building a production-quality local RAG system from scratch using open-source tools that run entirely on your machine.

What Is a Local RAG Pipeline?

A RAG pipeline combines a retrieval mechanism with a generative language model. Instead of relying on a model's internal memorized knowledge, RAG fetches relevant context from a document store and feeds it to the model at inference time. A local RAG pipeline means every piece of this system — the embedding model, the vector database, the retrieval logic, and the generative LLM — runs on your own hardware without sending data to any external service.

The core flow looks like this:

• Ingestion: Load documents, split them into chunks, and generate embeddings using a local embedding model.
• Storage: Store the embeddings and their associated text in a local vector database.
• Retrieval: When a user asks a question, embed the query and search the vector database for the most similar chunks.
• Generation: Pass the retrieved chunks as context to a local LLM, which generates an answer grounded in that context.

Why It Matters

Running RAG locally solves several real problems that cloud-based solutions create. First, data privacy is guaranteed — sensitive documents never leave your infrastructure, which matters for healthcare, legal, financial, and enterprise internal knowledge bases. Second, cost predictability is total — there are no per-token charges, so you can run thousands of queries without a growing bill. Third, offline capability means your application works in air-gapped environments or areas with unreliable internet. Finally, reproducibility and control are easier when you pin exact model versions and can inspect or modify any part of the pipeline.

Prerequisites and Tool Selection

For this tutorial, you will need Python 3.10 or higher and a machine with at least 16 GB of RAM. A GPU is helpful but not required — the smaller models we use run acceptably on CPU. We will use the following stack:

• sentence-transformers for local embeddings (model: all-MiniLM-L6-v2)
• ChromaDB as the local vector database
• Ollama to serve a local LLM (model: llama3.2 or mistral)
• LangChain for orchestration and document loading

Install the dependencies first:

pip install langchain langchain-community langchain-chroma
pip install sentence-transformers chromadb
pip install pypdf unstructured

Then install Ollama by following the instructions at ollama.com for your operating system. Once installed, pull a model:

ollama pull llama3.2
ollama pull nomic-embed-text

Step 1: Loading and Chunking Documents

The first stage is getting your source documents into a usable format. LangChain provides document loaders for PDFs, markdown files, web pages, and more. After loading, you split documents into chunks because embedding models have token limits and retrieval works best with focused, semantically coherent passages.

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Load all PDFs from a local folder
loader = DirectoryLoader(
    "./docs",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader,
    show_progress=True
)
documents = loader.load()

print(f"Loaded {len(documents)} document sections")

# Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""]
)

chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks")

The RecursiveCharacterTextSplitter tries to split on paragraph boundaries first, then line breaks, then sentences, preserving semantic structure. A chunk size of 500 characters with 50 characters of overlap is a reasonable starting point, but you should tune this based on your content.

Step 2: Generating Embeddings Locally

Embeddings are numerical vectors that capture the semantic meaning of text. We use sentence-transformers with the all-MiniLM-L6-v2 model, which produces 384-dimensional vectors and runs quickly even on CPU. Alternatively, you can use Ollama's nomic-embed-text model for a fully Ollama-based setup.

from langchain_huggingface import HuggingFaceEmbeddings

embedding_model = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    model_kwargs={"device": "cpu"},
    encode_kwargs={"normalize_embeddings": True}
)

# Test the embedding
sample_vector = embedding_model.embed_query("How does RAG work?")
print(f"Embedding dimension: {len(sample_vector)}")

Normalizing embeddings ensures cosine similarity calculations are accurate and efficient. The model downloads automatically on first use and caches locally, so subsequent runs start instantly.

Step 3: Storing Vectors in ChromaDB

ChromaDB is a lightweight, file-based vector database that persists to disk. It requires no server process and integrates cleanly with LangChain. We create a persistent collection that survives between sessions.

from langchain_chroma import Chroma

vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embedding_model,
    collection_name="local_rag_docs",
    persist_directory="./chroma_db"
)

# Verify retrieval works
results = vector_store.similarity_search_with_score(
    "What is retrieval-augmented generation?",
    k=3
)

for doc, score in results:
    print(f"Score: {score:.4f} | Source: {doc.metadata.get('source', 'unknown')}")
    print(f"Text: {doc.page_content[:150]}...\n")

The persist_directory argument means your indexed documents are saved to disk. On subsequent runs, you can load the existing store without re-embedding everything:

vector_store = Chroma(
    collection_name="local_rag_docs",
    embedding_function=embedding_model,
    persist_directory="./chroma_db"
)

Step 4: Connecting to a Local LLM via Ollama

Ollama runs an OpenAI-compatible API on localhost:11434. LangChain's Ollama integration makes it trivial to connect. We configure the model with a low temperature for factual answering and set a reasonable context window.

from langchain_ollama import OllamaLLM

llm = OllamaLLM(
    model="llama3.2",
    temperature=0.1,
    num_ctx=4096
)

# Quick test
response = llm.invoke("Explain what a vector database is in one sentence.")
print(response)

Step 5: Building the RAG Chain

Now we combine retrieval and generation into a single chain. The chain takes a user question, retrieves relevant chunks, formats them into a prompt, and calls the LLM. We use LangChain's LCEL (LangChain Expression Language) for a clean, composable pipeline.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

template = """You are a helpful assistant answering questions based strictly
on the provided context. If the context does not contain enough information
to answer the question, say "I don't have enough information to answer that."

Context:
{context}

Question: {question}

Answer:"""

prompt = ChatPromptTemplate.from_template(template)

def format_docs(docs):
    return "\n\n".join(
        f"[Source: {d.metadata.get('source', 'unknown')}]\n{d.page_content}"
        for d in docs
    )

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

Now you can query your documents:

question = "What are the main benefits of using RAG?"
answer = rag_chain.invoke(question)
print(answer)

Step 6: Adding Source Citations

A good RAG system tells users where its answers come from. We can modify the chain to return both the answer and the source documents used to generate it.

from langchain_core.runnables import RunnableParallel

rag_chain_with_sources = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
).assign(
    answer=(
        lambda x: {"context": format_docs(x["context"]), "question": x["question"]}
    )
    | prompt
    | llm
    | StrOutputParser()
)

result = rag_chain_with_sources.invoke("Explain the retrieval step in RAG.")

print("Answer:")
print(result["answer"])
print("\nSources used:")
for i, doc in enumerate(result["context"], 1):
    print(f"  {i}. {doc.metadata.get('source', 'unknown')} (page {doc.metadata.get('page', '?')})")

Step 7: Wrapping It Into a Reusable Class

To make this practical, encapsulate everything into a clean class that handles initialization, ingestion, and querying. This makes your RAG pipeline reusable across projects.

import os
from typing import List, Dict, Optional

class LocalRAG:
    def __init__(
        self,
        docs_dir: str = "./docs",
        db_dir: str = "./chroma_db",
        embedding_model_name: str = "sentence-transformers/all-MiniLM-L6-v2",
        llm_model: str = "llama3.2",
        chunk_size: int = 500,
        chunk_overlap: int = 50
    ):
        self.docs_dir = docs_dir
        self.db_dir = db_dir
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model_name,
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True}
        )

        self.llm = OllamaLLM(model=llm_model, temperature=0.1, num_ctx=4096)

        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

        self.vector_store = None
        self.retriever = None
        self.chain = None

    def ingest(self, file_glob: str = "**/*.pdf"):
        """Load and index documents from the docs directory."""
        loader = DirectoryLoader(
            self.docs_dir,
            glob=file_glob,
            loader_cls=PyPDFLoader,
            show_progress=True
        )
        documents = loader.load()
        chunks = self.text_splitter.split_documents(documents)

        self.vector_store = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            collection_name="local_rag_docs",
            persist_directory=self.db_dir
        )
        self._build_chain()
        print(f"Indexed {len(chunks)} chunks from {len(documents)} sections.")
        return self

    def load_existing(self):
        """Load an existing vector store without re-ingesting."""
        self.vector_store = Chroma(
            collection_name="local_rag_docs",
            embedding_function=self.embeddings,
            persist_directory=self.db_dir
        )
        self._build_chain()
        return self

    def _build_chain(self):
        self.retriever = self.vector_store.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 4}
        )
        template = """You are a helpful assistant answering questions based
strictly on the provided context. If the context does not contain enough
information, say "I don't have enough information to answer that."

Context:
{context}

Question: {question}

Answer:"""
        prompt = ChatPromptTemplate.from_template(template)

        self.chain = RunnableParallel(
            {"context": self.retriever, "question": RunnablePassthrough()}
        ).assign(
            answer=(
                lambda x: {
                    "context": format_docs(x["context"]),
                    "question": x["question"]
                }
            )
            | prompt
            | self.llm
            | StrOutputParser()
        )

    def query(self, question: str) -> Dict:
        """Ask a question and get an answer with sources."""
        if self.chain is None:
            raise RuntimeError("Pipeline not initialized. Call ingest() or load_existing() first.")
        return self.chain.invoke(question)

    def clear(self):
        """Delete the vector database."""
        import shutil
        if os.path.exists(self.db_dir):
            shutil.rmtree(self.db_dir)
        self.vector_store = None
        self.chain = None

Usage is now straightforward:

rag = LocalRAG(docs_dir="./docs", db_dir="./chroma_db")

# First run: ingest documents
rag.ingest()

# Subsequent runs: load existing index
# rag.load_existing()

result = rag.query("What are the key components of a RAG system?")
print("Answer:", result["answer"])
print("\nSources:")
for doc in result["context"]:
    print(f"  - {doc.metadata.get('source')}")

Best Practices

Building a RAG pipeline that actually works well requires attention to several details beyond the basic wiring. Here are the practices that separate toy demos from reliable systems:

• Tune chunk size carefully. Too small and you lose context; too large and retrieval precision drops. Start at 400–600 characters and measure retrieval quality on a small evaluation set.
• Use meaningful overlap. Overlap of 10–20% of chunk size prevents splitting critical information across chunk boundaries.
• Experiment with retrieval parameters. Try search_type="mmr" (Maximum Marginal Relevance) to reduce redundancy in retrieved chunks, and adjust k based on your LLM's context window.
• Add metadata filtering. Tag chunks with source, date, author, or category so users can scope queries to specific subsets of your corpus.
• Prompt engineering matters. Explicitly instruct the model to use only the provided context and to admit ignorance when context is insufficient. This dramatically reduces hallucinations.
• Evaluate retrieval separately from generation. A bad answer might come from poor retrieval, not a weak LLM. Log which chunks were retrieved and inspect them manually for a sample of queries.
• Re-embed when you change models. Different embedding models produce vectors in different spaces. If you swap embedding models, you must rebuild the entire vector store.
• Handle incremental updates. For production systems, track document hashes and only re-ingest changed files rather than rebuilding the whole index every time.
• Monitor memory usage. Local LLMs consume significant RAM. If you hit OOM errors, use a smaller model or quantized versions (Ollama handles quantization automatically).
• Build a small evaluation set. Create 20–50 question-answer pairs from your domain and measure answer accuracy as you tweak parameters. Without evaluation, tuning is just guessing.

Performance Considerations

On a modern laptop with no GPU, embedding a few hundred PDF pages takes a few minutes, and each query returns in 5–15 seconds depending on the LLM model size. If you need faster responses, consider using a smaller model like phi3 or gemma2:2b, or invest in a GPU. The embedding step is the main bottleneck during ingestion, but it only happens once per document set. For ongoing ingestion, process new documents in batches in the background.

You can also speed up retrieval by using ChromaDB's built-in HNSW indexing, which is enabled by default. For very large corpora (hundreds of thousands of chunks), consider migrating to a more scalable solution like FAISS or Qdrant running locally.

Conclusion

Building a local RAG pipeline without external APIs is entirely practical with today's open-source ecosystem. By combining sentence-transformers for embeddings, ChromaDB for vector storage, and Ollama for local LLM inference, you get a fully private, cost-free, offline-capable question-answering system that you control end to end. The architecture scales from a personal knowledge base to an enterprise internal docs assistant — the same code works, you just add more documents and optionally upgrade to larger models or GPU acceleration. Start with the class provided here, tune the chunk size and retrieval parameters to your content, build a small evaluation set, and iterate. The real value of local RAG emerges when you treat it as a system to continuously refine rather than a one-time setup, and having every component on your own machine makes that iteration fast, private, and free.