2.1 文档加载器

步骤4：API数据加载器

REST API加载器：

import requests
import json
from typing import List, Dict, Any, Optional
from haystack import Document
from urllib.parse import urljoin
import time

class APIDocumentLoader:
    """REST API文档加载器"""
    
    def __init__(self, base_url: str, auth_token: str = None, 
                 headers: Dict = None, timeout: int = 30):
        self.base_url = base_url.rstrip('/')
        self.auth_token = auth_token
        self.headers = headers or {}
        self.timeout = timeout
        
        if self.auth_token:
            self.headers['Authorization'] = f'Bearer {self.auth_token}'
        
        self.headers['Content-Type'] = 'application/json'
    
    def _make_request(self, endpoint: str, params: Dict = None, 
                     method: str = 'GET') -> Dict:
        """发送API请求"""
        url = urljoin(self.base_url, endpoint)
        
        try:
            if method.upper() == 'GET':
                response = requests.get(
                    url,
                    headers=self.headers,
                    params=params,
                    timeout=self.timeout
                )
            elif method.upper() == 'POST':
                response = requests.post(
                    url,
                    headers=self.headers,
                    json=params,
                    timeout=self.timeout
                )
            else:
                raise ValueError(f"不支持的HTTP方法: {method}")
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            print(f"API请求失败 {url}: {e}")
            return {}
    
    def load_from_paginated_api(self, endpoint: str, 
                              content_field: str,
                              id_field: str = 'id',
                              page_param: str = 'page',
                              size_param: str = 'size',
                              max_pages: int = 10) -> List[Document]:
        """从分页API加载文档"""
        documents = []
        page = 1
        
        while page <= max_pages:
            params = {
                page_param: page,
                size_param: 100  # 每页大小
            }
            
            data = self._make_request(endpoint, params)
            
            if not data:
                break
            
            items = self._extract_items(data, content_field)
            
            if not items:
                break
            
            for item in items:
                content = str(item.get(content_field, ''))
                if content.strip():
                    doc = Document(
                        content=content,
                        meta={
                            "source": f"api:{endpoint}",
                            "api_id": item.get(id_field),
                            "raw_data": item
                        }
                    )
                    documents.append(doc)
            
            page += 1
            time.sleep(1)  # 避免请求过于频繁
        
        return documents
    
    def _extract_items(self, data: Dict, content_field: str) -> List[Dict]:
        """从API响应中提取项目"""
        items = []
        
        # 处理不同的响应格式
        if isinstance(data, dict):
            # 检查常见的分页字段
            if 'data' in data:
                items.extend(data['data'])
            elif 'items' in data:
                items.extend(data['items'])
            elif 'results' in data:
                items.extend(data['results'])
            elif 'content' in data:
                items.extend(data['content'])
            elif 'hits' in data and 'hits' in data['hits']:
                items.extend(data['hits']['hits'])
            else:
                # 如果没有标准格式，尝试找到包含content_field的对象
                for key, value in data.items():
                    if isinstance(value, dict) and content_field in value:
                        items.append(value)
        elif isinstance(data, list):
            items.extend(data)
        
        return items
    
    def load_search_api(self, search_endpoint: str, 
                       query_param: str = 'q',
                       content_field: str = 'content',
                       filters: Dict = None) -> List[Document]:
        """从搜索API加载文档"""
        documents = []
        
        # 示例：加载多个搜索查询的结果
        search_queries = [
            "机器学习",
            "深度学习", 
            "自然语言处理",
            "计算机视觉",
            "人工智能"
        ]
        
        for query in search_queries:
            params = {query_param: query}
            
            if filters:
                params.update(filters)
            
            data = self._make_request(search_endpoint, params)
            items = self._extract_items(data, content_field)
            
            for item in items:
                content = str(item.get(content_field, ''))
                if content.strip():
                    doc = Document(
                        content=content,
                        meta={
                            "source": f"search:{search_endpoint}",
                            "query": query,
                            "search_result": item
                        }
                    )
                    documents.append(doc)
        
        return documents
    
    def load_webhook_data(self, webhook_url: str, 
                         secret_token: str = None,
                         payload: Dict = None) -> List[Document]:
        """从Webhook加载文档"""
        headers = self.headers.copy()
        
        if secret_token:
            headers['X-Webhook-Secret'] = secret_token
        
        try:
            response = requests.post(
                webhook_url,
                headers=headers,
                json=payload,
                timeout=self.timeout
            )
            response.raise_for_status()
            
            data = response.json()
            
            # 处理Webhook响应
            documents = []
            
            if isinstance(data, list):
                for item in data:
                    content = str(item.get('content', item.get('text', '')))
                    if content.strip():
                        doc = Document(
                            content=content,
                            meta={"source": "webhook", "raw_data": item}
                        )
                        documents.append(doc)
            elif isinstance(data, dict):
                content = str(data.get('content', data.get('text', '')))
                if content.strip():
                    doc = Document(
                        content=content,
                        meta={"source": "webhook", "raw_data": data}
                    )
                    documents.append(doc)
            
            return documents
            
        except Exception as e:
            print(f"Webhook加载失败: {e}")
            return []

