Tutorial Lengkap Unstructured: Preprocessing Dokumen untuk RAG dan Aplikasi LLM

Dalam membangun aplikasi Retrieval-Augmented Generation (RAG), salah satu tantangan terbesar bukan pada model LLM-nya, melainkan pada tahap preprocessing data. Dokumen bisnis hadir dalam berbagai format — PDF, Word, PowerPoint, HTML, gambar, dan email. Mengekstrak teks dari format-format ini secara akurat, mempertahankan struktur dokumen, dan menyiapkannya untuk embedding adalah proses yang kompleks. Di sinilah Unstructured hadir sebagai solusi.

Unstructured adalah library open-source Python yang dirancang khusus untuk mengekstrak, membersihkan, dan memproses data dari dokumen tidak terstruktur. Library ini mendukung lebih dari 25 format file dan menyediakan pipeline preprocessing yang siap digunakan untuk aplikasi LLM dan RAG.

Dalam tutorial ini, kita akan mempelajari cara menginstal, menggunakan, dan mengoptimalkan Unstructured untuk berbagai skenario preprocessing dokumen.

Mengapa Unstructured?

Sebelum kita mulai, mari pahami mengapa Unstructured menjadi pilihan utama banyak developer:

Instalasi

Instalasi Dasar

Untuk instalasi minimal yang mendukung format teks dasar:

pip install unstructured

Instalasi dengan Dependensi Tambahan

Untuk mendukung format dokumen yang lebih beragam, instal dengan extras yang sesuai:

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

Untuk semua format dokumen office
pip install "unstructured[docx,pptx,xlsx]"

Untuk semua format yang didukung
pip install "unstructured[all-docs]"

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

Instalasi Dependensi Sistem

Beberapa fitur memerlukan dependensi sistem tambahan:

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

macOS
brew install libmagic poppler tesseract libreoffice pandoc

Verifikasi Instalasi

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

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

Penggunaan Dasar

Memproses File dengan Auto-Partition

Fitur paling sederhana dan powerful dari Unstructured adalah fungsi partition() yang secara otomatis mendeteksi format file dan mengekstraknya:

from unstructured.partition.auto import partition

Proses file PDF
elements = partition(filename="dokumen.pdf")

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

Memahami Element Types

Unstructured mengekstrak dokumen menjadi elemen-elemen terstruktur. Setiap elemen memiliki tipe yang menunjukkan perannya dalam dokumen:

from unstructured.partition.auto import partition

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

Hitung distribusi tipe elemen
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")


Atau langsung dari 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")

Mengakses Metadata

Setiap elemen memiliki metadata yang kaya:

from unstructured.partition.auto import partition

elements = partition(filename="dokumen.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()

Penggunaan Lanjutan

Strategi Partisi PDF

Unstructured menawarkan beberapa strategi untuk memproses PDF, masing-masing dengan trade-off antara kecepatan dan akurasi:

from unstructured.partition.pdf import partitionpdf

Strategy: fast - menggunakan pdfminer saja (cepat, kurang akurat untuk layout kompleks)
elementsfast = partitionpdf(

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

Strategy: hires - menggunakan model deteksi layout (lambat, paling akurat)
elementshires = partitionpdf(
    filename="document.pdf",
    strategy="hires",

    modelname="yolox"
)

Strategy: ocronly - hanya OCR (untuk PDF yang di-scan)

elementsocr = partitionpdf(

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

)

Strategy: auto (default) - otomatis memilih strategi terbaik
elementsauto = partitionpdf(

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

from unstructured.partition.pdf import partitionpdf

elements = partitionpdf(

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

    infertablestructure=True

)

Filter hanya elemen tabel
tables = [el for el in elements if el.category == "Table"]

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

    # Akses HTML representation dari tabel
    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 berdasarkan judul (menjaga konteks semantik)
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: Ukuran maksimum setiap chunk
• newafternchars: Mulai chunk baru setelah N karakter (soft limit)
• combinetextundernchars: Gabungkan elemen kecil di bawah N karakter
• overlap: Jumlah karakter overlap antar chunk
• overlapall: Terapkan overlap ke semua chunk, bukan hanya yang berurutan

# Chunking dengan overlap untuk konteks yang lebih baik
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  "


Bersihkan whitespace berlebih
cleaned = cleanextrawhitespace(rawtext)
print(cleaned)

Cleaning komprehensif
cleaned = clean(
    rawtext,

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

)
print(cleaned)

Gabungkan paragraf yang terputus
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 ke JSON
elementstojson(elements, filename="output.json")


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

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

Integrasi dengan Vector Stores

Berikut contoh pipeline lengkap dari dokumen ke vector store:

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

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

Step 2: Chunk dokumen
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: Simpan ke 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")

Memproses Batch Dokumen

Untuk memproses banyak dokumen sekaligus:

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

def processdirectory(inputdir, outputdir):

    """Proses semua dokumen dalam direktori."""
    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()

# Dokumen teks biasa -> fast
elements = partition(filename="textdoc.pdf", strategy="fast")


Dokumen dengan layout kompleks -> hires
elements = partition(filename="annualreport.pdf", strategy="hires")

Dokumen scan/gambar -> ocronly

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

# Untuk Q&A yang membutuhkan konteks detail
chunks = chunkbytitle(elements, maxcharacters=500, overlap=100)


Untuk summarization yang membutuhkan konteks luas
chunks = chunkbytitle(elements, maxcharacters=2000, overlap=200)

Untuk search yang membutuhkan keseimbangan
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 dengan 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 ke strategi yang lebih sederhana
        try:
            return partition(filename=filename, strategy="fast")
        except Exception:
            return []

from unstructured.partition.auto import partition

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

Tambahkan metadata kustom
for element in elements:
    element.metadata.categorydepth = 0


    # Tambahkan 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 dengan caching untuk menghindari reprocessing."""
    os.makedirs(cachedir, existok=True)

    # Generate cache key dari 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)

    # Simpan ke 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):
        """Pipeline utama preprocessing dokumen."""
        # 1. Partition
        elements = partition(
            filename=filename,
            strategy="auto",
            includepagebreaks=False
        )

        # 2. Filter elemen yang tidak relevan
        elements = self.filterelements(elements)

        # 3. Bersihkan teks
        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):

        """Hapus elemen yang tidak berguna."""
        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):

        """Bersihkan teks setiap elemen."""
        for el in elements:
            el.apply(cleanextrawhitespace)

        return elements

    def prepareoutput(self, chunks):

        """Format output untuk 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
        ]

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


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

• Gunakan partition() untuk auto-detection format file, atau fungsi partisi spesifik untuk kontrol lebih
• Pilih strategi partisi sesuai jenis dokumen: fast untuk teks sederhana, hires untuk layout kompleks, ocronly untuk dokumen scan
• Manfaatkan chunkby_title() untuk chunking semantik yang mempertahankan konteks dokumen
• Gunakan cleaning functions untuk membersihkan teks sebelum embedding
• Implementasikan caching untuk menghindari reprocessing dokumen yang sama
• Pertimbangkan Unstructured Ingest CLI untuk pipeline batch processing yang lebih kompleks

Dengan menguasai Unstructured, Anda dapat membangun pipeline preprocessing dokumen yang robust dan siap produksi untuk aplikasi RAG dan LLM Anda.