3、数据源接入：常见数据源类型与适配器设计

聊到RAG系统的数据管道，第一个绕不开的坎就是——数据从哪来？

我见过不少团队，模型选型、向量库搭建都搞定了，结果卡在数据接入这一步。说白了，数据源接入是整个管道的「水龙头」。水龙头没拧对，后面再好的管道也白搭。

3.1 常见数据源类型

实际项目中，我接触到的数据源大概分四类。每一类都有它的脾气。

3.1.1 网页数据

网页是最常见的数据源。你想想看，公司官网、知识库、技术文档，大部分都在网页上。

我个人习惯用 requests + BeautifulSoup 做轻量抓取。但要注意，很多网站有反爬机制。我在项目中遇到过，某客户的知识库页面用了动态加载，直接请求拿不到内容。后来改用 Selenium 模拟浏览器才搞定。

import requests
from bs4 import BeautifulSoup

def fetch_web_page(url):
    headers = {'User-Agent': 'Mozilla/5.0'}
    resp = requests.get(url, headers=headers, timeout=10)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, 'html.parser')
    # 提取正文，去掉导航、广告等干扰
    content = soup.find('article') or soup.find('main')
    return content.get_text(strip=True) if content else ''

3.1.2 PDF 文档

PDF 是 RAG 系统里最头疼的数据源，没有之一。

为什么？因为 PDF 的「排版」和「内容」是混在一起的。有的 PDF 是扫描件，有的带表格，有的有复杂的页眉页脚。我建议优先用 PyMuPDF（也就是 fitz），它对文本提取的支持比较稳定。

import fitz  # PyMuPDF

def extract_pdf_text(pdf_path):
    doc = fitz.open(pdf_path)
    text = ""
    for page in doc:
        text += page.get_text()
    return text

3.1.3 数据库

数据库接入，说白了就是写 SQL。但这里有个坑——数据量。

我记得有一次，客户的生产库有上亿条记录。直接全量读取，内存直接爆了。后来我改成分批读取，每次取 10000 条，用 yield 做成生成器，问题就解决了。

import pymysql

def fetch_from_db(batch_size=10000):
    conn = pymysql.connect(host='...', user='...', password='...', db='...')
    cursor = conn.cursor()
    cursor.execute("SELECT id, content FROM documents")
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break
        for row in rows:
            yield {'id': row[0], 'content': row[1]}
    cursor.close()
    conn.close()

3.1.4 API 接口

API 接入是最「干净」的数据源。JSON 格式，结构清晰，不用清洗太多。

但要注意限流。我遇到过某个第三方 API，每分钟只能调用 60 次。解决方案是加一个 time.sleep(1)，或者用 ratelimit 库做限流装饰器。

import requests
from time import sleep

def fetch_from_api(api_url, api_key, max_retries=3):
    headers = {'Authorization': f'Bearer {api_key}'}
    for attempt in range(max_retries):
        resp = requests.get(api_url, headers=headers)
        if resp.status_code == 200:
            return resp.json()
        sleep(2 ** attempt)  # 指数退避
    raise Exception("API 请求失败")

3.2 数据源适配器设计

你想想看，如果每个数据源都写一套独立的接入逻辑，那代码会乱成什么样？

所以我们需要适配器模式。说白了，就是定义一个统一的接口，让所有数据源都「长一个样」。

3.2.1 适配器接口定义

我个人习惯先定义一个抽象基类，规定好必须实现的方法：

from abc import ABC, abstractmethod
from typing import Iterator, Dict

class DataSourceAdapter(ABC):
    @abstractmethod
    def connect(self) -> bool:
        """建立连接"""
        pass
    
    @abstractmethod
    def fetch(self) -> Iterator[Dict]:
        """获取数据，返回迭代器"""
        pass
    
    @abstractmethod
    def close(self):
        """关闭连接"""
        pass

3.2.2 具体适配器实现

有了接口，每个数据源各自实现自己的逻辑。比如网页适配器：

class WebAdapter(DataSourceAdapter):
    def __init__(self, url):
        self.url = url
        self.session = None
    
    def connect(self):
        self.session = requests.Session()
        return True
    
    def fetch(self):
        resp = self.session.get(self.url)
        soup = BeautifulSoup(resp.text, 'html.parser')
        yield {'source': self.url, 'content': soup.get_text()}
    
    def close(self):
        self.session.close()

PDF 适配器类似，只是内部调用 PyMuPDF。数据库适配器则封装 SQL 查询逻辑。

3.2.3 适配器注册与工厂

我建议再加一个工厂类，根据数据源类型自动选择适配器：

class AdapterFactory:
    _adapters = {
        'web': WebAdapter,
        'pdf': PDFAdapter,
        'db': DatabaseAdapter,
        'api': APIAdapter,
    }
    
    @classmethod
    def create(cls, source_type: str, **kwargs) -> DataSourceAdapter:
        adapter_cls = cls._adapters.get(source_type)
        if not adapter_cls:
            raise ValueError(f"不支持的数据源类型: {source_type}")
        return adapter_cls(**kwargs)

3.3 本章知识体系

下面这张图，是我自己画的数据源接入整体架构。你可以看到，适配器层就像「万能插座」，把各种数据源统一成标准格式，交给下游的清洗模块。

嗯，数据源接入这块，核心就这些。记住一句话：统一接口，隔离变化。下一层就是数据清洗了，到时候我们再细聊。

---

公众号:蓝海资料掘金营，微信deep3321