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

A developer built a complete retrieval-augmented generation (RAG) pipeline from scratch using LangChain, ChromaDB, and a local LLM via LM Studio, avoiding cloud costs and API keys. The system splits HR documents into chunks, embeds them with nomic-embed-text, stores vectors in ChromaDB, and retrieves relevant passages for a Qwen 9B model to answer employee questions. The developer highlighted key pitfalls, such as the need to use the same embedding model for indexing and querying and the tight coupling between chunk size and embeddings.

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. python 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. python 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. python 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. python 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. """ Initialise all four components processor = DocumentProcessor embeddings = EmbeddingService store = VectorStore embedding service=embeddings pipeline = RAGPipeline vector store=store Index documents — happens once documents = processor.process SAMPLE DOCUMENT, source="company policy" store.add documents documents Query — happens live for every user question 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. | Local | Azure Equivalent | |---|---| | nomic-embed-text via LM Studio | AzureOpenAIEmbeddings | | Qwen 9B via LM Studio | AzureChatOpenAI | | ChromaDB | Azure AI Search | 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.