RAG 文档分块策略实战：从固定长度到混合分块

可以通过该参数开启缓存。

当 enable_cache=true 时，系统会读取本地缓存。
如果缓存不存在，会回退到远程服务。

from langchain_text_splitters import CharacterTextSplitter

splitter = CharacterTextSplitter(
    separator="",        # 纯字符切分
    chunk_size=600,
    chunk_overlap=90,
)

chunks = splitter.split_text(text)

import re
from typing import List

def split_sentences_zh(text: str) -> List[str]:
    pattern = re.compile(r'([^。！？；]*[。！？；]+|[^。！？；]+$)')
    return [
        m.group(0).strip()
        for m in pattern.finditer(text)
        if m.group(0).strip()
    ]

def sentence_chunk(text: str, chunk_size: int = 600, overlap: int = 80) -> List[str]:
    sentences = split_sentences_zh(text)
    chunks = []
    buf = ""

    for sentence in sentences:
        if len(buf) + len(sentence) <= chunk_size:
            buf += sentence
        else:
            if buf:
                chunks.append(buf)

            tail = buf[-overlap:] if overlap > 0 and len(buf) > overlap else ""
            buf = tail + sentence

    if buf:
        chunks.append(buf)

    return chunks

chunks = sentence_chunk(text, chunk_size=600, overlap=90)

MIN_CHUNK_CHARS = 300
TARGET_CHUNK_CHARS = 700

from langchain_text_splitters import RecursiveCharacterTextSplitter

separators = [
    r"\n#{1,6}\s",             # Markdown 标题
    r"\n\d+(?:\.\d+)*\s",      # 编号标题，例如 1. / 2.3.
    "\n\n",                    # 段落
    "\n",                      # 行
    " ",                       # 空格
    "",                        # 兜底字符级
]

splitter = RecursiveCharacterTextSplitter(
    separators=separators,
    chunk_size=700,
    chunk_overlap=100,
    is_separator_regex=True,
)

chunks = splitter.split_text(text)

chunk_size = 400  # 到 800 之间试验
chunk_overlap = 80  # 约 10%～20%

RAG 指南 > 检索优化 > 重排策略

import re
from typing import Dict, List

HEADING_PATTERN = re.compile(r'^(#{1,6})\s+(.*)$')
FENCE_PATTERN = re.compile(r'^```')

def split_markdown_structure(
    text: str,
    chunk_size: int = 900,
    min_chunk: int = 250,
) -> List[Dict]:
    lines = text.splitlines()
    sections = []

    in_code = False
    path_stack = []
    current = {
        "level": 0,
        "title": "",
        "content": [],
        "path": [],
    }

    for line in lines:
        if FENCE_PATTERN.match(line):
            in_code = not in_code

        match = HEADING_PATTERN.match(line) if not in_code else None

        if match:
            if current["content"]:
                sections.append(current)

            level = len(match.group(1))
            title = match.group(2).strip()

            while path_stack and path_stack[-1][0] >= level:
                path_stack.pop()

            path_stack.append((level, title))
            breadcrumbs = [title for _, title in path_stack]

            current = {
                "level": level,
                "title": title,
                "content": [],
                "path": breadcrumbs,
            }
        else:
            current["content"].append(line)

    if current["content"]:
        sections.append(current)

    chunks = []

    def emit_chunk(text_block: str, section: Dict):
        clean_text = text_block.strip()
        if not clean_text:
            return

        breadcrumbs = section["path"]
        prefix = " > ".join(breadcrumbs[-3:])
        chunk_text = f"[{prefix}]\n{clean_text}" if prefix else clean_text

        chunks.append({
            "text": chunk_text,
            "meta": {
                "section_title": breadcrumbs[-1] if breadcrumbs else "",
                "breadcrumbs": breadcrumbs,
                "section_level": section["level"],
            }
        })

    for section in sections:
        raw = "\n".join(section["content"]).strip()
        if not raw:
            continue

        if len(raw) <= chunk_size:
            emit_chunk(raw, section)
            continue

        paragraphs = [p.strip() for p in raw.split("\n\n") if p.strip()]
        buf = ""

        for paragraph in paragraphs:
            if len(buf) + len(paragraph) + 2 <= chunk_size:
                buf = f"{buf}\n\n{paragraph}" if buf else paragraph
            else:
                if buf:
                    emit_chunk(buf, section)
                buf = paragraph

        if buf:
            emit_chunk(buf, section)

    merged = []
    for chunk in chunks:
        if not merged:
            merged.append(chunk)
            continue

        same_section = (
            merged[-1]["meta"]["breadcrumbs"] == chunk["meta"]["breadcrumbs"]
        )
        too_short = len(chunk["text"]) < min_chunk

        if same_section and too_short:
            merged[-1]["text"] += "\n\n" + chunk["text"]
        else:
            merged.append(chunk)

    return merged

