RAG Applications Without LangChain or LlamaIndex

Introduction

In recent months, Retrieval Augmented Generation (RAG) has emerged as a powerful pattern for enhancing Large Language Models (LLMs) with private or domain-specific knowledge. While frameworks like LangChain and LlamaIndex have made RAG implementation more accessible, they can sometimes feel like black boxes, making debugging and customization challenging.

Why Build RAG From Scratch?

When I created a tutorial on building RAG applications with LangChain, many developers reached out with a common challenge: debugging LangChain applications was becoming increasingly difficult. The layers of abstraction, while convenient, were obscuring the underlying mechanics of how RAG actually works.

Consider these common pain points with framework-based RAG implementations:

Debugging Complexity: When something goes wrong, tracing the issue through multiple layers of framework abstraction can be time-consuming
Documentation Overhead: Understanding framework-specific concepts often requires navigating extensive documentation, taking focus away from core RAG principles
Limited Control: Framework abstractions can make it harder to customize specific components or optimize for your use case
Version Dependencies: Framework updates can introduce breaking changes or compatibility issues

A Framework-Free Approach

In this guide, we'll build a complete RAG application from scratch using only fundamental libraries:

# Core dependencies
chromadb        # Vector database for document storage
openai          # LLM API access
pypdf2          # PDF document processing
python-docx     # Word document processing
sentence-transformers  # Text embeddings

Our implementation will include all essential RAG components:

Document Processing: Handle multiple document formats (PDF, DOCX, TXT)
Vector Storage: Implement semantic search using ChromaDB
LLM Integration: Direct interaction with OpenAI's API
Conversational Memory: Support for follow-up questions and context

What Makes This Approach Different?

By building without frameworks, you'll gain:

Deep Understanding: See how each component of RAG works and interacts
Complete Control: Customize any part of the pipeline to suit your needs
Simplified Debugging: Trace issues directly to their source
Easy Maintenance: No framework-specific knowledge required

Here's a high-level overview of our architecture:

This architecture shows how we'll:

Process and index documents into our knowledge base
Handle user queries through semantic search
Maintain conversation history
Generate contextual responses

Document Processing and Indexing

The foundation of any RAG system is its ability to process and index documents effectively. In this section, we'll build a robust document processing pipeline that can handle multiple file formats and prepare documents for semantic search.

Document Loading Utilities

First, let's create utilities to handle different document formats. Our implementation supports PDF, DOCX, and plain text files:

import docx
import PyPDF2
import os

def read_text_file(file_path: str):
    """Read content from a text file"""
    with open(file_path, 'r', encoding='utf-8') as file:
        return file.read()

def read_pdf_file(file_path: str):
    """Read content from a PDF file"""
    text = ""
    with open(file_path, 'rb') as file:
        pdf_reader = PyPDF2.PdfReader(file)
        for page in pdf_reader.pages:
            text += page.extract_text() + "\n"
    return text

def read_docx_file(file_path: str):
    """Read content from a Word document"""
    doc = docx.Document(file_path)
    return "\n".join([paragraph.text for paragraph in doc.paragraphs])

We then create a unified interface for document reading:

def read_document(file_path: str):
    """Read document content based on file extension"""
    _, file_extension = os.path.splitext(file_path)
    file_extension = file_extension.lower()

    if file_extension == '.txt':
        return read_text_file(file_path)
    elif file_extension == '.pdf':
        return read_pdf_file(file_path)
    elif file_extension == '.docx':
        return read_docx_file(file_path)
    else:
        raise ValueError(f"Unsupported file format: {file_extension}")

Text Chunking Strategy

Once we have the raw text, we need to split it into manageable chunks. This is crucial for two reasons:

More precise semantic search results
Staying within LLM context windows

Here's our implementation of a sentence-aware text chunker:

def split_text(text: str, chunk_size: int = 500):
    """Split text into chunks while preserving sentence boundaries"""
    sentences = text.replace('\n', ' ').split('. ')
    chunks = []
    current_chunk = []
    current_size = 0

    for sentence in sentences:
        sentence = sentence.strip()
        if not sentence:
            continue

        # Ensure proper sentence ending
        if not sentence.endswith('.'):
            sentence += '.'

        sentence_size = len(sentence)

        # Check if adding this sentence would exceed chunk size
        if current_size + sentence_size > chunk_size and current_chunk:
            chunks.append(' '.join(current_chunk))
            current_chunk = [sentence]
            current_size = sentence_size
        else:
            current_chunk.append(sentence)
            current_size += sentence_size

    # Add the last chunk if it exists
    if current_chunk:
        chunks.append(' '.join(current_chunk))

    return chunks

