Module 6: Multimodal RAG

6. Implementation with LangChain

Description

In the previous capsules you built each piece of multimodal RAG manually: embeddings, indexing, retrieval, document processing. It works, but you wrote a lot of infrastructure code. LangChain abstracts much of that infrastructure: document loaders, text splitters, vector stores, retrievers and chains are already implemented. In this capsule you'll re-implement multimodal RAG using LangChain, compare it with the manual implementation, and understand when each approach is worthwhile.

Why it matters: In production, you don't want to maintain custom chunking, embedding and retrieval code if a proven framework does it. LangChain gives you interchangeable components: you can switch from ChromaDB to Pinecone, or from OpenAI embeddings to Cohere, by changing one line. But you also need to understand what's happening underneath, which is why we did it manually first.

Connection with the module: LangChain is the implementation you'll use in the capsule 08 project. The concepts from capsules 02-05 (embeddings, indexing, retrieval, documents) map directly to LangChain components.

Setup

pip install langchain langchain-openai langchain-community chromadb pymupdf

from dotenv import load_dotenv
import os

load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY missing in .env"

LangChain Components for RAG

Component map

Manual (what you did)            →  LangChain equivalent
─────────────────────────────────────────────────────────
fitz.open() + get_text()         →  PyMuPDFLoader
chunk_text() / chunk_by_section  →  RecursiveCharacterTextSplitter
OpenAI embeddings.create()       →  OpenAIEmbeddings
chromadb.Client() + add()        →  Chroma.from_documents()
collection.query()               →  vectorstore.as_retriever()
client.chat.completions.create() →  ChatOpenAI + RetrievalQA chain

Every manual component you wrote has a LangChain equivalent. The advantage is that they're interchangeable: changing the vector store, the embedding model, or the LLM requires changing a single line.

Document Loaders

Load PDFs

from langchain_community.document_loaders import PyMuPDFLoader


def load_pdf(pdf_path: str) -> list:
    loader = PyMuPDFLoader(pdf_path)
    documents = loader.load()
    return documents


# docs = load_pdf("technical_manual.pdf")
# print(f"Pages loaded: {len(docs)}")
# print(f"First page ({len(docs[0].page_content)} chars):")
# print(docs[0].page_content[:200])
# print(f"Metadata: {docs[0].metadata}")

Each Document has:

page_content: the page's text
metadata: additional information (source, page, etc.)

Load multiple formats

from langchain_community.document_loaders import (
    PyMuPDFLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
)
from pathlib import Path


def load_document(file_path: str) -> list:
    path = Path(file_path)
    ext = path.suffix.lower()

    loaders = {
        ".pdf": PyMuPDFLoader,
        ".txt": TextLoader,
        ".md": UnstructuredMarkdownLoader,
    }

    loader_class = loaders.get(ext)
    if not loader_class:
        raise ValueError(f"Unsupported format: {ext}")

    loader = loader_class(file_path)
    return loader.load()


def load_directory(dir_path: str, glob_pattern: str = "**/*.pdf") -> list:
    from langchain_community.document_loaders import DirectoryLoader

    loader = DirectoryLoader(
        dir_path,
        glob=glob_pattern,
        loader_cls=PyMuPDFLoader,
        show_progress=True
    )
    return loader.load()

Custom loader for documents with images

LangChain doesn't have a native loader that extracts images from PDFs and generates descriptions. You can create one.

from langchain.schema import Document
from openai import OpenAI
import fitz
import base64

oai_client = OpenAI()


def load_pdf_with_images(
    pdf_path: str,
    describe_images: bool = True
) -> list[Document]:
    doc = fitz.open(pdf_path)
    documents = []

    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text().strip()

        if text:
            documents.append(Document(
                page_content=text,
                metadata={
                    "source": pdf_path,
                    "page": page_num + 1,
                    "type": "text",
                }
            ))

        if describe_images:
            img_list = page.get_images()
            for img_idx, img_ref in enumerate(img_list):
                xref = img_ref[0]
                try:
                    base_image = doc.extract_image(xref)
                    if not base_image or not base_image.get("image"):
                        continue

                    if len(base_image["image"]) < 5000:
                        continue

                    b64 = base64.b64encode(base_image["image"]).decode("utf-8")
                    ext = base_image.get("ext", "png")
                    mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"

                    response = oai_client.chat.completions.create(
                        model="gpt-4o-mini",
                        messages=[{
                            "role": "user",
                            "content": [
                                {"type": "text", "text": "Describe this image in 1-2 sentences for indexing."},
                                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
                            ]
                        }],
                        max_tokens=100
                    )
                    description = response.choices[0].message.content

                    documents.append(Document(
                        page_content=description,
                        metadata={
                            "source": pdf_path,
                            "page": page_num + 1,
                            "type": "image_description",
                            "image_index": img_idx,
                        }
                    ))
                except Exception as e:
                    print(f"Error processing image p.{page_num+1}: {e}")

    doc.close()
    return documents

