RAG设计之语义感知策略

又到了 RAG 优化环节了🥹，这次遇到了大文档有效信息截断的问题，分析很多 case 发现扩展 token+段落metadata 融合，也不咋好使，最终摸索实现“结构递归切分 + should_merge 语义合并 + embedding 相似度 + 大模型灰区判断”的 chunk 策略。

1. 背景

为什么需要语义感知切分？

当前项目的长期知识库链路是：

1
2
3
4
5
6
7
8


POST /upload_files
  -> execute_insert_process()
  -> chunk(file_path)
  -> parser 输出 sections / tables
  -> naive_merge() / naive_merge_docx()
  -> tokenize_chunks() / tokenize_table()
  -> generate_embedding()
  -> Elasticsearch 写入 chunk

没啥特殊的，这是经典做法：各类 parser 先把 PDF、DOCX、Markdown、HTML、TXT、JSON、Excel 转成 sections，然后由merge 模块按 token 阈值合并，再进入 tokenize_chunks()。

不同格式在当前链路中的差异如下：

格式	当前处理方式	语义切分改造重点
PDF	通过 `layout_recognize` 选择 `DeepDOC`、`Plain Text` 或 `Auto` 策略；文本走 `tokenize_chunks()`，表格走 `tokenize_table()`。	保留标题层级、页码、坐标和跨页段落；复杂 PDF 不应丢失引用定位。
DOCX	`Docx()` 输出段落、表格和图片，文本走 `naive_merge_docx()`。	保留 Heading 层级、列表项和图片上下文，语义合并时不能丢失图片拼接能力。
Excel	`ExcelParser()` 或 `html4excel` 输出行/表格文本。	表格类 chunk 应优先保持行列语义，不强行和正文段落混合。
TXT / 代码文本	`TxtParser()` 按分隔符和 token 阈值产生 sections。	缺少显式标题时，需要通过段落、列表和相似度补足边界判断。
Markdown	`Markdown()` 会抽取表格和剩余正文。	标题层级天然清晰，适合作为结构递归切分的优先输入。
HTML	`HtmlParser()` 抽取正文段落。	需要恢复 `h1` 到 `h6`、列表、表格等 DOM 结构。
JSON	`JsonParser(chunk_token_num)` 做结构化切分。	应保留 key path，避免把字段名和字段值拆开。

固定 token 合并能保证 chunk 大小可控，但它不理解语义边界，容易出现四类截断：

标题和正文被拆开：检索命中正文时，缺少章节主题。
前导段和列表被拆开：检索命中列表项时，不知道列表在说明什么。
转折和条件说明被拆开：答案缺少“但是”“除非”“需要注意”的限制条件。
同一知识点的连续段落被拆开：定义、步骤、例外、结果分布在多个 chunk。

RAG 的检索单元是 chunk。如果一个标准答案需要的证据被切散，后续无论 rerank 和大模型多强，都可能拿不到完整上下文。

因此切分策略要优先保证“一个 chunk 内尽量包含一个完整可回答的语义单元”。

就像刚才这段话，是有递进关系的，你人类读的时候，不可能拆散吧，所以语义节点提上日程。

2. 设计原则

语义感知切分千万不能把整篇文档交给大模型自由切分，有三个大问题：

成本高：每次上传文档都要对长文本调用模型，费用可顶不住。
不稳定：同一文档多次切分结果可能不一致，很难做回归评估。
难定位：切分错误时，不容易知道是结构解析、模型判断还是 prompt 问题。

所以必须通过工程化的措施去处理：

parser 负责恢复结构：章节、小节、段落、列表、表格、页码和坐标。
规则负责处理确定性强的边界：列表前导、转折词、指代词。
embedding 负责判断相邻段落是否仍属于同一语义主题。
大模型只处理灰区边界：相似度不上不下、规则无法判断时，返回 MERGE 或 SPLIT。
soft limit、hard limit 和 overlap 负责控制 chunk 大小，避免语义合并无限增长。

这样切分的话既有语义判断，又保持可复现、可解释、可评估。