Key features of our chunking strategy:

Preserves sentence boundaries
Configurable chunk size (default 500 characters)
Handles varying sentence lengths
Maintains readability of chunks

Setting Up ChromaDB

First, let's initialize ChromaDB with sentence transformers for embedding:

import chromadb
from chromadb.utils import embedding_functions

# Initialize ChromaDB client with persistence
client = chromadb.PersistentClient(path="chroma_db")

# Configure sentence transformer embeddings
sentence_transformer_ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2"
)

# Create or get existing collection
collection = client.get_or_create_collection(
    name="documents_collection",
    embedding_function=sentence_transformer_ef
)

Key components explained:

PersistentClient: Stores embeddings on disk for data persistence
all-MiniLM-L6-v2: A lightweight but effective embedding model
get_or_create_collection: Handles both initial creation and subsequent access

Inserting Data into ChromaDB

Now we'll create a complete pipeline that processes documents and prepares them for insertion into ChromaDB:

def process_document(file_path: str):
    """Process a single document and prepare it for ChromaDB"""
    try:
        # Read the document
        content = read_document(file_path)

        # Split into chunks
        chunks = split_text(content)

        # Prepare metadata
        file_name = os.path.basename(file_path)
        metadatas = [{"source": file_name, "chunk": i} for i in range(len(chunks))]
        ids = [f"{file_name}_chunk_{i}" for i in range(len(chunks))]

        return ids, chunks, metadatas
    except Exception as e:
        print(f"Error processing {file_path}: {str(e)}")
        return [], [], []

The process_document function:

Reads the document content
Splits it into chunks
Generates unique IDs for each chunk
Creates metadata including source file and chunk number

Batch Processing for Multiple Documents

To handle multiple documents efficiently:

def add_to_collection(collection, ids, texts, metadatas):
    """Add documents to collection in batches"""
    if not texts:
        return

    batch_size = 100
    for i in range(0, len(texts), batch_size):
        end_idx = min(i + batch_size, len(texts))
        collection.add(
            documents=texts[i:end_idx],
            metadatas=metadatas[i:end_idx],
            ids=ids[i:end_idx]
        )

def process_and_add_documents(collection, folder_path: str):
    """Process all documents in a folder and add to collection"""
    files = [os.path.join(folder_path, file) 
             for file in os.listdir(folder_path) 
             if os.path.isfile(os.path.join(folder_path, file))]

    for file_path in files:
        print(f"Processing {os.path.basename(file_path)}...")
        ids, texts, metadatas = process_document(file_path)
        add_to_collection(collection, ids, texts, metadatas)
        print(f"Added {len(texts)} chunks to collection")

Usage Example

Here's how to use the complete document processing pipeline:

# Initialize ChromaDB collection (we'll cover this in detail in the next section)
collection = client.get_or_create_collection(
    name="documents_collection",
    embedding_function=sentence_transformer_ef
)

# Process and add documents from a folder
folder_path = "/docs"
process_and_add_documents(collection, folder_path)

Example output:

Processing GreenGrow's EcoHarvest System_ A Revolution in Farming.pdf...
Added 6 chunks to collection
Processing Company_ QuantumNext Systems.docx...
Added 2 chunks to collection
Processing GreenGrow Innovations_ Company History.docx...
Added 5 chunks to collection
Processing Company_ TechWave Innovations.docx...
Added 1 chunks to collection
Processing Company_ GreenFields BioTech.docx...
Added 2 chunks to collection

Implementing Semantic Search

Now for the core functionality - semantic search to retrieve relevant documents:

def semantic_search(collection, query: str, n_results: int = 2):
    """Perform semantic search on the collection"""
    results = collection.query(
        query_texts=[query],
        n_results=n_results
    )
    return results

def get_context_with_sources(results):
    """Extract context and source information from search results"""
    # Combine document chunks into a single context
    context = "\n\n".join(results['documents'][0])

    # Format sources with metadata
    sources = [
        f"{meta['source']} (chunk {meta['chunk']})" 
        for meta in results['metadatas'][0]
    ]

    return context, sources

Example search usage:

