GraphRAG Chunking 模块全解析：Sentence / Token 两种切分器与工厂化配置实践

本文档深度讲解 GraphRAG 项目中 graphrag-chunking 子包的设计与使用，内容包括基于 NLTK 的句子级切分、基于 tokenizer 的定长切分、ChunkingConfig 配置模型以及工厂函数 create_chunker 的完整用法。读者将掌握该模块的两种内置分块策略、底层切分算法与配置默认值，并能够在自有文档处理链路中复用这套“策略可配置、实现可替换”的文本分块方案。

模块定位与整体结构

graphrag-chunking 是 GraphRAG 仓库中用于「把长文档切成语义可控的短片段」的独立 Python 包，当前版本为 3.1.1，声明依赖 graphrag-common==3.1.1 与 pydantic~=2.10，支持 Python 3.11～3.13（见 pyproject.toml）。在基于图结构的 RAG 流程中，文本需要先被拆分成可索引、可嵌入、可检索的最小单元，而本模块正是承载这一职责的统一入口。

该包提供三样核心资产（见 README）：

1. 一组文本切分器（text chunkers）：SentenceChunker 与 TokenChunker；
2. 一个核心配置模型：ChunkingConfig；
3. 一个负责获取切分器实例的工厂：create_chunker。

从源码结构看，整个包设计遵循「策略模式 + 工厂模式」：所有切分器继承统一的抽象基类，通过配置对象声明所需策略，工厂按需延迟注册并返回对应实例，从而把「切分成什么」与「怎么切」解耦。

包内文件布局如下（目录 graphrag_chunking）：

• chunker.py：抽象基类 Chunker；
• sentence_chunker.py：句子切分器实现；
• token_chunker.py：token 定长切分实现及底层算法 split_text_on_tokens；
• chunk_strategy_type.py：策略枚举 ChunkerType；
• chunking_config.py：Pydantic 配置模型 ChunkingConfig；
• chunker_factory.py：ChunkerFactory、register_chunker、create_chunker；
• create_chunk_results.py：把字符串列表包装为带位置信息的 TextChunk；
• text_chunk.py：TextChunk 数据结构；
• transformers.py：内置文本变换器 add_metadata；
• bootstrap_nltk.py：NLTK 资源自动初始化。

统一切分器抽象与产出物 TextChunk

所有切分器都实现同一个抽象基类 Chunker（见 chunker.py），它定义了两个抽象方法：

• __init__(self, **kwargs)：接受任意关键字参数，为不同切分策略的差异化参数留出空间；
• chunk(self, text, transform=None) -> list[TextChunk]：接收原始文本，返回 TextChunk 列表；transform 是一个可选的字符串函数，在最终产出前对每一段文本做变换（典型用途是注入元数据）。

TextChunk（见 text_chunk.py）是一个 dataclass，记录一份切分结果的关键信息：

• original：未经任何变换的原始文本片段；
• text：变换后的最终文本内容（未提供 transform 时与 original 一致）；
• index：该片段在源文档中的从 0 开始的下标；
• start_char / end_char：片段在源文档中的字符级起止位置（0 基索引）；
• token_count：最终文本的 token 数，仅在传入 encode 函数时计算，默认为 None。

这些字段由 create_chunk_results（见 create_chunk_results.py）填充。它按顺序遍历字符串列表，用「上一段 end_char + 1 作为下一段 start_char」的方式连续累计字符偏移，并假定各片段相对源文本未被 strip；transform 只作用于产出文本，字符定位仍基于 original。

策略类型由字符串枚举 ChunkerType（见 chunk_strategy_type.py）统一定义，目前仅包含两个成员：

枚举成员	字符串值	对应实现
ChunkerType.Tokens	"tokens"	TokenChunker
ChunkerType.Sentence	"sentence"	SentenceChunker

句子级切分：SentenceChunker 与 NLTK 引导

SentenceChunker（见 sentence_chunker.py）的目标是“把文本按句子边界拆成列表，每个元素是一个独立句子”，适合以句子为语义单元做更细粒度处理的场景。

NLTK 资源自动引导

构造 SentenceChunker 时会调用 bootstrap()（见 bootstrap_nltk.py）。该函数使用模块级布尔标志 initialized_nltk 保证整个进程只初始化一次，并依次下载 punkt、punkt_tab、averaged_perceptron_tagger、averaged_perceptron_tagger_eng、maxent_ne_chunker、maxent_ne_chunker_tab、words、wordnet 等语料资源，最后通过 wn.ensure_loaded() 确保 WordNet 就绪。此外它还顺带屏蔽了 numba 相关 warnings，避免噪音日志干扰。

