【LangChain】langchain_unstructured.document_loaders.UnstructuredLoader 类：依赖 Unstructured 库进行文档解析和分区

UnstructuredLoader 是 LangChain 的 langchain_unstructured 包中的文档加载器，基于 Unstructured 库（由 Unstructured.IO 提供）解析非结构化文档，支持多种文件类型，包括 PDF、Word、PowerPoint、HTML、Markdown、文本、图像等。它将文档分区为语义单元（如标题、叙述文本、表格），生成 LangCh

正义的彬彬侠

1751人浏览 · 2025-05-09 21:49:38

正义的彬彬侠 · 2025-05-09 21:49:38 发布

langchain_unstructured.document_loaders.UnstructuredLoader 类是 LangChain 生态中用于加载和处理多种类型非结构化文档（如 PDF、Word、HTML、Markdown 等）的工具类，依赖 Unstructured 库进行文档解析和分区。

本文基于 LangChain 0.3.x，提供 UnstructuredLoader 的定义、参数、方法和典型场景，并附带一个独立示例，展示如何使用 UnstructuredLoader 结合 ChatOpenAI 和向量存储处理人工智能主题的问答任务，示例加载 PDF 和文本文件，构建 RAG 链回答用户查询，突出 UnstructuredLoader 的多格式支持和分区能力。

`langchain_unstructured.document_loaders.UnstructuredLoader` 简介

UnstructuredLoader 是 LangChain 的 langchain_unstructured 包中的文档加载器，基于 Unstructured 库（由 Unstructured.IO 提供）解析非结构化文档，支持多种文件类型，包括 PDF、Word、PowerPoint、HTML、Markdown、文本、图像等。它将文档分区为语义单元（如标题、叙述文本、表格），生成 LangChain 的 Document 对象，适合 RAG、问答或语义搜索等任务。相比 langchain_community 中的 UnstructuredFileLoader（已于 0.2.8 废弃），UnstructuredLoader 提供更现代的接口，支持本地和 API 分区，并移除过时的 mode 参数（如 single、elements），改用 chunking_strategy 等配置。🔗

核心功能：

支持多种文件类型，自动检测 MIME 类型并分区。
提供本地分区（需要安装依赖）或通过 Unstructured API 分区（需 API 密钥）。
支持文档分块（chunking），生成语义单元或自定义大小的块。
提供元数据（如文件类型、语言、页面编号、来源 URL）。
支持惰性加载，优化内存使用。

适用场景：

从混合格式文档（如 PDF、Word、文本）提取文本，构建 RAG 系统。
处理复杂文档（如学术论文、报告），保留语义结构。
加载远程 URL 或本地文件的非结构化数据。
批量处理多格式文件，用于语义搜索或问答。

与 DirectoryLoader 对比：

DirectoryLoader：批量加载目录中的文件，需指定加载器（如 TextLoader、UnstructuredLoader）。
UnstructuredLoader：专注于单文件或文件列表的多格式解析，支持更复杂的语义分区。

最新变化：🔗🔗🔗

需安装 langchain_unstructured 和 unstructured-client：

pip install langchain-unstructured unstructured-client

推荐使用 Unstructured API（需 UNSTRUCTURED_API_KEY）或本地 Docker 部署。
不再支持 mode 参数，改用 chunking_strategy（如 basic、by_title）。
URL 加载需明确指定 web_url 参数。🔗

类定义和初始化

以下是 UnstructuredLoader 的定义，基于最新文档（UnstructuredLoader API) 。🔗

类签名

class UnstructuredLoader(BaseLoader):
    def __init__(
        self,
        file_path: Optional[Union[str, Path, List[Union[str, Path]]]] = None,
        *,
        file: Optional[Union[IO[bytes], List[IO[bytes]]]] = None,
        partition_via_api: bool = False,
        post_processors: Optional[List[Callable[[str], str]]] = None,
        api_key: Optional[str] = None,
        client: Optional[UnstructuredClient] = None,
        url: Optional[str] = None,
        web_url: Optional[str] = None,
        **kwargs: Any
    ) -> None