# Perform a search
query = "When was GreenGrow Innovations founded?"
results = semantic_search(collection, query)
results

{'ids': [['GreenGrow Innovations_ Company History.docx_chunk_0',
   'GreenGrow Innovations_ Company History.docx_chunk_4']],
 'embeddings': None,
 'documents': [['GreenGrow Innovations was founded in 2010 by Sarah Chen...',
   'Despite its growth, GreenGrow remains committed to its original mission of ...']],
 'uris': None,
 'data': None,
 'metadatas': [[{'chunk': 0,
    'source': 'GreenGrow Innovations_ Company History.docx'},
   {'chunk': 4, 'source': 'GreenGrow Innovations_ Company History.docx'}]],
 'distances': [[0.3241303612288011, 0.5925477286632909]],
 'included': [<IncludeEnum.distances: 'distances'>,
  <IncludeEnum.documents: 'documents'>,
  <IncludeEnum.metadatas: 'metadatas'>]}

Understanding Search Results

ChromaDB search results contain several key components:

def print_search_results(results):
    """Print formatted search results"""
    print("\nSearch Results:\n" + "-" * 50)

    for i in range(len(results['documents'][0])):
        doc = results['documents'][0][i]
        meta = results['metadatas'][0][i]
        distance = results['distances'][0][i]

        print(f"\nResult {i + 1}")
        print(f"Source: {meta['source']}, Chunk {meta['chunk']}")
        print(f"Distance: {distance}")
        print(f"Content: {doc}\n")

Search Results:
--------------------------------------------------

Result 1:
Source: GreenGrow Innovations_ Company History.docx, Chunk 0
Content: GreenGrow Innovations was founded in 2010 by Sarah Chen and Michael Rodriguez,
two agricultural engineers with a passion for sustainable farming. The company
started in a small garage in Portland, Oregon, with a simple mission: to make
farming more environmentally friendly and efficient. In its early days,
GreenGrow focused on developing smart irrigation systems that could
significantly reduce water usage in agriculture.

Result 2:
Source: GreenGrow Innovations_ Company History.docx, Chunk 4
Content: Despite its growth, GreenGrow remains committed to its original mission of
promoting sustainable farming practices. The company regularly partners with
universities and research institutions to advance the field of agricultural
technology and hosts annual conferences to share knowledge with farmers and
other industry professionals.

Result 3:
Source: GreenGrow Innovations_ Company History.docx, Chunk 3
Content: This system caught the attention of large-scale farmers across the United
States, propelling GreenGrow to national prominence. Today, GreenGrow
Innovations employs over 200 people and has expanded its operations to include
offices in California and Iowa. The company continues to focus on developing
sustainable agricultural technologies, with ongoing projects in vertical
farming, drought-resistant crop development, and AI-powered farm management
systems.

The results include:

documents: The actual text chunks
metadatas: Associated metadata (source, chunk number)
distances: Similarity scores (lower is better)

OpenAI Integration

Now that we have our document retrieval system in place, let's integrate OpenAI's API to create the generation part of our RAG system. We'll focus on crafting effective prompts and managing the conversation flow.

Setting Up OpenAI

First, let's configure the OpenAI client:

import os
from openai import OpenAI

# Initialize OpenAI client
client = OpenAI()

# Set your API key
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

Prompt Engineering

The effectiveness of RAG heavily depends on how we structure our prompts. Let's create a modular prompt system:

def get_prompt(context: str, conversation_history: str, query: str):
    """Generate a prompt combining context, history, and query"""
    prompt = f"""Based on the following context and conversation history, 
    please provide a relevant and contextual response. If the answer cannot 
    be derived from the context, only use the conversation history or say 
    "I cannot answer this based on the provided information."

    Context from documents:
    {context}

    Previous conversation:
    {conversation_history}

    Human: {query}

    Assistant:"""

    return prompt

Response Generation

Let's implement the core response generation function:

def generate_response(query: str, context: str, conversation_history: str = ""):
    """Generate a response using OpenAI with conversation history"""
    prompt = get_prompt(context, conversation_history, query)

    try:
        response = client.chat.completions.create(
            model="gpt-4",  # or gpt-3.5-turbo for lower cost
            messages=[
                {"role": "system", "content": "You are a helpful assistant that answers questions based on the provided context."},
                {"role": "user", "content": prompt}
            ],
            temperature=0,  # Lower temperature for more focused responses
            max_tokens=500
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"Error generating response: {str(e)}"

Perform RAG query

def rag_query(collection, query: str, n_chunks: int = 2):
    """Perform RAG query: retrieve relevant chunks and generate answer"""
    # Get relevant chunks
    results = semantic_search(collection, query, n_chunks)
    context, sources = get_context_with_sources(results)

    # Generate response
    response = generate_response(query, context)

    return response, sources

query = "When was GreenGrow Innovations founded?"
response, sources = rag_query(collection, query)

# Print results
print("\nQuery:", query)
print("\nAnswer:", response)
print("\nSources used:")
for source in sources:
    print(f"- {source}")

Query: When was GreenGrow Innovations founded?

Answer: GreenGrow Innovations was founded in 2010.

Sources used:
- GreenGrow Innovations_ Company History.docx (chunk 0)
- GreenGrow Innovations_ Company History.docx (chunk 4)

Building Conversational Memory

Conversational memory is crucial for RAG applications to handle follow-up questions and maintain context across interactions. We'll implement a robust system that manages conversation history and enables natural dialogue.

Session Management

First, let's implement basic session management to handle multiple conversations:

import uuid
from datetime import datetime
import json

# In-memory conversation store
conversations = {}

def create_session():
    """Create a new conversation session"""
    session_id = str(uuid.uuid4())
    conversations[session_id] = []
    return session_id

Message Management

Let's implement functions to add and retrieve messages:

def add_message(session_id: str, role: str, content: str):
    """Add a message to the conversation history"""
    if session_id not in conversations:
        conversations[session_id] = []

    conversations[session_id].append({
        "role": role,
        "content": content,
        "timestamp": datetime.now().isoformat()
    })

def get_conversation_history(session_id: str, max_messages: int = None):
    """Get conversation history for a session"""
    if session_id not in conversations:
        return []

    history = conversations[session_id]
    if max_messages:
        history = history[-max_messages:]

    return history

Formatting Conversation History

Create a formatter for the conversation history:

def format_history_for_prompt(session_id: str, max_messages: int = 5):
    """Format conversation history for inclusion in prompts"""
    history = get_conversation_history(session_id, max_messages)
    formatted_history = ""

    for msg in history:
        role = "Human" if msg["role"] == "user" else "Assistant"
        formatted_history += f"{role}: {msg['content']}\n\n"

    return formatted_history.strip()

Query Contextualization

For follow-up questions, we need to contextualize queries based on conversation history:

First Question : When was GreenGrow Innovations founded?

Follow up : Where is it headquartered?

contextualized_query : Where is GreenGrow Innovations headquartered?

def contextualize_query(query: str, conversation_history: str, client: OpenAI):
    """Convert follow-up questions into standalone queries"""
    contextualize_prompt = """Given a chat history and the latest user question 
    which might reference context in the chat history, formulate a standalone 
    question which can be understood without the chat history. Do NOT answer 
    the question, just reformulate it if needed and otherwise return it as is."""

    try:
        completion = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": contextualize_prompt},
                {"role": "user", "content": f"Chat history:\n{conversation_history}\n\nQuestion:\n{query}"}
            ]
        )
        return completion.choices[0].message.content
    except Exception as e:
        print(f"Error contextualizing query: {str(e)}")
        return query  # Fallback to original query

Combining RAG Components

def get_prompt(context, conversation_history, query):
  prompt = f"""Based on the following context and conversation history, please provide a relevant and contextual response.
    If the answer cannot be derived from the context, only use the conversation history or say "I cannot answer this based on the provided information."

    Context from documents:
    {context}

    Previous conversation:
    {conversation_history}

    Human: {query}

    Assistant:"""
  return prompt

# Updated generate response function with conversation history also passed for Chatbot Memory
def generate_response(query: str, context: str, conversation_history: str = ""):
    """Generate a response using OpenAI with conversation history"""
    prompt = get_prompt(context, conversation_history, query)
    # print(prompt)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": prompt}
        ],
        temperature=0,
        max_tokens=500
    )

    return response.choices[0].message.content

Now let's create the main RAG query function that combines everything:

