W8-D1 · RAG 地基代码详解：加载与分块

1
2
3

① 加载 ──► ② 分块 ──► ③ 嵌入 ──► ④ 入库 ──► ⑤ 检索 ──► ⑥ 生成
 Load      Split      Embed      Store      Retrieve    Generate
 └─今天─┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

day1_load_split.py
├── 配置区        CHUNK_SIZE / CHUNK_OVERLAP / SEPARATORS / HEADERS
├── 加载段        load_text_file / load_pdf_file / check_text_layer
├── 分块段        make_splitter / split_documents / split_one_text
│                split_by_headers / split_markdown_two_stage / drop_tiny_chunks
├── 度量段        measure_actual_overlap / expansion_ratio / formula_expansion
│                cut_at_punctuation
├── 报告段        report_split
├── 样例文档       SAMPLE_MD / SAMPLE_TXT / ensure_sample_docs
├── 三种模式      run_demo / run_real / run_compare
└── 入口          if __name__ == "__main__"

1
2
3
4

跑法：
  python w8/day1_load_split.py            # demo：自动造样例，0 准备
  python w8/day1_load_split.py real        # real：读 w8/docs/ 里的真实文件
  python w8/day1_load_split.py compare     # compare：chunk_size 选型对照

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from pathlib import Path

HERE = Path(__file__).resolve().parent
DOCS_DIR = HERE / "docs"

CHUNK_SIZE = 300                              # 上限，不是固定值
CHUNK_OVERLAP = 50                            # 约 size 的 15%
OVERLAP_RATIO = 0.15
SEPARATORS = ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
HEADERS = [("#", "h1"), ("##", "h2"), ("###", "h3")]
ENDS = ("。", "！", "？", "；", "，")

1
2
3
4
5

text = "A" * 800                                     # 800 字符，零分隔符
no_fb = ["\n\n", "\n", "。", "，"]                    # 故意不留 ""
# → 1 块，长度 800   ❌ 超预算 8 倍
# 加上 [""] 兜底后：
# → 8 块，最长 100   ✅

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

from langchain_core.documents import Document


def load_text_file(path: Path) -> Document:
    """读文本/Markdown。显式 encoding，避免中文乱码"""
    text = path.read_text(encoding="utf-8")
    return Document(page_content=text, metadata={"source": path.name})


def load_pdf_file(path: Path) -> list[Document]:
    """PDF：一页 = 一个 Document，metadata 里带 page"""
    from langchain_community.document_loaders import PyPDFLoader   # pip install pypdf
    return PyPDFLoader(str(path)).load()


def check_text_layer(pages: list[Document], min_chars: int = 20) -> list[int]:
    """扫描件探测：返回疑似没有文字层的页码（空列表 = 正常）"""
    return [i for i, d in enumerate(pages)
            if len(d.page_content.strip()) < min_chars]

1
2
3
4

Document(
    page_content="这一段正文……",                 # 内容
    metadata={"source": "手册.md", "page": 3}     # 从哪来 ★
)

1
2
3
4

pages = load_pdf_file(Path("manual.pdf"))
bad = check_text_layer(pages)
if bad:
    print(f"⚠️ 第 {bad} 页疑似扫描件，需要 OCR")

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

from langchain_text_splitters import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter


def make_splitter(size=CHUNK_SIZE, overlap=CHUNK_OVERLAP, separators=None):
    return RecursiveCharacterTextSplitter(
        chunk_size=size,
        chunk_overlap=overlap,
        separators=separators if separators is not None else SEPARATORS,
        keep_separator=True,
        length_function=len,
        add_start_index=True,
    )

1
2

块1 = "...到这里就结束了"        ← 结尾是"了"，不是标点
块2 = "。下一句从这里开始..."     ← 开头是句号

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

def split_documents(docs, *, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP, separators=None):
    """切多个 Document —— metadata 会自动复制到每一块"""
    return make_splitter(size, overlap, separators).split_documents(docs)


def split_one_text(text: str, *, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP,
                   metadata=None, separators=None) -> list[Document]:
    """切一段纯文本 —— 要手动传 metadata"""
    return make_splitter(size, overlap, separators).create_documents(
        [text], metadatas=[metadata or {}])