Text Splitters

RecursiveCharacterTextSplitter

The most-used splitter in LangChain. It splits by paragraphs, then by sentences, then by words, trying to keep chunks of uniform size.

from langchain.text_splitter import RecursiveCharacterTextSplitter


def split_documents(
    documents: list[Document],
    chunk_size: int = 1000,
    chunk_overlap: int = 200
) -> list[Document]:
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
        length_function=len,
    )

    text_docs = [d for d in documents if d.metadata.get("type") != "image_description"]
    image_docs = [d for d in documents if d.metadata.get("type") == "image_description"]

    split_text = text_splitter.split_documents(text_docs)
    all_docs = split_text + image_docs

    return all_docs

MarkdownTextSplitter

For markdown documents (or text with headers), it splits by sections.

from langchain.text_splitter import MarkdownTextSplitter


def split_markdown(
    documents: list[Document],
    chunk_size: int = 1000,
    chunk_overlap: int = 100
) -> list[Document]:
    splitter = MarkdownTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    return splitter.split_documents(documents)

Compare splitters

def compare_splitters(
    documents: list[Document],
    chunk_size: int = 1000
) -> dict:
    recursive = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size, chunk_overlap=200
    )
    markdown = MarkdownTextSplitter(
        chunk_size=chunk_size, chunk_overlap=100
    )

    text_docs = [d for d in documents if d.metadata.get("type") != "image_description"]

    recursive_chunks = recursive.split_documents(text_docs)
    markdown_chunks = markdown.split_documents(text_docs)

    def stats(chunks):
        lengths = [len(c.page_content) for c in chunks]
        return {
            "count": len(chunks),
            "avg_length": round(sum(lengths) / len(lengths)) if lengths else 0,
            "min_length": min(lengths) if lengths else 0,
            "max_length": max(lengths) if lengths else 0,
        }

    return {
        "recursive": stats(recursive_chunks),
        "markdown": stats(markdown_chunks),
    }

Vector Store with Chroma

Create a vector store from documents

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma


def create_vectorstore(
    documents: list[Document],
    persist_directory: str = "./chroma_langchain_db",
    collection_name: str = "multimodal_rag"
) -> Chroma:
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    vectorstore = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory=persist_directory,
        collection_name=collection_name,
    )

    return vectorstore

Load an existing vector store

def load_vectorstore(
    persist_directory: str = "./chroma_langchain_db",
    collection_name: str = "multimodal_rag"
) -> Chroma:
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    vectorstore = Chroma(
        persist_directory=persist_directory,
        collection_name=collection_name,
        embedding_function=embeddings,
    )

    return vectorstore

Add documents to an existing vector store

def add_documents_to_store(
    vectorstore: Chroma,
    new_documents: list[Document]
) -> None:
    vectorstore.add_documents(new_documents)

Retrievers

Basic retriever

def create_retriever(
    vectorstore: Chroma,
    k: int = 5,
    search_type: str = "similarity"
) -> object:
    retriever = vectorstore.as_retriever(
        search_type=search_type,
        search_kwargs={"k": k}
    )
    return retriever

Retriever with metadata filters

def create_filtered_retriever(
    vectorstore: Chroma,
    filter_dict: dict,
    k: int = 5
) -> object:
    retriever = vectorstore.as_retriever(
        search_kwargs={
            "k": k,
            "filter": filter_dict
        }
    )
    return retriever


# text_retriever = create_filtered_retriever(vectorstore, {"type": "text"}, k=5)
# image_retriever = create_filtered_retriever(vectorstore, {"type": "image_description"}, k=3)

Retriever with a score threshold

def create_threshold_retriever(
    vectorstore: Chroma,
    score_threshold: float = 0.7,
    k: int = 10
) -> object:
    retriever = vectorstore.as_retriever(
        search_type="similarity_score_threshold",
        search_kwargs={
            "k": k,
            "score_threshold": score_threshold
        }
    )
    return retriever

