RAG and LangChain

In this section, we will learn what Retrieval-Augmented Generation (RAG) is and how to use LangChain to build strong applications that connect language models with outside data sources. We will also look at how vector databases help with fast searching, and how to add RAG into your own applications using LangChain.

Retrieval-Augmented Generation (RAG)

Retrieval-Augmented Generation (RAG) is a technique that connects Large Language Models (LLMs) with outside data sources like databases, APIs, or any other place where information is stored. RAG lets an LLM look at useful information from these sources so it can give answers that are more correct and more related to what you asked.

Why do we need RAG?

LLMs can only handle a limited amount of text at one time - this is called a “context window.” Imagine you have 1000 pages of documents and you want to ask a question about them. Sending all 1000 pages to the LLM every time would cost a lot and would not work well. RAG solves this by picking out only the useful parts from those documents and giving just that to the LLM. This makes things faster and cheaper.

How does RAG work?

RAG has two phases:

1. Indexing Phase: In this step, we prepare the outside data so it can be searched later. We break the data into small pieces, turn each piece into a set of numbers called an “embedding” (this represents the meaning of the text), and save these in a vector database so we can find them quickly later.

2. Retrieval Phase: When a user asks something, the system looks through the indexed data and finds the pieces that match the question best. These pieces are then combined with the user’s question and sent to the LLM, which uses them to write an answer.

Vector Databases

Vector databases are special databases made to store and search through large amounts of “vectors” (lists of numbers) very quickly. They are important for RAG because they let us search by meaning, not just by matching exact words.

In simple words, a vector database stores embeddings and helps us quickly find pieces of text that have a similar meaning to what we are looking for. In RAG, it is used to find the most useful pieces of text for a user’s question so the LLM can give a better answer.

Think of a vector database like a search engine, but instead of matching exact words, it finds text that means something similar to what the user typed.

Popular Vector Databases

• Pinecone - Fully managed and easy for beginners.
• Qdrant - Open-source, fast, and easy to run on your own computer or server.
• Weaviate - Open-source and comes with built-in AI and search tools.
• Milvus - Built for very large-scale vector searching.
• PostgreSQL + pgvector - Lets you store and search embeddings inside PostgreSQL without needing a separate database.
• Chroma - Open-source and made specially for AI applications.

Example of setting up a Qdrant vector database using Docker:

service:
  vector-database:
    image: qdrant/qdrant
    ports:
      - "6333:6333"

    volumes:
      - ./qdrant_data:/qdrant/storage

LangChain

LangChain is a framework (a set of ready-made tools) that makes it easier to build AI applications using LLMs and outside data sources. It gives you many useful tools that most AI apps need, like prompt templates, memory, document loaders, vector database connections, output parsers, and ways to chain many LLM calls together. Instead of building everything from zero, LangChain gives you ready-made pieces so you can build faster.

LangChain Documentation: https://docs.langchain.com/docs/

Indexing Phase

In the indexing phase, we take data from outside sources and break it into small pieces, because large documents are hard to work with directly.

Each small piece is then sent to an embedding model, which turns the text into a list of numbers (a vector) that represents its meaning.

Finally, we save these vectors in a vector database, along with the original text piece. This makes it easy to find pieces later that have a meaning similar to a user’s question.

graph TD A[External Data Sources] --> B[Chunking] B --> C[Embedding Model] C --> D["Vector Database<br/>(Vectors + Chunks + Metadata)"]

Document Loaders

Document loaders are tools that help you load and read documents from different places, such as PDFs, Word files, web pages, and more. They are important for RAG apps because they make it easy to bring your data into the system.

LangChain gives you many document loaders that can handle different file types and sources. For example, you can use PyPDFLoader to load PDF files, UnstructuredURLLoader to load content from web pages, and TextLoader to load plain text files.

Install LangChain and the tools needed for loading PDF documents:

uv add langchain-community pypdf

# or
uv add langchain-community pymupdf # faster than pypdf

Example of using a document loader in LangChain:

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("./documents/sample.pdf")
docs = loader.load()

print(docs[0].page_content[:300])  # Print the content of the first page
print(docs[0].metadata)  # Print the metadata of the first page

Chunking

Chunking means breaking big documents into smaller, easier-to-handle pieces (chunks). This is needed because LLMs can only handle a limited amount of text, and giving them huge documents at once is slow and costly.

There are two common ways to chunk documents:

1. Overlap Chunking: In this method, each chunk shares some text with the next chunk. For example, if a document is 1000 words long, you might make chunks of 200 words, with 50 words overlapping between each chunk. So the first chunk would have words 1-200, the second chunk would have words 151-350, and so on. This helps make sure that important information which falls between two chunks does not get lost.

2. Non-Overlap Chunking: In this method, chunks do not share any text with each other. For example, if a document is 1000 words long, you might make chunks of 200 words with no overlap. So the first chunk would have words 1-200, the second chunk would have words 201-400, and so on. This method is faster but may lose some information that falls between two chunks.