切分逻辑与空白补偿

chunk 方法流程如下：

1. 对输入做 text.strip() 后调用 nltk.sent_tokenize 得到句子列表；
2. 调用 create_chunk_results 生成 TextChunk；
3. 由于 NLTK 句子分词器会裁掉句子首尾空白，它会对每个结果执行“回查校正”：在原始文本中从 start_char 起查找该句实际出现位置，若发现正偏移 delta，则把当前片段以及下一个片段的 start_char/end_char 整体平移 delta，从而保证字符定位不因空白丢失而错位。

该行为在 test_chunker.py 中得到验证：输入 " Sentence with spaces. Another one! " 时，两句的 start_char 分别是 3 与 25，正确跳过了前导空格，同时 end_char 也精确落在句末字符处。

Token 定长切分：TokenChunker 与滑动窗口算法

TokenChunker（见 token_chunker.py）不按句子边界，而是“按 token 数量切成固定大小、相邻块可重叠”的片段，这是 GraphRAG 索引文本单元阶段最典型的形态。

构造参数：

• size：每块目标 token 数；
• overlap：相邻块重叠的 token 数；
• encode: Callable[[str], list[int]]：把文本编码成 token id 列表；
• decode: Callable[[list[int]], str]：把 token id 列表还原成文本。

chunk 先调用模块级函数 split_text_on_tokens 完成原始切分，再交给 create_chunk_results 包装。核心算法 split_text_on_tokens 是一个带滑动步长的窗口循环：

1. 先 encode(text) 得到全部 token；
2. start_idx 从 0 开始，cur_idx = min(start_idx + size, len(tokens))，取 tokens[start_idx:cur_idx] 解码为一段文本；
3. 若已到末尾则终止，否则 start_idx += size - overlap，进入下一个窗口。

需要特别说明的是：滑动窗口的步长是 size - overlap，因此相邻块之间会有 overlap 个 token 是重复的；由于最后一个窗口若不足 size 会取到文本尾部余量，结果中可能出现尾部的短块。

单元测试对算法语义给出了精确示例（见 test_chunker.py）：当 chunk_size=2, chunk_overlap=1 且使用 o200k_base tokenizer 时，输入被切成 "This is"、" is a"、" a test"…… 每个后续块都以块大小减重叠数即 1 个新 token 向前推进；而当 chunk_overlap=0、chunk_size=3 时则得到完全无重叠的连续块。空文本 split_text_on_tokens("", ...) 返回空列表。

配置驱动：ChunkingConfig 的字段与默认值

ChunkingConfig（见 chunking_config.py）是一个 Pydantic v2 配置模型，直接把「选哪种策略、用多大块、重叠多少」沉淀成声明式配置，便于从 YAML/TOML 或环境配置中反序列化。其字段与默认值如下：

字段	类型	默认值	说明
type	str	ChunkerType.Tokens（即 "tokens"）	分块策略类型，见上文枚举
encoding_model	str | None	None	使用的编码模型名称
size	int	1200	每个 chunk 的目标 token 数量
overlap	int	100	相邻 chunk 之间的重叠 token 数
prepend_metadata	list[str] | None	None	需要前置到每个 chunk 上的源文档元数据字段名

此外，model_config = ConfigDict(extra="allow") 允许携带自定义扩展字段，以兼容用户自定义切分器的额外初始化参数。默认值 size=1200, overlap=100 意味着在未显式配置时，文本会按约 1200 token 切块、块间保留 100 token 重叠，兼顾上下文连贯性与总体成本。

工厂方法：create_chunker 与自定义策略注册

chunker_factory.py（见 chunker_factory.py）基于 graphrag_common.factory.factory.Factory 泛型基类构建了一个单例 ChunkerFactory[Chunker]，并导出两个函数：

• register_chunker(chunker_type, chunker_initializer, scope="transient")：把一个自定义切分器实现注册进工厂。chunker_type 是标识串，chunker_initializer 是返回 Chunker 的可调用对象，scope 默认为 "transient"（每次创建新实例）。
• create_chunker(config, encode=None, decode=None) -> Chunker：基于 ChunkingConfig 创建切分器。

create_chunker 的解析逻辑值得注意：

1. 先 config.model_dump() 把配置转成字典；
2. 若调用方显式传入 encode/decode 可调用对象，则覆盖到配置字典对应键上（这正是 TokenChunker 拿到 tokenizer 的途径）；
3. 以 config.type 作为策略键查询工厂；
4. 若该策略尚未注册，则按 ChunkerType 做延迟注册："tokens" 惰性导入 TokenChunker，"sentence" 惰性导入 SentenceChunker；
5. 对未知策略抛出 ValueError，消息中会列出当前工厂已注册的类型；
6. 最后用 chunker_factory.create(strategy, init_args=config_model) 完成实例化。

