2.2 文档加载器

2.2 文档加载器 — LangChain框架精通 数据处理与加载

本节导读：掌握LangChain中各种文档加载器的使用方法，学习从多种数据源加载、处理和转换文档内容的技术

学习目标

• 掌握LangChain文档加载器的核心概念和使用方法
• 学会从多种数据源加载文档内容
• 理解不同文档格式和加载策略
• 掌握文档预处理和转换技术
• 了解批量处理和性能优化方法

2.2.1 文档加载器概述

📚 什么是文档加载器

文档加载器是LangChain中负责从各种数据源读取并加载文档内容的核心组件。它们提供了统一的接口来处理不同格式的数据，包括文本文件、PDF、网页、数据库等。

文档加载器的核心价值

🏗️ 文档加载器架构

# 基础接口
from langchain.document_loaders.base import BaseLoader
from langchain.document_loaders.blob_loaders import BlobLoader
from langchain.schema import Document

# 文档加载器基类
class BaseDocumentLoader(BaseLoader):
    """文档加载器基类"""
    
    def load(self) -> List[Document]:
        """加载文档，返回Document对象列表"""
        pass
    
    def aload(self) -> Awaitable[List[Document]]:
        """异步加载文档"""
        pass

# 文档对象结构
class Document:
    """文档对象"""
    
    def __init__(self, page_content: str, metadata: dict = None):
        self.page_content = page_content  # 文档内容
        self.metadata = metadata or {}   # 元数据
​

2.2.2 文本文件加载器

📄 TextLoader

基础使用

from langchain.document_loaders import TextLoader

# 加载单个文本文件
loader = TextLoader("document.txt")
documents = loader.load()

# 查看文档内容
for doc in documents:
    print(f"内容: {doc.page_content[:100]}...")
    print(f"元数据: {doc.metadata}")

# 批量加载多个文本文件
loader = TextLoader(["doc1.txt", "doc2.txt", "doc3.txt"])
all_documents = loader.load()
​

高级配置

from langchain.document_loaders import TextLoader
from typing import List, Optional

class AdvancedTextLoader(TextLoader):
    """高级文本加载器"""
    
    def __init__(self, file_path: str, 
                 encoding: str = "utf-8",
                 errors: str = "ignore",
                 metadata: dict = None):
        super().__init__(file_path)
        self.encoding = encoding
        self.errors = errors
        self.metadata = metadata or {}
    
    def load(self) -> List[Document]:
        """加载文本文件"""
        try:
            with open(self.file_path, 'r', encoding=self.encoding, errors=self.errors) as f:
                content = f.read()
                
                # 合并元数据
                doc_metadata = {
                    "source": self.file_path,
                    "file_type": "text",
                    **self.metadata
                }
                
                return [Document(page_content=content, metadata=doc_metadata)]
        
        except Exception as e:
            return [Document(
                page_content=f"加载文件失败: {str(e)}",
                metadata={"error": str(e), "source": self.file_path}
            )]

# 使用示例
loader = AdvancedTextLoader(
    file_path="example.txt",
    encoding="utf-8",
    metadata={"author": "John Doe", "created": "2024-01-01"}
)
documents = loader.load()
​

📝 特定格式加载器

MarkdownLoader

from langchain.document_loaders import UnstructuredMarkdownLoader

# 加载Markdown文件
markdown_loader = UnstructuredMarkdownLoader("README.md")
markdown_docs = markdown_loader.load()

