从零开始：本地部署 Unstructured 实现 PDF 文档切片与向量转换

引言

在处理 PDF 等非结构化文档时，将其转换为可供大语言模型（LLM）和检索增强生成（RAG）系统使用的结构化数据，是 AI 应用开发中不可或缺的一环。Unstructured 是一个强大的开源 Python 库，专注于从非结构化数据中提取和预处理文本信息，广泛应用于 PDF、Word 文档、HTML 等多种格式的文件处理。其核心功能包括分区、清理、暂存和分块，能够将复杂的非结构化文档转换为结构化输出，为后续的自然语言处理任务提供高质量的数据支持。

本文将从零开始，详细介绍如何在本地环境中搭建 Unstructured，并完成对 PDF 文档的切片与向量转换，帮助读者构建一个完整的文档预处理流水线。

一、环境准备

1.1 系统要求

• Python 3.9 或更高版本
• 推荐使用虚拟环境管理依赖

1.2 安装 Unstructured 库

首先，创建并激活虚拟环境（以 macOS/Linux 为例）：

# 使用 uv 创建虚拟环境（推荐）
uv venv
source .venv/bin/activate

安装 Unstructured 库时，需要根据实际需求选择对应的扩展包。基础安装仅支持纯文本、HTML、XML 和电子邮件文件：

uv add unstructured

如需处理 PDF 文件，则需安装 PDF 扩展：

uv add "unstructured[pdf]"

如需同时支持多种文档类型，可以一次性安装多个扩展：

uv add "unstructured[pdf,docx,image]"

若希望使用本地推理（依赖模型）以及 OCR 等高级处理技术，建议进行完整版安装：

pip install "unstructured[local-inference]"

1.3 安装系统依赖

Unstructured 处理 PDF 文件时依赖多个底层系统工具，需要提前安装。

macOS 环境（使用 Homebrew）：

brew install libmagic poppler tesseract

Windows 环境：

1. 安装 poppler：从 poppler 官网 下载最新的压缩包，解压后将 bin 文件夹路径（如 C:\poppler-24.08.0-0\Library\bin）添加到系统环境变量 PATH 中。
2. 安装 Tesseract OCR：从 Tesseract 官网 下载安装程序。安装时建议勾选 "Additional language data" 以支持多种语言，并将安装路径（如 C:\Program Files\Tesseract-OCR）添加到系统环境变量 PATH 中。同时，新建环境变量 TESSDATA_PREFIX，值为 C:\Program Files\Tesseract-OCR\tessdata。

安装完成后，在终端中运行以下命令验证是否成功：

tesseract --version

1.4 Docker 部署（可选）

Unstructured 官方推荐使用 Docker 来确保所有系统依赖项都正确安装。以下是使用 Docker 部署 Unstructured API 的简要步骤：

# 拉取 Unstructured API 镜像
docker pull quay.io/unstructured-io/unstructured-api:latest

# 运行容器
docker run -p 8000:8000 quay.io/unstructured-io/unstructured-api:latest

启动后，可通过 http://localhost:8000 访问 Unstructured API 服务。对于生产环境或需要稳定依赖管理的场景，Docker 部署是更可靠的选择。

二、PDF 文档解析与分区

2.1 分区功能简介

Unstructured 的核心能力之一是分区（partitioning）——将原始文档分解为标准的结构化元素，例如从 PDF 中识别并提取标题、段落、表格等元素，准确率可达 90% 以上。

Unstructured 提供多种分区策略：

• auto：自动选择最优策略
• fast：快速处理，不进行 OCR
• hi_res：高精度处理，包含版面分析和 OCR

2.2 基础解析示例

以下代码展示了如何使用 Unstructured 解析 PDF 文件：

from unstructured.partition.auto import partition

# 解析 PDF 文件
elements = partition(
    filename="example.pdf",
    strategy="hi_res",  # 使用高精度策略
)

# 查看解析结果
for element in elements[:5]:
    print(f"类型: {element.category}")
    print(f"内容: {element.text[:100]}...")
    print(f"元数据: {element.metadata}")
    print("-" * 50)

2.3 解析输出说明

partition 函数返回一个由 Element 对象组成的列表。常见的元素类型包括：

• Title：文档标题
• NarrativeText：段落正文
• ListItem：列表项
• Table：表格
• Header/Footer：页眉/页脚
• Image：图片

