Building a RAG System From Scratch — Four Components, One Working Pipeline

Most RAG tutorials explain the concept. This one shows the code — a complete working pipeline using LangChain, ChromaDB, and a local LLM via LM Studio. No OpenAI API key. No cloud costs.

The business problem: A company has hundreds of pages of HR documentation. Employees ask questions. An AI answers accurately from the actual documents in seconds.

Stack: Python, LangChain, ChromaDB, nomic-embed-text, Qwen 9B via LM Studio

pip install langchain langchain-community langchain-chroma langchain-openai langchain-text-splitters langchain-core openai requests

Make sure LM Studio is running with both models loaded before running the code.

Splits raw text into chunks. Chunk size and overlap determine everything downstream — cut a paragraph wrong and retrieval suffers.

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List

class DocumentProcessor:
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

    def process(self, text: str, source: str = "") -> List[Document]:
        chunks = self.splitter.split_text(text)
        return [
            Document(
                page_content=chunk,
                metadata={"source": source, "chunk_index": i}
            )
            for i, chunk in enumerate(chunks)
        ]

⚠️ Changing chunk size later means reembedding everything from scratch. Your chunks and embeddings are tightly coupled. Choose carefully the first time.

Converts text into numerical vectors. LM Studio needs a direct requests call — the standard LangChain OpenAI wrapper sends the wrong input format and throws a 400 error.

from langchain_core.embeddings import Embeddings
import requests

LM_STUDIO_URL = "http://localhost:1234/v1"
EMBED_MODEL   = "nomic-embed-text"

class EmbeddingService(Embeddings):
    def __init__(self):
        self.url   = f"{LM_STUDIO_URL}/embeddings"
        self.model = EMBED_MODEL

    def _embed(self, text: str) -> List[float]:
        response = requests.post(
            self.url,
            json={"model": self.model, "input": text},
            headers={"Authorization": "Bearer lm-studio"}
        )
        response.raise_for_status()
        return response.json()["data"][0]["embedding"]

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return [self._embed(text) for text in texts]

    def embed_query(self, text: str) -> List[float]:
        return self._embed(text)

⚠️ Same model must be used for both indexing and querying. They must share the same vector space. Switching models means rebuilding the entire index.

Stores embeddings and retrieves closest matches by meaning — not by keyword. ChromaDB persists to disk so your index survives restarts without reprocessing documents.

from langchain_chroma import Chroma
from typing import Tuple

CHROMA_DIR      = "./chroma_db"
TOP_K           = 4
SCORE_THRESHOLD = 0.3

class VectorStore:
    def __init__(self, embedding_service: EmbeddingService):
        self.store = Chroma(
            collection_name="rag_collection",
            embedding_function=embedding_service,
            persist_directory=CHROMA_DIR
        )

    def add_documents(self, documents: List[Document]) -> None:
        self.store.add_documents(documents)

    def similarity_search(self, query: str) -> List[Tuple[Document, float]]:
        results = self.store.similarity_search_with_relevance_scores(
            query=query,
            k=TOP_K,
        )
        return [(doc, score) for doc, score in results if score >= SCORE_THRESHOLD]

💡 Score threshold of 0.3 works well for small document sets. In production with larger knowledge bases tune this to 0.6–0.75 to filter out loosely relevant chunks.

Orchestrates the full flow. Query in, embed it, retrieve closest chunks, build context, generate grounded answer, return sources.

The most important line in the entire pipeline is the system prompt. Without the strict instruction to answer only from context, the LLM falls back on its general training knowledge — defeating the entire purpose of RAG.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing import Dict

CHAT_MODEL = "qwen/qwen3.5-9b"