# 使用示例
api_loader = APIDocumentLoader(
    base_url="https://api.example.com",
    auth_token="your_auth_token_here"
)

# 1. 从分页API加载
paginated_docs = api_loader.load_from_paginated_api(
    endpoint="/articles",
    content_field="content",
    id_field="article_id",
    page_param="page",
    size_param="per_page"
)
print(f"分页API加载完成，{len(paginated_docs)} 个文档")

# 2. 从搜索API加载
search_docs = api_loader.load_search_api(
    search_endpoint="/search",
    query_param="query",
    content_field="snippet",
    filters={"type": "article", "status": "published"}
)
print(f"搜索API加载完成，{len(search_docs)} 个文档")

# 3. 从Webhook加载数据
webhook_docs = api_loader.load_webhook_data(
    webhook_url="https://your-webhook-endpoint.com/content",
    secret_token="webhook_secret",
    payload={"event": "new_content", "source": "cms"}
)
print(f"Webhook加载完成，{len(webhook_docs)} 个文档")
​

常见问题 FAQ

Q1：大文件处理时内存不足怎么办？

A： 大文件处理可能消耗大量内存，解决方案包括：

import pypdf
from typing import Iterator

def process_large_pdf(file_path: str, chunk_size: int = 100) -> Iterator[Document]:
    """分块处理大PDF文件"""
    try:
        with open(file_path, 'rb') as file:
            pdf_reader = pypdf.PdfReader(file)
            
            for page_num in range(len(pdf_reader.pages)):
                page = pdf_reader.pages[page_num]
                text = page.extract_text()
                
                if text.strip():
                    # 分块处理
                    words = text.split()
                    for i in range(0, len(words), chunk_size):
                        chunk = ' '.join(words[i:i + chunk_size])
                        
                        yield Document(
                            content=chunk,
                            meta={
                                "source": file_path,
                                "page": page_num + 1,
                                "chunk_start": i,
                                "chunk_end": min(i + chunk_size, len(words))
                            }
                        )
    except Exception as e:
        print(f"PDF处理失败: {e}")

# 使用示例
for doc in process_large_pdf("large_document.pdf"):
    print(f"处理页面 {doc.meta['page']}, 块 {doc.meta['chunk_start']}-{doc.meta['chunk_end']}")
​

Q2：如何处理编码问题？

A： 文档编码问题是常见问题，解决方案：

import chardet
from typing import Optional

def detect_encoding(file_path: str) -> Optional[str]:
    """检测文件编码"""
    try:
        with open(file_path, 'rb') as file:
            raw_data = file.read(1024)  # 读取前1KB用于检测
            result = chardet.detect(raw_data)
            return result['encoding']
    except Exception as e:
        print(f"编码检测失败: {e}")
        return None

def read_file_with_encoding(file_path: str, encoding: str = None) -> str:
    """使用指定编码读取文件"""
    if encoding is None:
        encoding = detect_encoding(file_path) or 'utf-8'
    
    try:
        with open(file_path, 'r', encoding=encoding) as file:
            return file.read()
    except UnicodeDecodeError:
        # 如果检测的编码不正确，尝试常见编码
        for alt_encoding in ['utf-8', 'gbk', 'gb2312', 'latin1']:
            try:
                with open(file_path, 'r', encoding=alt_encoding) as file:
                    return file.read()
            except UnicodeDecodeError:
                continue
    except Exception as e:
        print(f"文件读取失败: {e}")
    
    return ""

# 使用示例
encoding = detect_encoding("unknown_encoding_file.txt")
print(f"检测到的编码: {encoding}")

