数据管道是 RAG 系统的地基

RAG 系统的效果上限由数据质量决定。再好的 Embedding 模型和 LLM，如果喂进去的是垃圾数据，出来的也是垃圾答案。2026 年的 RAG 数据管道已经发展为一套完整的工程体系。

整体架构

数据源              解析层              处理层              索引层
┌─────────┐    ┌───────────┐    ┌───────────┐    ┌───────────┐
│  PDF    │───→│           │    │           │    │           │
│  HTML   │───→│  文档解析  │───→│  清洗+增强 │───→│  Embedding │
│  Word   │───→│  Unified  │    │  Pipeline  │    │  + Index  │
│  DB     │───→│  Parser   │    │           │    │           │
│  API    │───→│           │    │           │    │           │
│  Web    │───→│           │    │           │    │           │
└─────────┘    └───────────┘    └───────────┘    └───────────┘
                                      ↓
                               ┌───────────┐
                               │  质量监控  │
                               │  QC Layer │
                               └───────────┘

1. 多源文档解析

统一解析层

from abc import ABC, abstractmethod
from typing import List, Document
import fitz  # PyMuPDF
from bs4 import BeautifulSoup
from unstructured import partition_pdf

class DocumentParser(ABC):
    @abstractmethod
    def parse(self, file_path: str) -> List[Document]:
        pass

class PDFParser(DocumentParser):
    def parse(self, file_path: str) -> List[Document]:
        # 方案1: PyMuPDF（快，但结构感知弱）
        # 方案2: Unstructured（慢，但结构感知强）
        # 方案3: LLM 辅助解析（最准，但贵）
        
        elements = partition_pdf(
            file_path,
            strategy="hi_res",  # 高精度模式
            infer_table_structure=True,
            extract_image_block_types=["Image", "Table"],
            extract_image_block_output_dir="./images/"
        )
        
        documents = []
        for elem in elements:
            doc = Document(
                content=elem.text if hasattr(elem, 'text') else str(elem),
                metadata={
                    "source": file_path,
                    "page": elem.metadata.page_number if elem.metadata else 0,
                    "category": elem.category,
                    "type": self._map_type(elem.category)
                }
            )
            documents.append(doc)
        
        return documents

class HTMLParser(DocumentParser):
    def parse(self, file_path: str) -> List[Document]:
        with open(file_path, 'r', encoding='utf-8') as f:
            soup = BeautifulSoup(f.read(), 'html.parser')
        
        # 移除无关元素
        for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
            tag.decompose()
        
        documents = []
        
        # 按标题分块
        current_section = {"title": "", "content": []}
        
        for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'li', 'table', 'pre']):
            if element.name in ['h1', 'h2', 'h3']:
                if current_section["content"]:
                    documents.append(Document(
                        content=self._format_section(current_section),
                        metadata={"source": file_path, "section": current_section["title"]}
                    ))
                current_section = {"title": element.get_text(), "content": []}
            else:
                current_section["content"].append(self._extract_content(element))
        
        if current_section["content"]:
            documents.append(Document(
                content=self._format_section(current_section),
                metadata={"source": file_path, "section": current_section["title"]}
            ))
        
        return documents

class DatabaseParser(DocumentParser):
    def parse(self, conn_config: dict) -> List[Document]:
        """从数据库提取知识"""
        import sqlalchemy as sa
        
        engine = sa.create_engine(conn_config["url"])
        
        documents = []
        for table_name in conn_config["tables"]:
            query = f"SELECT * FROM {table_name}"
            df = pd.read_sql(query, engine)
            
            # 将每行转为一个文档
            for _, row in df.iterrows():
                content = self._row_to_text(row, table_name)
                documents.append(Document(
                    content=content,
                    metadata={
                        "source": f"db://{table_name}",
                        "table": table_name,
                        "primary_key": str(row.get('id', ''))
                    }
                ))
        
        return documents

class UnifiedParser:
    """统一解析入口"""
    def __init__(self):
        self.parsers = {
            ".pdf": PDFParser(),
            ".html": HTMLParser(),
            ".htm": HTMLParser(),
            ".docx": DocxParser(),
            ".md": MarkdownParser(),
            ".txt": TextParser(),
        }
    
    def parse(self, file_path: str) -> List[Document]:
        ext = Path(file_path).suffix.lower()
        parser = self.parsers.get(ext, TextParser())
        return parser.parse(file_path)

