Document loaders and text splitters

Loaders for PDF, HTML, CSV, Markdown and directories, keeping metadata intact, and choosing a chunk size and overlap you can defend.

Loading documents

from langchain_community.document_loaders import (
    PyPDFLoader, CSVLoader, UnstructuredMarkdownLoader,
    WebBaseLoader, DirectoryLoader, TextLoader,
)

# a PDF per page: page number becomes metadata, which is what citations need
pdf_docs = PyPDFLoader("handbook.pdf").load()
print(pdf_docs[0].metadata)
# {'source': 'handbook.pdf', 'page': 0, ...}

# every file in a tree, dispatched by extension, with parallel workers
loader = DirectoryLoader(
    "docs/",
    glob="**/*.md",
    loader_cls=UnstructuredMarkdownLoader,
    loader_kwargs={"mode": "single"},
    use_multithreading=True,
    show_progress=True,
)
md_docs = loader.load()

# HTML: strips tags but keeps the page title and URL
web_docs = WebBaseLoader(["https://example.com/pricing"]).load()

# CSV: one document per row, columns preserved in metadata
row_docs = CSVLoader("tickets.csv", source_column="ticket_id").load()
print(row_docs[0].page_content[:80], row_docs[0].metadata)

• Metadata is the part you will regret losing. Source path, page number, heading, and an ingestion timestamp should be on every document before it reaches a splitter.
• A loader that silently returns zero documents is the most common ingestion bug. Assert a non-zero count and log the total before indexing.
• PDF text extraction is unreliable for scanned pages, multi-column layouts and tables. Check a sample of extracted text by hand before trusting a whole corpus.
• Load lazily (lazy_load) for large corpora. Loading ten thousand documents into a list costs memory you probably do not need.

Splitting text

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter, TokenTextSplitter
)

# structure-first: split on markdown headings, then on size
headers = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
)
sections = headers.split_text(markdown_text)
print(sections[0].metadata)     # {'h1': 'Guide', 'h2': 'Installation'}

size_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=150,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = size_splitter.split_documents(sections)
print(len(chunks), [len(c.page_content) for c in chunks[:5]])

# token-based sizing when the model's context is the constraint
token_splitter = TokenTextSplitter(chunk_size=256, chunk_overlap=32)

# prepend the heading path so the chunk is self-describing after retrieval
for chunk in chunks:
    path = " > ".join(chunk.metadata.get(k, "") for k in ("h1", "h2") if chunk.metadata.get(k))
    chunk.page_content = f"[{path}]\n{chunk.page_content}" if path else chunk.page_content

print(len(chunks), "chunks prepared")

Chunk size is a retrieval parameter, not a formatting preference. Small chunks match precise questions; large chunks carry the context an answer needs. The only way to choose is to measure recall on questions you actually expect.

Ingestion quality checks

import hashlib
from collections import Counter

def stable_id(doc):
    key = doc.metadata.get("source", "") + str(doc.metadata.get("page", "")) + doc.page_content[:64]
    return hashlib.sha1(key.encode("utf-8")).hexdigest()[:16]

def audit(docs):
    counts = Counter()
    problems = []
    for doc in docs:
        text = doc.page_content
        if len(text.strip()) < 60:
            problems.append(("too_short", doc.metadata))
        elif len(text) > 4000 and not text.count("\n"):
            problems.append(("likely_extraction_failure", doc.metadata))
        if text.count(" ") / max(len(text), 1) < 0.05:
            problems.append(("no_word_boundaries", doc.metadata))
        counts["total"] += 1
    return {"counts": dict(counts), "problems": problems[:20]}

report = audit(chunks)
print(report["counts"], len(report["problems"]), "flagged")

# deduplicate by content hash before indexing
seen, unique = set(), []
for chunk in chunks:
    digest = hashlib.sha256(chunk.page_content.encode("utf-8")).hexdigest()
    if digest not in seen:
        seen.add(digest)
        unique.append(chunk)
print(len(chunks), "->", len(unique), "after deduplication")

• Flag chunks that are too short, too long without line breaks, or have no word boundaries. Those three checks catch most extraction and encoding failures.
• Deduplicate on a content hash before embedding. Duplicated chunks crowd out diverse results at query time and waste embedding cost.
• Keep a stable chunk id derived from source and position. Re-ingesting an updated document then becomes a delete-and-insert rather than a duplicate-insert.
• Re-run the audit after every loader or splitter change. An ingestion pipeline degrades silently, and the first symptom is a worse answer weeks later.

FAQ

Related

Retrieval-augmented generation Embeddings and vector stores

Last refreshed 2026-09-18.