content = read_file_with_encoding("unknown_encoding_file.txt")
print(f"文件内容长度: {len(content)}")
​

Q3：如何处理重复文档？

A： 文档重复问题可以通过多种方式解决：

import hashlib
from typing import List, Dict
from haystack import Document

class DuplicateRemover:
    """重复文档移除器"""
    
    def __init__(self, strategy: str = "content_hash"):
        self.strategy = strategy
        self.seen_hashes = set()
    
    def remove_duplicates(self, documents: List[Document]) -> List[Document]:
        """移除重复文档"""
        unique_docs = []
        
        for doc in documents:
            doc_hash = self._get_document_hash(doc)
            
            if doc_hash not in self.seen_hashes:
                unique_docs.append(doc)
                self.seen_hashes.add(doc_hash)
            else:
                print(f"发现重复文档: {doc.id}")
        
        return unique_docs
    
    def _get_document_hash(self, doc: Document) -> str:
        """获取文档哈希值"""
        if self.strategy == "content_hash":
            # 基于内容哈希
            content = doc.content.lower().strip()
            return hashlib.md5(content.encode()).hexdigest()
        
        elif self.strategy == "metadata_hash":
            # 基于元数据哈希
            metadata_str = str(sorted(doc.meta.items()))
            return hashlib.md5(metadata_str.encode()).hexdigest()
        
        elif self.strategy == "combined_hash":
            # 组合哈希
            combined = f"{doc.content.lower().strip()}|{str(sorted(doc.meta.items()))}"
            return hashlib.md5(combined.encode()).hexdigest()
        
        else:
            raise ValueError(f"不重复检测策略: {self.strategy}")

# 使用示例
remover = DuplicateRemover(strategy="content_hash")

# 测试重复文档
doc1 = Document(content="相同的文档内容", meta={"source": "file1.txt"})
doc2 = Document(content="相同的文档内容", meta={"source": "file2.txt"})
doc3 = Document(content="不同的文档内容", meta={"source": "file3.txt"})

unique_docs = remover.remove_duplicates([doc1, doc2, doc3])
print(f"去重后文档数: {len(unique_docs)}")
​

Q4：如何处理网络超时和重试？

A： 网络请求超时是常见问题，建议使用重试机制：

import time
import requests
from typing import Optional, Dict, Any
from requests.exceptions import RequestException

class APIClientWithRetry:
    """带重试机制的API客户端"""
    
    def __init__(self, base_url: str, max_retries: int = 3, 
                 initial_timeout: int = 30, backoff_factor: float = 2.0):
        self.base_url = base_url
        self.max_retries = max_retries
        self.initial_timeout = initial_timeout
        self.backoff_factor = backoff_factor
    
    def get(self, endpoint: str, params: Dict = None, 
            headers: Dict = None) -> Optional[Dict]:
        """带重试的GET请求"""
        url = f"{self.base_url}/{endpoint}"
        timeout = self.initial_timeout
        
        for attempt in range(self.max_retries + 1):
            try:
                response = requests.get(
                    url,
                    params=params,
                    headers=headers or {},
                    timeout=timeout
                )
                response.raise_for_status()
                return response.json()
                
            except RequestException as e:
                if attempt == self.max_retries:
                    print(f"API请求最终失败: {e}")
                    return None
                
                wait_time = self.backoff_factor ** attempt
                print(f"请求失败，{wait_time}秒后重试... (尝试 {attempt + 1}/{self.max_retries})")
                time.sleep(wait_time)
                timeout *= self.backoff_factor

# 使用示例
api_client = APIClientWithRetry(
    base_url="https://api.example.com",
    max_retries=3,
    initial_timeout=10
)

# 带重试的API调用
data = api_client.get("/articles", params={"page": 1})
if data:
    print(f"API调用成功: {len(data)} 条记录")
else:
    print("API调用失败")
​

Q5：如何优化批量处理性能？

A： 批量处理性能优化可以从多个方面入手：

import asyncio
import aiohttp
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
from haystack import Document