在测试（test_chunker.py）中可以看到完整用法：create_chunker(ChunkingConfig(type=ChunkerType.Sentence)) 一行即可得到句子切分器；而 token 场景则通过 create_chunker(config, mock_encoder.encode, mock_encoder.decode) 把自定义 encode/decode 注入 ChunkingConfig(size=5, overlap=1, encoding_model="fake-encoding", type=ChunkerType.Tokens)。由于默认 type 即为 "tokens"，即便只传一个空配置也能得到 token 切分器实例。

文本变换器：add_metadata 与 transform 钩子

chunk 方法签名中的 transform 参数配合 transformers.py（见 transformers.py）内置的 add_metadata 使用，可以在切分产出阶段把键值对元数据写入每段文本，增强下游检索时的可定位性。

add_metadata(metadata, delimiter=": ", line_delimiter="\n", append=False) 返回一个变换函数：

• 默认把多行 key: value 前置到文本前（如 message: hello\nThis is a test.）；
• append=True 时改为追加到文本末尾；
• delimiter 控制键值分隔符，line_delimiter 控制多对键值的换行符，均可自定义。

对应测试（test_prepend_metadata.py）覆盖了单行/多行元数据、前置/追加以及自定义分隔符四类场景，例如 add_metadata({"message": "hello", "tag": "first"}, delimiter="-", line_delimiter="_") 会把文本变为 message-hello_tag-first_This is a test.。

快速上手：三步完成一次分块

基于 example_notebooks 目录下的示例思路，可按如下步骤使用：

第一步：句子级切分（NLTK）

构造 SentenceChunker 或通过工厂按 "sentence" 策略创建，随后调用 chunk：

from graphrag_chunking.chunker_factory import create_chunker
from graphrag_chunking.chunk_strategy_type import ChunkerType
from graphrag_chunking.chunking_config import ChunkingConfig

chunker = create_chunker(ChunkingConfig(type=ChunkerType.Sentence))
chunks = chunker.chunk("This is a test. Another sentence. And a third one!")
# 每个 chunk.text 为一个完整句子，并带有 start_char/end_char/index 定位

完整可运行示例参见 basic_sentence_example.ipynb。

第二步：Token 定长切分

需要提供 encode/decode，实践中通常来自 graphrag_llm 的 tokenizer：

chunker = create_chunker(
    ChunkingConfig(
        type=ChunkerType.Tokens,
        size=1200,
        overlap=100,
        encoding_model="your-encoding-model",
    ),
    encode=tokenizer.encode,
    decode=tokenizer.decode,
)
chunks = chunker.chunk(long_text)

完整可运行示例参见 token_chunking_example.ipynb。

第三步：通过工厂统一获取实例

工厂方式是官方推荐路径：把 ChunkingConfig 与具体切分器解耦，策略变更时只需修改配置：

from graphrag_chunking.chunker_factory import create_chunker
from graphrag_chunking.chunking_config import ChunkingConfig

config = ChunkingConfig(type="tokens", size=1200, overlap=100)
chunker = create_chunker(config)

完整可运行示例参见 factory_helper_util_example.ipynb。

扩展自己的切分策略

若内置策略不满足需求，可按三步接入工厂体系：

1. 继承 Chunker，实现 __init__(**kwargs) 与 chunk(text, transform=None) -> list[TextChunk]；
2. 调用 register_chunker("my_strategy", MyChunker) 注册；
3. 用 ChunkingConfig(type="my_strategy", extra 参数由 extra="allow" 透传) 通过 create_chunker 获取实例。

register_chunker 的 scope 参数复用 graphrag_common.factory.factory 中的 ServiceScope（默认 "transient"），可依据生命周期需求决定实例是每次新建还是复用，从源码结构看这是为接入依赖注入容器预留的能力。

小结

graphrag-chunking 通过「抽象基类 + 两类内置实现 + Pydantic 配置 + 工厂延迟注册」四层结构，为 GraphRAG 场景提供了灵活而克制的文本分块能力：句子策略用 NLTK 做语义边界切分并自动补偿空白偏移；token 策略用滑动窗口实现可重叠的定长分块并严格遵循 size/overlap 语义；两者最终都产出带 index 与字符定位的 TextChunk，配合 add_metadata 变换器即可无缝对接下游索引与检索环节。