Document Parsers for Production RAG

from docling.document_converter import DocumentConverter
from docling.chunking import HybridChunker
from langchain_docling import DoclingLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 1. Parse — hierarchy, tables, reading order preserved
converter = DocumentConverter()
result = converter.convert("annual_report_2024.pdf")
doc = result.document

# 2. Inspect — DoclingDocument gives you structure for free
for table in doc.tables:
    print(table.export_to_html())  # TableFormer HTML

# 3. Semantic chunking — respects section boundaries
chunker = HybridChunker(tokenizer="BAAI/bge-small-en-v1.5", max_tokens=512)
chunks = list(chunker.chunk(doc))

# 4. Push to ChromaDB
texts = [c.text for c in chunks]
metas = [{"source": c.meta.origin.filename, "page": c.meta.page_no} for c in chunks]
Chroma.from_texts(texts, OpenAIEmbeddings(), metadatas=metas)

from llama_index.readers.docling import DoclingReader
from llama_index.node_parser.docling import DoclingNodeParser
from llama_index.core import VectorStoreIndex
import glob

reader = DoclingReader()  # no API key, fully local
docs = reader.load_data(glob.glob("contracts/*.pdf"))
parser = DoclingNodeParser()  # preserves IndexNode hierarchy
nodes = parser.get_nodes_from_documents(docs)
index = VectorStoreIndex.from_nodes(nodes)
response = index.as_query_engine().query(
    "What are the termination clauses across all contracts?"
)

from mineru.pipe.pdf_pipe import UnitPDFPipeline
from mineru.config import MinerUConfig

# 1. Configure — backend auto-selects GPU if available
cfg = MinerUConfig(
    parse_method="auto",  # txt | ocr | auto
    formula_enable=True,  # UniMERNet → LaTeX
    table_enable=True,    # structured HTML table output
    lang="en"            # or "zh", "ja", "auto"
)

# 2. Parse single paper
pipe = UnitPDFPipeline("attention_is_all_you_need.pdf", cfg)
md_content, images, layout_json = pipe.pipe_parse()

# 3. Formulas come out as LaTeX inline
print(md_content[:500])
# → "...the attention function $\text{Attention}(Q,K,V) =
# \text{softmax}\left(\frac{QK^T}{\sqrt{d_k}}\right)V$..."

import subprocess

# CLI batch — parallel GPU workers
subprocess.run([
    "mineru", "-p", "docs/",
    "-o", "output/",
    "--method", "ocr",
    "--lang", "auto"  # PaddleOCR auto-detects language
])

# LangChain integration (MinerU MCP server)
from langchain_mineru import MineruLoader
loader = MineruLoader("research_papers/", formula=True)
loader = loader.load()

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunk_by_title
from pathlib import Path

# 1. One API for every format
results = {}
for fpath in Path("inbox/").iterdir():
    elements = partition(str(fpath), strategy="hi_res")
    results[fpath.name] = elements

# 2. Filter by element type for targeted retrieval
tables = [e for batch in results.values()
          for e in batch if e.category == "Table"]

# 3. Structure-aware chunking
all_elements = [e for batch in results.values() for e in batch]
chunks = chunk_by_title(all_elements, max_characters=1500)
print(f"{len(chunks)} chunks across {len(results)} files")

from haystack_integrations.components.converters.unstructured import UnstructuredFileConverter
from haystack.components.preprocessors import DocumentSplitter
from haystack.core.pipeline import Pipeline

converter = UnstructuredFileConverter(
    api_url="http://localhost:8000"  # self-hosted OSS server
)
splitter = DocumentSplitter(
    split_by="word", split_length=300, split_overlap=20
)
pipe = Pipeline()
pipe.add_component("converter", converter)
pipe.add_component("splitter", splitter)
pipe.connect("converter.documents", "splitter.documents")
pipe.run({"converter": {"paths": ["s3://my-bucket/docs/"]}})