Multi-retriever: combine results from multiple retrievers

from langchain.retrievers import EnsembleRetriever


def create_ensemble_retriever(
    vectorstore: Chroma,
    weights: list[float] = None
) -> EnsembleRetriever:
    text_retriever = create_filtered_retriever(
        vectorstore, {"type": "text"}, k=5
    )
    image_retriever = create_filtered_retriever(
        vectorstore, {"type": "image_description"}, k=3
    )

    if weights is None:
        weights = [0.6, 0.4]

    ensemble = EnsembleRetriever(
        retrievers=[text_retriever, image_retriever],
        weights=weights
    )

    return ensemble

RAG Chains

RetrievalQA: the simplest chain

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI


def create_qa_chain(
    vectorstore: Chroma,
    model: str = "gpt-4o",
    k: int = 5
) -> RetrievalQA:
    llm = ChatOpenAI(model=model, temperature=0)
    retriever = vectorstore.as_retriever(search_kwargs={"k": k})

    qa = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
    )

    return qa


# qa = create_qa_chain(vectorstore)
# result = qa.invoke({"query": "How do the microservices connect?"})
# print(result["result"])
# for doc in result["source_documents"]:
#     print(f"  Source: {doc.metadata}")

Chain with a custom prompt

from langchain.prompts import PromptTemplate


RAG_PROMPT = PromptTemplate(
    template=(
        "You are an assistant that answers questions based ONLY on the provided context.\n"
        "The context may include image descriptions marked with type 'image_description'.\n"
        "If you find information from images, mention it in your answer.\n"
        "If you can't find the answer in the context, say you don't have enough information.\n"
        "Cite the sources (page, content type) at the end of your answer.\n\n"
        "Context:\n{context}\n\n"
        "Question: {question}\n\n"
        "Answer:"
    ),
    input_variables=["context", "question"]
)


def create_custom_qa_chain(
    vectorstore: Chroma,
    model: str = "gpt-4o",
    k: int = 5
) -> RetrievalQA:
    llm = ChatOpenAI(model=model, temperature=0)
    retriever = vectorstore.as_retriever(search_kwargs={"k": k})

    qa = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
        chain_type_kwargs={"prompt": RAG_PROMPT},
    )

    return qa

Chain with LCEL (LangChain Expression Language)

LCEL is the modern way to build chains in LangChain. More flexible and composable.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser


def create_lcel_rag_chain(
    vectorstore: Chroma,
    model: str = "gpt-4o",
    k: int = 5
):
    retriever = vectorstore.as_retriever(search_kwargs={"k": k})
    llm = ChatOpenAI(model=model, temperature=0)

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "You are an assistant that answers questions using the provided context. "
            "The context may include text and image descriptions from documents. "
            "Answer precisely and cite the sources."
        )),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])

    def format_docs(docs):
        formatted = []
        for doc in docs:
            doc_type = doc.metadata.get("type", "text")
            page = doc.metadata.get("page", "?")
            prefix = f"[{doc_type.upper()} - p.{page}]"
            formatted.append(f"{prefix}\n{doc.page_content}")
        return "\n\n---\n\n".join(formatted)

    chain = (
        {
            "context": retriever | format_docs,
            "question": RunnablePassthrough(),
        }
        | prompt
        | llm
        | StrOutputParser()
    )

    return chain


# chain = create_lcel_rag_chain(vectorstore)
# answer = chain.invoke("What does the architecture diagram show?")
# print(answer)

Chain with explicit sources

from langchain_core.runnables import RunnableParallel


def create_rag_with_sources(
    vectorstore: Chroma,
    model: str = "gpt-4o",
    k: int = 5
):
    retriever = vectorstore.as_retriever(search_kwargs={"k": k})
    llm = ChatOpenAI(model=model, temperature=0)

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "Answer the question based on the context. "
            "If it includes image descriptions, reference them. "
            "At the end, list the sources used."
        )),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])

    def format_docs(docs):
        parts = []
        for doc in docs:
            page = doc.metadata.get("page", "?")
            doc_type = doc.metadata.get("type", "text")
            parts.append(f"[{doc_type} p.{page}] {doc.page_content}")
        return "\n\n".join(parts)

    def get_sources(docs):
        sources = []
        for doc in docs:
            sources.append({
                "page": doc.metadata.get("page"),
                "type": doc.metadata.get("type"),
                "source": doc.metadata.get("source"),
                "preview": doc.page_content[:100],
            })
        return sources

    chain = RunnableParallel(
        answer=(
            {
                "context": retriever | format_docs,
                "question": RunnablePassthrough(),
            }
            | prompt
            | llm
            | StrOutputParser()
        ),
        sources=retriever | get_sources,
    )

    return chain


