【RAG实战】 文档处理 的清洗 到 上下文增强
·
RAG 文档处理 Pipeline 设计详解：从清洗到上下文增强
项目背景：基于 FastAPI + LlamaIndex 构建的企业级 RAG 知识库管理系统，核心处理链路采用 LlamaIndex IngestionPipeline 编排，实现清洗→切片→后处理→上下文增强→元数据增强→入库收集的完整流水线。
文章目录
一、整体架构概览
1.1 设计目标
| 目标 | 实现方式 |
|---|---|
| 组件解耦 | 每个环节封装为独立的 TransformComponent，组件无状态 |
| 策略可插拔 | 注册式工厂模式，新增策略无需修改工厂代码 |
| 全异步执行 | CPU 密集型放入线程池，I/O 密集型纯异步 |
| 配置统一传递 | PipelineContext 作为唯一数据源，通过 kwargs 透传 |
| 可追溯性 | 每个环节记录统计信息到 ctx.state，支持审计 |
1.2 Pipeline 完整组件链
Documents（外部读取）
│
▼
┌──────────────────────────────┐
│ 1. DocumentCleanerComponent │ ← 文档清洗（CPU 密集型，线程池）
│ 输入: Document[] │
│ 输出: Document[] │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ 2. ChunkingComponent │ ← 文档切片（CPU 密集型，线程池）
│ 输入: Document[] │
│ 输出: TextNode[] │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ 3. ChunkPostProcessorComponent │ ← 切片后处理（CPU 密集型，线程池）
│ 注入: prev/next_chunk_id │
│ 注入: heading_path（MD） │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ 4. ContextEnhancementComponent │ ← 上下文增强（I/O 密集型，纯异步）
│ 可选，默认启用 │
│ HyDE → Summary → Title │
│ 最终组装向量化文本 │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ 5. MetadataEnrichmentComponent │ ← 元数据增强（CPU 密集型，线程池）
│ 注入: doc_id, kb_id 等 │
│ 注入: chunk_index, 页码 │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ 6. StorageContextCollectorComponent │ ← 入库收集（轻量，直接执行）
│ 记录最终 Node 数量 │
│ 透传 Node[] 给外部 │
└──────────────────────────────┘
│
▼
最终 Node[] 列表
（向量化入库 + PG 持久化）
1.3 核心设计原则
1. Controller 轻量 → 仅负责参数传递
2. Service 核心 → 所有业务逻辑集中在 Service 层
3. 组件无状态 → PipelineComponent 不持有任何配置，从 PipelineContext 读取
4. 策略可插拔 → 注册式工厂，新增策略零修改工厂代码
5. 异步优先 → 所有 I/O 操作使用 async/await，CPU 密集型放入线程池
二、PipelineContext：统一配置中心
2.1 设计理念
PipelineContext 是整个 Pipeline 的唯一数据源，所有组件的配置参数以分组属性的形式内联在其中。组件通过 get_pipeline_context(kwargs) 从 LlamaIndex 透传的 kwargs 中提取。
# Pipeline 启动时
pipeline.arun(documents=documents, pipeline_context=self.context)
↓
# LlamaIndex 内部遍历组件时
await transform.acall(nodes, pipeline_context=context, **kwargs)
↓
# 组件内部提取
ctx = get_pipeline_context(kwargs) # 从 kwargs 中取出 PipelineContext
2.2 完整字段定义
class PipelineContext(BaseModel):
# ── 清洗配置（cleaning_*） ──────────────────────────
cleaning_strategies: list[str] | None = None
# 清洗策略名称列表，按顺序执行；None 时使用 CleaningService 默认策略
cleaning_params_map: dict[str, dict[str, Any]] = {}
# 清洗策略参数映射，key 为策略名称，value 为该策略的参数
# ── 切片配置（chunking_*） ──────────────────────────
chunking_strategy_type: ChunkStrategyType = ChunkStrategyType.SENTENCE
# 切片策略类型：sentence / semantic / window / hierarchical / markdown / json / auto
chunking_params: dict[str, Any] = {}
# 切片策略参数，如 {"chunk_size": 512, "chunk_overlap": 50}
chunking_strategy_name: str = "basic"
# 切片策略名称标识
# ── 上下文增强配置（enhancement_*） ─────────────────
enhancement_enabled: bool = True
# 是否启用上下文增强
enhancement_strategies: list[str] = ["hyde", "summary", "title_injection"]
# 启用的增强策略列表，按顺序执行
enhancement_params: dict[str, Any] = {}
# 增强策略参数，如 {"hyde_num_questions": 3, "hyde_batch_size": 5}
# ── 元数据配置（meta_*） ────────────────────────────
meta_doc_id: str | None = None # 文档唯一标识
meta_kb_id: str | None = None # 所属知识库 ID
meta_doc_type: str | None = None # 文档类型
meta_file_name: str | None = None # 原始文件名
meta_file_type: str | None = None # 原始文件类型
meta_file_size: int = 0 # 原始文件大小（字节）
meta_extra: dict[str, Any] = {} # 额外自定义元数据
meta_enrichment_params: dict[str, Any] = {} # 元数据增强策略参数
meta_enrichment_strategy: str = "basic" # 元数据增强策略名称
# ── 运行时数据 ──────────────────────────────────────
rag_document: Any = None # 原始 RagDocument 对象
documents: list[Document] = [] # 文件读取后的 Document 列表
state: dict[str, Any] = {} # 组件共享的中间状态
extra: dict[str, Any] = {} # 额外扩展数据
2.3 配置传递链路
process_single_document()
│
├── 构建 PipelineContext
│ ├── chunking_strategy_type = 从数据库 JSONB 解析
│ ├── chunking_params = 从数据库 JSONB 解析
│ ├── meta_doc_id = document.id
│ ├── meta_kb_id = document.kb_id
│ └── ...
│
├── RAGIngestionPipeline(context=pipeline_context)
│ └── 组装 TransformComponent 列表
│
└── pipeline.arun(documents)
└── self.pipeline.arun(documents, pipeline_context=self.context)
└── LlamaIndex 遍历每个组件:
await transform.acall(nodes, pipeline_context=context)
三、文档清洗阶段
3.1 架构设计
清洗采用多策略链式执行模式，前一个策略的输出作为后一个的输入，遵循「先基础后语义、先去噪后修复、先内容后结构」的原则。
原始 Document[]
│
▼
CleaningService.process_cleaning(documents, strategy_names, params_map)
│
├── basic → 换行符统一、空白压缩、控制字符移除
├── special_char → 乱码行、Emoji、颜文字、连续符号处理
├── privacy_redaction → 隐私脱敏（手机号/邮箱/身份证/银行卡）
├── hyperlink_handling → 超链接处理（保留文本移除 URL）
├── document_artifact → 页码/页眉页脚/水印/目录移除
├── boilerplate → 免责声明/版本信息/版权声明
├── meaningless_filter → 纯符号行/空标题/重复字符
├── footnote_reference → 脚注标记/参考文献条目
├── text_repair → 断行合并、连字符断词修复
├── chinese → 中文规范化（全角半角/繁简/标点/间距）
├── markdown → 图片/HTML/失效链接/标题层级
├── code_block → 围栏修复/空块移除/LaTeX
├── table_repair → MD表格/HTML转MD/跨页合并
├── structure_recovery → 标题层级/列表标记/空行保证
├── paragraph_merge → 短段落合并、长段落拆分
└── dedup → 段落级/句子级去重
│
▼
清洗后 Document[]
3.2 五阶段执行顺序
| 阶段 | 策略 | 核心职责 |
|---|---|---|
| 阶段 1：基础规范化 | basic → special_char | 统一编码格式，移除乱码 |
| 阶段 2：噪音移除 | privacy → hyperlink → artifact → boilerplate → meaningless → footnote | 移除无关内容 |
| 阶段 3：文本修复 | text_repair → chinese | 修复断行，规范化中文 |
| 阶段 4：格式修复 | markdown → code_block → table_repair → structure_recovery | 修复 Markdown 结构 |
| 阶段 5：后处理 | paragraph_merge → dedup | 合并短段落，去除重复 |
3.3 注册式工厂模式
# 工厂：CleaningStrategyFactory
class CleaningStrategyFactory:
_strategies: dict[str, type[CleaningStrategy]] = {}
@classmethod
def register(cls, name: str, strategy_class: type[CleaningStrategy]):
cls._strategies[name] = strategy_class
@classmethod
def get_strategy(cls, name: str) -> CleaningStrategy:
strategy_class = cls._strategies.get(name)
if not strategy_class:
raise ValueError(f"Unsupported cleaning strategy: {name}")
return strategy_class()
# 策略模块：strategies/__init__.py
# 导入即注册，新增策略只需在此添加导入
from .basic import BasicCleaningStrategy
from .special_char import SpecialCharCleaningStrategy
# ... 共 16 个策略
3.4 清洗策略抽象基类
class CleaningStrategy(ABC):
@abstractmethod
def process(self, documents: list[Document], params: dict[str, Any]) -> CleaningResult:
"""执行清洗操作，返回清洗后文档 + 统计信息"""
@property
@abstractmethod
def name(self) -> str:
"""策略名称标识"""
3.5 Pipeline 组件封装
class DocumentCleanerComponent(TransformComponent):
def __call__(self, nodes, **kwargs):
documents = [n for n in nodes if isinstance(n, Document)]
ctx = get_pipeline_context(kwargs)
# 委托给 CleaningService 执行链式清洗
response = CleaningService.process_cleaning(
documents,
strategy_names=ctx.cleaning_strategies,
params_map=ctx.cleaning_params_map,
)
# 记录统计信息到共享状态
ctx.state["cleaned_count"] = len(response.documents)
ctx.state["cleaning_stats"] = [
{"strategy": s.strategy_name, "stats": s.stats}
for s in response.strategy_stats
]
return response.documents
async def acall(self, nodes, **kwargs):
# CPU 密集型，放入线程池避免阻塞事件循环
return await asyncio.to_thread(self.__call__, nodes, **kwargs)
四、文档切片阶段
4.1 策略类型总览
| 策略 | 枚举值 | 适用场景 | 核心原理 |
|---|---|---|---|
| Sentence | sentence |
短文本、FAQ、词条 | 按句子边界切分，控制 chunk_size |
| Semantic | semantic |
散文、报告、会议记录 | 按语义相似度断句 |
| SentenceWindow | window |
FAQ 问答、操作手册 | 句子级切片 + 滑动窗口扩展上下文 |
| Hierarchical | hierarchical |
超长文档、财报、白皮书 | 多粒度切片（4096→2048→1024→512） |
| Markdown | markdown |
法规、合同、API 文档 | 按标题层级切分，保留结构 |
| JSON | json |
JSON 结构化数据 | 按 JSON 路径切分 |
| Auto | auto |
未知文档类型 | 自动分析文档特征推荐策略 |
4.2 自动策略推荐引擎
当用户选择 auto 模式时，系统会分析文档结构特征，自动推荐最优策略：
文档内容
│
▼
MDStructureAnalyzer.analyze(md_text)
│ 提取特征：
│ - 标题层级/密度
│ - 段落长度分布
│ - 表格/列表占比
│ - FAQ 模式/编号步骤
│ - 语义连贯性指标
│
▼
AutoChunkStrategyRecommender._score_strategies(features)
│ 规则引擎打分 → Top 3 候选
│
▼
AutoChunkStrategyRecommender._llm_select_strategy(candidates, features, sample)
│ LLM 从 Top 3 中选最优
│
▼
StrategyRecommendation(strategy_type, strategy_params, reason, confidence)
规则优先级（从高到低）：
| 优先级 | 规则 | 推荐策略 | 适用条件 |
|---|---|---|---|
| 1 | 超长文档 | hierarchical | > 10 万字符 |
| 2 | 多级标题 | markdown | ≥ 2 级标题，密度合理 |
| 3 | 表格/列表密集 | markdown | 表格行 > 15% 或列表行 > 20% |
| 3.5 | FAQ 密集 | window | FAQ 段落 > 15% |
| 3.6 | 步骤型文本 | window | 编号步骤 > 10% |
| 4 | 语义连贯 | semantic | 无标题，段落长且连贯 |
| 5 | 短文本 | sentence | < 5000 字符 |
| 6 | 兜底 | sentence | 默认安全选择 |
4.3 切片策略抽象基类
class BaseChunkStrategy(ABC):
strategy_type: ClassVar["ChunkStrategyType"]
@abstractmethod
def get_parser(self, params: dict) -> NodeParser:
"""获取 LlamaIndex 的 NodeParser 实例"""
def execute(self, documents: list[Document], params: dict) -> list:
"""执行切片并自动注入 strategy_type 元数据"""
parser = self.get_parser(params)
nodes = parser.get_nodes_from_documents(documents)
strategy_value = self.__class__.strategy_type.value
for node in nodes:
node.metadata["strategy_type"] = strategy_value
return nodes
4.4 Pipeline 组件封装
class ChunkingComponent(TransformComponent):
def __call__(self, nodes, **kwargs):
documents = [n for n in nodes if isinstance(n, Document)]
ctx = get_pipeline_context(kwargs)
# 委托给 ChunkingService
response = ChunkingService.process_chunking(
documents,
strategy_type=ctx.chunking_strategy_type,
params=ctx.chunking_params,
)
# 记录中间结果
ctx.state["chunking_response"] = response.model_dump(mode="json")
# 转换为 TextNode 列表
text_nodes = [
TextNode(
id_=chunk.node_id,
text=chunk.text,
metadata=chunk.metadata,
start_char_idx=chunk.start_char_idx,
end_char_idx=chunk.end_char_idx,
)
for chunk in response.chunks
]
return text_nodes
async def acall(self, nodes, **kwargs):
return await asyncio.to_thread(self.__call__, nodes, **kwargs)
五、切片后处理阶段
5.1 职责说明
切片后处理组件负责注入切片间关系元数据，为后续的检索和导航提供支撑：
| 注入字段 | 适用策略 | 说明 |
|---|---|---|
prev_chunk_id |
所有策略 | 前一个切片的 ID |
next_chunk_id |
所有策略 | 后一个切片的 ID |
heading_path |
markdown | 标题层级路径，如 “H1标题 > H2标题 > H3标题” |
heading_level |
markdown | 当前标题层级数字 |
5.2 标题路径构建算法
@staticmethod
def _process_markdown_heading_paths(nodes: list[BaseNode]) -> None:
"""构建 markdown 标题层级路径栈"""
heading_stack: list[tuple[int, str]] = []
for node in nodes:
heading = node.metadata.get("heading", "")
if not heading:
continue
# 解析标题层级（# 号数量）
level = 0
for ch in heading:
if ch == '#':
level += 1
else:
break
level = max(level, 1)
title = heading.lstrip("#").strip()
# 弹出所有 >= 当前层级的标题（维护栈的单调性）
while heading_stack and heading_stack[-1][0] >= level:
heading_stack.pop()
heading_stack.append((level, title))
# 构建路径：如 "安装指南 > 环境配置 > Docker 部署"
path = " > ".join(t for _, t in heading_stack)
node.metadata["heading_path"] = path
node.metadata["heading_level"] = level
示例：
Node 1: heading="# 第一章" → heading_path="第一章"
Node 2: heading="## 1.1 概述" → heading_path="第一章 > 1.1 概述"
Node 3: heading="### 细节" → heading_path="第一章 > 1.1 概述 > 细节"
Node 4: heading="## 1.2 实践" → heading_path="第一章 > 1.2 实践" (弹出 "1.1 概述" 和 "细节")
六、上下文增强阶段（核心）
6.1 设计目标
上下文增强是提升 RAG 检索质量的核心环节。通过为每个 chunk 注入额外的语义信息（摘要、假设性问题、标题上下文），使得向量化后的 embedding 更接近用户查询的语义空间。
6.2 增强器架构
ContextEnhancementComponent.acall()
│
├── 1. HyDEEnhancer（I/O 密集，LLM 调用）
│ └── 为每个 chunk 生成假设性问题
│ └── 写入 metadata["hypothetical_questions"]
│
├── 2. SummaryEnhancer（I/O 密集，LLM 调用）
│ └── 为每个 chunk 生成摘要
│ └── 写入 metadata["chunk_summary"]
│
├── 3. TitleInjectionEnhancer（零成本，无 LLM）
│ └── 提取文档标题
│ └── 写入 metadata["doc_title"]
│
└── 4. _assemble_enhanced_text()
└── 读取所有 metadata 字段
└── 按固定格式拼接最终向量化文本
└── 替换 node.text
6.3 增强器抽象基类
class BaseEnhancer(ABC):
"""所有增强器必须实现 async enhance()"""
@property
@abstractmethod
def name(self) -> str:
"""增强器名称标识"""
@abstractmethod
async def enhance(
self,
nodes: list[BaseNode],
context: MetadataContext,
params: dict[str, Any],
) -> list[BaseNode]:
"""执行增强（异步），写入 metadata，不修改 text"""
6.4 注册式工厂
class ContextEnhancementFactory:
_enhancers: dict[str, BaseEnhancer] = {}
@classmethod
def register(cls, name: str, enhancer: BaseEnhancer):
cls._enhancers[name] = enhancer
@classmethod
def get_enhancer(cls, name: str) -> BaseEnhancer:
enhancer = cls._enhancers.get(name)
if enhancer is None:
raise ValueError(f"不支持的增强器: {name}")
return enhancer
自动注册机制（strategies/__init__.py）：
# 导入即注册
from .hyde import HyDEEnhancer # → ContextEnhancementFactory.register("hyde", HyDEEnhancer())
from .summary import SummaryEnhancer # → ContextEnhancementFactory.register("summary", SummaryEnhancer())
from .keyword import KeywordEnhancer # → ContextEnhancementFactory.register("keyword", KeywordEnhancer())
from .title_injection import TitleInjectionEnhancer # → ContextEnhancementFactory.register("title_injection", ...)
6.5 四大增强器详解
6.5.1 HyDE 增强器（假设性问题生成）
核心原理：HyDE（Hypothetical Document Embeddings）是检索增强的核心手段。为每个 chunk 生成"用户可能会问的问题"，这些问题的 embedding 与用户查询的 embedding 更接近，从而显著提升召回率。
class HyDEEnhancer(BaseEnhancer):
HYDE_PROMPT = """基于以下文本片段，生成 {num_questions} 个用户可能会问的问题。
要求：
1. 问题必须能用文本片段中的信息回答
2. 问题应覆盖不同的角度（是什么、为什么、怎么做、什么条件）
3. 每个问题一行，用换行分隔
文本片段：
{text}
问题："""
async def enhance(self, nodes, context, params):
return await self._llm_generation(nodes, params)
async def _llm_generation(self, nodes, params):
num_questions = params.get("hyde_num_questions", 3)
batch_size = params.get("hyde_batch_size", 5)
llm = get_llm()
# 分批并发处理
for i in range(0, len(nodes), batch_size):
batch = nodes[i:i + batch_size]
tasks = [self._generate_questions_for_node(llm, node, num_questions) for node in batch]
results = await asyncio.gather(*tasks, return_exceptions=True)
for node, result in zip(batch, results, strict=True):
if isinstance(result, Exception):
node.metadata["hypothetical_questions"] = []
else:
node.metadata["hypothetical_questions"] = result
async def _generate_questions_for_node(self, llm, node, num_questions):
prompt = self.HYDE_PROMPT.format(text=node.text[:1000], num_questions=num_questions)
# LLM 是同步 SDK，必须放入线程池
response = await asyncio.to_thread(llm.complete, prompt)
questions = [q.strip() for q in str(response).strip().split("\n") if q.strip()]
return questions[:num_questions]
注入字段：metadata["hypothetical_questions"] → list[str]
6.5.2 Summary 增强器（切片摘要生成）
核心原理：为每个 chunk 生成高质量摘要，用于融入向量化文本提升检索精度，同时可用于前端快速预览。
class SummaryEnhancer(BaseEnhancer):
SUMMARY_PROMPT = """请用简洁的中文为以下文本片段生成摘要。
要求：
1. 摘要不超过 3 句话
2. 保留关键信息和核心观点
3. 直接输出摘要，不要加任何前缀
文本片段：
{text}
摘要："""
async def enhance(self, nodes, context, params):
return await self._llm_summary(nodes, params)
async def _llm_summary(self, nodes, params):
batch_size = params.get("summary_batch_size", 5)
llm = get_llm()
for i in range(0, len(nodes), batch_size):
batch = nodes[i:i + batch_size]
tasks = [self._generate_summary_for_node(llm, node) for node in batch]
results = await asyncio.gather(*tasks, return_exceptions=True)
for node, result in zip(batch, results, strict=True):
if isinstance(result, Exception):
node.metadata["chunk_summary"] = ""
else:
node.metadata["chunk_summary"] = result
async def _generate_summary_for_node(self, llm, node):
prompt = self.SUMMARY_PROMPT.format(text=node.text[:1000])
response = await asyncio.to_thread(llm.complete, prompt)
return str(response).strip()
注入字段：metadata["chunk_summary"] → str
6.5.3 Keyword 增强器（关键词提取）
核心原理：基于 jieba 的 TF-IDF / TextRank 算法提取关键词，用于 BM25 检索增强和 Milvus 标量索引过滤。零成本，不依赖 LLM。
class KeywordEnhancer(BaseEnhancer):
async def enhance(self, nodes, context, params):
top_k = params.get("keyword_top_k", 5)
mode = params.get("keyword_mode", "tfidf")
for node in nodes:
keywords = self._extract_keywords(node.text or "", top_k, mode)
node.metadata["keywords"] = keywords
return nodes
@staticmethod
def _extract_keywords(text, top_k, mode):
import jieba.analyse
if mode == "textrank":
return jieba.analyse.textrank(text, topK=top_k)
else:
return jieba.analyse.extract_tags(text, topK=top_k)
注入字段：metadata["keywords"] → list[str]
6.5.4 TitleInjection 增强器（标题上下文注入）
核心原理：提取文档标题（去掉扩展名）存入 metadata，供最终文本组装使用。零成本，不依赖 LLM。
class TitleInjectionEnhancer(BaseEnhancer):
async def enhance(self, nodes, context, params):
include_file_name = params.get("include_file_name", True)
for node in nodes:
if include_file_name and context.file_name:
node.metadata["doc_title"] = os.path.splitext(context.file_name)[0]
# heading_path 已由 ChunkPostProcessorComponent 注入，此处无需重复处理
return nodes
注入字段：metadata["doc_title"] → str
6.6 最终文本组装
所有增强器执行完毕后，_assemble_enhanced_text() 统一组装最终向量化文本：
@staticmethod
def _assemble_enhanced_text(nodes: list[BaseNode]) -> list[BaseNode]:
for node in nodes:
parts = []
# 1. 标题上下文（顶部）
doc_title = node.metadata.get("doc_title")
heading_path = node.metadata.get("heading_path")
title_parts = []
if doc_title:
title_parts.append(f"[文档: {doc_title}]")
if heading_path:
title_parts.append(f"[章节: {heading_path}]")
if title_parts:
parts.append(" ".join(title_parts))
# 2. 摘要
chunk_summary = node.metadata.get("chunk_summary")
if chunk_summary:
parts.append(chunk_summary)
# 3. 原文（用 <original_text> 标记包裹，便于检索端提取）
original_text = node.text or ""
if original_text:
parts.append(f"<original_text>\n{original_text}\n</original_text>")
# 4. 假设性问题（底部）
questions = node.metadata.get("hypothetical_questions", [])
if questions:
parts.append("[假设性问题]")
parts.extend(questions)
# 拼接并替换 node.text
if len(parts) > 1 or (len(parts) == 1 and parts[0] != original_text):
node.text = "\n".join(parts)
return nodes
6.7 固定拼接格式
最终向量化文本的完整格式：
[文档: RAG知识库系统设计方案] [章节: 第三章 > 3.2 向量检索优化]
本文介绍了向量检索的优化方案，包括索引策略调整和缓存机制改进...
<original_text>
向量检索是 RAG 系统的核心环节，通过 Milvus 的 HNSW 索引...
（此处为原始切片文本）
</original_text>
[假设性问题]
RAG 系统的向量检索如何优化？
Milvus HNSW 索引的参数如何调整？
向量检索的缓存机制是怎样的？
6.8 检索端原文提取
def extract_original_text(enhanced_text: str) -> str:
"""从增强文本中提取纯净原文"""
match = re.search(r"<original_text>\s*(.*?)\s*</original_text>", enhanced_text, re.DOTALL)
if match:
return match.group(1).strip()
return enhanced_text.strip()
使用方式：
- 向量化：使用完整增强文本（含标记），提升召回率
- 给 LLM 的上下文：调用
extract_original_text()提取纯净原文
七、元数据增强阶段
7.1 职责说明
元数据增强为每个 Node 注入文档级业务元数据，用于 Milvus 标量过滤、业务展示和溯源。
7.2 MetadataContext 数据模型
class MetadataContext(BaseModel):
doc_id: str | None = None # 文档唯一标识
kb_id: str | None = None # 所属知识库 ID
doc_type: str | None = None # 文档类型
file_name: str | None = None # 原始文件名
file_type: str | None = None # 原始文件类型
file_size: int = 0 # 原始文件大小（字节）
extra: dict[str, Any] = {} # 额外自定义元数据
7.3 Basic 策略注入的元数据字段
| 字段 | 来源 | 用途 |
|---|---|---|
doc_id |
MetadataContext | Milvus 标量过滤 |
kb_id |
MetadataContext | Milvus 标量过滤（知识库级别） |
doc_type |
MetadataContext | 文档类型过滤 |
file_name |
MetadataContext | 溯源展示 |
file_type |
MetadataContext | 文件类型过滤 |
file_size |
MetadataContext | 文件大小信息 |
chunk_index |
自动生成（枚举序号） | 切片顺序标识 |
page_number / page_start / page_end |
清洗阶段记录的页码边界 | 页码定位 |
7.4 页码注入算法
@staticmethod
def _inject_page_numbers(node: BaseNode) -> None:
"""基于页码边界和字符位置，二分查找所属页码"""
page_boundaries = node.metadata.get("page_boundaries")
if not page_boundaries:
return
offsets = [b["char_offset"] for b in page_boundaries]
pages = [b["page"] for b in page_boundaries]
start_char = node.start_char_idx or 0
end_char = node.end_char_idx or start_char
page_start = _lookup_page(start_char, offsets, pages)
page_end = _lookup_page(end_char, offsets, pages)
if page_start is not None:
if page_start == page_end:
node.metadata["page_number"] = page_start
else:
node.metadata["page_start"] = page_start
node.metadata["page_end"] = page_end
# 清理中间数据，不写入向量库
node.metadata.pop("page_boundaries", None)
def _lookup_page(char_idx: int, offsets: list[int], pages: list[int]) -> int | None:
"""二分查找所属页码"""
if not offsets:
return None
idx = bisect.bisect_right(offsets, char_idx) - 1
if idx < 0:
idx = 0
return pages[idx]
八、全异步执行模型
8.1 异步策略分类
| 组件类型 | 操作性质 | acall() 实现 | 说明 |
|---|---|---|---|
| DocumentCleanerComponent | CPU 密集型 | asyncio.to_thread(self.__call__, ...) |
正则匹配、文本处理 |
| ChunkingComponent | CPU 密集型 | asyncio.to_thread(self.__call__, ...) |
NodeParser 切片 |
| ChunkPostProcessorComponent | CPU 密集型 | asyncio.to_thread(self.__call__, ...) |
关系注入、标题路径构建 |
| ContextEnhancementComponent | I/O 密集型 | 纯异步 await enhancer.enhance() |
LLM API 调用 |
| MetadataEnrichmentComponent | CPU 密集型 | asyncio.to_thread(self.__call__, ...) |
元数据写入 |
| StorageContextCollectorComponent | 轻量操作 | self.__call__(nodes, **kwargs) |
仅记录数量 |
8.2 LlamaIndex 的 acall 陷阱
LlamaIndex TransformComponent 的 acall() 默认实现：
# LlamaIndex 源码（schema.py）
class TransformComponent(BaseComponent):
@abstractmethod
def __call__(self, nodes, **kwargs):
"""Transform nodes."""
async def acall(self, nodes, **kwargs):
"""Async transform nodes."""
return self.__call__(nodes, **kwargs) # ⚠️ 本质是同步的！
问题：默认的 acall() 虽然声明为 async def，但内部直接调用同步的 __call__()，会阻塞事件循环。
解决：所有组件必须自己重写 acall()：
- CPU 密集型 →
asyncio.to_thread(self.__call__, ...) - I/O 密集型 → 纯异步实现
- 轻量操作 → 直接调用
self.__call__()
8.3 执行时序图
时间轴 →
[DocumentCleanerComponent]
└─ asyncio.to_thread() ──────────────────┐
(线程池中执行清洗策略链) │
↓
[ChunkingComponent]
└─ asyncio.to_thread() ──────────────────┐
(线程池中执行切片) │
↓
[ChunkPostProcessorComponent]
└─ asyncio.to_thread() ──────────────────┐
(线程池中注入关系元数据) │
↓
[ContextEnhancementComponent]
└─ await HyDEEnhancer.enhance()
└─ asyncio.gather( │
to_thread(llm.complete, ...), │ ← 并发 LLM 调用
to_thread(llm.complete, ...), │
... │
) │
└─ await SummaryEnhancer.enhance()
└─ asyncio.gather( │
to_thread(llm.complete, ...), │ ← 并发 LLM 调用
... │
) │
└─ await TitleInjectionEnhancer.enhance()│ ← 零成本，直接执行
└─ _assemble_enhanced_text() │ ← 组装最终文本
↓
[MetadataEnrichmentComponent]
└─ asyncio.to_thread() ──────────────────┐
(线程池中注入业务元数据) │
↓
[StorageContextCollectorComponent]
└─ 直接执行（轻量操作）
九、Metadata 字段完整生命周期
9.1 字段流转全景图
阶段 写入的 metadata 字段
─────────────────────────────────────────────────────────────
切片策略执行 strategy_type
切片后处理 prev_chunk_id, next_chunk_id, heading_path, heading_level
HyDE 增强器 hypothetical_questions
Summary 增强器 chunk_summary
TitleInjection 增强器 doc_title
Keyword 增强器 keywords
元数据增强（Basic） doc_id, kb_id, doc_type, file_name, file_type,
file_size, chunk_index, page_number/page_start/page_end
9.2 字段分类与用途
| 字段 | 写入阶段 | Milvus 标量索引 | 检索用途 |
|---|---|---|---|
doc_id |
元数据增强 | ✅ | 文档级别过滤 |
kb_id |
元数据增强 | ✅ | 知识库级别过滤 |
doc_type |
元数据增强 | ✅ | 文档类型过滤 |
file_name |
元数据增强 | ❌ | 溯源展示 |
chunk_index |
元数据增强 | ❌ | 切片排序 |
page_number |
元数据增强 | ✅ | 页码定位 |
strategy_type |
切片阶段 | ❌ | 策略溯源 |
prev_chunk_id |
后处理 | ❌ | 上下文导航 |
next_chunk_id |
后处理 | ❌ | 上下文导航 |
heading_path |
后处理 | ❌ | 章节定位 |
doc_title |
上下文增强 | ❌ | 向量化文本组装 |
chunk_summary |
上下文增强 | ❌ | 向量化文本组装 |
hypothetical_questions |
上下文增强 | ❌ | 向量化文本组装 |
keywords |
上下文增强 | ✅ | BM25 检索增强 |
十、注册式工厂模式总结
10.1 统一模式
项目中所有策略模块都采用注册式工厂模式：
┌────────────────────────────────────────────────────┐
│ 工厂（Factory） │
│ _strategies: dict[name, StrategyClass] │
│ register(name, cls) / get_strategy(name) │
└───────────────────────┬────────────────────────────┘
│ 注册
┌───────────────┼───────────────┐
↓ ↓ ↓
strategies/basic.py strategies/xxx.py ...
│
└── 文件末尾: Factory.register("name", XxxStrategy)
strategies/__init__.py:
from .basic import XxxStrategy # 导入触发注册
10.2 四大工厂一览
| 工厂 | 注册键 | 策略数量 | 触发注册位置 |
|---|---|---|---|
CleaningStrategyFactory |
策略名称（str） | 16 个 | cleaning/strategies/__init__.py |
ChunkStrategyFactory |
ChunkStrategyType（枚举） | 6 个 | chunking/strategies/__init__.py |
ContextEnhancementFactory |
增强器名称（str） | 4 个 | context_enhancement/strategies/__init__.py |
MetadataEnrichmentStrategyFactory |
策略名称（str） | 1 个 | metadata_enrichment/strategies/__init__.py |
10.3 新增策略的标准流程
以新增一个清洗策略为例：
# 1. 创建策略文件：cleaning/strategies/my_new_strategy.py
class MyNewCleaningStrategy(CleaningStrategy):
@property
def name(self) -> str:
return "my_new_strategy"
def process(self, documents, params) -> CleaningResult:
# 清洗逻辑
return CleaningResult(documents=cleaned_docs, stats={"modified": count})
# 文件末尾自动注册
CleaningStrategyFactory.register("my_new_strategy", MyNewCleaningStrategy)
# 2. 在 __init__.py 中添加导入：cleaning/strategies/__init__.py
from .my_new_strategy import MyNewCleaningStrategy
# 3. 完成！无需修改工厂代码、Service 代码
十一、完整执行链路示例
11.1 从任务入口到最终入库
# 1. 任务入口（document_parse_task.py）
async def document_parse_job(doc_ids=None):
documents = await fetch_pending_documents(db)
for document in documents:
await process_single_document(document)
# 2. 单文档处理
async def process_single_document(document):
# 2.1 文件读取（同步阻塞，放入线程池）
documents = await asyncio.to_thread(common_read_file_content, file_path)
# 2.2 解析切片策略
strategy_type, strategy_params = get_chunk_strategy_params(document.chunk_strategy)
# 2.3 自动模式：分析文档特征推荐策略
if strategy_type == ChunkStrategyType.AUTO:
recommendation = AutoChunkStrategyRecommender.recommend(documents)
strategy_type = recommendation.strategy_type
strategy_params = recommendation.strategy_params
# 2.4 构建 PipelineContext
pipeline_context = PipelineContext(
chunking_strategy_type=strategy_type,
chunking_params=strategy_params,
meta_doc_id=document.id,
meta_kb_id=document.kb_id,
# ...
)
# 2.5 执行 Pipeline
nodes = await _run_ingestion_pipeline(documents, pipeline_context)
# 2.6 向量化入库（Milvus）
await StorageService.ingest_nodes(nodes, collection_name=kb_info.milvus_collection)
# 2.7 切片元数据持久化（PostgreSQL）
await save_chunks_to_db(session, chunk_vos, doc_id, kb_id, doc_type)
# 3. Pipeline 执行
async def _run_ingestion_pipeline(documents, pipeline_context):
pipeline = RAGIngestionPipeline(context=pipeline_context)
nodes = await pipeline.arun(documents=documents)
return nodes
11.2 Pipeline 内部执行流程
RAGIngestionPipeline.arun(documents)
│
└── self.pipeline.arun(documents, pipeline_context=self.context)
│
│ LlamaIndex IngestionPipeline 内部遍历 transformations：
│
├── [1] DocumentCleanerComponent.acall()
│ └── asyncio.to_thread(__call__)
│ └── CleaningService.process_cleaning()
│ └── 16 个策略链式执行
│
├── [2] ChunkingComponent.acall()
│ └── asyncio.to_thread(__call__)
│ └── ChunkingService.process_chunking()
│ └── ChunkStrategyFactory.get_strategy().execute()
│
├── [3] ChunkPostProcessorComponent.acall()
│ └── asyncio.to_thread(__call__)
│ └── 注入 prev/next_chunk_id + heading_path
│
├── [4] ContextEnhancementComponent.acall() ← 纯异步
│ ├── await HyDEEnhancer.enhance()
│ │ └── asyncio.gather(to_thread(llm.complete, ...) × N)
│ ├── await SummaryEnhancer.enhance()
│ │ └── asyncio.gather(to_thread(llm.complete, ...) × N)
│ ├── await TitleInjectionEnhancer.enhance()
│ └── _assemble_enhanced_text() ← 组装最终向量化文本
│
├── [5] MetadataEnrichmentComponent.acall()
│ └── asyncio.to_thread(__call__)
│ └── MetadataEnrichmentService.process_enrichment()
│ └── BasicMetadataEnrichmentStrategy.process()
│
└── [6] StorageContextCollectorComponent.acall()
└── 直接执行，记录 final_node_count
十二、关键设计决策
12.1 为什么上下文增强器只写 metadata，不直接改 text？
设计决策：增强器的 enhance() 方法只写入 metadata 字段（如 chunk_summary、hypothetical_questions），不修改 node.text。最终由 _assemble_enhanced_text() 统一组装。
原因：
- 可组合性：各增强器独立执行，互不干扰，最终统一组装
- 可追溯性：原文始终保留在
node.text中，增强内容在 metadata 中 - 灵活性：组装格式集中管理，修改拼接模板不影响增强器逻辑
- 检索端友好：通过
<original_text>标记，检索端可以精确提取原文给 LLM
12.2 为什么 CPU 密集型组件用 asyncio.to_thread 而不是直接异步？
原因：
- 清洗、切片等操作涉及大量正则匹配、字符串处理，是纯 CPU 密集型
- Python GIL 限制了 CPU 密集型操作在 asyncio 事件循环中会阻塞其他协程
asyncio.to_thread()将操作放入线程池执行，事件循环可以继续处理其他请求- 项目全局线程池
max_workers=50，足以支撑并发处理
12.3 为什么 ContextEnhancementComponent 的 call 抛异常？
设计决策：ContextEnhancementComponent.__call__() 抛出 RuntimeError，强制要求使用 arun()。
原因：
- 上下文增强的核心逻辑是调用 LLM API，是 I/O 密集型操作
- 没有同步逻辑可以执行，
__call__()无法提供有意义的实现 - 强制异步可以避免开发者误用同步调用导致事件循环阻塞
12.4 为什么使用 <original_text> 标记？
设计决策：在向量化文本中用 <original_text>...</original_text> 标记包裹原文。
原因：
- 向量化时：使用完整增强文本（含标记），让 embedding 包含摘要、假设性问题的语义
- 给 LLM 时：调用
extract_original_text()提取纯净原文，避免把假设性问题传给 LLM - 简单可靠：基于正则匹配提取，不需要额外的存储字段
十三、目录结构总览
rag_common/
├── cleaning/ # 文档清洗模块
│ ├── strategies/ # 16 个清洗策略实现
│ │ ├── basic.py # 基础规范化
│ │ ├── special_char.py # 乱码/Emoji 处理
│ │ ├── privacy_redaction.py # 隐私脱敏
│ │ ├── hyperlink_handling.py # 超链接处理
│ │ ├── document_artifact.py # 页面产物移除
│ │ ├── boilerplate.py # 模板套话移除
│ │ ├── meaningless_filter.py # 无意义内容过滤
│ │ ├── footnote_reference.py # 脚注/参考文献
│ │ ├── text_repair.py # 断行/连字符修复
│ │ ├── chinese.py # 中文规范化
│ │ ├── markdown.py # Markdown 格式修复
│ │ ├── code_block.py # 代码块修复
│ │ ├── table_repair.py # 表格修复
│ │ ├── structure_recovery.py # 结构恢复
│ │ ├── paragraph_merge.py # 段落合并/拆分
│ │ └── dedup.py # 去重
│ ├── factory.py # 注册式工厂
│ ├── schemas.py # 数据模型（CleanRequest/Response）
│ ├── service.py # 清洗服务（链式执行）
│ └── strategy.py # 抽象基类
│
├── chunking/ # 文档切片模块
│ ├── strategies/ # 6 个切片策略实现
│ │ ├── sentence.py # 按句切片
│ │ ├── semantic.py # 语义切片
│ │ ├── window.py # 句子窗口
│ │ ├── hierarchical.py # 层次化切片
│ │ ├── markdown.py # Markdown 标题切片
│ │ └── json_parser.py # JSON 结构切片
│ ├── auto_strategy.py # 自动策略推荐引擎
│ ├── factory.py # 注册式工厂
│ ├── schemas.py # 数据模型（ChunkStrategyType 枚举）
│ ├── service.py # 切片服务
│ └── strategy.py # 抽象基类
│
├── context_enhancement/ # 上下文增强模块
│ ├── strategies/ # 4 个增强器实现
│ │ ├── hyde.py # HyDE 假设性问题
│ │ ├── summary.py # 切片摘要
│ │ ├── keyword.py # 关键词提取
│ │ └── title_injection.py # 标题注入
│ ├── base.py # 增强器抽象基类
│ └── factory.py # 注册式工厂
│
├── metadata_enrichment/ # 元数据增强模块
│ ├── strategies/
│ │ └── basic.py # 基础元数据注入策略
│ ├── factory.py # 注册式工厂
│ ├── schemas.py # 数据模型（MetadataContext）
│ ├── service.py # 元数据增强服务
│ └── strategy.py # 抽象基类
│
├── pipeline/ # Pipeline 编排层
│ ├── components/ # 6 个 TransformComponent
│ │ ├── _common.py # 公共工具（get_pipeline_context）
│ │ ├── cleaner.py # 清洗组件
│ │ ├── chunker.py # 切片组件
│ │ ├── chunk_post_processor.py # 切片后处理组件
│ │ ├── context_enhancement.py # 上下文增强组件
│ │ ├── metadata_enrichment.py # 元数据增强组件
│ │ └── collector.py # 入库收集组件
│ ├── context.py # PipelineContext 定义
│ └── pipeline.py # RAGIngestionPipeline 组装
│
└── ...（storage/, ragas_eva/, utils/ 等）
十四、总结
14.1 核心架构优势
| 特性 | 实现方式 | 收益 |
|---|---|---|
| 组件解耦 | TransformComponent + PipelineContext | 各组件独立开发测试 |
| 策略可插拔 | 注册式工厂 + 导入即注册 | 新增策略零修改工厂 |
| 全异步执行 | asyncio.to_thread + 纯异步增强器 | 不阻塞事件循环 |
| 配置统一 | PipelineContext 分组属性 | 避免参数透传地狱 |
| 可追溯 | ctx.state 记录中间结果 | 审计和问题排查 |
| 检索增强 | HyDE + Summary + Title + 固定拼接 | 显著提升召回率 |
14.2 关键数字
| 指标 | 数值 |
|---|---|
| 清洗策略数量 | 16 个（5 阶段链式执行） |
| 切片策略数量 | 6 个 + 1 个自动推荐 |
| 上下文增强器数量 | 4 个（2 个 LLM + 2 个零成本） |
| Pipeline 组件数量 | 6 个 |
| 最终 metadata 字段数 | 20+ 个 |
文档版本：v1.0
适用项目：RAG-FastAPI-Backend
核心依赖：LlamaIndex IngestionPipeline + FastAPI + Milvus
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
3
0- 0
已为社区贡献3条内容
所有评论(0)