# 自定义Markdown处理
class CustomMarkdownLoader:
    """自定义Markdown加载器"""
    
    def __init__(self, file_path: str):
        self.file_path = file_path
    
    def load(self) -> List[Document]:
        import re
        
        with open(self.file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # 提取标题和内容
        sections = []
        current_section = None
        
        for line in content.split('\n'):
            # 检测标题
            if line.startswith('# '):
                if current_section:
                    sections.append(current_section)
                current_section = {
                    "title": line[2:],
                    "content": ""
                }
            elif current_section:
                current_section["content"] += line + "\n"
        
        if current_section:
            sections.append(current_section)
        
        # 创建文档
        documents = []
        for i, section in enumerate(sections):
            doc_content = f"# {section['title']}\n\n{section['content']}"
            metadata = {
                "title": section['title'],
                "section_number": i + 1,
                "source": self.file_path
            }
            documents.append(Document(page_content=doc_content, metadata=metadata))
        
        return documents

# 使用示例
md_loader = CustomMarkdownLoader("documentation.md")
documents = md_loader.load()
​

2.2.3 PDF文档加载器

📖 PyPDFLoader

基础使用

from langchain.document_loaders import PyPDFLoader

# 加载单个PDF文件
pdf_loader = PyPDFLoader("report.pdf")
pdf_documents = pdf_loader.load()

# 查看PDF文档
for i, doc in enumerate(pdf_documents):
    print(f"第{i+1}页: {doc.page_content[:100]}...")
    print(f"元数据: {doc.metadata}")

# 多页PDF处理
def process_pdf(file_path: str, chunk_size: int = 1000):
    """处理大型PDF文件"""
    loader = PyPDFLoader(file_path)
    documents = loader.load()
    
    # 分块处理
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=200
    )
    
    chunks = splitter.split_documents(documents)
    return chunks

# 使用示例
pdf_chunks = process_pdf("large_document.pdf", chunk_size=800)
print(f"PDF分块数量: {len(pdf_chunks)}")
​

高级PDF处理

import fitz  # PyMuPDF
from typing import List, Dict, Any
from langchain.schema import Document

class AdvancedPDFLoader:
    """高级PDF加载器"""
    
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.doc = fitz.open(file_path)
    
    def load_with_metadata(self) -> List[Document]:
        """加载PDF并包含详细元数据"""
        documents = []
        
        for page_num in range(len(self.doc)):
            page = self.doc[page_num]
            
            # 获取页面文本
            text = page.get_text()
            
            # 获取页面元数据
            page_metadata = {
                "page_number": page_num + 1,
                "total_pages": len(self.doc),
                "file_path": self.file_path,
                "page_size": page.rect.size,
                "rotation": page.rotation
            }
            
            # 提取链接
            links = page.get_links()
            if links:
                page_metadata["links"] = links
            
            # 提取图像
            images = page.get_images()
            if images:
                page_metadata["images_count"] = len(images)
            
            document = Document(
                page_content=text,
                metadata=page_metadata
            )
            documents.append(document)
        
        return documents
    
    def extract_tables(self) -> List[Dict]:
        """提取PDF中的表格"""
        tables = []
        
        for page_num in range(len(self.doc)):
            page = self.doc[page_num]
            
            # 查找表格
            tables_page = page.find_tables()
            for table in tables_page:
                table_data = {
                    "page_number": page_num + 1,
                    "rows": len(table),
                    "cols": len(table[0]) if table else 0,
                    "content": table.extract(),
                    "bbox": table.bbox
                }
                tables.append(table_data)
        
        return tables
    
    def search_text(self, search_term: str) -> List[Dict]:
        """搜索PDF中的文本"""
        results = []
        
        for page_num in range(len(self.doc)):
            page = self.doc[page_num]
            
            # 搜索文本
            search_results = page.search_for(search_term)
            
            for result in search_results:
                result_data = {
                    "page_number": page_num + 1,
                    "text": search_term,
                    "bbox": result,
                    "context": self._get_context(page, result)
                }
                results.append(result_data)
        
        return results
    
    def _get_context(self, page, bbox) -> str:
        """获取搜索结果的上下文"""
        # 获取文本区域
        text_area = page.get_textarea(bbox)
        return text_area
    
    def __del__(self):
        """清理资源"""
        if hasattr(self, 'doc'):
            self.doc.close()

# 使用示例
pdf_loader = AdvancedPDFLoader("document.pdf")

# 加载文档
documents = pdf_loader.load_with_metadata()

# 提取表格
tables = pdf_loader.extract_tables()

# 搜索文本
search_results = pdf_loader.search_text("机器学习")
​

多PDF批量处理

import asyncio
from typing import List
from concurrent.futures import ThreadPoolExecutor

