【RAG 实战】基于文档结构特征分析的切片策略自动推荐
·
【RAG 实战】基于文档结构特征分析的切片策略自动推荐引擎设计与实现
摘要：在 RAG（Retrieval-Augmented Generation）系统中，文档切片策略的选择直接影响检索质量。本文介绍了一种基于文档结构特征分析的切片策略自动推荐引擎，通过"规则引擎 + LLM 轻量判定"的两阶段架构，自动为不同类型的文档推荐最优切片策略。文章涵盖架构设计、特征提取、规则打分、LLM 决策、参数自动计算等完整实现细节，并附带可直接运行的代码示例。
文章目录
📌 一、问题背景
1.1 切片策略对 RAG 的影响
在 RAG 系统中，文档切片（Chunking） 是决定检索质量的核心环节。不同的文档类型适合不同的切片策略：
| 文档类型 | 推荐策略 | 原因 |
|---|---|---|
| Markdown 格式文档（法规、合同、API 文档） | markdown |
按标题层级切片，保持章节完整性 |
| 语义连贯的长文（报告、会议记录） | semantic |
按语义相似度切片，避免切断语义 |
| 表格/列表密集型文档（数据报告） | markdown |
保持结构化元素完整性 |
| FAQ/问答文档（客服话术） | window |
句子窗口切片保留问答上下文 |
| 超长文档（财报、白皮书） | hierarchical |
多粒度层级切片，支持粗/细粒度检索 |
| 短文本/碎片（词条说明） | sentence |
按句切片获得更精确粒度 |
1.2 痛点
用户往往不知道该选哪种策略，只能凭经验盲猜。选错策略会导致：
- 表格被截断 → 检索到不完整的表格
- 语义被切断 → 检索到断章取义的片段
- 超长文档直接语义切分 → 触碰 Embedding Token 上限报错
1.3 解决方案
构建一个自动推荐引擎，基于文档的结构特征分析，自动推荐最适合的切片策略。
🏗️ 二、整体架构设计
2.1 设计理念
采用 “规则引擎 + LLM 轻量判定” 结合的方式：
- 规则引擎先打分：基于文档结构特征，运行多条规则，筛选 Top 3 候选策略
- LLM 做最终决策：从 Top 3 候选中，结合文档内容语义选出最优策略
- 特征提取与策略判定分离：便于独立测试和调优
- 支持用户手动覆盖：用户指定策略时直接返回，跳过自动判定
2.2 核心组件
┌─────────────────────────────────────────────────────────────────┐
│ AutoChunkStrategyRecommender │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ MDStructureAnalyzer（特征提取器） │ │
│ │ - 标题层级分析 - 段落长度统计 │ │
│ │ - 表格/列表占比 - FAQ/步骤型模式检测 │ │
│ └───────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ DocumentStructureFeatures（特征数据类） │ │
│ │ - total_chars / heading_density │ │
│ │ - table_ratio / list_ratio │ │
│ │ - faq_pattern_ratio / numbered_steps_ratio │ │
│ └───────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ 规则引擎（Rule Engine） │ │
│ │ - 超长文档 → hierarchical │ │
│ │ - 多级标题 → markdown │ │
│ │ - 表格/列表密集 → markdown │ │
│ │ - FAQ 密集 → sentence_window │ │
│ │ - 步骤型文本 → sentence_window │ │
│ │ - 语义连贯 → semantic │ │
│ │ - 短文本 → sentence │ │
│ └───────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ LLM 轻量判定（从 Top 3 中选择最优） │ │
│ │ - 失败时回退到规则引擎第一候选 │ │
│ └───────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ StrategyRecommendation（推荐结果） │ │
│ │ - strategy_type / strategy_params / confidence / reason │ │
│ └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
2.3 执行流程
输入文档 → 特征提取 → 规则打分(7条) → 排序去重 → Top 3 候选
↓
LLM 选择最优
↓
输出推荐结果
(策略+参数+原因)
📊 三、数据模型设计
3.1 文档结构特征集合
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class DocumentStructureFeatures:
"""文档结构特征集合
所有字段均为从 MD 文本中提取的统计指标，用于策略判定。
"""
# --- 基础统计 ---
total_chars: int = 0
"""文档总字符数"""
total_lines: int = 0
"""文档总行数（非空行）"""
total_paragraphs: int = 0
"""文档总段落数（以空行分隔）"""
# --- 标题特征 ---
heading_counts: dict = field(default_factory=dict)
"""各级标题数量，如 {1: 5, 2: 12, 3: 8, 4: 3}"""
max_heading_level: int = 0
"""最大标题层级深度（如 3 表示最深到 ###）"""
total_headings: int = 0
"""标题总数"""
heading_density: float = 0.0
"""标题密度 = 标题总数 / 段落数"""
# --- 段落特征 ---
avg_paragraph_length: float = 0.0
"""平均段落长度（字符数）"""
max_paragraph_length: float = 0.0
"""最长段落长度（字符数）"""
short_paragraph_ratio: float = 0.0
"""短段落占比（段落长度 < 50 字符的比例）"""
# --- 结构化元素占比 ---
table_line_count: int = 0
"""表格行数（包含 | 的行）"""
table_ratio: float = 0.0
"""表格行占总非空行的比例"""
list_line_count: int = 0
"""列表行数（以 - / * / 数字. 开头的行）"""
list_ratio: float = 0.0
"""列表行占总非空行的比例"""
code_block_count: int = 0
"""代码块数量（```包裹的块）"""
image_count: int = 0
"""图片数量（![...] 格式）"""
link_count: int = 0
"""链接数量（[...]() 格式，不含图片）"""
# --- 分隔符与特殊元素 ---
horizontal_rule_count: int = 0
"""水平分隔线数量（--- / *** / ___）"""
blockquote_count: int = 0
"""引用块行数（> 开头）"""
# --- 语义连贯性指标 ---
avg_sentence_length: float = 0.0
"""平均句子长度（字符数，按 。！？.!? 分句）"""
sentence_count: int = 0
"""句子总数"""
# --- 场景特征 ---
faq_pattern_count: int = 0
"""FAQ 模式段落数（问句+答句交替）"""
faq_pattern_ratio: float = 0.0
"""FAQ 模式段落占比"""
numbered_steps_count: int = 0
"""编号步骤数（1. 2. 3. 或 一、二、三、）"""
numbered_steps_ratio: float = 0.0
"""编号步骤占比"""
设计说明：
- 特征分为 6 大类：基础统计、标题特征、段落特征、结构化元素、分隔符、语义连贯性、场景特征
heading_density（标题密度）是关键指标：标题多且密度合理 → 适合按标题切片faq_pattern_ratio和numbered_steps_ratio是场景特征，用于识别特定文档类型
3.2 策略推荐结果
@dataclass
class StrategyRecommendation:
"""策略推荐结果"""
strategy_type: ChunkStrategyType
"""推荐的切片策略类型"""
strategy_params: dict = field(default_factory=dict)
"""推荐的切片策略参数"""
features: Optional[DocumentStructureFeatures] = None
"""提取到的文档结构特征"""
reason: str = ""
"""推荐原因说明"""
confidence: str = "medium"
"""推荐置信度：high / medium / low"""
🔍 四、特征提取器实现
MDStructureAnalyzer 负责从 Markdown 文本中提取各类结构特征。
4.1 正则模式定义
import re
class MDStructureAnalyzer:
"""Markdown 文档结构分析器
从 MD 文本中提取各类结构特征，供策略推荐引擎使用。
"""
# --- 正则模式 ---
_HEADING_RE = re.compile(r'^(#{1,6})\s+(.+)$', re.MULTILINE)
_TABLE_LINE_RE = re.compile(r'^\s*\|.*\|', re.MULTILINE)
_LIST_LINE_RE = re.compile(r'^\s*(?:[-*+]|\d+[.)])\s+', re.MULTILINE)
_CODE_BLOCK_RE = re.compile(r'^```', re.MULTILINE)
_IMAGE_RE = re.compile(r'!\[.*?\]\(.*?\)')
_LINK_RE = re.compile(r'(?<!!)\[.*?\]\(.*?\)')
_HR_RE = re.compile(r'^\s*(?:---+|\*\*\*+|___+)\s*$', re.MULTILINE)
_BLOCKQUOTE_RE = re.compile(r'^\s*>\s*', re.MULTILINE)
_SENTENCE_SPLIT_RE = re.compile(r'[。！？.!?]+')
# 场景特征检测
_FAQ_QUESTION_RE = re.compile(r'^[^\n]*[？?]\s*$', re.MULTILINE)
_NUMBERED_STEP_RE = re.compile(
r'^(?:\d+|[\u4e00\u4e8c\u4e09\u56db\u4e94\u516d\u4e03\u516b\u4e5d\u5341])[、.．]\s*',
re.MULTILINE,
)
各正则的作用：
| 正则 | 匹配目标 | 示例 |
|---|---|---|
_HEADING_RE |
Markdown 标题 | ## 二级标题 |
_TABLE_LINE_RE |
表格行 | | 列1 | 列2 | |
_LIST_LINE_RE |
列表行 | - 项目 / 1. 步骤 |
_CODE_BLOCK_RE |
代码块标记 | ```python |
_IMAGE_RE |
图片 |  |
_LINK_RE |
链接（排除图片） | [text](url) |
_HR_RE |
水平分隔线 | --- / *** |
_BLOCKQUOTE_RE |
引用块 | > 引用内容 |
_SENTENCE_SPLIT_RE |
句子分隔符 | 。！？.!? |
_FAQ_QUESTION_RE |
问句（以问号结尾的行） | 什么是 RAG？ |
_NUMBERED_STEP_RE |
编号步骤（中英文） | 1. 第一步 / 一、概述 |
4.2 特征提取主方法
@classmethod
def analyze(cls, md_text: str) -> DocumentStructureFeatures:
"""分析 MD 文本，提取结构特征
:param md_text: Markdown 格式的纯文本
:return: 文档结构特征集合
"""
if not md_text or not md_text.strip():
return DocumentStructureFeatures()
features = DocumentStructureFeatures()
features.total_chars = len(md_text)
# 非空行
all_lines = md_text.split('\n')
non_empty_lines = [line for line in all_lines if line.strip()]
features.total_lines = len(non_empty_lines)
# 段落（以空行分隔）
paragraphs = cls._split_paragraphs(md_text)
features.total_paragraphs = max(len(paragraphs), 1)
# --- 标题特征 ---
heading_matches = cls._HEADING_RE.findall(md_text)
heading_counts = {}
for hashes, _ in heading_matches:
level = len(hashes)
heading_counts[level] = heading_counts.get(level, 0) + 1
features.heading_counts = heading_counts
features.total_headings = sum(heading_counts.values())
features.max_heading_level = max(heading_counts.keys()) if heading_counts else 0
features.heading_density = features.total_headings / features.total_paragraphs
# --- 段落特征 ---
para_lengths = [len(p.strip()) for p in paragraphs if p.strip()]
if para_lengths:
features.avg_paragraph_length = sum(para_lengths) / len(para_lengths)
features.max_paragraph_length = max(para_lengths)
short_count = sum(1 for length in para_lengths if length < 50)
features.short_paragraph_ratio = short_count / len(para_lengths)
# --- 结构化元素 ---
features.table_line_count = len(cls._TABLE_LINE_RE.findall(md_text))
features.table_ratio = features.table_line_count / max(features.total_lines, 1)
features.list_line_count = len(cls._LIST_LINE_RE.findall(md_text))
features.list_ratio = features.list_line_count / max(features.total_lines, 1)
code_fence_count = len(cls._CODE_BLOCK_RE.findall(md_text))
features.code_block_count = code_fence_count // 2 # 每个代码块有开闭两个标记
features.image_count = len(cls._IMAGE_RE.findall(md_text))
features.link_count = len(cls._LINK_RE.findall(md_text))
# --- 分隔符与特殊元素 ---
features.horizontal_rule_count = len(cls._HR_RE.findall(md_text))
features.blockquote_count = len(cls._BLOCKQUOTE_RE.findall(md_text))
# --- 语义连贯性 ---
sentences = [s.strip() for s in cls._SENTENCE_SPLIT_RE.split(md_text) if s.strip()]
features.sentence_count = max(len(sentences), 1)
total_sentence_len = sum(len(s) for s in sentences)
features.avg_sentence_length = total_sentence_len / features.sentence_count
# --- 场景特征 ---
# FAQ 模式检测：以问号结尾的行
faq_questions = cls._FAQ_QUESTION_RE.findall(md_text)
features.faq_pattern_count = len(faq_questions)
features.faq_pattern_ratio = features.faq_pattern_count / max(features.total_paragraphs, 1)
# 编号步骤检测
numbered_steps = cls._NUMBERED_STEP_RE.findall(md_text)
features.numbered_steps_count = len(numbered_steps)
features.numbered_steps_ratio = features.numbered_steps_count / max(features.total_lines, 1)
return features
4.3 段落分割辅助方法
@staticmethod
def _split_paragraphs(text: str) -> List[str]:
"""按空行分割段落"""
paragraphs = []
current = []
for line in text.split('\n'):
if line.strip():
current.append(line)
else:
if current:
paragraphs.append('\n'.join(current))
current = []
if current:
paragraphs.append('\n'.join(current))
return paragraphs
🧠 五、策略推荐引擎实现
5.1 可调阈值集中管理
所有阈值定义为类属性，便于后续调优：
class AutoChunkStrategyRecommender:
"""切片策略自动推荐引擎"""
# --- 可调阈值（集中管理，便于后续调优） ---
# 超长文档阈值（字符数）
VERY_LONG_DOC_THRESHOLD = 100_000 # > 10 万字符视为超长文档
LONG_DOC_THRESHOLD = 50_000 # > 5 万字符视为长文档
# 标题结构判定
MIN_HEADING_LEVELS_FOR_MARKDOWN = 2 # 至少 2 级标题才适合 markdown 切片
MIN_HEADING_DENSITY = 0.05 # 标题密度下限（太低说明标题稀疏）
MAX_HEADING_DENSITY = 0.5 # 标题密度上限（太高说明标题过密）
# 结构化元素判定
HIGH_TABLE_RATIO = 0.15 # 表格行占比 > 15% 视为表格密集
HIGH_LIST_RATIO = 0.20 # 列表行占比 > 20% 视为列表密集
# 语义连贯性判定
MIN_AVG_PARA_LENGTH_FOR_SEMANTIC = 100 # 平均段落长度 > 100 字符才适合语义切分
MAX_SHORT_PARA_RATIO_FOR_SEMANTIC = 0.3 # 短段落占比 < 30% 才适合语义切分
MIN_AVG_SENTENCE_LENGTH = 15 # 平均句长 > 15 字符
# 短文本判定
SHORT_DOC_THRESHOLD = 5_000 # < 5000 字符视为短文档
VERY_SHORT_DOC_THRESHOLD = 2_000 # < 2000 字符视为极短文档
# 场景特征阈值
FAQ_PATTERN_RATIO_THRESHOLD = 0.15 # FAQ 模式段落占比 > 15% 视为 FAQ 密集
NUMBERED_STEPS_RATIO_THRESHOLD = 0.10 # 编号步骤占比 > 10% 视为步骤型文档
5.2 推荐主入口
@classmethod
def recommend(
cls,
source: str | List[Document],
user_strategy: Optional[ChunkStrategyType] = None,
) -> StrategyRecommendation:
"""根据文档内容推荐切片策略
流程：提取特征 → 规则引擎打分取 Top 3 → LLM 从候选中选最优
:param source: MinerU 输出的 Markdown 文本，或 LlamaIndex Document 列表
:param user_strategy: 用户手动指定的策略（非 None 时直接返回，跳过自动判定）
:return: 策略推荐结果
"""
# 用户手动覆盖优先
if user_strategy is not None:
return StrategyRecommendation(
strategy_type=user_strategy,
reason=f'用户手动指定: {user_strategy.value}',
confidence='high',
)
# Document 列表 → 拼接为纯文本
md_text = (
'\n\n'.join(doc.text or '' for doc in source)
if isinstance(source, list)
else source
)
features = MDStructureAnalyzer.analyze(md_text)
# 空文档兜底
if features.total_chars == 0:
return StrategyRecommendation(
strategy_type=ChunkStrategyType.SENTENCE,
strategy_params={},
features=features,
reason='文档为空，使用默认 sentence 策略',
confidence='low',
)
# 规则引擎打分 → 排序去重取 Top 3
top = cls._score_strategies(features)
if not top:
top = [cls._rule_fallback(features)]
for i, rec in enumerate(top, 1):
logger.info(
f"[AUTO_STRATEGY] 候选 #{i}: {rec.strategy_type.value}"
f" (置信度={rec.confidence}) | {rec.reason}"
)
# LLM 从 Top 3 中选最优，失败则回退到规则引擎第一候选
try:
sample = md_text[:_LLM_SAMPLE_MAX_CHARS]
llm_result = cls._llm_select_strategy(top, features, sample)
if llm_result is not None:
llm_result.features = features
return llm_result
except Exception as e:
logger.warning(f"[AUTO_STRATEGY] LLM 判定失败，使用规则引擎第一候选: {e}")
result = top[0]
result.features = features
return result
关键设计点：
- 用户手动覆盖优先：如果用户指定了策略，直接返回，不做分析
- 支持两种输入：纯文本
str或 LlamaIndexList[Document] - 空文档保护：空文档直接返回
sentence兜底 - LLM 失败回退：LLM 调用失败时，使用规则引擎的第一候选
5.3 规则打分与排序
# 常量定义
_TOP_K = 3
_CONFIDENCE_WEIGHT: dict[str, int] = {'high': 3, 'medium': 2, 'low': 1}
@classmethod
def _score_strategies(cls, features: DocumentStructureFeatures) -> list[StrategyRecommendation]:
"""运行所有规则，按置信度排序去重后返回 Top K 候选"""
results: list[StrategyRecommendation] = []
for rule_method in [
cls._rule_very_long_document,
cls._rule_heading_structure,
cls._rule_structured_elements,
cls._rule_faq_dense,
cls._rule_procedural_text,
cls._rule_semantic_text,
cls._rule_short_text,
]:
result = rule_method(features)
if result is not None:
results.append(result)
if not results:
results.append(cls._rule_fallback(features))
# 按置信度降序排序，同一策略类型去重（保留最高置信度）
results.sort(key=lambda r: _CONFIDENCE_WEIGHT.get(r.confidence, 0), reverse=True)
seen: set[str] = set()
unique: list[StrategyRecommendation] = []
for rec in results:
key = rec.strategy_type.value
if key not in seen:
seen.add(key)
unique.append(rec)
return unique[:_TOP_K]
设计要点：
- 运行所有 7 条规则，收集命中的结果
- 按置信度降序排序（
high > medium > low） - 同一策略类型去重，保留最高置信度的那条
- 取 Top 3 候选给 LLM 选择
📏 六、规则链详解
规则按优先级从高到低排列，每条规则返回 Optional[StrategyRecommendation]。
6.1 规则 1：超长文档 → hierarchical
@classmethod
def _rule_very_long_document(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""超长文档使用 hierarchical 多粒度切片
适用场景：财报、白皮书、长篇技术文档
原因：超长文档需要多粒度检索（粗粒度定位章节，细粒度检索段落）
"""
if f.total_chars >= cls.VERY_LONG_DOC_THRESHOLD:
return StrategyRecommendation(
strategy_type=ChunkStrategyType.HIERARCHICAL,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.HIERARCHICAL, f),
reason=(
f'超长文档（{f.total_chars:,} 字符），'
f'使用 hierarchical 多粒度切片以支持章节级和段落级检索'
),
confidence='high',
)
return None
判定逻辑：文档总字符数 > 100,000 → 推荐 hierarchical
为什么：超长文档如果只用单层切片，要么粒度太粗丢失细节，要么粒度太细丢失全局。hierarchical 策略生成多层级的 Node（父-子关系），检索时可以粗粒度定位章节，再下钻到具体段落。
6.2 规则 2：多级标题结构 → markdown
@classmethod
def _rule_heading_structure(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""有清晰多级标题的文档使用 markdown 按标题切片"""
if f.max_heading_level < cls.MIN_HEADING_LEVELS_FOR_MARKDOWN:
return None
if not (cls.MIN_HEADING_DENSITY <= f.heading_density <= cls.MAX_HEADING_DENSITY):
return None
# 构建标题层级描述
level_desc = ', '.join(
f'H{level}: {count}'
for level, count in sorted(f.heading_counts.items())
)
return StrategyRecommendation(
strategy_type=ChunkStrategyType.MARKDOWN,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.MARKDOWN, f),
reason=(
f'文档具有 {f.max_heading_level} 级标题结构'
f'（{level_desc}），标题密度 {f.heading_density:.2f}，'
f'使用 markdown 按标题层级切片'
),
confidence='high',
)
判定逻辑：
- 至少 2 级标题层级（
max_heading_level >= 2） - 标题密度在合理范围内（
0.05 <= heading_density <= 0.5）
为什么：有清晰标题结构的文档，按标题切片能保持章节完整性，检索时还能带上标题路径（如 第三章 > 3.2 核心架构）。
6.3 规则 3：表格/列表密集 → markdown
@classmethod
def _rule_structured_elements(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""表格或列表密集的文档使用 markdown 切片"""
reasons = []
if f.table_ratio >= cls.HIGH_TABLE_RATIO:
reasons.append(f'表格密集（表格行占比 {f.table_ratio:.1%}）')
if f.list_ratio >= cls.HIGH_LIST_RATIO:
reasons.append(f'列表密集（列表行占比 {f.list_ratio:.1%}）')
if reasons:
return StrategyRecommendation(
strategy_type=ChunkStrategyType.MARKDOWN,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.MARKDOWN, f),
reason='，'.join(reasons) + '，使用 markdown 切片以保持结构化元素完整性',
confidence='medium',
)
return None
判定逻辑：
- 表格行占比 > 15% → 表格密集
- 列表行占比 > 20% → 列表密集
为什么：semantic 或 sentence 策略可能会把一个完整的表格从中间切断，导致检索到半截表格。markdown 策略能识别并保护结构化元素。
6.4 规则 3.5：FAQ/问答密集 → window
@classmethod
def _rule_faq_dense(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""FAQ/问答密集文档使用 window 切片"""
if f.faq_pattern_ratio < cls.FAQ_PATTERN_RATIO_THRESHOLD:
return None
return StrategyRecommendation(
strategy_type=ChunkStrategyType.SENTENCE_WINDOW,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.SENTENCE_WINDOW, f),
reason=(
f'FAQ 模式密集（{f.faq_pattern_ratio:.0%} 段落为问答对），'
f'使用 window 切片以保留完整问答上下文'
),
confidence='high',
)
判定逻辑：以问号结尾的行占总段落比例 > 15%
为什么：FAQ 文档中，问句和答句通常紧密相邻。window 策略在检索时会自动扩展上下文窗口——问句命中后能展示完整答案，而非孤立的句子。
6.5 规则 3.6：步骤型文本 → window
@classmethod
def _rule_procedural_text(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""步骤型文档使用 window 切片"""
if f.numbered_steps_ratio < cls.NUMBERED_STEPS_RATIO_THRESHOLD:
return None
# 如果有大量标题，更适合 markdown
if f.max_heading_level >= cls.MIN_HEADING_LEVELS_FOR_MARKDOWN:
return None
return StrategyRecommendation(
strategy_type=ChunkStrategyType.SENTENCE_WINDOW,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.SENTENCE_WINDOW, f),
reason=(
f'步骤型文本（编号步骤占比 {f.numbered_steps_ratio:.1%}），'
f'使用 window 切片保持步骤上下文连贯'
),
confidence='medium',
)
判定逻辑：
- 编号步骤占比 > 10%
- 且没有多级标题（有标题的优先走 markdown）
为什么：操作手册、SOP 中的步骤间有强上下文依赖（"第 3 步"依赖"第 2 步"的结果），窗口扩展能避免切断步骤链。
6.6 规则 4：语义连贯无标题 → semantic
@classmethod
def _rule_semantic_text(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""语义连贯但无标题结构的文档使用 semantic 按语义切片"""
if f.heading_density >= cls.MIN_HEADING_DENSITY:
return None
if f.avg_paragraph_length < cls.MIN_AVG_PARA_LENGTH_FOR_SEMANTIC:
return None
if f.short_paragraph_ratio > cls.MAX_SHORT_PARA_RATIO_FOR_SEMANTIC:
return None
if f.avg_sentence_length < cls.MIN_AVG_SENTENCE_LENGTH:
return None
return StrategyRecommendation(
strategy_type=ChunkStrategyType.SEMANTIC,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.SEMANTIC, f),
reason=(
f'文档语义连贯但无明显标题结构'
f'（标题密度 {f.heading_density:.2f}，'
f'平均段落长度 {f.avg_paragraph_length:.0f} 字符），'
f'使用 semantic 按语义相似度切片'
),
confidence='medium',
)
判定逻辑（需同时满足）：
- 标题密度 < 0.05（标题极少）
- 平均段落长度 > 100 字符（段落较长且连贯）
- 短段落占比 < 30%（不是碎片文本）
- 平均句长 > 15 字符（不是超短句子）
为什么：没有标题结构但语义连贯的文档（如会议记录、报告），按语义相似度切分能在主题转换处自然断开。
6.7 规则 5：短文本 → sentence
@classmethod
def _rule_short_text(cls, f: DocumentStructureFeatures) -> Optional[StrategyRecommendation]:
"""短文档使用 sentence 按句切片"""
if f.total_chars >= cls.SHORT_DOC_THRESHOLD:
return None
return StrategyRecommendation(
strategy_type=ChunkStrategyType.SENTENCE,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.SENTENCE, f),
reason=(
f'短文档（{f.total_chars:,} 字符），'
f'使用 sentence 按句切片以获得更精确的检索粒度'
),
confidence='medium',
)
判定逻辑：文档总字符数 < 5,000
为什么：短文档用语义切分粒度太粗（可能整个文档就一个语义段），按句切片更精确。
6.8 规则 6：兜底 → sentence
@classmethod
def _rule_fallback(cls, f: DocumentStructureFeatures) -> StrategyRecommendation:
"""兜底策略：使用 sentence 切片"""
return StrategyRecommendation(
strategy_type=ChunkStrategyType.SENTENCE,
strategy_params=cls._compute_strategy_params(ChunkStrategyType.SENTENCE, f),
reason=(
f'未匹配到明确的策略特征'
f'（标题层级 {f.max_heading_level}，'
f'标题密度 {f.heading_density:.2f}，'
f'平均段落长度 {f.avg_paragraph_length:.0f}），'
f'使用 sentence 兜底策略'
),
confidence='low',
)
为什么选 sentence 兜底：sentence 策略对各类文档都有基本适应性，不会出现严重问题（如表格截断、Token 超限等）。
🤖 七、LLM 轻量判定
规则引擎输出 Top 3 候选后，调用 LLM 做最终决策。
7.1 LLM 选择逻辑
_LLM_SAMPLE_MAX_CHARS = 2000 # LLM 判定时的文档内容采样长度
@classmethod
def _llm_select_strategy(
cls,
candidates: list[StrategyRecommendation],
features: DocumentStructureFeatures,
sample: str,
) -> Optional[StrategyRecommendation]:
"""调用 LLM 从规则引擎 Top 3 候选中选择最优策略"""
from module_rag.config.settings import get_llm
llm = get_llm()
# 构建候选描述
candidate_desc = '\n'.join(
f'{i+1}. {c.strategy_type.value}（置信度: {c.confidence}，原因: {c.reason}）'
for i, c in enumerate(candidates)
)
# 构建特征摘要
feature_summary = (
f'文档总字符: {features.total_chars:,}，'
f'总段落: {features.total_paragraphs}，'
f'最大标题层级: {features.max_heading_level}，'
f'标题密度: {features.heading_density:.2f}，'
f'平均段落长度: {features.avg_paragraph_length:.0f}字符，'
f'表格行占比: {features.table_ratio:.1%}，'
f'列表行占比: {features.list_ratio:.1%}，'
f'FAQ段落占比: {features.faq_pattern_ratio:.0%}，'
f'编号步骤占比: {features.numbered_steps_ratio:.1%}'
)
valid_names = '/'.join(c.strategy_type.value for c in candidates)
prompt = f"""你是一个文档切片策略选择专家。请根据以下文档特征和内容摘要，
从候选策略中选择最适合的切片策略。
## 文档特征摘要
{feature_summary}
## 文档内容摘要（前 {_LLM_SAMPLE_MAX_CHARS} 字符）
{sample[:1500]}
## 候选策略（由规则引擎预筛选）
{candidate_desc}
## 要求
1. 请仅从候选策略中选择一个最合适的策略
2. 综合考虑文档结构、内容特点、检索场景
3. 严格按以下 JSON 格式回复，不要添加其他内容：
{{"strategy": "策略名称", "reason": "选择原因的简要说明"}}
可选策略名称: {valid_names}"""
response = llm.complete(prompt)
response_text = str(response).strip()
# 解析 LLM 响应（提取 JSON）
import json
json_match = re.search(r'\{[^}]+\}', response_text)
if not json_match:
return None
try:
result = json.loads(json_match.group())
except json.JSONDecodeError:
return None
strategy_name = result.get('strategy', '')
reason = result.get('reason', '')
# 验证策略名称
try:
strategy_type = ChunkStrategyType(strategy_name)
except ValueError:
return None
# 计算对应策略的参数
strategy_params = cls._compute_strategy_params(strategy_type, features)
return StrategyRecommendation(
strategy_type=strategy_type,
strategy_params=strategy_params,
reason=f'[LLM] {reason}',
confidence='high',
)
设计要点：
- 轻量判定：只采样前 2000 字符，不传全文，降低 Token 消耗
- 约束选择范围：LLM 只能从 Top 3 候选中选择，不能凭空发明
- JSON 格式输出：便于解析，容错处理（正则提取 JSON）
- 失败回退：LLM 调用失败或解析失败时，回退到规则引擎第一候选
⚙️ 八、参数自动计算
推荐引擎不仅推荐策略类型，还会根据文档特征动态计算策略的内部参数。
@classmethod
def _compute_strategy_params(
cls,
strategy_type: ChunkStrategyType,
f: DocumentStructureFeatures,
) -> dict:
"""根据文档特征为指定策略计算最优参数"""
# --- Sentence 策略参数 ---
if strategy_type == ChunkStrategyType.SENTENCE:
avg_para = f.avg_paragraph_length
if avg_para < 80:
chunk_size = 256
elif avg_para < 150:
chunk_size = 384
elif avg_para < 300:
chunk_size = 512
elif avg_para < 500:
chunk_size = 768
else:
chunk_size = 1024
# 极短文档用更小的 chunk_size
if f.total_chars < 2000 and chunk_size > 256:
chunk_size = 256
return {'chunk_size': chunk_size, 'chunk_overlap': max(20, chunk_size // 10)}
# --- Semantic 策略参数 ---
if strategy_type == ChunkStrategyType.SEMANTIC:
# 短段落多 → 降低阈值（更敏感地切分）
if f.short_paragraph_ratio > 0.5:
threshold = 98
elif f.short_paragraph_ratio > 0.3:
threshold = 95
elif f.avg_paragraph_length > 200:
threshold = 90
else:
threshold = 93
return {
'threshold': threshold,
'buffer_size': 2 if f.avg_sentence_length > 30 else 1,
}
# --- Hierarchical 策略参数 ---
if strategy_type == ChunkStrategyType.HIERARCHICAL:
total = f.total_chars
if total >= 500_000:
return {'chunk_sizes': [4096, 2048, 1024, 512]}
if total >= 200_000:
return {'chunk_sizes': [2048, 1024, 512]}
return {'chunk_sizes': [1024, 512, 256]}
# --- Window 策略参数 ---
if strategy_type == ChunkStrategyType.SENTENCE_WINDOW:
# FAQ 密集或段落很短 → 更大的窗口
if f.faq_pattern_ratio > 0.3 or f.avg_paragraph_length < 80:
return {'window_size': 3}
return {'window_size': 2}
return {}
参数自适应逻辑：
| 策略 | 参数 | 自适应逻辑 |
|---|---|---|
sentence |
chunk_size |
段落越长 → chunk_size 越大（256~1024） |
sentence |
chunk_overlap |
chunk_size 的 1/10，最小 20 |
semantic |
threshold |
短段落越多 → 阈值越高（更敏感切分） |
semantic |
buffer_size |
句子越长 → buffer 越大（上下文窗口） |
hierarchical |
chunk_sizes |
文档越长 → 层级越多、每层越大 |
window |
window_size |
FAQ 密集或段落短 → 窗口更大（3 vs 2） |
🔗 九、与切片策略工厂的集成
9.1 注册式工厂模式
# factory.py
class ChunkStrategyFactory:
"""切片策略工厂 - 注册式机制"""
_strategies: Dict[ChunkStrategyType, Type[BaseChunkStrategy]] = {}
@classmethod
def register(cls, strategy_type: ChunkStrategyType,
strategy_class: Type[BaseChunkStrategy]) -> None:
cls._strategies[strategy_type] = strategy_class
@classmethod
def get_strategy(cls, strategy_type: ChunkStrategyType) -> BaseChunkStrategy:
strategy_class = cls._strategies.get(strategy_type)
if not strategy_class:
raise ValueError(f"Unsupported chunk strategy: {strategy_type}")
return strategy_class()
9.2 策略基类
# strategy.py
class BaseChunkStrategy(ABC):
"""切片策略抽象基类"""
strategy_type: ClassVar["ChunkStrategyType"]
@abstractmethod
def get_parser(self, params: dict) -> NodeParser:
"""获取 LlamaIndex 的 NodeParser 实例"""
pass
def execute(self, documents: List[Document], params: dict) -> list:
"""执行切片并返回 Node 列表，自动注入 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
9.3 策略自注册示例（以 Window 策略为例）
# strategies/window.py
class SentenceWindowStrategy(BaseChunkStrategy):
strategy_type: ClassVar[ChunkStrategyType] = ChunkStrategyType.SENTENCE_WINDOW
def get_parser(self, params: dict):
return SentenceWindowNodeParser(
window_size=params.get("window_size", 2),
sentence_splitter=chinese_sentence_splitter,
window_metadata_key="window",
original_text_metadata_key="original_text"
)
def execute(self, documents: List[Document], params: dict) -> list:
nodes = super().execute(documents, params)
window_size = params.get("window_size", 2)
for node in nodes:
node.metadata["window_size"] = window_size
# 将 window 句子列表合并为可直接检索的文本
window_sentences = node.metadata.get("window", [])
if isinstance(window_sentences, list) and window_sentences:
node.metadata["window_text"] = "".join(window_sentences)
return nodes
# 自注册
ChunkStrategyFactory.register(ChunkStrategyType.SENTENCE_WINDOW, SentenceWindowStrategy)
9.4 在文档解析任务中的集成
# tasks/document_parse_task.py
from module_rag.rag_common.chunking.auto_strategy import AutoChunkStrategyRecommender
# 获取用户配置的策略
strategy_type, strategy_params = get_chunk_strategy_params(document.chunk_strategy)
# 自动模式：基于文档内容分析推荐切片策略
if strategy_type == ChunkStrategyType.AUTO:
recommendation = AutoChunkStrategyRecommender.recommend(documents)
logger.info(
f'[AUTO_STRATEGY] doc_id={doc_id} 自动推荐: '
f'{recommendation.strategy_type.value} | {recommendation.reason}'
)
strategy_type = recommendation.strategy_type
strategy_params = recommendation.strategy_params
# 后续使用推荐的策略执行切片...
📋 十、统一元数据协议
所有策略切出的 Node 必须遵循统一元数据协议，确保下游（检索、展示、上下文增强）能统一处理：
@dataclass
class ChunkMetadataProtocol:
"""切片元数据统一协议"""
# === 策略级元数据（由各切片策略自行注入）===
strategy_type: str # "sentence" / "window" / "hierarchical" / ...
# === 关系元数据（由 ChunkPostProcessor 注入）===
prev_chunk_id: Optional[str] = None
next_chunk_id: Optional[str] = None
parent_chunk_id: Optional[str] = None
child_chunk_ids: List[str] = field(default_factory=list)
# === 位置元数据 ===
heading_path: Optional[str] = None # "第一章 > 1.2 核心架构"
heading_level: Optional[int] = None
page_number: Optional[int] = None
page_start: Optional[int] = None
page_end: Optional[int] = None
# === Window 专用 ===
window_text: Optional[str] = None
window_size: Optional[int] = None
# === Hierarchical 专用 ===
chunk_sizes: Optional[List[int]] = None
has_children: bool = False
# === 上下文增强元数据 ===
chunk_summary: Optional[str] = None
keywords: List[str] = field(default_factory=list)
hypothetical_questions: List[str] = field(default_factory=list)
元数据注入责任分工：
Strategy.execute() → strategy_type + 策略专有字段
ChunkPostProcessor → prev/next (所有策略) + heading_path (markdown)
MetadataEnrichment → doc_id, kb_id, file_name 等文档级字段
ContextEnhancement → chunk_summary, keywords 等增强字段
🚀 十一、完整使用示例
11.1 基础使用
from module_rag.rag_common.chunking.auto_strategy import AutoChunkStrategyRecommender
# 方式 1：传入 Markdown 文本
md_text = open("document.md", encoding="utf-8").read()
result = AutoChunkStrategyRecommender.recommend(md_text)
print(f"推荐策略: {result.strategy_type.value}")
print(f"推荐参数: {result.strategy_params}")
print(f"置信度: {result.confidence}")
print(f"原因: {result.reason}")
11.2 传入 Document 列表
from llama_index.core import Document
from module_rag.rag_common.chunking.auto_strategy import AutoChunkStrategyRecommender
documents = [
Document(text="第一章 概述\n..."),
Document(text="第二章 架构设计\n..."),
]
result = AutoChunkStrategyRecommender.recommend(documents)
print(f"推荐策略: {result.strategy_type.value}")
11.3 用户手动覆盖
from module_rag.rag_common.chunking.schemas import ChunkStrategyType
# 用户强制指定策略
result = AutoChunkStrategyRecommender.recommend(
md_text,
user_strategy=ChunkStrategyType.SEMANTIC
)
# 直接返回用户指定的策略，跳过自动分析
11.4 配合 ChunkingService 使用
from module_rag.rag_common.chunking.service import ChunkingService
from module_rag.rag_common.chunking.auto_strategy import AutoChunkStrategyRecommender
# 1. 自动推荐策略
result = AutoChunkStrategyRecommender.recommend(md_text)
# 2. 使用推荐的策略执行切片
response = ChunkingService.process_chunking(
documents=documents,
strategy_type=result.strategy_type,
params=result.strategy_params,
)
print(f"切片数: {response.total_chunks}")
print(f"耗时: {response.cost_time_ms}ms")
for chunk in response.chunks:
print(f" [{chunk.node_id}] {chunk.text[:50]}...")
📊 十二、效果验证
12.1 各类文档的推荐结果
| 文档类型 | 文档特征 | 推荐策略 | 置信度 |
|---|---|---|---|
| API 文档（多级标题） | H1:3, H2:15, H3:28, 密度=0.12 | markdown |
high |
| 财报（超长+结构化） | 120,000 字符, 表格占比 18% | hierarchical |
high |
| FAQ 问答文档 | FAQ 占比 25%, 短段落为主 | window |
high |
| 操作手册（无标题） | 步骤占比 15%, 无标题层级 | window |
medium |
| 会议记录（长文连贯） | 标题密度 0.02, 平均段落 180 字符 | semantic |
medium |
| 短词条说明 | 总字符 1,200 | sentence |
medium |
12.2 日志输出示例
[AUTO_STRATEGY] 候选 #1: markdown (置信度=high) | 文档具有 3 级标题结构（H1: 3, H2: 15, H3: 28），标题密度 0.12，使用 markdown 按标题层级切片
[AUTO_STRATEGY] 候选 #2: hierarchical (置信度=medium) | 表格密集（表格行占比 18.5%），使用 markdown 切片以保持结构化元素完整性
[AUTO_STRATEGY] 候选 #3: semantic (置信度=medium) | 文档语义连贯但无明显标题结构（标题密度 0.02，平均段落长度 180 字符），使用 semantic 按语义相似度切片
[AUTO_STRATEGY] LLM 判定结果: markdown | 文档具有清晰的多级标题结构，且表格占比较高，markdown 策略能同时保护标题层级和表格完整性
🎯 十三、设计亮点总结
13.1 规则引擎 + LLM 混合决策
- 规则引擎保证确定性：不会因为 LLM 幻觉推荐出不存在的策略
- LLM 提供灵活性：能在候选中根据语义做更精细的判断
- 失败回退保证鲁棒性：LLM 挂了也不影响使用
13.2 参数自适应
不仅推荐策略类型，还根据文档特征动态计算策略参数：
- 段落越长 → chunk_size 越大
- 文档越长 → hierarchical 层级越多
- FAQ 越密集 → window_size 越大
13.3 可扩展性
- 新增规则只需添加一个
_rule_xxx方法，注册到_score_strategies的规则列表中 - 新增策略只需继承
BaseChunkStrategy并调用ChunkStrategyFactory.register() - 阈值集中管理，便于后续基于实际数据调优
13.4 统一元数据协议
所有策略切出的 Node 遵循统一的 ChunkMetadataProtocol，确保下游（检索、展示、上下文增强）能统一处理，不需要针对不同策略做特殊适配。
📦 十四、项目结构
module_rag/rag_common/chunking/
├── __init__.py # 模块导出
├── schemas.py # 数据模型（枚举、VO、元数据协议）
├── strategy.py # 策略抽象基类 BaseChunkStrategy
├── factory.py # 注册式策略工厂
├── auto_strategy.py # 自动推荐引擎（本文核心）
├── service.py # 切片服务入口
├── strategies/ # 具体策略实现
│ ├── __init__.py # 触发所有策略自动注册
│ ├── sentence.py # SentenceStrategy
│ ├── semantic.py # SemanticStrategy（含超长文档保护）
│ ├── window.py # SentenceWindowStrategy
│ ├── hierarchical.py # HierarchicalStrategy
│ ├── markdown.py # MarkdownStrategy
│ └── json_parser.py # JsonStrategy
└── ...
📝 十五、总结
本文介绍了一种基于文档结构特征分析的切片策略自动推荐引擎，核心思路是：
- 特征提取：从 Markdown 文本中提取 6 大类结构特征（标题、段落、表格、列表、FAQ、步骤）
- 规则打分：7 条规则按优先级运行，输出 Top 3 候选策略
- LLM 决策：LLM 从 Top 3 中选择最优策略（失败回退到规则引擎）
- 参数自适应：根据文档特征动态计算策略内部参数
- 统一协议：所有策略遵循
ChunkMetadataProtocol，确保下游兼容
这种"规则 + LLM"的混合架构，既保证了决策的确定性，又具备一定的灵活性，能够适应各种类型的文档。
作者：RAG 工程实践
标签：
RAGLangChainLlamaIndex文档切片智能推荐Python参考：LlamaIndex 官方文档、Semantic Chunking 论文
为武汉地区的开发者提供学习、交流和合作的平台。社区聚集了众多技术爱好者和专业人士，涵盖了多个领域，包括人工智能、大数据、云计算、区块链等。社区定期举办技术分享、培训和活动，为开发者提供更多的学习和交流机会。
更多推荐
8
0- 0
已为社区贡献1条内容
所有评论(0)