{
  "doc_id": "manual-001",
  "breadcrumbs": ["用户指南", "安装", "Docker 部署"],
  "section_title": "Docker 部署",
  "block_type": "paragraph",
  "start_offset": 1024,
  "end_offset": 1840,
  "source": "docs/install.md"
}

from typing import Dict, List

def chunk_dialogue(
    turns: List[Dict],
    max_turns: int = 10,
    max_chars: int = 900,
    overlap_turns: int = 2,
) -> List[Dict]:
    """
    turns 示例：
    [
        {"speaker": "User", "text": "订单为什么还没发货？", "ts_start": 10},
        {"speaker": "Agent", "text": "我帮您查询一下物流状态。", "ts_start": 15}
    ]
    """
    chunks = []
    i = 0

    while i < len(turns):
        j = i
        char_count = 0
        speakers = set()

        while j < len(turns):
            turn = turns[j]
            text_len = len(turn["text"])

            if (j - i + 1) > max_turns or (char_count + text_len) > max_chars:
                break

            char_count += text_len
            speakers.add(turn["speaker"])
            j += 1

        window = turns[i:j] if j > i else [turns[i]]
        text = "\n".join(
            f'{turn["speaker"]}: {turn["text"]}'
            for turn in window
        )

        chunks.append({
            "text": text,
            "meta": {
                "speakers": list(speakers),
                "turns_range": (i, j - 1),
                "ts_start": window[0].get("ts_start"),
                "ts_end": window[-1].get("ts_end"),
            }
        })

        if j >= len(turns):
            break

        next_start = i + len(window) - overlap_turns
        i = max(next_start, i + 1)

    return chunks

import re
from typing import Dict, List

import numpy as np
from sentence_transformers import SentenceTransformer

def split_sentences_zh(text: str) -> List[str]:
    pattern = re.compile(r'([^。！？；]*[。！？；]+|[^。！？；]+$)')
    return [
        m.group(0).strip()
        for m in pattern.finditer(text)
        if m.group(0).strip()
    ]

def semantic_chunk(
    text: str,
    model_name: str = "BAAI/bge-m3",
    window_size: int = 2,
    min_chars: int = 350,
    max_chars: int = 1100,
    lambda_std: float = 0.8,
) -> List[Dict]:
    sentences = split_sentences_zh(text)
    if not sentences:
        return []

    model = SentenceTransformer(model_name)
    embeddings = model.encode(
        sentences,
        normalize_embeddings=True,
        batch_size=64,
        show_progress_bar=False,
    )
    embeddings = np.asarray(embeddings)

    novelties = []
    for i in range(len(sentences)):
        if i == 0:
            novelties.append(0.0)
            continue

        start = max(0, i - window_size)
        ref = embeddings[start:i].mean(axis=0)
        ref = ref / (np.linalg.norm(ref) + 1e-8)

        novelty = 1.0 - float(np.dot(embeddings[i], ref))
        novelties.append(novelty)

    novelties = np.asarray(novelties)
    threshold = float(novelties.mean() + lambda_std * novelties.std())

    chunks = []
    buf = ""
    start_idx = 0

    def flush(end_idx: int):
        nonlocal buf, start_idx

        if buf.strip():
            chunks.append({
                "text": buf.strip(),
                "meta": {
                    "start_sent": start_idx,
                    "end_sent": end_idx - 1,
                }
            })

        buf = ""
        start_idx = end_idx

    for i, sentence in enumerate(sentences):
        over_max = len(buf) + len(sentence) > max_chars
        can_split = len(buf) >= min_chars

        if over_max and can_split:
            flush(i)

        buf += sentence

        semantic_shift = novelties[i] > threshold
        if can_split and semantic_shift:
            flush(i + 1)

    if buf:
        flush(len(sentences))

    return chunks