def split_by_headers(md_text: str) -> list[Document]:
    """Markdown 一级切：按标题切章节，标题信息进 metadata"""
    sp = MarkdownHeaderTextSplitter(headers_to_split_on=HEADERS, strip_headers=False)
    return sp.split_text(md_text)


def split_markdown_two_stage(md_text: str, *, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
    """两级切：先按标题，再按长度（heading metadata 会继承到每块）"""
    return split_documents(split_by_headers(md_text), size=size, overlap=overlap)

1
2
3

一级：按标题切（# / ## / ###）  → 保证语义边界 + 章节归属
二级：超长章节再按长度递归切    → 保证不超预算
要求：二级切完，一级的 h1/h2/h3 必须还在

1
2
3

def drop_tiny_chunks(docs: list[Document], min_len: int = 10) -> list[Document]:
    """过滤碎块：切分可能产出 1 个字符的块（例如一个孤立句号），进库就是噪声"""
    return [d for d in docs if len(d.page_content.strip()) >= min_len]

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

def measure_actual_overlap(chunks: list[str]) -> list[int]:
    """实测相邻块的真实重叠字符数"""
    out = []
    for a, b in zip(chunks, chunks[1:]):
        n = 0
        for k in range(min(len(a), len(b)), 0, -1):
            if a[-k:] == b[:k]:
                n = k
                break
        out.append(n)
    return out


def expansion_ratio(original_len: int, chunks: list[str]) -> float:
    """冗余膨胀率 = 所有块字符数 / 原文字符数 - 1"""
    return 0.0 if not original_len else sum(len(c) for c in chunks) / original_len - 1


def formula_expansion(size: int, overlap: int) -> float:
    """理论膨胀率 ≈ overlap / (size - overlap)"""
    return float("inf") if size <= overlap else overlap / (size - overlap)

1

50 / (300 - 50) ≈ 20%

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

def cut_at_punctuation(chunks: list[Document], original: str,
                       ends: tuple[str, ...] = ENDS) -> float | None:
    """切分点是否落在标点上——用原文坐标判断，与 keep_separator 取值无关。

    keep_separator=True  → 标点在下一块开头（original[o]）
    keep_separator="end" → 标点在上一块末尾（original[o-1]）
    """
    offs = [c.metadata.get("start_index") for c in chunks]
    if len(offs) < 2 or any(o is None for o in offs):
        return None                      # 坐标缺失 → 交给调用方报错
    hit = 0
    for o in offs[1:]:
        if o <= 0:
            continue
        if original[o - 1:o] in ends or original[o:o + 1] in ends:
            hit += 1
    return hit / (len(offs) - 1)

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

def report_split(label: str, original: str, docs: list[Document]) -> None:
    lens = [len(d.page_content) for d in docs]
    chunks = [d.page_content for d in docs]
    print(f"\n=== {label} ===")
    if not docs:
        print("  0 块（原文为空？）")
        return
    print(f"  原文 {len(original)} 字符 -> {len(docs)} 块")
    print(f"  块长：最短 {min(lens)}  最长 {max(lens)}  平均 {sum(lens) // len(lens)}")
    print(f"  超预算的块：{[l for l in lens if l > CHUNK_SIZE] or '无'}")
    print(f"  切分点落在标点上：{cut_at_punctuation(docs, original)}")
    print(f"  膨胀率 实测 {expansion_ratio(len(original), chunks):.1%}"
          f"   公式预测 {formula_expansion(CHUNK_SIZE, CHUNK_OVERLAP):.1%}")
    print("  --- 前 2 块预览 ---")
    for i, d in enumerate(docs[:2]):
        print(f"  #{i} ({len(d.page_content)}字) "
              f"{d.page_content[:60]}".replace("\n", " ") + "...")
    print(f"  metadata[0] = {docs[0].metadata}")

1
2

all_docs: list[Document] = []
for p in files: ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "demo"
    if mode == "demo":
        ensure_sample_docs()
        run_demo()
    elif mode == "real":
        run_real()
    elif mode == "compare":
        run_compare()
    else:
        print(f"未知模式：{mode}（用法：python w8/day1_load_split.py [demo|real|compare]）")