Setup for chunking in LangChain:

uv add langchain-text-splitters

Example of using a text splitter for chunking in LangChain:

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=200)
chunks = text_splitter.split_text(document_text)
print(chunks[0])  # Print the first chunk

Embedding and Vector Database Storage

After chunking the documents, the next step is to create embeddings for each chunk and save them in a vector database. This lets us quickly find the right chunks when a user asks a question.

To create embeddings, you can use a ready-made embedding model such as OpenAI’s text-embedding-3-small, or any other embedding model that fits your needs. Once you have the embeddings, you can store them in a vector database like Qdrant, Pinecone, or Weaviate.

Setup for embedding and vector database storage in LangChain:

uv add langchain-openai langchain-qdrant

docker run -d --name qdrant -p 6333:6333 qdrant/qdrant

Example of creating embeddings and saving them in a Qdrant vector database using LangChain:

from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore

# Create embeddings for each chunk
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = QdrantVectorStore.from_documents(
    documents=chunks,
    embedding=embedding_model,
    url="http://localhost:6333",
    collection_name="my_collection"
)

API key required

Make sure to set your OpenAI API key in your environment variables before using the embedding model. If you don’t, you will get an error when trying to create embeddings.

OPENAI_API_KEY="your_api_key_here"

from dotenv import load_dotenv

load_dotenv()

Full Code

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
from dotenv import load_dotenv

load_dotenv()

# Load the PDF document
loader = PyPDFLoader("./documents/sample.pdf")
docs = loader.load()

# Chunk the document
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=200)
chunks = text_splitter.split_text(docs)

# Create embeddings for each chunk
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")

# Store embeddings in Qdrant vector database
vector_store = QdrantVectorStore.from_documents(
    documents=chunks,
    embedding=embedding_model,
    url="http://localhost:6333",
    collection_name="my_collection"
)

Retrieval Phase

This phase starts the moment a user asks a question.

First, the user’s question is turned into a vector, using the same embedding model that was used during indexing.

That question vector is then used to search the vector database for chunks that have a similar meaning to the user’s question.

Once the most useful chunks are found, they are combined with the user’s question and sent to the LLM as extra context (background information).

The LLM then uses both the user’s question and this extra information to write a more accurate answer.

graph TD A[User Query] --> B[Embedding Model] B --> C[Query Vector] C --> D["Vector Database<br/>(Find Similar Chunks)"] D --> E[Relevant Chunks] E --> F[Query + Retrieved Chunks] F --> G[LLM] G --> H[Generated Response]

Getting the Query

To get the question from the user, you can use a simple input function in Python. For example:

user_query = input("Please enter your question: ")

Searching for Similar Text in the Vector Database

To search for similar chunks in the vector database, you can use the similarity_search method that comes with the vector store. For example, if you are using Qdrant, you can do:

from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore

embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = QdrantVectorStore.from_existing_collection(
    url="http://localhost:6333",
    collection_name="my_collection"
    embedding=embedding_model
)

similar_chunks = vector_store.similarity_search(query=user_query)

Combining the Query and the Retrieved Chunks

Once you have the useful chunks, you can combine them with the user’s question to make a prompt for the LLM. For example:

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

content = "\n\n\n".join(
    f"Page content: {chunk.page_content}\nPage Number: {chunk.metadata['page_label']}\nFile Location: {chunk.metadata['source']}"
    for chunk in similar_chunks
)

SYSTEM_PROMPT = f"""
You are a helpful assistant that answers questions based on the following retrieved
information from documents. Use the retrieved information to answer the user's
question as accurately as possible. If you don't know the answer, say you don't know.
if you know the answer it and include page number and file location in your answer.

Context:
{content}
"""

input_prompt = input("Please enter your question: ")

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": input_prompt}
    ]
)

print(response.choices[0].message.content)

Full Code

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# Create embedding model and vector store
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = QdrantVectorStore.from_existing_collection(
    url="http://localhost:6333",
    collection_name="my_collection",
    embedding=embedding_model
)

# Get user query
user_query = input("Please enter your question: ")


# Search for similar chunks in the vector database
similar_chunks = vector_store.similarity_search(query=user_query)

# Combine the retrieved chunks with the user's query to create a prompt for the LLM
content = "\n\n\n".join(
    f"Page content: {chunk.page_content}\nPage Number: {chunk.metadata['page_label']}\nFile Location: {chunk.metadata['source']}"
    for chunk in similar_chunks
)

SYSTEM_PROMPT = f"""
You are a helpful assistant that answers questions based on the following retrieved
information from documents. Use the retrieved information to answer the user's
question as accurately as possible. If you don't know the answer, say you don't know.
if you know the answer it and include page number and file location in your answer.

Context:
{content}
"""

input_prompt = input("Please enter your question: ")

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": input_prompt}
    ]
)

print(response.choices[0].message.content)

Asynchronous/Background RAG