参数：
- file_path（Optional[Union[str, Path, List[Union[str, Path]]]]，默认 None）：本地文件路径或路径列表。
- file（Optional[Union[IO[bytes], List[IO[bytes]]]]，默认 None）：文件字节流或字节流列表。
- partition_via_api（bool，默认 False）：是否使用 Unstructured API 分区（需 api_key）。
- post_processors（Optional[List[Callable[[str], str]]]，默认 None）：后处理函数列表，修改提取的文本。
- api_key（Optional[str]，默认 None）：Unstructured API 密钥。
- client（Optional[UnstructuredClient]，默认 None）：自定义 Unstructured 客户端。
- url（Optional[str]，默认 None）：Unstructured API 的 URL（默认 https://api.unstructuredapp.io/general/v0/general）。
- web_url（Optional[str]，默认 None）：远程文件 URL（需明确指定）。
- **kwargs：传递给 Unstructured 分区函数的参数（如 chunking_strategy、max_characters）。
关键要求：
- 必须指定 file_path、file 或 web_url 中的一个（）。
- 使用 API 分区需设置 UNSTRUCTURED_API_KEY：
```
export UNSTRUCTURED_API_KEY="your-api-key"
```
- 本地分区需安装依赖：
```
pip install unstructured
```
功能：
- 自动检测文件类型，调用 Unstructured 分区函数。
- 生成 Document 对象，包含 page_content 和 metadata（如 filetype、languages、page_number）。

初始化示例

from langchain_unstructured.document_loaders import UnstructuredLoader
loader = UnstructuredLoader(
    file_path="example_data/ai_paper.pdf",
    partition_via_api=True,
    api_key="your-api-key",
    chunking_strategy="basic",
    max_characters=1000
)

常用方法

UnstructuredLoader 继承自 langchain_core.document_loaders.base.BaseLoader，提供以下核心方法。

1. `load`

def load(self) -> List[Document]

功能：同步加载文件，返回 Document 列表。
输出：List[Document]，每个 Document 包含 page_content 和 metadata。

示例：

docs = loader.load()
print(docs[0].page_content)  # 文档内容
print(docs[0].metadata)      # {'source': 'ai_paper.pdf', 'filetype': 'application/pdf', ...}

2. `lazy_load`

def lazy_load(self) -> Iterator[Document]

功能：惰性加载文件，逐个生成 Document，节省内存。
输出：Iterator[Document]，迭代器逐个返回文档。

示例：

for doc in loader.lazy_load():
    print(doc.page_content)

3. `alazy_load`（异步）

async def alazy_load(self) -> AsyncIterator[Document]

功能：异步惰性加载，适合高并发场景。

示例：

import asyncio
async def load_docs():
    async for doc in loader.alazy_load():
        print(doc.page_content)
asyncio.run(load_docs())

使用方式

以下是使用 UnstructuredLoader 的步骤。🔗

1. 安装依赖

pip install --upgrade langchain langchain-unstructured unstructured-client langchain-openai faiss-cpu

若本地分区，安装完整 unstructured：
```
pip install unstructured
```

或使用 Docker 部署 Unstructured：

docker run -p 8000:8000 -d --rm --name unstructured-api downloads.unstructured.io/unstructured-io/unstructured-api:latest --port 8000 --host 0.0.0.0

2. 获取 Unstructured API 密钥

生成 API 密钥并设置环境变量：

export UNSTRUCTURED_API_KEY="your-api-key"

或在代码中传递：
```
loader = UnstructuredLoader(api_key="your-api-key")
```
仅调用在线API分区（文件解析）需要注册和设置。如果安装了pip install unstructured支持本机调用（自己的电脑解析文件），不需要设置。

3. 准备文件

创建 example_data 目录，包含：

ai_paper.pdf：学术论文，内容如“人工智能是计算机科学的一个分支…”。
ai_notes.txt：文本笔记，内容如“AI 在医疗中的应用包括诊断和药物研发”。

4. 初始化 UnstructuredLoader

from langchain_unstructured.document_loaders import UnstructuredLoader
loader = UnstructuredLoader(
    file_path=["example_data/ai_paper.pdf", "example_data/ai_notes.txt"],
    partition_via_api=True,
    chunking_strategy="by_title",
    max_characters=1000
)

5. 加载文档

docs = loader.load()

6. 集成到 RAG 链

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())
retriever = vectorstore.as_retriever()
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("根据以下上下文回答问题：\n{context}\n问题：{question}")
chain = {"context": retriever, "question": lambda x: x["question"]} | prompt | llm

7. 调用链

result = chain.invoke({"question": "人工智能是什么？"})
print(result.content)

使用 `UnstructuredLoader` 的示例

以下是一个独立示例，展示如何使用 UnstructuredLoader 加载 PDF 和文本文件，结合 FAISS 向量存储和 ChatOpenAI 构建 RAG 链，回答用户关于人工智能的查询。示例使用 Unstructured API 分区，突出多格式支持和语义分块。

准备环境：

获取 OpenAI API 密钥：OpenAI Platform。
获取 Unstructured API 密钥：unstructured.io。

设置环境变量：

export OPENAI_API_KEY="your-openai-key"
export UNSTRUCTURED_API_KEY="your-unstructured-key"

安装依赖：

pip install --upgrade langchain langchain-unstructured unstructured-client langchain-openai faiss-cpu