import re
from typing import Dict, List

import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans

def split_sentences_zh(text: str) -> List[str]:
    pattern = re.compile(r'([^。！？；]*[。！？；]+|[^。！？；]+$)')
    return [
        m.group(0).strip()
        for m in pattern.finditer(text)
        if m.group(0).strip()
    ]

def topic_chunk(
    text: str,
    k_topics: int = 5,
    min_chars: int = 500,
    max_chars: int = 1400,
    smooth_window: int = 2,
    model_name: str = "BAAI/bge-m3",
) -> List[Dict]:
    sentences = split_sentences_zh(text)
    if not sentences:
        return []

    model = SentenceTransformer(model_name)
    embeddings = model.encode(
        sentences,
        normalize_embeddings=True,
        batch_size=64,
        show_progress_bar=False,
    )
    embeddings = np.asarray(embeddings)

    n_clusters = min(k_topics, len(sentences))
    kmeans = KMeans(n_clusters=n_clusters, n_init="auto", random_state=42)
    labels = kmeans.fit_predict(embeddings)

    smoothed = labels.copy()
    for i in range(len(labels)):
        start = max(0, i - smooth_window)
        end = min(len(labels), i + smooth_window + 1)

        values, counts = np.unique(labels[start:end], return_counts=True)
        smoothed[i] = int(values[np.argmax(counts)])

    chunks = []
    buf = ""
    start_idx = 0
    current_label = smoothed[0]

    def flush(end_idx: int):
        nonlocal buf, start_idx

        if buf.strip():
            chunks.append({
                "text": buf.strip(),
                "meta": {
                    "start_sent": start_idx,
                    "end_sent": end_idx - 1,
                    "topic": int(current_label),
                }
            })

        buf = ""
        start_idx = end_idx

    for i, sentence in enumerate(sentences):
        switched = smoothed[i] != current_label
        under_min = len(buf) < min_chars
        over_max = len(buf) + len(sentence) > max_chars

        if switched and not under_min:
            flush(i)
            current_label = smoothed[i]

        if over_max and not under_min:
            flush(i)

        buf += sentence

    if buf:
        flush(len(sentences))

    return chunks

# 离线阶段：
# 1. 建小块索引：句子、短段、子句
# 2. 存父块内容：段落、小节、章节
# 3. 保存 child_id -> parent_id 映射

small_hits = small_index.search(embed(query), top_k=30)

groups = group_by_parent_id(small_hits)

scored_parents = score_groups(
    groups,
    strategy="max_mean_coverage",
)

candidate_parent_ids = top_m(scored_parents, m=3)

rerank_inputs = [
    (query, parent_store[parent_id]["text"])
    for parent_id in candidate_parent_ids
]

reranked_parent_ids = cross_encoder_rerank(rerank_inputs)

contexts = []
for parent_id in reranked_parent_ids:
    hit_children = groups[parent_id]
    context = build_local_window(
        parent_text=parent_store[parent_id]["text"],
        hit_children=hit_children,
        window_sents=1,
    )
    contexts.append(context)

