Complete Unstructured Tutorial: Document Preprocessing for RAG and LLM Applications

When building Retrieval-Augmented Generation (RAG) applications, one of the biggest challenges isn't the LLM model itself — it's the data preprocessing stage. Business documents come in various formats: PDFs, Word documents, PowerPoints, HTML pages, images, and emails. Accurately extracting text from these formats, preserving document structure, and preparing them for embedding is a complex process. This is where Unstructured comes in.

Unstructured is an open-source Python library specifically designed for extracting, cleaning, and processing data from unstructured documents. It supports over 25 file formats and provides ready-to-use preprocessing pipelines for LLM and RAG applications.

In this tutorial, we'll learn how to install, use, and optimize Unstructured for various document preprocessing scenarios.

Why Unstructured?

Before we begin, let's understand why Unstructured has become the go-to choice for many developers:

Installation

Basic Installation

For a minimal installation that supports basic text formats:

pip install unstructured

Installation with Additional Dependencies

To support a wider range of document formats, install with the appropriate extras:

# For PDF processing
pip install "unstructured[pdf]"

For all office document formats
pip install "unstructured[docx,pptx,xlsx]"

For all supported formats
pip install "unstructured[all-docs]"

For local inference (OCR, table detection, layout analysis)
pip install "unstructured[local-inference]"

System Dependencies

Some features require additional system dependencies:

# Ubuntu/Debian
sudo apt-get install -y libmagic-dev poppler-utils tesseract-ocr libreoffice pandoc

macOS
brew install libmagic poppler tesseract libreoffice pandoc

Verify Installation

import unstructured
print(f"Unstructured version: {unstructured.version}")

from unstructured.partition.auto import partition
print("Partition module loaded successfully!")

Basic Usage

Processing Files with Auto-Partition

The simplest and most powerful feature of Unstructured is the partition() function, which automatically detects file formats and extracts content:

from unstructured.partition.auto import partition

Process a PDF file
elements = partition(filename="document.pdf")

View extracted elements
for element in elements:
    print(f"Type: {type(element).name}")
    print(f"Text: {element.text[:100]}...")
    print(f"Metadata: {element.metadata}")
    print("---")

Understanding Element Types

Unstructured extracts documents into structured elements. Each element has a type indicating its role in the document:

from unstructured.partition.auto import partition

elements = partition(filename="report.pdf")

Count element type distribution
from collections import Counter
typecounts = Counter(type(el).name for el in elements)

for elementtype, count in typecounts.mostcommon():
    print(f"{elementtype}: {count}")

from unstructured.partition.auto import partition
from unstructured.partition.pdf import partitionpdf
from unstructured.partition.docx import partitiondocx

from unstructured.partition.html import partitionhtml
from unstructured.partition.pptx import partitionpptx


PDF
pdfelements = partitionpdf(filename="report.pdf")


Word Document
docxelements = partitiondocx(filename="document.docx")


HTML
htmlelements = partitionhtml(filename="page.html")


Or directly from a URL
htmlelements = partitionhtml(url="https://example.com/article")


PowerPoint
pptxelements = partitionpptx(filename="presentation.pptx")


Markdown
from unstructured.partition.md import partitionmd
mdelements = partitionmd(filename="README.md")

Accessing Metadata

Each element comes with rich metadata:

from unstructured.partition.auto import partition

elements = partition(filename="document.pdf")

for element in elements[:5]:
    meta = element.metadata
    print(f"Text: {element.text[:50]}...")
    print(f"  Page Number: {meta.pagenumber}")

    print(f"  Filename: {meta.filename}")
    print(f"  File Directory: {meta.filedirectory}")
    print(f"  Coordinates: {meta.coordinates}")
    print(f"  Category: {element.category}")
    print()

Advanced Usage

PDF Partition Strategies

Unstructured offers several strategies for processing PDFs, each with trade-offs between speed and accuracy:

from unstructured.partition.pdf import partitionpdf

Strategy: fast - uses pdfminer only (fast, less accurate for complex layouts)
elementsfast = partitionpdf(

    filename="document.pdf",
    strategy="fast"
)