2. 数据清洗 Pipeline

class DataCleaner:
    def __init__(self):
        self.steps = [
            self._remove_boilerplate,    # 去除样板文字
            self._normalize_whitespace,  # 规范化空白
            self._fix_encoding,          # 修复编码
            self._remove_duplicates,     # 去重
            self._filter_quality,        # 质量过滤
            self._add_metadata,          # 补充元数据
        ]
    
    def clean(self, documents: List[Document]) -> List[Document]:
        for step in self.steps:
            documents = step(documents)
        return documents
    
    def _remove_boilerplate(self, docs):
        """去除样板文字"""
        boilerplate_patterns = [
            r"版权所有.*?保留所有权利",
            r"Cookie.*?设置",
            r"订阅我们的.*?通讯",
            r"© \d{4}.*?",
        ]
        
        cleaned = []
        for doc in docs:
            text = doc.content
            for pattern in boilerplate_patterns:
                text = re.sub(pattern, "", text, flags=re.IGNORECASE)
            doc.content = text.strip()
            if len(doc.content) > 50:  # 过滤过短的
                cleaned.append(doc)
        
        return cleaned
    
    def _filter_quality(self, docs):
        """质量过滤"""
        filtered = []
        for doc in docs:
            # 检查文本质量
            quality_score = self._assess_quality(doc.content)
            if quality_score > 0.5:
                doc.metadata["quality_score"] = quality_score
                filtered.append(doc)
        
        return filtered
    
    def _assess_quality(self, text: str) -> float:
        score = 1.0
        
        # 过短
        if len(text) < 50:
            score -= 0.3
        
        # 重复内容过多
        words = text.split()
        if len(words) > 0:
            unique_ratio = len(set(words)) / len(words)
            if unique_ratio < 0.3:
                score -= 0.3
        
        # 特殊字符过多
        special_chars = sum(1 for c in text if not c.isalnum() and not c.isspace())
        if len(text) > 0 and special_chars / len(text) > 0.2:
            score -= 0.2
        
        # 乱码检测
        if re.search(r'[\ufffd]{3,}', text):
            score -= 0.5
        
        return max(0, score)

3. 数据增强

class DataAugmentor:
    """为原始文档增加结构化信息"""
    
    def augment(self, documents: List[Document]) -> List[Document]:
        for doc in documents:
            # 1. 生成摘要
            doc.metadata["summary"] = self._generate_summary(doc.content)
            
            # 2. 提取关键词
            doc.metadata["keywords"] = self._extract_keywords(doc.content)
            
            # 3. 生成问答对（用于 FAQ 式检索）
            doc.metadata["qa_pairs"] = self._generate_qa_pairs(doc.content)
            
            # 4. 实体标注
            doc.metadata["entities"] = self._extract_entities(doc.content)
            
            # 5. 生成假设性问题（HyDE 优化）
            doc.metadata["hypothetical_questions"] = self._generate_hyde(doc.content)
        
        return documents
    
    def _generate_hyde(self, content: str) -> List[str]:
        """生成该内容可能回答的问题，用于提升检索召回"""
        prompt = f"""
基于以下内容，生成 3 个用户可能会问的问题，这些问题可以由这段内容回答：

内容：{content[:1000]}

输出 JSON 列表：["问题1", "问题2", "问题3"]
"""
        return llm.generate(prompt, response_format="json")

4. 分块与索引

class ChunkingAndIndexing:
    def __init__(self):
        self.chunker = HybridChunker()
        self.embedder = EmbeddingModel("bge-m3")
        self.sparse_embedder = BM25Encoder()
    
    def process(self, documents: List[Document]):
        all_chunks = []
        
        for doc in documents:
            # 1. 智能分块
            chunks = self.chunker.chunk(doc.content)
            
            # 2. 为每个 chunk 添加上下文
            for i, chunk in enumerate(chunks):
                chunk.metadata = {
                    **doc.metadata,
                    "chunk_index": i,
                    "total_chunks": len(chunks),
                    "context_before": chunks[i-1].text[-100:] if i > 0 else "",
                    "context_after": chunks[i+1].text[:100] if i < len(chunks)-1 else ""
                }
                
                # 3. 生成 Embedding（稠密 + 稀疏）
                chunk.dense_embedding = self.embedder.encode(chunk.text)
                chunk.sparse_embedding = self.sparse_embedder.encode(chunk.text)
                
                # 4. HyDE embedding（用假设问题做额外 embedding）
                if "hypothetical_questions" in doc.metadata:
                    hyde_embeddings = [
                        self.embedder.encode(q)
                        for q in doc.metadata["hypothetical_questions"]
                    ]
                    chunk.hyde_embeddings = hyde_embeddings
                
                all_chunks.append(chunk)
        
        # 5. 写入向量数据库
        self._write_to_index(all_chunks)
        
        return all_chunks

