Build a LangChain Document Q&A System (PDF, HTML, DOCX)

pip install langchain langchain-openai langchain-community
pip install pypdf unstructured python-docx
pip install faiss-cpu  # or chromadb
pip install beautifulsoup4 lxml

# For table extraction from PDFs
pip install pdfplumber

# For better unstructured document processing
pip install "unstructured[all-docs]"

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("documents/annual_report_2025.pdf")
pages = loader.load()

print(f"Loaded {len(pages)} pages")
print(f"First page metadata: {pages[0].metadata}")
# {'source': 'documents/annual_report_2025.pdf', 'page': 0}

print(f"First 500 chars: {pages[0].page_content[:500]}")

# PyPDFLoader loads per-page — each Document is one page
# Good for: standard text PDFs, preserves page numbers in metadata
# Bad for: scanned PDFs (no OCR), complex tables

from langchain_community.document_loaders import PDFPlumberLoader

loader = PDFPlumberLoader("documents/financial_report.pdf")
docs = loader.load()

# PDFPlumber preserves table structure better
# Tables are extracted as text with spacing preserved
print(docs[0].page_content[:1000])

# For explicit table extraction:
import pdfplumber

def extract_tables_from_pdf(pdf_path: str) -> list:
    """Extract tables as list of lists."""
    tables = []
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            page_tables = page.extract_tables()
            for table in page_tables:
                tables.append({
                    "page": page_num + 1,
                    "data": table
                })
    return tables

tables = extract_tables_from_pdf("documents/financial_report.pdf")
for t in tables:
    print(f"Page {t['page']}: {len(t['data'])} rows x {len(t['data'][0])} cols")

from langchain_community.document_loaders import UnstructuredPDFLoader

# Basic mode
loader = UnstructuredPDFLoader(
    "documents/complex_layout.pdf",
    mode="elements"  # "single" or "elements" or "paged"
)
elements = loader.load()

# "elements" mode gives you granular control — each text element is a separate Document
for elem in elements[:5]:
    print(f"Type: {elem.metadata.get('category', 'unknown')}")
    print(f"Content: {elem.page_content[:100]}")
    print("---")

# For scanned PDFs (requires tesseract)
# pip install pytesseract pillow
loader_ocr = UnstructuredPDFLoader(
    "documents/scanned_contract.pdf",
    strategy="hi_res",  # enables OCR
    languages=["eng"]
)
docs_ocr = loader_ocr.load()

from langchain_community.document_loaders import UnstructuredHTMLLoader, BSHTMLLoader

# BSHTMLLoader — faster, uses BeautifulSoup
loader = BSHTMLLoader("documents/product_manual.html")
docs = loader.load()
print(f"Loaded {len(docs)} documents from HTML")
print(docs[0].page_content[:500])

# UnstructuredHTMLLoader — better structure preservation
unstructured_loader = UnstructuredHTMLLoader("documents/complex_page.html")
docs = unstructured_loader.load()

# Loading from URL
from langchain_community.document_loaders import WebBaseLoader

web_loader = WebBaseLoader(
    web_paths=["https://docs.langchain.com/docs/"],
    bs_kwargs={
        "parse_only": None,  # parse everything
    }
)
web_docs = web_loader.load()
print(f"Loaded {len(web_docs)} web pages")

# Multiple URLs
multi_loader = WebBaseLoader([
    "https://example.com/page1",
    "https://example.com/page2"
])

from langchain_community.document_loaders import UnstructuredWordDocumentLoader, Docx2txtLoader

# Docx2txt — simple and reliable
loader = Docx2txtLoader("documents/meeting_notes.docx")
docs = loader.load()
print(docs[0].page_content[:500])

# Unstructured — preserves more structure (tables, headers, etc.)
unstructured_loader = UnstructuredWordDocumentLoader(
    "documents/contract.docx",
    mode="elements"
)
elements = unstructured_loader.load()

# Check element types
categories = set(e.metadata.get("category") for e in elements)
print("Element types found:", categories)
# {'Title', 'NarrativeText', 'Table', 'ListItem', 'Header'}

# Process different element types differently
tables = [e for e in elements if e.metadata.get("category") == "Table"]
headings = [e for e in elements if e.metadata.get("category") == "Title"]
print(f"Found {len(tables)} tables, {len(headings)} headings")

import os
from pathlib import Path
from typing import List
from langchain_core.documents import Document

def load_documents_from_directory(directory: str) -> List[Document]:
    """Load all supported document types from a directory."""
    from langchain_community.document_loaders import (
        PyPDFLoader, Docx2txtLoader, BSHTMLLoader,
        TextLoader, CSVLoader
    )
    
    loaders = {
        ".pdf": PyPDFLoader,
        ".docx": Docx2txtLoader,
        ".html": BSHTMLLoader,
        ".txt": TextLoader,
        ".csv": CSVLoader
    }
    
    all_docs = []
    directory_path = Path(directory)
    
    for file_path in directory_path.rglob("*"):
        if file_path.is_file():
            suffix = file_path.suffix.lower()
            if suffix in loaders:
                try:
                    loader_class = loaders[suffix]
                    loader = loader_class(str(file_path))
                    docs = loader.load()
                    
                    # Add filename to metadata
                    for doc in docs:
                        doc.metadata["filename"] = file_path.name
                        doc.metadata["file_type"] = suffix
                    
                    all_docs.extend(docs)
                    print(f"Loaded {len(docs)} docs from {file_path.name}")
                    
                except Exception as e:
                    print(f"Failed to load {file_path.name}: {e}")
    
    return all_docs

