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

# 基于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]}"

# 基础加载器类。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)

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个字符...

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())

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())