# result = chain.invoke("How does authentication work?")
# print(f"Answer: {result['answer']}")
# print(f"Sources: {result['sources']}")

Complete Pipeline with LangChain

End-to-end: PDF → index → query → answer

def build_multimodal_rag(
    pdf_paths: list[str],
    persist_dir: str = "./chroma_langchain_db",
    collection_name: str = "multimodal_rag",
    chunk_size: int = 1000,
    describe_images: bool = True
):
    all_documents = []
    for pdf_path in pdf_paths:
        if describe_images:
            docs = load_pdf_with_images(pdf_path, describe_images=True)
        else:
            docs = load_pdf(pdf_path)
        all_documents.extend(docs)

    split_docs = split_documents(all_documents, chunk_size=chunk_size)

    print(f"Total documents: {len(split_docs)}")
    text_count = sum(1 for d in split_docs if d.metadata.get("type") != "image_description")
    image_count = sum(1 for d in split_docs if d.metadata.get("type") == "image_description")
    print(f"  Text: {text_count}, Images: {image_count}")

    vectorstore = create_vectorstore(
        split_docs,
        persist_directory=persist_dir,
        collection_name=collection_name
    )

    chain = create_rag_with_sources(vectorstore)

    return chain, vectorstore


# chain, vs = build_multimodal_rag(["doc1.pdf", "doc2.pdf"])
# result = chain.invoke("What does the main diagram show?")

Comparison: Manual vs LangChain

Code for the same task

TASK: Index a PDF with images and answer questions

Manual (capsules 02-05):
  - extract_text_chunks_from_pdf()     ~ 25 lines
  - extract_images_from_pdf()          ~ 20 lines
  - describe_image_for_indexing()      ~ 15 lines
  - add_text_chunks()                  ~ 20 lines
  - add_image_descriptions()           ~ 25 lines
  - search_by_text()                   ~ 15 lines
  - hybrid_retrieve()                  ~ 20 lines
  - generate_answer()                  ~ 25 lines
  Total: ~165 lines

LangChain:
  - load_pdf_with_images()             ~ 40 lines (custom loader)
  - split_documents()                  ~ 10 lines
  - Chroma.from_documents()            ~ 3 lines
  - create_lcel_rag_chain()            ~ 20 lines
  Total: ~73 lines

When to use each approach

Criterion	Manual	LangChain
Fine-grained control	Total — you decide every detail	Limited by abstractions
Development speed	Slow — you write everything	Fast — ready-made components
Debugging	Direct — you see every step	More opaque — abstractions hide details
Interchangeability	You rewrite code when swapping components	You change one line
Dependencies	Minimal (openai, chromadb)	LangChain + its dependencies
Production	More predictable, less magic	Faster to iterate
Learning	You understand the fundamentals	You understand the framework

Recommendation

Quick prototype → LangChain
  - You want to test an idea
  - The defaults are enough
  - You're going to iterate a lot

Production with specific requirements → Manual (or LangChain + customization)
  - You need fine-grained cost control
  - The retrieval logic is custom
  - The pipeline has non-standard steps

Learning → Both
  - First manual to understand
  - Then LangChain to be productive

Troubleshooting

Error: "Collection already exists"

When re-running the script, you try to create a vector store that already exists.

vectorstore = load_vectorstore(persist_dir, collection_name)
# If you need to recreate:
# import shutil
# shutil.rmtree(persist_dir)
# vectorstore = create_vectorstore(docs, persist_dir, collection_name)

The retriever doesn't find image documents

Verify that the image documents were indexed correctly.

def debug_vectorstore(vectorstore: Chroma) -> dict:
    collection = vectorstore._collection
    all_docs = collection.get(include=["metadatas"])

    types = {}
    for meta in all_docs["metadatas"]:
        t = meta.get("type", "unknown")
        types[t] = types.get(t, 0) + 1

    return {"total": len(all_docs["ids"]), "by_type": types}

LangChain deprecation warnings

LangChain evolves quickly. Use the current imports.

