数据管道是 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 数据管道是系统工程,核心原则是:
...