3. 主流程

  
flowchart TD
  A["原始文档"] --> B["parser 解析<br/>PDF / DOCX / MD / HTML / TXT"]
  B --> C["结构化 Block<br/>章节 / 小节 / 段落 / 列表 / 表格"]
  C --> D["递归切分<br/>章节 -> 小节 -> 段落 -> 句子"]
  D --> E["相邻 Block 边界判断"]
  E --> F["should_merge 规则<br/>列表前导 / 转折词 / 指代词"]
  F --> G["embedding 相似度<br/>cosine > 0.75"]
  G --> H["灰区判断<br/>0.55 <= sim <= 0.75 调 LLM"]
  H --> I["chunk 组装<br/>soft limit / hard limit / overlap"]
  I --> J["补充 metadata<br/>section_path / page / position / merge_reasons"]
  J --> K["tokenize_chunks()"]
  K --> L["generate_embedding()"]
  L --> M["Elasticsearch 入库"]

这个实现和之前的实现有点小区别，从直接切分加了中间层：

1

sections -> naive_merge() -> tokenize_chunks()

升级为：

1

sections -> build_blocks() -> recursive_split() -> semantic_merge() -> tokenize_chunks()

4. 数据结构设计

4.1 Block

Block 是 parser 输出和 chunk 之间的中间结构。它不直接入库，而是用于保留文档结构和边界判断依据。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


from dataclasses import dataclass, field

@dataclass
class Block:
    text: str
    block_type: str
    level: int
    section_path: list[str] = field(default_factory=list)
    page_num_int: list[int] = field(default_factory=list)
    position_int: list[tuple[int, int, int, int, int]] = field(default_factory=list)
    children: list["Block"] = field(default_factory=list)
    image: object | None = None

字段含义：

字段	说明
`text`	当前结构块文本。
`block_type`	`heading`、`paragraph`、`list_item`、`table`、`code` 等。
`level`	章节层级，一级标题为 1，二级标题为 2，普通段落可设为 99。
`section_path`	当前块所属标题路径，例如 `["第 3 章用户认证", "3.1 登录流程"]`。
`page_num_int`	PDF 页码，用于引用定位。
`position_int`	PDF 坐标，用于引用高亮。
`children`	结构树子节点。
`image`	DOCX 或 PDF 截图上下文，兼容现有 `naive_merge_docx()` 能力。

4.2 ChunkDraft

ChunkDraft 是语义合并后的待入库 chunk。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


@dataclass
class ChunkDraft:
    text: str
    blocks: list[Block]
    section_path: list[str]
    block_types: list[str]
    page_num_int: list[int]
    position_int: list[tuple[int, int, int, int, int]]
    semantic_merge_reasons: list[str]
    prev_chunk_id: str | None = None
    next_chunk_id: str | None = None

最终进入 ES 前，需要把这些 metadata 合并到现有 doc 字段中。当前 tokenize_chunks() 能从 PDF parser 回填 page_num_int、position_int 和 top_int。

新增的 section_path、block_types、semantic_merge_reasons 可作为 keyword/text 字段随 chunk 一起保存，或者先保存在 content_with_weight 的 header 中，等 mapping 稳定后再结构化入库。

这块倒是影响不大。

5. 结构递归切分

结构递归切分的目标是先找自然边界，再用 token 限制兜底。

优先级固定为：

1

章节 -> 小节 -> 段落 -> 句子

伪代码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


SOFT_LIMIT = 500
HARD_LIMIT = 800

def recursive_split(block: Block) -> list[Block]:
    if token_len(block.text) <= SOFT_LIMIT:
        return [block]

    if block.children:
        result = []
        for child in block.children:
            result.extend(recursive_split(child))
        return result

    if block.block_type == "paragraph":
        return split_long_paragraph(block)

    return split_by_sentence(block)

如果整章不超过 SOFT_LIMIT，整章保留。
如果章节太长，按小节递归。
如果小节太长，按段落递归。
如果段落太长，才按句号、分号、换行等句子边界切。
如果单句仍超过 HARD_LIMIT，按 token 硬切，并在下一 chunk 加 overlap。