class BatchPDFProcessor:
    """PDF批量处理器"""
    
    def __init__(self, max_workers: int = 4):
        self.max_workers = max_workers
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    def process_multiple_pdfs(self, pdf_paths: List[str], process_func=None) -> List[List[Document]]:
        """批量处理多个PDF文件"""
        if process_func is None:
            process_func = lambda x: PyPDFLoader(x).load()
        
        futures = []
        
        # 提交任务
        for pdf_path in pdf_paths:
            future = self.executor.submit(process_func, pdf_path)
            futures.append(future)
        
        # 收集结果
        results = []
        for future in futures:
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                results.append([Document(
                    page_content=f"处理失败: {str(e)}",
                    metadata={"error": str(e)}
                )])
        
        return results
    
    async def async_process_pdfs(self, pdf_paths: List[str]) -> List[List[Document]]:
        """异步批量处理PDF"""
        semaphore = asyncio.Semaphore(self.max_workers)
        tasks = []
        
        for pdf_path in pdf_paths:
            task = self._process_pdf_async(pdf_path, semaphore)
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理异常
        processed_results = []
        for result in results:
            if isinstance(result, Exception):
                processed_results.append([Document(
                    page_content=f"异步处理失败: {str(result)}",
                    metadata={"error": str(result)}
                )])
            else:
                processed_results.append(result)
        
        return processed_results
    
    async def _process_pdf_async(self, pdf_path: str, semaphore: asyncio.Semaphore) -> List[Document]:
        """异步处理单个PDF"""
        async with semaphore:
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(
                self.executor, 
                lambda: PyPDFLoader(pdf_path).load()
            )

# 使用示例
processor = BatchPDFProcessor(max_workers=4)

pdf_files = ["doc1.pdf", "doc2.pdf", "doc3.pdf"]

# 同步批量处理
all_documents = processor.process_multiple_pdfs(pdf_files)

# 异步批量处理
async def async_process():
    results = await processor.async_process_pdfs(pdf_files)
    return results

# 在实际异步环境中使用
# asyncio.run(async_process())
​

2.2.4 网页内容加载器

🌐 WebBaseLoader

基础使用

from langchain.document_loaders import WebBaseLoader

# 加载单个网页
web_loader = WebBaseLoader(["https://example.com"])
web_docs = web_loader.load()

# 加载多个网页
multi_loader = WebBaseLoader([
    "https://site1.com/article1",
    "https://site2.com/blog2", 
    "https://site3.com/docs3"
])
multi_docs = multi_loader.load()

# 查看网页内容
for doc in web_docs:
    print(f"URL: {doc.metadata.get('source', 'Unknown')}")
    print(f"标题: {doc.metadata.get('title', 'No title')}")
    print(f"内容: {doc.page_content[:200]}...")
​

网页内容提取优化

from bs4 import BeautifulSoup
from urllib.request import urlopen
from typing import List, Dict
import requests