class RAGPipeline:
    def __init__(self, vector_store: VectorStore):
        self.vector_store = vector_store
        self.llm = ChatOpenAI(
            model=CHAT_MODEL,
            openai_api_base=LM_STUDIO_URL,
            openai_api_key="lm-studio",
            max_tokens=1000,
            temperature=0.1
        )

    def _build_context(self, documents: List[Tuple[Document, float]]) -> str:
        parts = []
        for doc, score in documents:
            parts.append(
                f"Source: {doc.metadata.get('source', 'unknown')}\n"
                f"Relevance: {score:.2f}\n"
                f"Content: {doc.page_content}"
            )
        return "\n\n---\n\n".join(parts)

    def query(self, question: str) -> Dict:
        results = self.vector_store.similarity_search(question)

        if not results:
            return {
                "answer": "I could not find relevant information to answer this question.",
                "sources": [],
                "chunks_used": 0
            }

        context = self._build_context(results)

        messages = [
            SystemMessage(content=(
                "You are a precise assistant that answers questions "
                "based solely on the provided context. "
                "If the answer is not in the context, say so clearly. "
                "Do not use your general knowledge to supplement the context."
            )),
            HumanMessage(content=(
                f"Context:\n{context}\n\n"
                f"Question: {question}\n\n"
                f"Answer based only on the context above:"
            ))
        ]

        response = self.llm(messages)

        return {
            "answer": response.content,
            "sources": list(set(doc.metadata.get("source") for doc, _ in results)),
            "chunks_used": len(results)
        }

💡 temperature=0.1 keeps answers factual and consistent. Low temperature means the model stays close to what the context says rather than being creative with it.

if __name__ == "__main__":

    SAMPLE_DOCUMENT = """
    Employee Leave Policy

    Sick Leave:
    Employees are entitled to 10 days of paid sick leave per calendar year.
    Sick leave resets on January 1st each year.
    Unused sick leave cannot be carried over to the next year.
    To apply for sick leave, submit a request through the HR portal.
    A medical certificate is required for sick leave exceeding 3 consecutive days.

    Annual Leave:
    Employees receive 25 days of annual leave per year.
    Annual leave must be approved by the line manager at least 2 weeks in advance.
    Up to 5 unused annual leave days can be carried over to the following year.

    Remote Work Policy:
    Employees may work remotely up to 3 days per week.
    Remote work requires a stable internet connection and a dedicated workspace.
    Core hours of 10:00 to 16:00 must be maintained regardless of location.

    Expense Policy:
    Business travel expenses must be approved before travel.
    Receipts are required for all expenses above 25 euros.
    Expense reports must be submitted within 30 days of the expense.
    Maximum meal allowance is 50 euros per day during business travel.
    """

    processor  = DocumentProcessor()
    embeddings = EmbeddingService()
    store      = VectorStore(embedding_service=embeddings)
    pipeline   = RAGPipeline(vector_store=store)

    documents = processor.process(SAMPLE_DOCUMENT, source="company_policy")
    store.add_documents(documents)

    questions = [
        "How many sick days am I entitled to per year?",
        "Can I carry over unused annual leave?",
        "How many days can I work remotely?",
        "What is the maximum meal allowance during business travel?",
        "Do I need a medical certificate for sick leave?",
    ]

    for q in questions:
        result = pipeline.query(q)
        print(f"Q: {q}")
        print(f"A: {result['answer']}")
        print(f"   Sources: {result['sources']}")
        print(f"   Chunks used: {result['chunks_used']}")
        print("-" * 60)

This is the actual output from running this pipeline locally:

Q: How many sick days am I entitled to per year?
A: Based on the provided context, employees are entitled to 10 days
   of paid sick leave per calendar year.
   Sources: ['company_policy'] · Chunks used: 4

Q: Can I carry over unused annual leave?
A: Yes, up to 5 unused annual leave days can be carried over to
   the following year.
   Sources: ['company_policy'] · Chunks used: 4

Q: How many days can I work remotely?
A: Based on the provided context, employees may work remotely up
   to 3 days per week.
   Sources: ['company_policy'] · Chunks used: 4

Q: What is the maximum meal allowance during business travel?
A: The maximum meal allowance during business travel is 50 euros
   per day.
   Sources: ['company_policy'] · Chunks used: 4

Q: Do I need a medical certificate for sick leave?
A: A medical certificate is required for sick leave exceeding
   3 consecutive days.
   Sources: ['company_policy'] · Chunks used: 4

Every answer accurate. Every answer sourced. No hallucination. No API key. No cloud cost.

Replace SAMPLE_DOCUMENT

with your own content and you have a working RAG system on your documents in minutes.

Every component maps directly to an Azure service. The pipeline logic stays identical — only the initialisation changes.

This implementation is intentionally minimal. In production add:

These are the patterns covered in the next post — where RAG goes from working to production-ready.