docs = load_documents_from_directory("./documents")
print(f"Total documents loaded: {len(docs)}")

from langchain.text_splitter import RecursiveCharacterTextSplitter

# RecursiveCharacterTextSplitter is the right default for most use cases
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,       # characters per chunk
    chunk_overlap=150,    # overlap to preserve context at boundaries
    length_function=len,
    separators=[
        "\n\n",    # Try paragraph breaks first
        "\n",      # Then line breaks
        ". ",      # Then sentence boundaries
        " ",       # Then words
        ""         # Finally characters
    ]
)

chunks = splitter.split_documents(docs)
print(f"Split {len(docs)} documents into {len(chunks)} chunks")
print(f"Average chunk size: {sum(len(c.page_content) for c in chunks) / len(chunks):.0f} chars")

# Inspect a chunk
print("\nSample chunk:")
print(chunks[10].page_content)
print("\nMetadata:", chunks[10].metadata)

from langchain.text_splitter import MarkdownTextSplitter, PythonCodeTextSplitter

# Markdown-aware splitting
md_splitter = MarkdownTextSplitter(chunk_size=1000, chunk_overlap=100)
md_chunks = md_splitter.create_documents([markdown_content])

# Python code splitting
code_splitter = PythonCodeTextSplitter(chunk_size=500, chunk_overlap=50)
code_chunks = code_splitter.create_documents([python_code])

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS, Chroma
import os

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

# FAISS — good for local development and medium-sized collections
vectorstore = FAISS.from_documents(
    documents=chunks,
    embedding=embeddings
)

# Save and reload
vectorstore.save_local("faiss_index")
# Later: vectorstore = FAISS.load_local("faiss_index", embeddings)

# Chroma — persistent, good for larger collections
vectorstore_chroma = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="chroma_db",
    collection_name="document_qa"
)

# Create retriever
retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximal Marginal Relevance — reduces redundancy
    search_kwargs={
        "k": 6,                # number of chunks to retrieve
        "fetch_k": 20,         # candidates before MMR filtering
        "lambda_mult": 0.5     # diversity vs relevance balance
    }
)

# Test retrieval
test_docs = retriever.invoke("What was the revenue in Q3 2025?")
for doc in test_docs:
    print(f"Source: {doc.metadata.get('source', 'unknown')}, Page: {doc.metadata.get('page', 'N/A')}")
    print(f"Content: {doc.page_content[:200]}\n")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Prompt with explicit instructions for source-grounded answers
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an expert at answering questions based on provided documents.
    
Answer the question using ONLY the information from the context below.
If the answer isn't in the context, say "I don't have enough information in the provided documents to answer this."

Always cite your sources by mentioning the document name and page number when available.
Be precise and factual."""),
    ("human", """Context:
{context}

Question: {question}""")
])

def format_docs(docs):
    """Format retrieved documents with source attribution."""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("filename", doc.metadata.get("source", "unknown"))
        page = doc.metadata.get("page", "N/A")
        formatted.append(f"[Source {i}: {source}, Page {page}]\n{doc.page_content}")
    return "\n\n".join(formatted)

# Build the complete RAG chain
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | qa_prompt
    | llm
    | StrOutputParser()
)

# Query the system
answer = rag_chain.invoke("What were the main findings of the Q3 2025 report?")
print(answer)

from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# Contextualize the question given chat history
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", """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."""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_prompt
)

# QA prompt with history
qa_with_history_prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer the question based only on the context provided.
    If you don't know the answer, say so.
    
Context: {context}"""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

question_answer_chain = create_stuff_documents_chain(llm, qa_with_history_prompt)
rag_chain_with_history = create_retrieval_chain(
    history_aware_retriever,
    question_answer_chain
)

# Use with conversation history
chat_history = []

def chat(question: str) -> str:
    result = rag_chain_with_history.invoke({
        "input": question,
        "chat_history": chat_history
    })
    
    chat_history.extend([
        HumanMessage(content=question),
        AIMessage(content=result["answer"])
    ])
    
    return result["answer"]

# Multi-turn conversation
print(chat("What is the total revenue mentioned in the documents?"))
print(chat("What about the expenses?"))  # "expenses" understood in context of revenue
print(chat("Calculate the profit margin based on those numbers."))

# Metadata filtering for large document collections
from langchain_community.vectorstores import Chroma

# Store with rich metadata
chunks_with_metadata = []
for chunk in chunks:
    chunk.metadata.update({
        "department": "finance",
        "year": 2025,
        "document_type": "annual_report"
    })
    chunks_with_metadata.append(chunk)

vectorstore = Chroma.from_documents(chunks_with_metadata, embeddings)

# Filter retrieval by metadata
filtered_retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {
            "department": "finance",
            "year": 2025
        }
    }
)