# OLD (deprecated)
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings

# NEW
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

The chain doesn't return sources

Make sure to use return_source_documents=True in RetrievalQA or build the chain with explicit sources using LCEL.

Excessive memory with large documents

def process_in_batches(
    documents: list[Document],
    vectorstore: Chroma,
    batch_size: int = 100
) -> None:
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        vectorstore.add_documents(batch)
        print(f"Batch {i//batch_size + 1}: {len(batch)} docs indexed")

Exercises

Exercise 1: Retriever with a dynamic filter

Create a retriever that filters by content type based on the query. If the query mentions "diagram", "image" or "figure", search only in image_description. Otherwise, search everything.

See solution

def smart_retriever(
    vectorstore: Chroma,
    query: str,
    k: int = 5
) -> list[Document]:
    visual_keywords = {"diagram", "image", "figure", "chart", "table", "screenshot", "photo"}
    query_words = set(query.lower().split())

    if query_words & visual_keywords:
        retriever = create_filtered_retriever(
            vectorstore, {"type": "image_description"}, k=k
        )
    else:
        retriever = vectorstore.as_retriever(search_kwargs={"k": k})

    return retriever.invoke(query)


# docs = smart_retriever(vectorstore, "show the architecture diagram")
# for doc in docs:
#     print(f"  [{doc.metadata.get('type')}] {doc.page_content[:60]}...")

Exercise 2: RAG with conversation history

Extend the chain so it keeps history and can handle follow-up questions.

See solution

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain


def create_conversational_rag(
    vectorstore: Chroma,
    model: str = "gpt-4o",
    k: int = 5
):
    llm = ChatOpenAI(model=model, temperature=0)
    retriever = vectorstore.as_retriever(search_kwargs={"k": k})

    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True,
        output_key="answer"
    )

    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        return_source_documents=True,
    )

    return chain


# chain = create_conversational_rag(vectorstore)
# r1 = chain.invoke({"question": "What is the API Gateway?"})
# print(r1["answer"])
# r2 = chain.invoke({"question": "How does it connect to the database?"})
# print(r2["answer"])

Exercise 3: Compare manual vs LangChain results

Given the same query, run the manual retrieval (capsule 04) and the LangChain retrieval. Compare which documents each one retrieves.

See solution

def compare_retrieval(
    manual_collection,
    langchain_vectorstore: Chroma,
    query: str,
    k: int = 5
) -> dict:
    manual_results = manual_collection.query(
        query_texts=[query],
        n_results=k
    )
    manual_ids = manual_results["ids"][0]

    lc_retriever = langchain_vectorstore.as_retriever(search_kwargs={"k": k})
    lc_docs = lc_retriever.invoke(query)
    lc_previews = [d.page_content[:80] for d in lc_docs]

    print(f"Query: {query}\n")
    print("Manual retrieval:")
    for i, (doc_id, doc_text) in enumerate(zip(manual_ids, manual_results["documents"][0])):
        print(f"  {i+1}. {doc_id}: {doc_text[:60]}...")

    print("\nLangChain retrieval:")
    for i, (doc, preview) in enumerate(zip(lc_docs, lc_previews)):
        print(f"  {i+1}. [{doc.metadata.get('type')}] {preview}...")

    return {
        "manual_count": len(manual_ids),
        "langchain_count": len(lc_docs),
    }

Summary

LangChain abstracts the RAG pipeline into interchangeable components: loaders, splitters, vector stores, retrievers, chains.
Document loaders load PDFs, markdown, text. For images, you need a custom loader.
Text splitters split documents into chunks. RecursiveCharacterTextSplitter is the most versatile.
Chroma integrates natively with LangChain to create persistent vector stores.
Retrievers search for relevant documents. You can filter by metadata, use score thresholds, or combine multiple retrievers with EnsembleRetriever.
LCEL is the modern way to build chains: composable, typed, and with streaming support.
Manual vs LangChain: manual gives total control, LangChain gives development speed. Both are valid.
For production, LangChain with customization (custom loaders, specific prompts) is usually the sweet spot.

Additional Resources

LangChain RAG Tutorial — Official RAG tutorial
LangChain Document Loaders — All available loaders
LangChain Retrievers — Types of retrievers
Chroma + LangChain — ChromaDB integration
LCEL (LangChain Expression Language) — LCEL guide

← Previous5. Documents with Images Next →7. Limitations and Optimization