class BatchProcessor:
    """批量处理器"""
    
    def __init__(self, max_workers: int = 4, batch_size: int = 100):
        self.max_workers = max_workers
        self.batch_size = batch_size
    
    def process_documents_batch(self, documents: List[Document]) -> List[Document]:
        """批量处理文档"""
        processed_docs = []
        
        for i in range(0, len(documents), self.batch_size):
            batch = documents[i:i + self.batch_size]
            
            # 并行处理当前批次
            with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                futures = [
                    executor.submit(self._process_single_doc, doc) 
                    for doc in batch
                ]
                
                for future in as_completed(futures):
                    try:
                        processed_doc = future.result()
                        if processed_doc:
                            processed_docs.append(processed_doc)
                    except Exception as e:
                        print(f"文档处理失败: {e}")
        
        return processed_docs
    
    def _process_single_doc(self, document: Document) -> Document:
        """处理单个文档"""
        # 简单的内容清洗
        content = document.content.strip()
        if len(content) > 10000:
            content = content[:10000]  # 截断过长内容
        
        # 更新文档
        document.content = content
        document.meta['processed_at'] = self._get_current_time()
        
        return document
    
    def _get_current_time(self) -> str:
        """获取当前时间"""
        from datetime import datetime
        return datetime.now().isoformat()

# 异步批量处理
async def async_batch_processing(documents: List[Document], 
                                 batch_size: int = 50) -> List[Document]:
    """异步批量处理"""
    processed_docs = []
    
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        
        # 异步处理批次
        tasks = [process_document_async(doc) for doc in batch]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for result in results:
            if isinstance(result, Document):
                processed_docs.append(result)
            elif isinstance(result, Exception):
                print(f"处理失败: {result}")
    
    return processed_docs

async def process_document_async(document: Document) -> Document:
    """异步处理单个文档"""
    # 模拟异步处理
    await asyncio.sleep(0.1)
    
    content = document.content.strip()
    if len(content) > 5000:
        content = content[:5000]
    
    document.content = content
    document.meta['async_processed'] = True
    
    return document

# 使用示例
processor = BatchProcessor(max_workers=4, batch_size=50)

# 创建测试文档
test_docs = [Document(content=f"文档内容 {i}", meta={"id": i}) for i in range(100)]

# 批量处理
processed = processor.process_documents_batch(test_docs)
print(f"批量处理完成，共 {len(processed)} 个文档")
​

最佳实践与避坑

实践1：性能优化策略

• 批量处理：使用批量操作减少I/O开销
• 并行处理：利用多线程/异步提高处理效率
• 缓存机制：对重复访问的数据进行缓存
• 内存管理：及时释放不再使用的资源

实践2：错误处理健壮性

• 分级重试：对不同类型的错误采用不同的重试策略
• 错误隔离：单个文档处理失败不影响整体流程
• 日志记录：详细记录错误信息和处理状态
• 优雅降级：在服务不可用时提供基本功能

实践3：配置管理

• 环境配置：区分开发、测试、生产环境
• 参数验证：启动时验证所有配置参数
• 动态更新：支持运行时更新配置
• 配置加密：敏感信息加密存储

坑点1：文件权限问题

• 问题表现：无法读取某些文件
• 解决方案：检查文件权限，使用适当的访问权限
• 预防措施：统一文件权限标准，定期检查

坑点2：网络连接不稳定

• 问题表现：API请求频繁失败
• 解决方案：实现重试机制和超时控制
• 预防措施：建立连接池，监控网络状态

坑点3：数据格式不一致

• 问题表现：不同来源的数据格式差异很大
• 解决方案：建立数据标准化流程
• 预防措施：定义统一的数据模型和验证规则

本节小结

本节详细介绍了Haystack文档加载器的完整实现，从基础的文件加载到复杂的多源数据统一处理。我们学习了如何处理不同格式的文档，如何构建企业级的文档处理系统，以及在实际应用中需要注意的各种问题。

文档加载器是RAG系统的数据基础，其质量直接影响后续的检索和生成效果。通过本节的学习，我们掌握了构建高效、可靠的文档处理系统的关键技术和最佳实践。

下一节我们将学习2.2文档预处理技术，深入研究如何对加载的文档进行清洗、分割和质量控制，为后续的检索和生成做好准备。

延伸阅读

• Haystack官方文档 - Document Loaders
• pypdf文档处理库
• python-docx Word文档处理
• SQLAlchemy数据库连接
• Requests HTTP客户端库

关键词：文档加载器, 多源数据处理, 文件格式, 数据库集成, API加载器, 企业级处理
难度：进阶
预计阅读：25分钟