每个元素都包含 text 属性（元素文本内容）、category 属性（元素类型）和 metadata 属性（来源页码、文件路径等元信息）。

2.4 处理包含图片和表格的 PDF

对于包含扫描图片或复杂排版的 PDF，需要使用 hi_res 策略并结合 Tesseract OCR 进行文字识别：

from unstructured.partition.pdf import partition_pdf

# 使用高精度策略解析包含图片的 PDF
elements = partition_pdf(
    filename="scanned_document.pdf",
    strategy="hi_res",
    extract_images_in_pdf=True,
    infer_table_structure=True,  # 自动推断表格结构
)

for element in elements:
    if element.category == "Table":
        print(f"表格内容:\n{element.text}")
    elif element.category == "Image":
        print(f"图片描述: {element.metadata.image_description}")

三、文本切片

3.1 切片的基本概念

切片（chunking）是将长文档分割成更小块的过程，以便在 RAG 应用和相似性搜索中使用。与传统的基于纯文本特征的切片方法不同，Unstructured 的切片机制是在分区生成的文档元素之上进行的：它尽可能将连续的整元素组合成块，仅当单个元素超过最大尺寸时才进行文本切分。这种方法能够保留分区阶段建立的语义单元的完整性。

Unstructured 支持两种切片策略：

• basic：尽可能用整元素填满每个块，达到指定大小限制。当单个元素超过硬性最大长度时，会通过文本切分将其拆分为多个块。表格元素始终独立成块，不与其他元素合并。
• by_title：在 basic 策略的基础上，遵循章节边界。遇到标题元素时，即使当前块还有空间也会关闭当前块并开启新块，从而确保同一块内不会包含跨章节的内容。

3.2 核心切片参数

以下是在切片过程中常用的参数：

3.3 切片示例

方法一：在分区时同步切片

from unstructured.partition.auto import partition

# 在 partition 函数中通过 chunking_strategy 参数执行切片
elements = partition(
    filename="example.pdf",
    strategy="hi_res",
    chunking_strategy="by_title",
    max_characters=1000,
    new_after_n_chars=800,
    overlap=50,
    multipage_sections=False,
)

# 切片后的结果包含 CompositeElement、Table 或 TableChunk 类型的元素
for chunk in elements:
    print(f"类型: {chunk.category}")
    print(f"长度: {len(chunk.text)} 字符")
    print(f"内容: {chunk.text[:200]}...")
    print("-" * 50)

方法二：单独执行切片

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

# 先分区，后切片
elements = partition(filename="example.pdf", strategy="hi_res")
chunks = chunk_by_title(
    elements,
    max_characters=1000,
    new_after_n_chars=800,
    overlap=50,
)

for chunk in chunks:
    print(chunk.text)

3.4 切片参数调优建议

在实际应用中，切片参数的选择直接影响后续检索的效果：

• 块大小：max_characters 通常在 500-1500 之间较为合理。过小的块可能丢失上下文信息，过大的块则可能包含过多不相关内容。
• 重叠设置：对于超大元素被文本切分的情况，设置 overlap 有助于避免语义单元在切分边界处被割裂。
• 策略选择：对于结构清晰的文档（如论文、技术手册），推荐使用 by_title 策略以保留章节边界；对于结构松散的文档，basic 策略效率更高。

四、向量转换

4.1 向量转换概述

向量转换（embedding）是将文本块转换为数值向量的过程，这些向量能够捕捉文本的语义信息，使计算机能够“理解”和处理文本的含义。经过向量转换后的数据可以存储到向量数据库中，用于语义搜索、相似性匹配和 RAG 应用。

4.2 使用 sentence-transformers 生成向量

sentence-transformers 是一个广泛使用的 Python 库，支持在本地运行各种嵌入模型，无需调用外部 API。以下示例演示如何为切片后的文本块生成向量：

from sentence_transformers import SentenceTransformer
from unstructured.partition.auto import partition
from unstructured.chunking.title import chunk_by_title

# 1. 加载本地嵌入模型
model = SentenceTransformer('BAAI/bge-small-zh-v1.5')  # 中文模型
# 或使用英文模型: model = SentenceTransformer('all-MiniLM-L6-v2')

