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. Component 1 — Document Processor 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. Component 2 — Embedding Service 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. Component 3 — Vector Store 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. Component 4 — RAG Pipeline 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. Run 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) Real Output 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. Moving to Azure 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 What to Tune Next This implementation is intentionally minimal. In production add: Higher score threshold — 0.6 to 0.75 for larger document sets Reranking — a second model pass to reorder chunks by true relevance Query expansion — generate multiple question phrasings before searching Conversation history — maintain context across multiple turns Observability — log every retrieval and generation for debugging These are the patterns covered in the next post — where RAG goes from working to production-ready.