这样可以避免一开始就按固定长度切断结构。

6. should_merge 规则

should_merge(prev, curr) 决定相邻两个 block 是否应放入同一个 chunk。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


def should_merge(prev: Block, curr: Block, sim: float | None = None) -> tuple[bool, list[str]]:
    reasons = []

    if is_list_continuation(prev, curr):
        reasons.append("list_leader")

    if starts_with_transition(curr.text):
        reasons.append("transition")

    if has_coreference_without_subject(curr.text):
        reasons.append("coreference")

    if sim is not None and sim > 0.75:
        reasons.append("semantic_similarity_gt_075")

    return bool(reasons), reasons

6.1 列表前导合并

需要合并的典型文本：

1
2
3
4


停车场规划验收流程包括以下步骤：
1. xxx
2. xxx
3. xxx

如果切分后只命中列表项，回答会缺少“这是停车场规划验收流程”的上下文。

判断规则：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


LEADER_SUFFIXES = [
    "如下", "如下：", "包括", "包括：", "步骤", "步骤：",
    "条件", "条件：", "规则", "规则：", "分别为", "分别为："
    ...
]

LIST_PATTERNS = [
    r"^\s*[-*]\s+",
    r"^\s*\d+[\.、]\s+",
    r"^\s*[（(][一二三四五六七八九十\d]+[）)]",
    r"^\s*[一二三四五六七八九十]+[、.]\s*"
]

def is_list_continuation(prev: Block, curr: Block) -> bool:
    prev_text = prev.text.strip()
    curr_text = curr.text.strip()
    return (
        curr.block_type == "list_item"
        or any(re.match(pattern, curr_text) for pattern in LIST_PATTERNS)
        or any(prev_text.endswith(suffix) for suffix in LEADER_SUFFIXES)
    )

6.2 转折和承接词合并

需要合并的典型文本：

1
2
3


绿波交通需要前一个交通信号灯信号。
但是，如果前一个交通灯故障，则应视情况而定。
因此，需要先检查交通灯状态。

第二、三句依赖上一段，不适合独立成为 chunk。

1
2
3
4
5
6
7
8


TRANSITIONS = [
    "但是", "然而", "不过", "因此", "所以", "同时",
    "另外", "此外", "需要注意", "综上", "也就是说"
]

def starts_with_transition(text: str) -> bool:
    stripped = text.strip()
    return any(stripped.startswith(word) for word in TRANSITIONS)

6.3 指代词合并

需要合并的典型文本：

1
2


停车场收费策略xxx
该收费策略会根据 xxx。

该xxx 依赖上一段定义。如果单独检索到第二段，会丢失指代对象。

1
2
3
4
5
6
7


COREFERENCE_WORDS = ["该", "其", "上述", "前者", "后者", "这些", "此时"]

def has_coreference_without_subject(text: str) -> bool:
    stripped = text.strip()
    has_coref = any(word in stripped[:12] for word in COREFERENCE_WORDS)
    has_clear_subject = bool(re.match(r"^[\u4e00-\u9fa5A-Za-z0-9_]{2,12}(是|为|用于|表示)", stripped))
    return has_coref and not has_clear_subject

6.4 embedding 相似度合并

对于没有明显规则信号的相邻段落，用 embedding 判断是否仍在讲同一主题。

当前工程已有 app/service/core/rag/nlp/model.py::generate_embedding()，可以复用它批量生成向量，再用 cosine similarity 判断：

1
2
3
4
5
6
7
8
9


import numpy as np

def cosine_similarity(a: list[float], b: list[float]) -> float:
    va = np.array(a, dtype=float)
    vb = np.array(b, dtype=float)
    denom = np.linalg.norm(va) * np.linalg.norm(vb)
    if denom == 0:
        return 0.0
    return float(np.dot(va, vb) / denom)

阈值固定为：

1
2
3


sim > 0.75：认为语义连续，合并。
0.55 <= sim <= 0.75：进入灰区，交给大模型判断。
sim < 0.55：默认切分。