# 2. 解析 PDF 并进行切片
elements = partition(filename="example.pdf", strategy="hi_res")
chunks = chunk_by_title(elements, max_characters=1000)

# 3. 提取切片文本并生成向量
chunk_texts = [chunk.text for chunk in chunks]
embeddings = model.encode(chunk_texts, normalize_embeddings=True)

# 4. 查看结果
print(f"生成了 {len(embeddings)} 个向量")
print(f"每个向量的维度: {embeddings[0].shape}")
print(f"第一个向量的前10个值: {embeddings[0][:10]}")

4.3 使用 LangChain 集成

LangChain 提供了与 Unstructured 的深度集成，可以通过 UnstructuredLoader 加载 PDF 并与其他组件协同工作。

from langchain_community.document_loaders import UnstructuredPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS

# 1. 加载 PDF 文档
loader = UnstructuredPDFLoader(
    "example.pdf",
    mode="elements",  # 以元素模式加载，保留文档结构
    strategy="hi_res",
)
documents = loader.load()

# 2. 文本切片
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)
chunks = text_splitter.split_documents(documents)

# 3. 生成向量并存储
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    model_kwargs={'device': 'cpu'},
)
vectorstore = FAISS.from_documents(chunks, embeddings)

# 4. 向量检索测试
query = "什么是检索增强生成？"
results = vectorstore.similarity_search(query, k=3)
for doc in results:
    print(doc.page_content[:200])
    print("-" * 50)

4.4 常用的本地嵌入模型

4.5 向量存储

生成向量后，需要将其存储到向量数据库中以供检索。以下是使用 FAISS（Facebook AI Similarity Search）的完整示例：

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

# 生成向量后存储到 FAISS
def store_to_faiss(embeddings, chunk_texts, save_path="faiss_index"):
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)  # L2 距离索引
    index.add(embeddings.astype(np.float32))
    
    # 保存索引和对应的文本
    faiss.write_index(index, f"{save_path}.index")
    with open(f"{save_path}_texts.txt", "w", encoding="utf-8") as f:
        for text in chunk_texts:
            f.write(text + "\n---\n")
    
    print(f"FAISS 索引已保存到 {save_path}.index")
    return index

# 使用 FAISS 进行检索
def search_faiss(query, model, index, chunk_texts, k=3):
    query_embedding = model.encode([query], normalize_embeddings=True)
    distances, indices = index.search(query_embedding.astype(np.float32), k)
    
    results = []
    for i, idx in enumerate(indices[0]):
        results.append({
            "text": chunk_texts[idx],
            "distance": distances[0][i]
        })
    return results

五、完整流水线示例

以下是一个从 PDF 到切片再到向量的完整处理流水线，涵盖了解析、切片、向量生成和检索的全过程：

import os
from unstructured.partition.auto import partition
from unstructured.chunking.title import chunk_by_title
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