创建 example_data 目录，包含：
- ai_paper.pdf：内容如“人工智能是计算机科学的一个分支，旨在模拟人类智能…”。
- ai_notes.txt：内容如“AI 在医疗中的应用包括诊断、药物研发和个性化治疗。”。

代码：

from langchain_unstructured.document_loaders import UnstructuredLoader
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

# 初始化 UnstructuredLoader
loader = UnstructuredLoader(
    file_path=["example_data/ai_paper.pdf", "example_data/ai_notes.txt"],
    partition_via_api=True,
    api_key=os.environ.get("UNSTRUCTURED_API_KEY"),
    chunking_strategy="by_title",
    max_characters=1000,
    include_orig_elements=False
)

# 加载文档
docs = loader.load()

# 初始化向量存储
vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 初始化 ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 定义提示模板
prompt = ChatPromptTemplate.from_template(
    "根据以下上下文回答问题：\n{context}\n问题：{question}"
)

# 定义输出解析器
parser = StrOutputParser()

# 创建 RAG 链
chain = RunnableParallel({
    "context": retriever,
    "question": lambda x: x["question"]
}) | prompt | llm | parser

# 测试 UnstructuredLoader 和 RAG 链
print("测试 UnstructuredLoader 和 RAG 链：")
try:
    question = "人工智能是什么？"
    result = chain.invoke({"question": question})
    
    print(f"输入问题: {question}")
    print(f"回答: {result}")
    
    # 显示加载的文档
    print("\n加载的文档（前 2 个）：")
    for i, doc in enumerate(docs[:2]):
        print(f"文档 {i+1}:")
        print(f"内容: {doc.page_content[:100]}...")  # 截取前 100 字符
        print(f"元数据: {doc.metadata}")
except Exception as e:
    print(f"错误: {e}")

输出示例（实际输出取决于模型和 API 响应）：

测试 UnstructuredLoader 和 RAG 链：
输入问题: 人工智能是什么？
回答: 人工智能（AI）是计算机科学的一个分支，旨在模拟人类智能，如学习、推理和问题解决。

加载的文档（前 2 个）：
文档 1:
内容: 人工智能是计算机科学的一个分支，旨在模拟人类智能... ...
元数据: {'source': 'example_data/ai_paper.pdf', 'filetype': 'application/pdf', 'languages': ['eng'], 'page_number': 1, 'category': 'Title'}
文档 2:
内容: AI 在医疗中的应用包括诊断、药物研发和个性化治疗。... ...
元数据: {'source': 'example_data/ai_notes.txt', 'filetype': 'text/plain', 'languages': ['eng'], 'category': 'NarrativeText'}

代码说明

UnstructuredLoader：
- 初始化 UnstructuredLoader，加载 PDF 和文本文件，使用 API 分区。
- 设置 chunking_strategy="by_title"，按标题分块，max_characters=1000 限制块大小。
- 设置 include_orig_elements=False，简化输出。🔗
向量存储：
- 使用 FAISS 和 OpenAIEmbeddings 创建向量存储。
- 配置 retriever 检索前 2 个最相似文档。
LLM 初始化：
- 使用 ChatOpenAI 调用 gpt-3.5-turbo，设置 temperature=0.7。
提示模板：
- ChatPromptTemplate 组合检索上下文和用户问题。
RAG 链：
- 使用 RunnableParallel 同时传递 context（检索结果）和 question。
- LCEL 链连接 prompt、llm 和 parser，生成回答。
测试：
- 测试 RAG 链，回答 AI 定义问题。
- 显示加载的文档，展示 page_content 和 metadata。
错误处理：
- 使用 try-except 捕获文件、API 或检索错误。

运行要求：

有效的 OpenAI 和 Unstructured API 密钥：

export OPENAI_API_KEY="your-openai-key"
export UNSTRUCTURED_API_KEY="your-unstructured-key"

安装依赖：

pip install --upgrade langchain langchain-unstructured unstructured-client langchain-openai faiss-cpu

准备 example_data 目录和文件。
网络连接：访问 https://api.openai.com 和 https://api.unstructuredapp.io.

注意事项

API 密钥：

确保设置 UNSTRUCTURED_API_KEY：🔗
```
echo $UNSTRUCTURED_API_KEY
```

或在代码中传递：

loader = UnstructuredLoader(api_key="your-api-key")

OpenAI 密钥同理：

llm = ChatOpenAI(api_key="your-openai-key")

文件准备：
- 确保文件存在：
```
ls example_data
```
- 支持格式包括 PDF、TXT、MD、DOCX、PPTX、HTML 等。🔗

分区模式：

API 分区：推荐，需密钥，减少本地依赖：

loader = UnstructuredLoader(partition_via_api=True)

本地分区：需安装 unstructured 和依赖：
```
pip install unstructured[all-docs]
```

Docker 部署：

docker run -p 8000:8000 downloads.unstructured.io/unstructured-io/unstructured-api:latest