Strategy: hires - uses layout detection model (slow, most accurate)
elementshires = partitionpdf(
    filename="document.pdf",
    strategy="hires",

    modelname="yolox"
)

Strategy: ocronly - OCR only (for scanned PDFs)

elementsocr = partitionpdf(

    filename="scanneddocument.pdf",
    strategy="ocronly"

)

Strategy: auto (default) - automatically selects the best strategy
elementsauto = partitionpdf(

    filename="document.pdf",
    strategy="auto"
)

from unstructured.partition.pdf import partitionpdf

elements = partitionpdf(

    filename="financialreport.pdf",
    strategy="hires",

    infertablestructure=True

)

Filter only table elements
tables = [el for el in elements if el.category == "Table"]

for i, table in enumerate(tables):
    print(f"\nTable {i+1}:")
    print(table.text)

    # Access HTML representation of the table
    if table.metadata.textashtml:

        print(f"\nHTML:\n{table.metadata.textashtml}")

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunkbytitle

from unstructured.chunking.basic import chunkelements

elements = partition(filename="longdocument.pdf")


Chunk by title (preserves semantic context)
chunks = chunkbytitle(

    elements,
    maxcharacters=1500,
    newafternchars=1000,

    combinetextundernchars=200

)

print(f"Original elements: {len(elements)}")
print(f"After chunking: {len(chunks)}")

for i, chunk in enumerate(chunks[:3]):
    print(f"\nChunk {i+1} ({type(chunk).name}):")
    print(f"  Length: {len(chunk.text)} chars")
    print(f"  Text: {chunk.text[:200]}...")

• maxcharacters: Maximum size of each chunk
• newafternchars: Start a new chunk after N characters (soft limit)
• combinetextundernchars: Combine small elements under N characters
• overlap: Number of overlapping characters between chunks
• overlapall: Apply overlap to all chunks, not just adjacent ones

# Chunking with overlap for better context
chunkswithoverlap = chunkbytitle(
    elements,
    maxcharacters=1500,

    newafternchars=1000,
    overlap=200,
    overlapall=False

)

from unstructured.cleaners.core import (
    clean,
    cleanextrawhitespace,

    cleannonasciichars,
    cleanbullets,

    cleandashes,
    replaceunicodequotes,
    groupbrokenparagraphs
)

rawtext = "  This   has    extra   spaces   and\n\nbroken\nparagraphs  "


Clean excess whitespace
cleaned = cleanextrawhitespace(rawtext)
print(cleaned)

Comprehensive cleaning
cleaned = clean(
    rawtext,

    extrawhitespace=True,
    dashes=True,
    bullets=True,
    trailingpunctuation=False

)
print(cleaned)

Join broken paragraphs
textwithbreaks = "This is a sentence that was\nbroken across multiple\nlines in the PDF."

fixed = groupbrokenparagraphs(textwithbreaks)

print(fixed)

from unstructured.partition.auto import partition
from unstructured.staging.base import elementstojson, elementsfromjson

from unstructured.staging.base import converttodataframe


elements = partition(filename="document.pdf")

Export to JSON
elementstojson(elements, filename="output.json")


Import from JSON
loadedelements = elementsfromjson(filename="output.json")

Convert to pandas DataFrame
df = converttodataframe(elements)
print(df.head())
print(df.columns.tolist())

Integration with Vector Stores

Here's a complete pipeline from document to vector store:

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunkbytitle
from unstructured.embed.openai import OpenAIEmbeddingEncoder

Step 1: Partition the document
elements = partition(filename="knowledgebase.pdf", strategy="hires")

Step 2: Chunk the document
chunks = chunkbytitle(
    elements,
    maxcharacters=1500,

    newafternchars=1000,
    overlap=150
)

Step 3: Generate embeddings
embeddingencoder = OpenAIEmbeddingEncoder(

    apikey="your-openai-api-key",
    modelname="text-embedding-3-small"

)