class PDFVectorPipeline:
    """PDF 文档向量化处理流水线"""
    
    def __init__(self, embedding_model="BAAI/bge-small-zh-v1.5"):
        self.model = SentenceTransformer(embedding_model)
        self.index = None
        self.chunk_texts = []
        self.metadata_list = []
    
    def process_pdf(self, pdf_path, chunk_max_char=1000, chunk_overlap=50):
        """处理单个 PDF 文件"""
        print(f"正在处理: {pdf_path}")
        
        # 1. 分区
        print("  - 执行文档分区...")
        elements = partition(
            filename=pdf_path,
            strategy="hi_res",
            infer_table_structure=True,
        )
        
        # 2. 切片
        print("  - 执行文本切片...")
        chunks = chunk_by_title(
            elements,
            max_characters=chunk_max_char,
            overlap=chunk_overlap,
            multipage_sections=False,
        )
        
        # 3. 提取文本和元数据
        for chunk in chunks:
            self.chunk_texts.append(chunk.text)
            self.metadata_list.append({
                "file": pdf_path,
                "type": chunk.category,
                "page": chunk.metadata.page_number if hasattr(chunk.metadata, "page_number") else None,
            })
        
        print(f"  - 生成 {len(chunks)} 个文本块")
        return len(chunks)
    
    def build_index(self):
        """构建 FAISS 索引"""
        if not self.chunk_texts:
            raise ValueError("请先处理 PDF 文档")
        
        print("正在生成向量嵌入...")
        embeddings = self.model.encode(
            self.chunk_texts, 
            normalize_embeddings=True,
            show_progress_bar=True,
        )
        
        dimension = embeddings.shape[1]
        self.index = faiss.IndexFlatIP(dimension)  # 使用内积（余弦相似度）
        self.index.add(embeddings.astype(np.float32))
        
        print(f"索引构建完成，共 {self.index.ntotal} 个向量，维度 {dimension}")
        return self.index
    
    def search(self, query, k=3):
        """检索相似文本块"""
        if self.index is None:
            raise ValueError("请先构建索引")
        
        query_embedding = self.model.encode([query], normalize_embeddings=True)
        scores, indices = self.index.search(query_embedding.astype(np.float32), k)
        
        results = []
        for i, idx in enumerate(indices[0]):
            results.append({
                "text": self.chunk_texts[idx],
                "score": scores[0][i],
                "metadata": self.metadata_list[idx],
            })
        return results
    
    def save(self, save_dir="pipeline_output"):
        """保存流水线结果"""
        os.makedirs(save_dir, exist_ok=True)
        
        # 保存 FAISS 索引
        if self.index:
            faiss.write_index(self.index, f"{save_dir}/faiss.index")
        
        # 保存文本块和元数据
        with open(f"{save_dir}/chunks.txt", "w", encoding="utf-8") as f:
            for text in self.chunk_texts:
                f.write(text + "\n\n---\n\n")
        
        import json
        with open(f"{save_dir}/metadata.json", "w", encoding="utf-8") as f:
            json.dump(self.metadata_list, f, ensure_ascii=False, indent=2)
        
        print(f"结果已保存到 {save_dir}")


# 使用示例
if __name__ == "__main__":
    # 初始化流水线
    pipeline = PDFVectorPipeline(embedding_model="BAAI/bge-small-zh-v1.5")
    
    # 处理 PDF 文件
    pipeline.process_pdf("example.pdf", chunk_max_char=1000)
    
    # 构建向量索引
    pipeline.build_index()
    
    # 执行检索
    query = "文档的主要研究内容是什么？"
    results = pipeline.search(query, k=3)
    
    print(f"\n查询: {query}")
    for i, result in enumerate(results):
        print(f"\n结果 {i+1} (相似度: {result['score']:.4f}):")
        print(f"来源: {result['metadata']['file']}")
        print(f"文本预览: {result['text'][:200]}...")

六、常见问题与解决方案

6.1 Poppler 未找到

问题：运行代码时报错 pdf2image.exceptions.PDFInfoNotInstalledError: Unable to get page count. Is poppler installed and in PATH?

解决：确保 poppler 已正确安装并添加到系统 PATH。Windows 用户需确认将 bin 文件夹路径添加到环境变量。

6.2 YOLOX 模型下载失败

问题：使用 hi_res 策略时，YOLOX 模型无法从 Hugging Face 下载。

解决：可以手动下载模型文件后放置到缓存目录。Hugging Face 的默认缓存目录为 ~/.cache/huggingface/hub/，在其中找到 models--unstructuredio-- 开头的目录并放入模型文件。

6.3 Tesseract OCR 未生效

问题：OCR 功能不工作，图片中的文字未被提取。

解决：确认 Tesseract 已正确安装，并设置了 TESSDATA_PREFIX 环境变量指向 tessdata 目录。此外，可在代码中显式指定语言：

elements = partition(
    filename="scanned.pdf",
    strategy="hi_res",
    languages=["chi_sim", "eng"],  # 中文简体 + 英文
)

七、总结

本文详细介绍了从零开始在本地环境中搭建 Unstructured 文档处理流水线的完整流程，涵盖以下核心环节：

1. 环境搭建：安装 Unstructured 库、系统依赖（poppler、Tesseract）及可选配置
2. PDF 解析与分区：使用 partition 函数将 PDF 转换为结构化文档元素，支持多种分区策略
3. 文本切片：通过 basic 或 by_title 策略对文档元素进行智能切片，保留语义边界
4. 向量转换：使用 sentence-transformers 等本地嵌入模型将文本块转换为向量，并存储到 FAISS 等向量数据库

通过这一完整的流水线，读者可以将任意 PDF 文档转换为可直接用于 RAG 应用和语义搜索的向量数据，为后续的 AI 应用开发奠定坚实的基础。