5. 质量监控

class PipelineQualityMonitor:
    def __init__(self):
        self.metrics = {}
    
    def monitor(self, pipeline_run):
        report = {
            "input": {
                "total_documents": len(pipeline_run.input_docs),
                "total_size_mb": pipeline_run.input_size,
            },
            "parsing": {
                "success_rate": pipeline_run.parsed / pipeline_run.total,
                "failed_files": pipeline_run.failed,
                "avg_parse_time": pipeline_run.avg_parse_time,
            },
            "cleaning": {
                "retention_rate": len(pipeline_run.cleaned) / len(pipeline_run.parsed),
                "avg_quality_score": np.mean([d.metadata.get("quality_score", 0) for d in pipeline_run.cleaned]),
            },
            "chunking": {
                "total_chunks": len(pipeline_run.chunks),
                "avg_chunk_size": np.mean([len(c.text) for c in pipeline_run.chunks]),
                "chunks_per_doc": len(pipeline_run.chunks) / len(pipeline_run.cleaned),
            },
            "indexing": {
                "index_time": pipeline_run.index_time,
                "index_size_mb": pipeline_run.index_size,
                "vectors_written": len(pipeline_run.chunks),
            }
        }
        
        # 告警
        if report["parsing"]["success_rate"] < 0.95:
            alert("解析成功率低于 95%")
        if report["cleaning"]["avg_quality_score"] < 0.7:
            alert("平均质量分数低于 0.7")
        if report["chunking"]["avg_chunk_size"] > 2000:
            alert("平均块大小过大")
        
        return report

6. 增量更新与调度

class IncrementalPipeline:
    """支持增量更新的数据管道"""
    
    def run(self, data_source: str):
        # 1. 发现新增/修改/删除的文件
        changes = self._detect_changes(data_source)
        
        if not changes["added"] and not changes["modified"] and not changes["deleted"]:
            return "No changes detected"
        
        # 2. 处理新增和修改
        new_docs = []
        for file_path in changes["added"] + changes["modified"]:
            docs = self.parser.parse(file_path)
            docs = self.cleaner.clean(docs)
            docs = self.augmentor.augment(docs)
            new_docs.extend(docs)
        
        # 3. 处理删除
        for file_path in changes["deleted"]:
            self.index.delete_by_source(file_path)
        
        # 4. 更新索引
        if new_docs:
            self.indexer.process(new_docs)
        
        # 5. 记录变更
        self._update_file_registry(changes)
        
        # 6. 质量检查
        quality_report = self.monitor.monitor(pipeline_run)
        
        return quality_report
    
    def _detect_changes(self, source: str) -> dict:
        """检测文件变更"""
        current_files = self._scan_files(source)
        registered_files = self._get_registered_files()
        
        added = set(current_files) - set(registered_files)
        deleted = set(registered_files) - set(current_files)
        modified = {
            f for f in current_files if f in registered_files
            and current_files[f] != registered_files[f]  # 比较 hash
        }
        
        return {"added": list(added), "modified": list(modified), "deleted": list(deleted)}

总结

RAG 数据管道是系统工程，核心原则是：

  1. 统一解析：一套解析器处理所有格式
  2. 多层清洗：从样板去除到质量评估
  3. 数据增强：摘要、关键词、HyDE 等附加信息大幅提升检索效果
  4. 增量更新：不要全量重建，只处理变更
  5. 质量监控：每个环节都要有度量和告警

2026 年的趋势是数据工程占 RAG 项目工作量的 60-70%，模型选型只占 10%。把数据管道做好，RAG 系统就成功了 80%。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。