final_context = pack_under_budget(contexts, token_budget=3000)

{
  "parent": {
    "parent_id": "doc-001#section-03",
    "text": "完整小节文本",
    "breadcrumbs": ["部署指南", "缓存配置"],
    "start_offset": 1200,
    "end_offset": 2600
  },
  "child": {
    "child_id": "doc-001#section-03#sent-05",
    "parent_id": "doc-001#section-03",
    "text": "当 enable_cache=true 时，系统会优先读取本地缓存。",
    "start": 410,
    "end": 460
  }
}

score_parent =
    α * max(child_scores)
  + (1 - α) * mean(child_scores)
  + β * coverage

density = sum(exp(score_i)) / len(parent_text)

系统指令：
你是一个 RAG 分块器。目标是创建高内聚、可追溯的文本块。

规则：
1. 不得在代码块、表格、公式中间切分。
2. 每块长度控制在 400～1000 字。
3. 保留标题路径。
4. 尽量让“定义 + 解释 + 示例”留在同一块。
5. 只输出 JSON。

输出格式：
{
  "segments": [
    {
      "start": 0,
      "end": 812,
      "title_path": ["指南", "安装"],
      "reason": "完整安装步骤和注意事项"
    }
  ]
}

from typing import Callable, Dict, List

def hybrid_chunk(
    doc_text: str,
    parse_structure: Callable,
    recursive_splitter: Callable,
    semantic_splitter: Callable,
    dialogue_splitter: Callable,
    max_coarse_len: int = 1100,
    min_chunk_len: int = 320,
) -> List[Dict]:
    """
    parse_structure 返回示例：
    [
        {
            "type": "text|code|table|formula|dialogue",
            "text": "...",
            "breadcrumbs": ["指南", "安装"],
            "anchor": "install"
        }
    ]
    """
    blocks = parse_structure(doc_text)
    chunks = []

    def emit(text: str, meta: Dict):
        clean_text = text.strip()
        if not clean_text:
            return

        breadcrumbs = meta.get("breadcrumbs", [])
        prefix = " > ".join(breadcrumbs[-3:])
        final_text = f"[{prefix}]\n{clean_text}" if prefix else clean_text

        chunks.append({
            "text": final_text,
            "meta": meta,
        })

    for block in blocks:
        block_type = block.get("type", "text")
        text = block.get("text", "")

        if block_type in {"code", "table", "formula"}:
            emit(text, {**block, "splitter": "atomic"})
            continue

        if block_type == "dialogue":
            for chunk in dialogue_splitter(block.get("turns", [])):
                emit(chunk["text"], {**block, "splitter": "dialogue"})
            continue

        if len(text) <= max_coarse_len:
            sub_chunks = recursive_splitter(text)

            buf = ""
            for sub_chunk in sub_chunks:
                sub_text = sub_chunk["text"]

                if len(buf) + len(sub_text) < min_chunk_len:
                    buf += sub_text
                else:
                    emit(buf or sub_text, {**block, "splitter": "recursive"})
                    buf = ""

            if buf:
                emit(buf, {**block, "splitter": "recursive"})
        else:
            for chunk in semantic_splitter(text):
                emit(chunk["text"], {**block, "splitter": "semantic"})

    return chunks

[
  {
    "question": "如何开启本地缓存？",
    "gold_doc_id": "cache-guide",
    "gold_answer_spans": [
      "当 enable_cache=true 时，系统会优先读取本地缓存。"
    ]
  }
]

lengths = [len(chunk["text"]) for chunk in chunks]

print("chunk count:", len(lengths))
print("min:", min(lengths))
print("max:", max(lengths))
print("avg:", sum(lengths) / len(lengths))

import matplotlib.pyplot as plt

plt.hist(lengths, bins=30)
plt.xlabel("chunk length")
plt.ylabel("count")
plt.title("Chunk Length Distribution")
plt.show()