LangChain Document Loaders实战：如何高效加载和处理各种文档数据

发布日期：2024-07-10 07:03:50 浏览次数： 3780

作者：智能体AI

微信搜一搜，关注“智能体AI”

在实际业务场景中，我们经常需要从各种数据源加载数据。LangChain 提供了一套强大的文档加载器模块，帮助开发者轻松地将数据源中的内容加载为文档对象。本文将详细介绍 LangChain 的 Document Loaders 核心模块，并结合实际业务场景和代码示例，展示如何高效地加载和处理文档数据。

一、Document 类详解

Document 类是 LangChain 的核心组件之一，定义了一个文档对象的结构，包括文本内容和相关的元数据。该类允许用户与文档内容进行交互，如查看文档段落、摘要，以及使用查找功能来查询特定字符串。以下是 Document 类源码：

# 基于BaseModel定义的文档类。class Document(BaseModel):"""接口，用于与文档进行交互。"""
# 文档的主要内容。page_content: str# 用于查找的字符串。lookup_str: str = ""# 查找的索引，初次默认为0。lookup_index = 0# 用于存储任何与文档相关的元数据。metadata: dict = Field(default_factory=dict)
@propertydef paragraphs(self) -> List[str]:"""页面的段落列表。"""# 使用"\n\n"将内容分割为多个段落。return self.page_content.split("\n\n")
@propertydef summary(self) -> str:"""页面的摘要（即第一段）。"""# 返回第一个段落作为摘要。return self.paragraphs[0]
# 这个方法模仿命令行中的查找功能。def lookup(self, string: str) -> str:"""在页面中查找一个词，模仿cmd-F功能。"""# 如果输入的字符串与当前的查找字符串不同，则重置查找字符串和索引。if string.lower() != self.lookup_str:self.lookup_str = string.lower()self.lookup_index = 0else:# 如果输入的字符串与当前的查找字符串相同，则查找索引加1。self.lookup_index += 1# 找出所有包含查找字符串的段落。lookups = [p for p in self.paragraphs if self.lookup_str in p.lower()]# 根据查找结果返回相应的信息。if len(lookups) == 0:return "No Results"elif self.lookup_index >= len(lookups):return "No More Results"else:result_prefix = f"(Result {self.lookup_index + 1}/{len(lookups)})"return f"{result_prefix} {lookups[self.lookup_index]}"

二、BaseLoader 类详解

BaseLoader 类定义了如何从不同的数据源加载文档，并提供了一个可选的方法来分割加载的文档。开发者可以基于该类为特定数据源创建自定义加载器。以下是 BaseLoader 类的定义：

# 基础加载器类。class BaseLoader(ABC):"""基础加载器类定义。"""
# 抽象方法，所有子类必须实现此方法。@abstractmethoddef load(self) -> List[Document]:"""加载数据并将其转换为文档对象。"""
# 该方法可以加载文档，并将其分割为更小的块。def load_and_split(self, text_splitter: Optional[TextSplitter] = None) -> List[Document]:"""加载文档并分割成块。"""# 如果没有提供特定的文本分割器，使用默认的字符文本分割器。if text_splitter is None:_text_splitter: TextSplitter = RecursiveCharacterTextSplitter()else:_text_splitter = text_splitter# 先加载文档。docs = self.load()# 然后使用_text_splitter来分割每一个文档。return _text_splitter.split_documents(docs)

三、使用 TextLoader 加载 Txt 文件

TextLoader 是用于加载简单 .txt 文件的加载器。以下是一个加载 .txt 文件的示例：

class TextLoader(BaseLoader):def __init__(self, filepath):self.filepath = filepath
def load(self):with open(self.filepath, 'r', encoding='utf-8') as file:text = file.read()return [Document(text)]
# 示例使用text_loader = TextLoader('example.txt')documents = text_loader.load()print(documents[0].summary())# 输出文件的前100个字符...

四、使用 ArxivLoader 加载 ArXiv 论文

ArxivLoader 类专门用于从 ArXiv 平台获取文档。用户提供一个搜索查询，加载器与 ArXiv API 交互，检索相关文档列表并返回 Document 对象。以下是一个示例：

class ArxivLoader(BaseLoader):def __init__(self, query):self.query = query
def load(self):# 模拟与 ArXiv API 的交互results = self.query_arxiv(self.query)return [Document(result['summary'], result) for result in results]
def query_arxiv(self, query):# 伪代码，实际需要调用 ArXiv APIreturn [{'summary': '论文1摘要...', 'title': '论文1标题', 'authors': '作者1'},{'summary': '论文2摘要...', 'title': '论文2标题', 'authors': '作者2'}]
# 示例使用arxiv_loader = ArxivLoader('deep learning')documents = arxiv_loader.load()for doc in documents:print(doc.summary())

五、使用 UnstructuredURLLoader 加载网页内容

UnstructuredURLLoader 使用非结构化分区函数（Unstructured）来检测 MIME 类型并将文件路由到适当的分区器。它支持两种模式："single" 和 "elements"。以下是一个加载网页内容的示例：

from unstructured.partition.auto import partition
class UnstructuredURLLoader(BaseLoader):def __init__(self, urls, mode='single', continue_on_failure=True, **kwargs):self.urls = urlsself.mode = modeself.continue_on_failure = continue_on_failureself.kwargs = kwargs
def load(self):documents = []for url in self.urls:try:partitioned_text = partition(url, **self.kwargs)if self.mode == 'single':documents.append(Document(partitioned_text))elif self.mode == 'elements':for element in partitioned_text:documents.append(Document(element))except Exception as e:if not self.continue_on_failure:raise eprint(f"Failed to load {url}: {e}")return documents
# 示例使用urls = ['https://react-lm.github.io/']url_loader = UnstructuredURLLoader(urls, mode='elements')documents = url_loader.load()for doc in documents:print(doc.summary())