embeddedchunks = embeddingencoder.embeddocuments(chunks)

Step 4: Store in ChromaDB
import chromadb

client = chromadb.Client()
collection = client.createcollection("documents")


for i, chunk in enumerate(embeddedchunks):
    collection.add(
        ids=[f"chunk{i}"],

        embeddings=[chunk.embeddings],
        documents=[chunk.text],
        metadatas=[{
            "pagenumber": str(chunk.metadata.pagenumber or ""),

            "filename": chunk.metadata.filename or "",
            "category": chunk.category
        }]
    )

print(f"Indexed {len(embeddedchunks)} chunks to ChromaDB")

Batch Document Processing

To process multiple documents at once:

import os
from pathlib import Path
from unstructured.partition.auto import partition
from unstructured.chunking.title import chunkbytitle

def processdirectory(inputdir, outputdir):

    """Process all documents in a directory."""
    inputpath = Path(inputdir)

    outputpath = Path(outputdir)

    outputpath.mkdir(parents=True, existok=True)


    supportedextensions = {
        '.pdf', '.docx', '.pptx', '.html', '.htm',
        '.txt', '.md', '.csv', '.xlsx', '.eml', '.msg'
    }

    allchunks = []


    for filepath in inputpath.rglob(""):

        if filepath.suffix.lower() not in supportedextensions:

            continue

        print(f"Processing: {filepath.name}")

        try:
            elements = partition(filename=str(filepath))


            chunks = chunkbytitle(

                elements,
                maxcharacters=1500,
                newafternchars=1000

            )

            allchunks.extend(chunks)
            print(f"  Extracted {len(chunks)} chunks")

        except Exception as e:
            print(f"  Error processing {filepath.name}: {e}")


    print(f"\nTotal chunks from all documents: {len(allchunks)}")
    return allchunks


chunks = processdirectory("./documents", "./output")

# Install ingest dependencies
pip install "unstructured-ingest[s3,pinecone]"

unstructured-ingest \
  local \
    --input-path ./documents \
    --partition-by-api \
    --api-key $UNSTRUCTUREDAPIKEY \
  pinecone \
    --api-key $PINECONEAPIKEY \
    --index-name my-index \
    --environment us-east1-gcp

unstructured-ingest \
  s3 \
    --remote-url s3://my-bucket/documents/ \
    --anonymous \
  local \
    --output-dir ./output

from unstructuredingest.v2.pipeline.pipeline import Pipeline
from unstructuredingest.v2.interfaces import ProcessorConfig
from unstructuredingest.v2.processes.connectors.local import (

    LocalIndexerConfig,
    LocalDownloaderConfig,
    LocalConnectionConfig,
)
from unstructuredingest.v2.processes.partitioner import PartitionerConfig
from unstructuredingest.v2.processes.chunker import ChunkerConfig


pipeline = Pipeline.fromconfigs(
    context=ProcessorConfig(),
    indexerconfig=LocalIndexerConfig(inputpath="./documents"),
    downloaderconfig=LocalDownloaderConfig(),

    sourceconnectionconfig=LocalConnectionConfig(),

    partitionerconfig=PartitionerConfig(
        partitionbyapi=False,
        strategy="hires"

    ),
    chunkerconfig=ChunkerConfig(
        chunkingstrategy="bytitle",
        chunkmaxcharacters=1500,
        chunknewafternchars=1000
    )
)

pipeline.run()

# Plain text documents -> fast
elements = partition(filename="textdoc.pdf", strategy="fast")


Documents with complex layouts -> hires
elements = partition(filename="annualreport.pdf", strategy="hires")

Scanned documents/images -> ocronly

elements = partition(filename="scanned.pdf", strategy="ocronly")

# For Q&A requiring detailed context
chunks = chunkbytitle(elements, maxcharacters=500, overlap=100)


For summarization requiring broad context
chunks = chunkbytitle(elements, maxcharacters=2000, overlap=200)

For search requiring balance
chunks = chunkbytitle(elements, maxcharacters=1000, overlap=150)

from unstructured.partition.auto import partition
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