The problem with the code above is that it is “synchronous,” meaning it can only handle one request at a time. While the app is waiting for the LLM to send back a response, it cannot do anything else. If the response takes a long time, users have to wait longer too.

To fix this, we can process requests in a way where many things happen at the same time. There are two common ways to do this:

• Async IO: This uses Python’s asyncio library to handle tasks that involve waiting, like calling an LLM API, searching a vector database, or reading from a database. Instead of sitting and doing nothing while waiting, the app can work on other requests at the same time. This makes the app feel faster and lets it serve many users at once. But asyncio does not work well for tasks that use a lot of CPU power, like heavy document processing, creating embeddings on the CPU, or complex reranking, because these tasks block everything else from running. Also, if a task fails, asyncio will not automatically try it again - you would need to write that logic yourself or use a retry library.

• Using a Queue: This means sending tasks to a queue, which are then handled in the background. Tools like Celery, RQ, or Dramatiq send tasks to separate “worker” processes that run them on their own, away from the main app. This works better for tasks that take a long time or use a lot of CPU power, because the main app stays fast and responsive while the workers do the heavy work. Queues also come with useful features like automatic retries, task scheduling, handling failures, and the ability to add more workers to handle more load.

Note: Use Async IO when the user is waiting right there for a quick response (like a chat app, question answering, or an API request). Use a Queue when the task takes a long time, uses a lot of CPU, or can run in the background (like loading documents, creating embeddings, indexing, batch jobs, or making reports). Queues are also better when you need automatic retries, scheduling, or want to spread the work across many worker machines.

Here we will use a Redis queue to handle RAG requests in the background.

Setup

1. Install Redis using Docker:

   # docker-compose.yml
   services:
   redis:
     image: redis:latest
     ports:
       - "6379:6379"
     volumes:
       - ./.redis/data:/data

2. Start the Redis container:

   docker-compose up -d

3. Install the Python packages needed for Redis and RQ:

   uv add redis rq

4. Create a queue and a worker to handle RAG tasks in the background. Make a file named worker.py:

   # worker.py
   from redis import Redis
   from rq import Queue
   
   q = Queue(connection=Redis(
       host='localhost',
       port=6379,
   ))

5. Add a RAG task to the queue. Make a file named enqueue_task.py:

   # enqueue_task.py
   from worker import q
   from your_rag_module import process_rag_task  # Import your RAG processing function
   
   # Enqueue the RAG task
   job = q.enqueue(process_rag_task, user_query)
   print(f"Task enqueued with job ID: {job.id}")

6. Start the worker so it can process tasks from the queue:

   uv run rq worker --with-scheduler

Example FastAPI Application with Background RAG

# embedding.py
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
from dotenv import load_dotenv

load_dotenv()

# Load the PDF document
loader = PyPDFLoader("./documents/sample.pdf")
docs = loader.load()

# Chunk the document
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=200)
chunks = text_splitter.split_text(docs)

# Create embeddings for each chunk
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")

# Store embeddings in Qdrant vector database
vector_store = QdrantVectorStore.from_documents(
    documents=chunks,
    embedding=embedding_model,
    url="http://localhost:6333",
    collection_name="my_collection"
)

# process_query.py

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore

from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# Create embedding model and vector store
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = QdrantVectorStore.from_existing_collection(
    url="http://localhost:6333",
    collection_name="my_collection",
    embedding=embedding_model
)


def query_task(query: str):
    # Search for similar chunks in the vector database
    similar_chunks = vector_store.similarity_search(query=query)

    # Combine the retrieved chunks with the user's query to create a prompt for the LLM
    content = "\n\n\n".join(
        f"Page content: {chunk.page_content}\nPage Number: {chunk.metadata['page_label']}\nFile Location: {chunk.metadata['source']}"
        for chunk in similar_chunks
    )

    SYSTEM_PROMPT = f"""
    You are a helpful assistant that answers questions based on the following retrieved
    information from documents. Use the retrieved information to answer the user's
    question as accurately as possible. If you don't know the answer, say you don't know.
    if you know the answer it and include page number and file location in your answer.

    Context:
    {content}
    """

    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": query}
        ]
    )

    return response.choices[0].message.content

# worker.py
from redis import Redis
from rq import Queue

q = Queue(connection=Redis(
    host='localhost',
    port=6379,
))

# server.py
from fastapi import FastAPI, Query
from worker import q
from process_query import query_task


app = FastAPI()

@app.post("/query")
async def handle_query(user_query: str = Query(..., description="The user's question")):
    # Enqueue the RAG task to the background queue
    job = q.enqueue(query_task, user_query)
    return {"message": "Your query is being processed.", "job_id": job.id}

@app.get("/job_status/{job_id}")
async def get_job_status(job_id: str):
    job = q.fetch_job(job_id)
    if job is None:
        return {"error": "Job not found."}
    return {"job_id": job.id, "status": job.get_status(), "result": job.result}