Python unstructured 库：非结构化数据解析与结构化处理指南

输出示例：

Title: Introduction
NarrativeText: This is the first paragraph...
ListItem: - Item 1

对于复杂 PDF（含表格），建议使用高分辨率策略，但这需要安装 Tesseract 和 Poppler：

from unstructured.partition.pdf import partition_pdf

# hi_res 策略利用计算机视觉提取表格
elements = partition_pdf(filename="example.pdf", strategy="hi_res")

2. 清理：净化文本数据

提取后的文本往往包含多余标点或格式噪音，可以使用清理模块标准化。

from unstructured.cleaners.core import clean, remove_punctuation

text = "Hello, World!!! This is a test..."
cleaned_text = clean(text, lowercase=True)  # 转小写
cleaned_text = remove_punctuation(cleaned_text)  # 去标点

print(cleaned_text)
# 输出：hello world this is a test

3. 格式化：生成下游输入

将元素列表转为 JSON 或其他格式，便于 LLM 消费。

from unstructured.staging.base import convert_to_dict

json_data = convert_to_dict(elements)
print(json_data[:2])

输出示例：

[
  {"type": "Title", "text": "Introduction", "metadata": {...}},
  {"type": "NarrativeText", "text": "This is the first paragraph...", "metadata": {...}}
]

与 LangChain 集成

LangChain 提供了 UnstructuredLoader，能无缝接入现有的 RAG 管道。

from langchain_unstructured import UnstructuredLoader

# 本地加载
loader = UnstructuredLoader(file_path="example.pdf")
docs = loader.load()
print(docs[0].page_content[:100])

# 或使用 Serverless API（需 API Key）
loader_api = UnstructuredLoader(
    file_path="example.pdf",
    api_key="your_api_key",
    strategy="hi_res"
)
docs_api = loader_api.load()

支持多种文件类型的 Loader，如 UnstructuredCSVLoader、UnstructuredHTMLLoader 等。

生产环境部署

Serverless API

通过 API 处理可减少本地依赖，适合高并发场景。

from unstructured_client import UnstructuredClient

client = UnstructuredClient(api_key_auth="your_api_key")
with open("example.pdf", "rb") as f:
    response = client.general.partition(file=f, strategy="hi_res")
    print(response.elements[:2])

Docker 部署

在容器内运行脚本可避免环境冲突：

from unstructured.partition.auto import partition

# 挂载数据卷后访问 /data/example.pdf
elements = partition(filename="/data/example.pdf")
print([str(el) for el in elements[:5]])

注意事项与优化

1. 依赖管理：本地处理 PDF/图片必须安装 tesseract 和 poppler，否则解析会失败。若只处理单一格式，建议安装特定依赖包以减少开销。
2. 性能权衡：strategy="hi_res" 虽然精度高，但计算成本较大。对纯文本 PDF 可使用默认策略提速。
3. 解析局限：DOCX 列表项有时会被误识为标题，大型文档可能缺乏父子关系标注，影响 LLM 上下文理解，建议后处理验证。
4. 隐私设置：库包含轻量级分析 Ping，可通过环境变量禁用：

   export DO_NOT_TRACK=true
   export SCARF_NO_ANALYTICS=true
   

5. 替代方案：若追求极致速度，可关注 extractous；若专注布局分析，LayoutParser 也是不错的选择。

综合示例

下面是一个完整的处理流程，涵盖解析、清理、保存及日志记录：

from unstructured.partition.auto import partition
from unstructured.cleaners.core import clean, remove_punctuation
from unstructured.staging.base import convert_to_dict
from langchain_unstructured import UnstructuredLoader
import json
from loguru import logger

logger.add("app.log", rotation="1 MB", level="INFO")

try:
    logger.info("Starting PDF processing")
    # 解析 PDF，使用 hi_res 策略提取表格
    elements = partition(filename="sample.pdf", strategy="hi_res")
except Exception as e:
    logger.exception("Failed to process PDF")
    raise

# 清理文本
cleaned_elements = []
for element in elements:
    text = clean(element.text, lowercase=True)
    text = remove_punctuation(text)
    cleaned_elements.append({"type": element.category, "text": text})

logger.info("Text cleaning completed")

# 保存为 JSON
json_data = convert_to_dict(cleaned_elements)
with open("output.json", "w") as f:
    json.dump(json_data, f, indent=2)

logger.info("JSON output saved")

# LangChain 集成
loader = UnstructuredLoader(file_path="sample.pdf", strategy="hi_res")
docs = loader.load()
logger.info(f"Loaded {len(docs)} documents")
print(docs[0].page_content[:100])

参考资料

• 官方文档：https://docs.unstructured.io/
• GitHub 仓库：https://github.com/Unstructured-IO/unstructured
• PyPI 页面：https://pypi.org/project/unstructured/
• LangChain 集成：https://python.langchain.com/docs/integrations/document_loaders/unstructured/