def safepartition(filename, kwargs):

    """Partition with error handling."""
    try:
        elements = partition(filename=filename, kwargs)
        if not elements:
            logger.warning(f"No elements extracted from {filename}")
        return elements
    except FileNotFoundError:
        logger.error(f"File not found: {filename}")
        return []
    except Exception as e:
        logger.error(f"Error processing {filename}: {e}")
        # Fallback to simpler strategy
        try:
            return partition(filename=filename, strategy="fast")
        except Exception:
            return []

from unstructured.partition.auto import partition

elements = partition(filename="contract.pdf")

Add custom metadata
for element in elements:
    element.metadata.categorydepth = 0


    # Add source information
    element.metadata.datasource = {
        "sourcetype": "contract",

        "department": "legal",
        "year": 2026
    }

import hashlib
import json
import os
from unstructured.partition.auto import partition
from unstructured.staging.base import elementstojson, elementsfromjson


def partitionwithcache(filename, cachedir="./cache", kwargs):

    """Partition with caching to avoid reprocessing."""
    os.makedirs(cachedir, existok=True)

    # Generate cache key from file hash
    with open(filename, "rb") as f:
        filehash = hashlib.md5(f.read()).hexdigest()


    cachefile = os.path.join(cachedir, f"{filehash}.json")

    if os.path.exists(cachefile):

        print(f"Loading from cache: {cachefile}")
        return elementsfromjson(filename=cachefile)


    print(f"Processing: {filename}")
    elements = partition(filename=filename, kwargs)

    # Save to cache
    elementstojson(elements, filename=cachefile)

    return elements

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunkbytitle
from unstructured.cleaners.core import cleanextrawhitespace
from unstructured.documents.elements import NarrativeText, Title, Table

class DocumentProcessor:
    def init(self, chunksize=1500, chunkoverlap=150):
        self.chunksize = chunksize
        self.chunkoverlap = chunkoverlap

    def process(self, filename):
        """Main document preprocessing pipeline."""
        # 1. Partition
        elements = partition(
            filename=filename,
            strategy="auto",
            includepagebreaks=False
        )

        # 2. Filter irrelevant elements
        elements = self.filterelements(elements)

        # 3. Clean text
        elements = self.cleanelements(elements)

        # 4. Chunk
        chunks = chunkbytitle(
            elements,
            maxcharacters=self.chunksize,
            newafternchars=int(self.chunksize  0.7),
            overlap=self.chunkoverlap

        )

        # 5. Prepare output
        return self.prepareoutput(chunks)


    def filterelements(self, elements):

        """Remove unhelpful elements."""
        return [
            el for el in elements
            if el.text and len(el.text.strip()) > 10
            and el.category not in ("Header", "Footer", "PageBreak")
        ]

    def cleanelements(self, elements):

        """Clean text in each element."""
        for el in elements:
            el.apply(cleanextrawhitespace)

        return elements

    def prepareoutput(self, chunks):

        """Format output for vector store."""
        return [
            {
                "text": chunk.text,
                "metadata": {
                    "pagenumber": chunk.metadata.pagenumber,

                    "filename": chunk.metadata.filename,
                    "category": chunk.category,
                    "charcount": len(chunk.text)
                }
            }
            for chunk in chunks
        ]

Usage
processor = DocumentProcessor(chunksize=1500, chunkoverlap=150)
results = processor.process("annualreport.pdf")


for r in results[:3]:
    print(f"[{r['metadata']['category']}] {r['text'][:100]}...")

• Use partition() for auto-detection of file formats, or specific partition functions for more control
• Choose the right partition strategy for your documents: fast for simple text, hires for complex layouts, ocronly for scanned documents
• Leverage chunkby_title() for semantic chunking that preserves document context
• Use cleaning functions to preprocess text before embedding
• Implement caching to avoid reprocessing the same documents
• Consider the Unstructured Ingest CLI for more complex batch processing pipelines

By mastering Unstructured, you can build robust, production-ready document preprocessing pipelines for your RAG and LLM applications.