6.5 大模型灰区判断

大模型只处理灰区，不做全量切分。触发条件：

1
2
3


0.55 <= similarity <= 0.75
且列表、转折、指代规则都没有明确命中
且当前 chunk 接近 SOFT_LIMIT

Prompt 固定为：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


你是 RAG 文档切分边界判断器。
请判断 A 段和 B 段是否应该放在同一个检索 chunk 中。

判断标准：
1. 如果 B 依赖 A 的定义、条件、上下文、列表前导或限制说明，返回 MERGE。
2. 如果 B 是新主题、新章节、新流程或可独立回答，返回 SPLIT。
3. 只返回 JSON，不要输出其他内容。

A:
{prev_text}

B:
{curr_text}

返回格式：
{"decision":"MERGE 或 SPLIT","reason":"一句话原因"}

结果需要缓存，缓存 key 可以由文档名、相邻 block 文本 hash、模型名组成：

1

cache_key = f"semantic-boundary:{doc_name}:{hash_text(prev.text)}:{hash_text(curr.text)}:{model_name}"

缓存可以优先放 Redis，避免同一文档重复上传或评估回放时重复调用模型。

7. chunk 组装策略

chunk 大小策略固定为：

1
2
3


SOFT_LIMIT = 500 tokens
HARD_LIMIT = 800 tokens
OVERLAP = 130 tokens

组装逻辑：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33


def build_chunks(blocks: list[Block]) -> list[ChunkDraft]:
    chunks = []
    current = new_chunk()

    embeddings = generate_embedding([block.text for block in blocks])

    for i, block in enumerate(blocks):
        if current.is_empty():
            current.add(block)
            continue

        prev = current.blocks[-1]
        sim = cosine_similarity(embeddings[i - 1], embeddings[i])
        merge, reasons = should_merge(prev, block, sim)
        merged_tokens = current.token_len + token_len(block.text)

        if merged_tokens <= SOFT_LIMIT:
            current.add(block, reasons)
            continue

        if merge and merged_tokens <= HARD_LIMIT:
            current.add(block, reasons)
            continue

        chunks.append(current)
        current = new_chunk(with_overlap=tail_tokens(chunks[-1], OVERLAP))
        current.add(block)

    if not current.is_empty():
        chunks.append(current)

    link_neighbors(chunks)
    return chunks

关键点：

小于 SOFT_LIMIT 时正常合并，减少碎片。
超过 SOFT_LIMIT 但 should_merge=True 时，允许合并到 HARD_LIMIT。
超过 HARD_LIMIT 时必须切分，但新 chunk 带上上一 chunk 的 OVERLAP。
每个 chunk 前面补 section_path，保证标题上下文不丢。
semantic_merge_reasons 记录合并原因，方便评估和排查。

最终 chunk 格式：

1
2
3
4
5
6
7
8


标题路径：第 3 章 用户认证 > 3.1 登录流程

登录需要校验账号密码。
包括以下步骤：
1. 校验用户名
2. 校验密码
3. 生成 token
但是，如果用户被冻结，则拒绝登录。

新增语义解析层：

1
2
3
4


app/service/core/rag/chunking/
  semantic_chunker.py
  semantic_rules.py
  semantic_types.py

为了更好的测试，也要兼容策略：

1
2
3
4


if parser_config.get("chunk_strategy") == "semantic":
    chunks = semantic_chunk(sections, parser_config)
else:
    chunks = naive_merge(sections, chunk_token_num, delimiter)

这样可以保留原策略作为 baseline，方便在评估系统中做 baseline/current 对比。

9. 评估指标

语义切分不能只看 chunk 数量，要看证据是否完整、检索是否改善、答案是否更准。

9.1 基础 chunk 质量

指标	作用
chunk 空值率	检查 parser 或合并策略是否产生空 chunk。
平均 token	判断 chunk 是否过碎或过长。
最大 token	检查是否突破 `HARD_LIMIT = 800`。
标题继承率	检查 chunk 是否带上 `section_path`。
表格保留率	检查表格是否独立保留，而不是被正文切散。
PDF 坐标保留率	检查 `page_num_int`、`position_int` 是否从 parser 透传到最终 chunk。

