数据管道

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

RAG 数据管道全景 RAG 系统的效果，70% 取决于数据质量，20% 取决于检索策略，10% 取决于模型能力。大多数人把精力花在那 10% 上，而忽略了 70% 的基础。 原始数据 → 接入 → 清洗 → 标准化 → 分块 → 元数据 → 嵌入 → 向量库 → 检索 → 生成 ──────────── ETL ──────────── ──── 索引 ──── 本文聚焦管道前半段：从原始数据到高质量向量索引。 数据源接入 多源数据统一接入 from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Iterator import hashlib @dataclass class Document: id: str content: str source: str source_type: str # pdf, html, markdown, database, api metadata: dict content_hash: str updated_at: str class DataSource(ABC): @abstractmethod def fetch(self) -> Iterator[Document]: pass class WebPageSource(DataSource): def __init__(self, urls: list, cache_dir: str): self.urls = urls self.cache = cache_dir def fetch(self) -> Iterator[Document]: for url in self.urls: html = self._fetch_with_cache(url) text = self._html_to_markdown(html) yield Document( id=self._url_hash(url), content=text, source=url, source_type="web", metadata={"title": self._extract_title(html)}, content_hash=hashlib.md5(text.encode()).hexdigest(), updated_at=datetime.now().isoformat() ) class PDFSource(DataSource): def fetch(self) -> Iterator[Document]: for pdf_path in self.pdf_files: pages = self._extract_pdf_pages(pdf_path) for i, page_text in enumerate(pages): yield Document( id=f"{pdf_path}_p{i+1}", content=page_text, source=pdf_path, source_type="pdf", metadata={"page": i+1, "total_pages": len(pages)}, content_hash=hashlib.md5(page_text.encode()).hexdigest(), updated_at=datetime.now().isoformat() ) class DatabaseSource(DataSource): def __init__(self, conn_str: str, query: str): self.conn_str = conn_str self.query = query def fetch(self) -> Iterator[Document]: import psycopg2 with psycopg2.connect(self.conn_str) as conn: with conn.cursor() as cur: cur.execute(self.query) for row in cur: yield Document( id=f"db_{row[0]}", content=row[1], source=f"database:{row[0]}", source_type="database", metadata={"table": row[2]} if len(row) > 2 else {}, content_hash=hashlib.md5(row[1].encode()).hexdigest(), updated_at=datetime.now().isoformat() ) 清洗与标准化 class DocumentCleaner: """文档清洗流水线""" def clean(self, doc: Document) -> Document: steps = [ self._remove_html_tags, self._normalize_whitespace, self._remove_boilerplate, self._fix_encoding, self._normalize_punctuation, ] for step in steps: doc.content = step(doc.content) return doc def _remove_html_tags(self, text: str) -> str: import re # 保留代码块中的内容 code_blocks = re.findall(r'```.*?```', text, re.DOTALL) text = re.sub(r'<[^>]+>', '', text) # 恢复代码块 for i, block in enumerate(code_blocks): text = text.replace(f'CODE_BLOCK_{i}', block, 1) return text def _normalize_whitespace(self, text: str) -> str: import re # 合并连续空行 text = re.sub(r'\n{3,}', '\n\n', text) # 去除行尾空格 text = '\n'.join(line.rstrip() for line in text.split('\n')) # 合并连续空格 text = re.sub(r' {2,}', ' ', text) return text.strip() def _remove_boilerplate(self, text: str) -> str: """移除样板文字""" boilerplate_patterns = [ r'Cookie.*?声明', r'版权所有.*?保留', r'本文档仅供参考', r'Last updated:.*?$', ] for pattern in boilerplate_patterns: text = re.sub(pattern, '', text, flags=re.IGNORECASE) return text def _fix_encoding(self, text: str) -> str: # 修复常见编码问题 replacements = { '\ufeff': '', # BOM '\u200b': '', # 零宽空格 '\u200c': '', # 零宽非连接符 '\u00a0': ' ', # 不间断空格 ''': "'", ''': "'", '"': '"', '"': '"', '—': '-', '–': '-', } for old, new in replacements.items(): text = text.replace(old, new) return text 分块策略：最关键的决策 三种分块策略对比 策略 原理 优点 缺点 适用场景 固定长度 按token数切割 简单可控 可能切断语义 FAQ、短文本 语义分块 按句子/段落边界切 保持语义完整 块大小不均 文章、文档 递归分块 先按段落，太长再按句子 平衡大小和语义 实现复杂 混合内容 递归分块实现 from typing import List import tiktoken class RecursiveChunker: def __init__( self, chunk_size: int = 512, chunk_overlap: int = 50, model: str = "gpt-4o" ): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.encoder = tiktoken.encoding_for_model(model) # 分隔符按优先级排序 self.separators = [ "\n## ", # Markdown 二级标题 "\n### ", # Markdown 三级标题 "\n\n", # 段落 "\n", # 行 "。", # 中文句号 ". ", # 英文句号 " ", # 空格 "", # 字符级（最后手段） ] def chunk(self, text: str) -> List[str]: return self._recursive_split(text, self.separators) def _recursive_split(self, text: str, separators: list) -> List[str]: if self._count_tokens(text) <= self.chunk_size: return [text] if text.strip() else [] for i, sep in enumerate(separators): if sep == "": return self._force_split(text) if sep not in text: continue parts = text.split(sep) chunks = [] current = "" for part in parts: candidate = current + sep + part if current else part if self._count_tokens(candidate) <= self.chunk_size: current = candidate else: if current: chunks.append(current) # 如果单个 part 就超长，用更细的分隔符递归 if self._count_tokens(part) > self.chunk_size: sub_chunks = self._recursive_split(part, separators[i+1:]) chunks.extend(sub_chunks) current = "" else: current = part if current: chunks.append(current) # 添加重叠 return self._add_overlap(chunks) return self._force_split(text) def _add_overlap(self, chunks: List[str]) -> List[str]: if len(chunks) <= 1 or self.chunk_overlap <= 0: return chunks result = [chunks[0]] for i in range(1, len(chunks)): # 从前一个 chunk 末尾取 overlap 个 token prev_tokens = self.encoder.encode(chunks[i-1]) overlap_text = self.encoder.decode( prev_tokens[-self.chunk_overlap:] ) result.append(overlap_text + chunks[i]) return result def _count_tokens(self, text: str) -> int: return len(self.encoder.encode(text)) def _force_split(self, text: str) -> List[str]: """按 token 强制切割""" tokens = self.encoder.encode(text) chunks = [] for i in range(0, len(tokens), self.chunk_size): chunk_tokens = tokens[i:i + self.chunk_size] chunks.append(self.encoder.decode(chunk_tokens)) return chunks 语义分块（基于嵌入） class SemanticChunker: """基于句子嵌入相似度的语义分块""" def __init__(self, model, threshold: float = 0.5): self.model = model # SentenceTransformer self.threshold = threshold def chunk(self, text: str) -> List[str]: # 1. 按句子分割 sentences = self._split_sentences(text) if len(sentences) <= 1: return [text] # 2. 计算相邻句子嵌入的余弦相似度 embeddings = self.model.encode(sentences) similarities = [ cosine_similarity([embeddings[i]], [embeddings[i+1]])[0][0] for i in range(len(embeddings) - 1) ] # 3. 在相似度低谷处分割 chunks = [] current = [sentences[0]] for i, sim in enumerate(similarities): if sim < self.threshold: chunks.append(" ".join(current)) current = [sentences[i + 1]] else: current.append(sentences[i + 1]) if current: chunks.append(" ".join(current)) return chunks 元数据提取 class MetadataExtractor: """从文档中提取结构化元数据""" def extract(self, doc: Document) -> dict: return { **doc.metadata, "language": self._detect_language(doc.content), "doc_type": self._classify_doc_type(doc.content), "key_phrases": self._extract_key_phrases(doc.content), "has_code": "```" in doc.content or "def " in doc.content, "has_table": "|" in doc.content and "---" in doc.content, "reading_time_min": max(1, len(doc.content) // 500), "summary": self._generate_summary(doc.content), } def _detect_language(self, text: str) -> str: # 简单启发式 chinese_ratio = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') / max(len(text), 1) if chinese_ratio > 0.3: return "zh" return "en" def _classify_doc_type(self, text: str) -> str: if "```" in text: return "technical" if any(kw in text for kw in ["第", "章", "节"]): return "document" if "?" in text and len(text) < 500: return "faq" return "article" def _extract_key_phrases(self, text: str) -> list: # 使用 TextRank 或 YAKE 提取关键词 from keybert import KeyBERT kw_model = KeyBERT() keywords = kw_model.extract_keywords(text, top_n=5) return [kw[0] for kw in keywords] 嵌入生成与批量入库 import asyncio from typing import List, Tuple class EmbeddingPipeline: def __init__(self, model_name: str = "text-embedding-3-large"): self.model_name = model_name self.client = AsyncOpenAI() self.batch_size = 100 # OpenAI 批量限制 async def embed_chunks(self, chunks: List[str]) -> List[List[float]]: # 分批处理 all_embeddings = [] for i in range(0, len(chunks), self.batch_size): batch = chunks[i:i + self.batch_size] response = await self.client.embeddings.create( model=self.model_name, input=batch ) all_embeddings.extend([d.embedding for d in response.data]) return all_embeddings async def embed_with_retry(self, text: str, max_retries: int = 3): for attempt in range(max_retries): try: response = await self.client.embeddings.create( model=self.model_name, input=text ) return response.data[0].embedding except Exception as e: if attempt == max_retries - 1: raise await asyncio.sleep(2 ** attempt) class VectorIndexer: def __init__(self, pinecone_client): self.pinecone = pinecone_client self.index_name = "rag-knowledge" async def upsert(self, chunks: List[dict]): """ chunks: [{id, content, embedding, metadata}] """ index = self.pinecone.Index(self.index_name) # 批量 upsert batch_size = 100 for i in range(0, len(chunks), batch_size): batch = chunks[i:i + batch_size] vectors = [ { "id": c["id"], "values": c["embedding"], "metadata": { "content": c["content"][:1000], # Pinecone metadata 大小限制 **c["metadata"] } } for c in batch ] await index.upsert(vectors=vectors) 增量更新 class IncrementalUpdater: """只更新变化的部分，避免全量重建""" def __init__(self, vector_indexer, storage): self.indexer = vector_indexer self.storage = storage # 记录文档状态的数据库 async def update(self, new_docs: List[Document]): added, updated, deleted = 0, 0, 0 for doc in new_docs: existing = await self.storage.get(doc.id) if existing is None: # 新文档 await self._index_doc(doc) await self.storage.save(doc) added += 1 elif existing.content_hash != doc.content_hash: # 内容变更: 删旧建新 await self._delete_doc_chunks(doc.id) await self._index_doc(doc) await self.storage.save(doc) updated += 1 # content_hash 相同则跳过 # 检查删除: 源中不再存在的文档 all_ids = {d.id for d in new_docs} stored_ids = await self.storage.get_all_ids() deleted_ids = stored_ids - all_ids for doc_id in deleted_ids: await self._delete_doc_chunks(doc_id) await self.storage.delete(doc_id) deleted += 1 return {"added": added, "updated": updated, "deleted": deleted} 质量监控 class PipelineMonitor: """数据管道质量监控""" async def check_quality(self, docs: List[Document]) -> dict: checks = { "empty_docs": 0, "too_short": 0, # < 50 字符 "too_long": 0, # > 100k 字符 "encoding_issues": 0, "duplicate_content": 0, "low_density": 0, # 信息密度低 } contents = [d.content for d in docs] seen_hashes = set() for doc in docs: if not doc.content.strip(): checks["empty_docs"] += 1 elif len(doc.content) < 50: checks["too_short"] += 1 elif len(doc.content) > 100000: checks["too_long"] += 1 if doc.content_hash in seen_hashes: checks["duplicate_content"] += 1 seen_hashes.add(doc.content_hash) # 信息密度: 非空非标点字符比例 alpha_ratio = sum(c.isalnum() for c in doc.content) / max(len(doc.content), 1) if alpha_ratio < 0.3: checks["low_density"] += 1 # 检索质量测试 retrieval_score = await self._test_retrieval(docs) return { "total_docs": len(docs), "quality_issues": checks, "retrieval_test_score": retrieval_score, "timestamp": datetime.now().isoformat() } async def _test_retrieval(self, docs) -> float: """用标准查询测试检索质量""" test_queries = load_test_queries() # 预定义的测试查询 correct = 0 for q in test_queries: results = await self.search(q["query"], top_k=5) if q["expected_doc_id"] in [r["id"] for r in results]: correct += 1 return correct / len(test_queries) 分块参数调优建议 文档类型 chunk_size overlap 分隔符优先级 FAQ 256 0 问题边界 技术文档 512 50 标题→段落→句子 代码 1024 100 函数/类边界 法律文本 768 80 条款→段落 对话记录 384 30 话轮边界 总结 RAG 数据管道的核心决策在分块策略：太大切不到语义，太小丢失上下文，重叠太少检索不连续，太大浪费 token。递归分块是最通用的方案，语义分块效果最好但成本高。增量更新避免全量重建，质量监控防止垃圾数据入库。记住：检索质量的天花板在数据管道阶段就定了。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...