Unstructured.io: 将任何文档转换为LLM就绪数据块的预处理流水线 — 2026指南

python -m venv venv_unstructured
source venv_unstructured/bin/activate

# 安装基础包
pip install "unstructured[pdf]==0.17.0"

# 完整文档支持（安装量更大）
pip install "unstructured[all-docs]==0.17.0"

from unstructured.partition.auto import partition

elements = partition(filename="test.pdf")
print(f"提取了 {len(elements)} 个元素")
for el in elements[:5]:
    print(f"  {el.category}: {str(el)[:60]}...")

# 拉取预构建镜像
docker pull downloads.unstructured.io/unstructured-io/unstructured-api:latest

# 使用GPU支持运行 hi_res 分区
docker run -d \
  --name unstructured-api \
  -p 8000:8000 \
  --gpus all \
  downloads.unstructured.io/unstructured-io/unstructured-api:latest

# 验证健康状态
curl http://localhost:8000/healthcheck

docker run -d \
  --name unstructured-api-cpu \
  -p 8000:8000 \
  downloads.unstructured.io/unstructured-io/unstructured-api-cpu:latest

import requests

with open("annual_report.pdf", "rb") as f:
    response = requests.post(
        "http://localhost:8000/general/v0/general",
        files={"files": ("annual_report.pdf", f)},
        data={
            "strategy": "hi_res",
            "chunking_strategy": "by_title",
            "max_characters": 1500,
            "new_after_n_chars": 1200,
            "overlap": 150,
            "output_format": "application/json"
        }
    )

elements = response.json()
print(f"获取了 {len(elements)} 个块")

from langchain_community.document_loaders import UnstructuredFileLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 一步完成加载和分区
loader = UnstructuredFileLoader(
    "quarterly_earnings.pdf",
    mode="elements",           # 保留元素类型
    strategy="hi_res",
    post_processors=["chunk_by_title_characters"],
)

documents = loader.load()  # 返回 Document 对象列表

# 每个文档都有丰富的元数据
print(documents[0].metadata)
# {'source': 'quarterly_earnings.pdf', 'page_number': 1,
#  'category': 'NarrativeText', 'element_id': '...', 'parent_id': '...'}

# 直接存入向量库
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=OpenAIEmbeddings(),
)

from llama_index.readers.unstructured import UnstructuredReader
from llama_index.core import VectorStoreIndex

reader = UnstructuredReader(
    api_url="http://localhost:8000",
    partition_kwargs={
        "strategy": "hi_res",
        "chunking_strategy": "by_title",
        "max_characters": 1500,
        "overlap": 200,
    }
)

documents = reader.load_data("whitepaper.pdf")
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

response = query_engine.query("第3节提到的关键风险有哪些？")
print(response)

import chromadb
from unstructured.chunking.title import chunk_by_title
from unstructured.partition.pdf import partition_pdf
from sentence_transformers import SentenceTransformer

# 分区
raw_elements = partition_pdf("contract.pdf", strategy="hi_res")

# 保留章节的分块
chunks = chunk_by_title(
    raw_elements,
    max_characters=1200,
    new_after_n_chars=1000,
    overlap=200,
)

# 嵌入并存储
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("contracts")

model = SentenceTransformer("all-MiniLM-L6-v2")
for i, chunk in enumerate(chunks):
    embedding = model.encode(str(chunk)).tolist()
    collection.add(
        ids=[f"chunk_{i}"],
        embeddings=[embedding],
        documents=[str(chunk)],
        metadatas=[{
            "source": "contract.pdf",
            "page": chunk.metadata.page_number,
            "type": chunk.category,
        }]
    )

from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title
from unstructured.cleaners.core import clean

# 第1步：使用 hi_res 进行分区以检测布局
elements = partition_pdf(
    "complex_report.pdf",
    strategy="hi_res",
    extract_images_in_pdf=True,        # 保存嵌入图像
    infer_table_structure=True,         # 表格输出HTML
    max_partition=2000,                  # 每批元素数
)

# 第2步：过滤不需要的元素
filtered = [
    el for el in elements
    if el.category not in ["Header", "Footer", "PageBreak"]
]

# 第3步：清洗文本内容
for el in filtered:
    el.text = clean(
        el.text,
        extra_whitespace=True,
        dashes=True,           # 统一破折号
        trailing_punctuation=True,
    )

# 第4步：带重叠的分块
chunks = chunk_by_title(
    filtered,
    max_characters=1500,
    new_after_n_chars=1200,
    overlap_all=True,          # 所有块之间重叠
    overlap=200,
)

print(f"{len(elements)} 原始 → {len(filtered)} 过滤 → {len(chunks)} 块")

import concurrent.futures
from pathlib import Path
from unstructured.partition.auto import partition

def process_file(path: Path) -> dict:
    try:
        elements = partition(
            filename=str(path),
            strategy="fast",
        )
        return {
            "file": path.name,
            "elements": len(elements),
            "status": "success",
        }
    except Exception as e:
        return {
            "file": path.name,
            "elements": 0,
            "status": "error",
            "error": str(e),
        }

# 使用8个工作器处理500个PDF
pdf_dir = Path("./documents")
pdf_files = list(pdf_dir.glob("*.pdf"))

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
    results = list(executor.map(process_file, pdf_files))

success = sum(1 for r in results if r["status"] == "success")
print(f"成功处理了: {success}/{len(results)} 个文件")

import json
import hashlib
from pathlib import Path
from unstructured.staging.base import elements_to_dicts, dicts_to_elements

def partition_with_cache(file_path: str, strategy: str = "hi_res"):
    file_hash = hashlib.md5(open(file_path, "rb").read()).hexdigest()
    cache_path = Path(f"./cache/{file_hash}_{strategy}.json")
    cache_path.parent.mkdir(exist_ok=True)

    if cache_path.exists():
        return dicts_to_elements(json.load(open(cache_path)))

    elements = partition_pdf(file_path, strategy=strategy)
    cache_path.write_text(json.dumps(elements_to_dicts(elements), indent=2))
    return elements

# unstructured-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: unstructured-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: unstructured-api
  template:
    metadata:
      labels:
        app: unstructured-api
    spec:
      containers:
      - name: api
        image: downloads.unstructured.io/unstructured-io/unstructured-api:latest
        ports:
        - containerPort: 8000
        resources:
          limits:
            nvidia.com/gpu: 1
            memory: "8Gi"
          requests:
            memory: "4Gi"
---
apiVersion: v1
kind: Service
metadata:
  name: unstructured-api
spec:
  selector:
    app: unstructured-api
  ports:
  - port: 80
    targetPort: 8000