9.2 截断率

定义：

1
2
3
4


截断 case = 标准答案所需 evidence 不能被单个 chunk 覆盖，
也不能被 overlap 邻接 chunk 恢复。

截断率 = 截断 case 数 / evidence case 总数

所谓的阶段率不能是主观体验，必须是基于金标 evidence case 的覆盖结果。

9.3 证据完整率

定义：

1

证据完整率 = 完整覆盖标准 evidence 的 case 数 / evidence case 总数

如果问题答案需要“定义 + 条件 + 例外”，只有 chunk 同时覆盖这些要点，才算完整。

9.4 邻接覆盖率

定义：

1

邻接覆盖率 = 通过 prev/next chunk 或 overlap 能恢复完整 evidence 的 case 数 / evidence case 总数

这个指标用于评估 OVERLAP = xxx 是否足够。如果邻接覆盖率高但单 chunk 完整率低，说明 chunk 可能切得偏碎。

9.5 检索排序指标

复用现有 RAG 评估系统：

指标	作用
`Recall@K`	正确 chunk 是否进入 TopK。
`MRR`	正确 chunk 排名是否靠前。
`nDCG@K`	多证据场景下排序质量。
空召回率	新切分是否导致召回失败。

9.6 答案和引用指标

指标	作用
答案要点覆盖率	模型回答是否覆盖标准答案要点。
幻觉率	回答是否包含 evidence 不支持的内容。
引用准确率	引用 chunk 是否真实支撑答案。
引用召回率	标准 evidence 对应引用是否被展示。

10. 评估与反馈闭环

  
flowchart TD
  A["金标集<br/>question / answer / evidence"] --> B["baseline<br/>naive_merge"]
  A --> C["current<br/>semantic_chunk"]
  B --> D["检索与答案评估"]
  C --> D
  D --> E["指标报告<br/>截断率 / 证据完整率 / Recall / MRR / 引用"]
  E --> F{"失败类型归因"}
  F --> G["parser 问题<br/>标题/列表/表格未识别"]
  F --> H["chunk 问题<br/>should_merge 规则缺失"]
  F --> I["embedding 问题<br/>阈值或向量质量不稳"]
  F --> J["LLM 判断问题<br/>prompt 或缓存错误"]
  F --> K["overlap 问题<br/>硬切后上下文不足"]
  G --> L["修 parser 或 Block 构造"]
  H --> M["补规则和测试 case"]
  I --> N["调 0.75 / 0.55 阈值"]
  J --> O["改 prompt / 加置信度"]
  K --> P["调 OVERLAP 或 HARD_LIMIT"]
  L --> Q["失败 case 回流金标集"]
  M --> Q
  N --> Q
  O --> Q
  P --> Q
  Q --> A

反馈闭环的重点是把失败 case 分层归因：

parser 问题：章节、小节、表格、列表没有被识别成结构。
chunk 问题：规则没有把相关段落合并。
embedding 问题：相似段落分数偏低，或无关段落分数偏高。
LLM 判断问题：灰区判断错，或者输出不符合 JSON。
overlap 问题：硬切后下一 chunk 缺少前文。

每次修复后，把失败 case 加入金标集，重新跑 baseline/current。只有截断率、证据完整率、Recall@K、引用准确率同时不退化，才认为策略有效。

文章目录

1. 背景

2. 设计原则

3. 主流程

4. 数据结构设计

4.1 Block

4.2 ChunkDraft

5. 结构递归切分

6. should_merge 规则

6.1 列表前导合并

6.2 转折和承接词合并

6.3 指代词合并

6.4 embedding 相似度合并

6.5 大模型灰区判断

7. chunk 组装策略

9. 评估指标

9.1 基础 chunk 质量

9.2 截断率

9.3 证据完整率

9.4 邻接覆盖率

9.5 检索排序指标

9.6 答案和引用指标

10. 评估与反馈闭环