class EnhancedWebLoader:
    """增强网页加载器"""
    
    def __init__(self, urls: List[str], 
                 timeout: int = 30,
                 headers: Dict[str, str] = None):
        self.urls = urls
        self.timeout = timeout
        self.headers = headers or {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        }
    
    def load(self) -> List[Document]:
        """加载网页内容"""
        documents = []
        
        for url in self.urls:
            try:
                doc = self._load_single_url(url)
                documents.append(doc)
            except Exception as e:
                error_doc = Document(
                    page_content=f"加载网页失败: {str(e)}",
                    metadata={"url": url, "error": str(e)}
                )
                documents.append(error_doc)
        
        return documents
    
    def _load_single_url(self, url: str) -> Document:
        """加载单个网页"""
        
        # 发送HTTP请求
        response = requests.get(
            url,
            headers=self.headers,
            timeout=self.timeout,
            allow_redirects=True
        )
        response.raise_for_status()
        
        # 解析HTML
        soup = BeautifulSoup(response.content, 'html.parser')
        
        # 提取主要内容
        content = self._extract_main_content(soup)
        
        # 提取元数据
        metadata = self._extract_metadata(soup, url)
        
        return Document(
            page_content=content,
            metadata=metadata
        )
    
    def _extract_main_content(self, soup: BeautifulSoup) -> str:
        """提取主要内容"""
        
        # 移除不需要的元素
        for element in soup(['script', 'style', 'nav', 'header', 'footer', 'aside']):
            element.decompose()
        
        # 尝试找到主要内容区域
        content_selectors = [
            'article', 'main', '[role="main"]', 
            '.content', '.main-content', '.post-content'
        ]
        
        content = ""
        for selector in content_selectors:
            element = soup.select_one(selector)
            if element:
                content = element.get_text(strip=True)
                break
        
        # 如果没有找到主要内容区域，提取body内容
        if not content:
            content = soup.body.get_text(strip=True) if soup.body else ""
        
        # 清理文本
        content = self._clean_text(content)
        
        return content
    
    def _extract_metadata(self, soup: BeautifulSoup, url: str) -> Dict:
        """提取网页元数据"""
        
        metadata = {
            "source": url,
            "title": "",
            "description": "",
            "author": "",
            "keywords": [],
            "language": "",
            "published_date": ""
        }
        
        # 提取标题
        title = soup.find('title')
        if title:
            metadata["title"] = title.get_text(strip=True)
        
        # 提取描述
        description = soup.find('meta', attrs={'name': 'description'})
        if description:
            metadata["description"] = description.get('content', '')
        
        # 提取作者
        author = soup.find('meta', attrs={'name': 'author'})
        if author:
            metadata["author"] = author.get('content', '')
        
        # 提取关键词
        keywords = soup.find('meta', attrs={'name': 'keywords'})
        if keywords:
            metadata["keywords"] = keywords.get('content', '').split(',')
        
        # 提取语言
        html = soup.find('html')
        if html:
            metadata["language"] = html.get('lang', '')
        
        # 提取发布日期
        published_date = soup.find('meta', attrs={'property': 'article:published_time'})
        if published_date:
            metadata["published_date"] = published_date.get('content', '')
        
        return metadata
    
    def _clean_text(self, text: str) -> str:
        """清理文本内容"""
        import re
        
        # 移除多余的空白字符
        text = re.sub(r'\s+', ' ', text)
        
        # 移除特殊字符
        text = re.sub(r'[^\w\s\u4e00-\u9fff.,;:!?()-]', '', text)
        
        # 移除URL
        text = re.sub(r'https?://\S+', '', text)
        
        # 移除邮件地址
        text = re.sub(r'\S+@\S+', '', text)
        
        return text.strip()

# 使用示例
web_loader = EnhancedWebLoader([
    "https://example.com/article",
    "https://news.example.com/tech"
])

documents = web_loader.load()
for doc in documents:
    print(f"标题: {doc.metadata['title']}")
    print(f"作者: {doc.metadata['author']}")
    print(f"内容长度: {len(doc.page_content)} 字符")
​

📰 新闻和博客加载器

from langchain.document_loaders import RSSFeedLoader

class NewsFeedLoader:
    """新闻RSS加载器"""
    
    def __init__(self, rss_urls: List[str]):
        self.rss_urls = rss_urls
    
    def load_news(self) -> List[Document]:
        """加载新闻内容"""
        documents = []
        
        for rss_url in self.rss_urls:
            try:
                loader = RSSFeedLoader(rss_url)
                news_docs = loader.load()
                
                # 添加RSS源信息
                for doc in news_docs:
                    if "metadata" not in doc.metadata:
                        doc.metadata["source_rss"] = rss_url
                        doc.metadata["source_type"] = "news"
                
                documents.extend(news_docs)
                
            except Exception as e:
                error_doc = Document(
                    page_content=f"RSS加载失败: {str(e)}",
                    metadata={"rss_url": rss_url, "error": str(e)}
                )
                documents.append(error_doc)
        
        return documents
    
    def filter_by_keywords(self, keywords: List[str]) -> List[Document]:
        """按关键词过滤新闻"""
        all_news = self.load_news()
        
        filtered_news = []
        for doc in all_news:
            # 检查标题和内容是否包含关键词
            content_lower = doc.page_content.lower()
            title_lower = doc.metadata.get("title", "").lower()
            
            if any(keyword.lower() in title_lower or keyword.lower() in content_lower 
                   for keyword in keywords):
                filtered_news.append(doc)
        
        return filtered_news

# 使用示例
news_loader = NewsFeedLoader([
    "https://fe
​