分块配置：
- 使用 chunking_strategy：🔗
  - "basic"：简单分块，类似旧 mode="single"。
  - "by_title"：按标题分块，保留语义结构。
- 设置 max_characters 控制块大小：
```
loader = UnstructuredLoader(chunking_strategy="by_title", max_characters=1000)
```

性能优化：

惰性加载：处理大文件：

for doc in loader.lazy_load():
    process(doc)

异步加载：高并发场景：

async for doc in loader.alazy_load():
    process(doc)

缓存向量：持久化 FAISS：

vectorstore.save_local("faiss_index")
vectorstore = FAISS.load_local("faiss_index", OpenAIEmbeddings(), allow_dangerous_deserialization=True)

错误调试：

API 错误：🔗

检查密钥格式：

print(os.environ.get("UNSTRUCTURED_API_KEY"))

验证 API 访问：

curl -X POST https://api.unstructuredapp.io/general/v0/general -H "Authorization: Bearer $UNSTRUCTURED_API_KEY"

文件错误：

检查路径：

import os
print(os.path.exists("example_data/ai_paper.pdf"))

支持多文件：

loader = UnstructuredLoader(file_path=["file1.pdf", "file2.txt"])

分块失败：

检查 chunking_strategy：

print(loader.kwargs.get("chunking_strategy"))

增加 max_characters：

loader = UnstructuredLoader(max_characters=5000)

常见问题

Q1：如何加载远程 URL？
A：使用 web_url 参数：🔗🔗

loader = UnstructuredLoader(
    web_url="https://example.com/document.pdf",
    partition_via_api=True,
    api_key="your-api-key"
)

注：url 参数用于指定 API 服务器，非文档 URL。🔗

Q2：如何处理多种文件类型？
A：UnstructuredLoader 自动检测类型：🔗

loader = UnstructuredLoader(
    file_path=["doc.pdf", "notes.txt", "slides.pptx"],
    chunking_strategy="by_title"
)

Q3：如何与少样本提示结合？
A：加载文档作为示例：

from langchain_core.prompts import FewShotPromptTemplate
docs = loader.load()
examples = [{"content": doc.page_content} for doc in docs]
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template("文档：{content}"),
    prefix="根据文档回答：",
    suffix="问题：{question}",
    input_variables=["question"]
)

Q4：如何使用开源模型？
A：替换为 ChatOllama：

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3")
chain = {"context": retriever, "question": lambda x: x["question"]} | prompt | llm | parser

Q5：如何模拟旧的 mode="single"？
A：设置 chunking_strategy="basic" 和大 max_characters（）：

loader = UnstructuredLoader(
    file_path="ai_paper.pdf",
    chunking_strategy="basic",
    max_characters=1000000,
    include_orig_elements=False
)

总结

langchain_unstructured.document_loaders.UnstructuredLoader 是 LangChain 中处理非结构化文档的现代工具，核心功能包括：

定义：解析多种格式文档，分区为语义单元。
初始化：配置 file_path、web_url、partition_via_api、chunking_strategy 等。
常用方法：load（同步）、lazy_load（惰性）、alazy_load（异步）。
适用场景：RAG、问答、多格式文档处理。

示例代码展示了使用 UnstructuredLoader 加载 PDF 和文本文件，构建 RAG 链回答 AI 问题。

参考：

LangChain UnstructuredLoader 文档：文档1、文档2、文档3
Unstructured 官方文档：docs.unstructured.io 🔗
GitHub 问题跟踪：https://github.com/langchain-ai/langchain/issues/26230

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

【免费下载】 index-tts-vllm：高性能文本转语音解决方案

`index-tts-vllm` 是一款开源的文本转语音（TTS）项目，基于 `index-tts` 进行了优化，通过引入 `vllm` 库重新实现了 GPT 模型的推理过程，大幅提升了推理速度和并发处理能力。项目旨在为开发者提供一个高效、稳定、易于集成的 TTS 解决方案。## 项目技术分析`index-tts-vllm` 采用了以下核心技术：- **vllm 库**：用于加速 GP

AI Agent技术社区

2025年8月SCI-炭黑四角蜂优化算法 Tetragonula carbonaria Optimization Algorithm-附Matlab免费代码

AI Agent技术社区

AgentScope示例应用：多智能体对话

在人工智能快速发展的今天，单一智能体已经难以满足复杂任务的需求。想象一下这样的场景：一个项目团队需要讨论技术方案，不同专业背景的成员各抒己见，最终达成共识。这正是多智能体对话要解决的问题——让多个AI智能体像真实团队成员一样进行有意义的交流。AgentScope作为面向开发者的多智能体编程框架，提供了强大的多智能体对话能力。本文将深入探讨如何使用AgentScope构建高效的多智能体对话系统。..