def conversational_rag_query(
    collection,
    query: str,
    session_id: str,
    n_chunks: int = 3
):
    """Perform RAG query with conversation history"""
    # Get conversation history
    conversation_history = format_history_for_prompt(session_id)

    # Handle follo up questions
    query = contextualize_query(query, conversation_history, client)
    print("Contextualized Query:", query)

    # Get relevant chunks
    context, sources = get_context_with_sources(
        semantic_search(collection, query, n_chunks)
    )
    print("Context:", context)
    print("Sources:", sources)


    response = generate_response(query, context, conversation_history)

    # Add to conversation history
    add_message(session_id, "user", query)
    add_message(session_id, "assistant", response)

    return response, sources

Example Usage

Here's how to use the complete system:

# Create a new conversation session
session_id = create_session()

# First question
query = "When was GreenGrow Innovations founded?"
response, sources = conversational_rag_query(
            collection,
            query,
            session_id
)
print(response)

Example output:

ontextualized Query: When did GreenGrow Innovations come into existence?
Context: GreenGrow Innovations was founded in 2010 by .....
Sources: ['GreenGrow Innovations_ Company History.docx (chunk 0)', 'GreenGrow Innovations_ Company History.docx (chunk 3)', 'GreenGrow Innovations_ Company History.docx (chunk 4)']
GreenGrow Innovations was founded in 2010.

query = "Where is it located?"
response, sources = conversational_rag_query(
            collection,
            query,
            session_id
)
print(response)

Contextualized Query: What is the location of GreenGrow Innovations?
Context: GreenGrow Innovations was founded in 2010 by Sarah Chen and Michael Rodriguez, two agricultural engineers with a passion for sustainable farming. The company started in a small garage in Portland, Oregon, with a simple mission......
Sources: ['GreenGrow Innovations_ Company History.docx (chunk 0)', 'GreenGrow Innovations_ Company History.docx (chunk 3)', 'GreenGrow Innovations_ Company History.docx (chunk 4)']
GreenGrow Innovations started in Portland, Oregon, and has since expanded its operations to include offices in California and Iowa.

Conclusion:

This tutorial shows you how to build RAG without LangChain or LlamaIndex when you need direct control over your implementation. You'll learn to process documents, perform semantic search, and handle conversations using just ChromaDB and OpenAI's API. All code is production-ready and debuggable – no black boxes.

Explore How We Can Help Your Business:

Case Studies: Visit FutureSmart AI Case Studies to see real-world applications of our NLP solutions
Contact Us: Have a project in mind? Reach out at contact@futuresmart.ai

Let us help you transform your business with custom AI solutions tailored to your specific needs.

Code: https://github.com/PradipNichite/Youtube-Tutorials/tree/main/Vanilla%20RAG

https://youtu.be/8sSHg1034r0

Building RAG Applications Without LangChain or LlamaIndex

Comments

More from this blog

Building AI Knowledge Graph Using Graphiti & Neo4j

AI Agents Memory: Mem0 + LangGraph Agent Integration

LangGraph RAG Agent Tutorial | Basics to Advanced Multi-Agent AI Chatbot

Building a RAG System with Async FastAPI, Qdrant, Langchain and OpenAI

Building an Async Similarity Search System from scratch with FastAPI and Qdrant VectorDB

Introduction

Why Build RAG From Scratch?

A Framework-Free Approach

What Makes This Approach Different?

Document Processing and Indexing

Document Loading Utilities

Text Chunking Strategy

Setting Up ChromaDB

Inserting Data into ChromaDB

Batch Processing for Multiple Documents

Usage Example

Implementing Semantic Search

Understanding Search Results

OpenAI Integration

Setting Up OpenAI

Prompt Engineering

Response Generation

Perform RAG query

Building Conversational Memory

Session Management

Message Management

Formatting Conversation History

Query Contextualization

Combining RAG Components

Example Usage

Conclusion:

Explore How We Can Help Your Business:

Command Palette

Comments

More from this blog

Introduction

Why Build RAG From Scratch?

A Framework-Free Approach

What Makes This Approach Different?

Document Processing and Indexing

Document Loading Utilities

Text Chunking Strategy

Setting Up ChromaDB

Inserting Data into ChromaDB

Batch Processing for Multiple Documents

Usage Example

Implementing Semantic Search

Understanding Search Results

OpenAI Integration

Setting Up OpenAI

Prompt Engineering

Response Generation

Perform RAG query

Building Conversational Memory

Session Management

Message Management

Formatting Conversation History

Query Contextualization

Combining RAG Components

Example Usage

Conclusion:

Explore How We Can Help Your Business: