ChunkingService 参考

ChunkingService 用于将长文本切分为带重叠的文本块，便于嵌入与检索。系统支持多种分块策略，可根据文档类型和场景灵活选择。

概览

分块是 RAG 摄取流程的关键步骤：

嵌入模型存在上下文长度限制
更小的块更利于高精度检索
适度重叠可避免语义断裂

系统提供五种分块策略：

策略	标识	适用场景
固定大小	`fixed_size`	通用场景，均匀分块
句子级	`sentence`	保持句子完整，适合文章
段落级	`paragraph`	保持段落完整，适合长文档
递归层级	`recursive`	多层级切分，语义完整
语义分割	`semantic`	基于相似度，实验性

配置

通用配置

参数	类型	默认值	说明	环境变量
`chunking.size`	int	512	默认分块字符数	`CHUNK_SIZE`
`chunking.overlap`	int	50	默认重叠字符数	`CHUNK_OVERLAP`
`chunking.default_strategy`	str	`fixed_size`	默认分块策略	`CHUNK_DEFAULT_STRATEGY`

句子策略配置

参数	类型	默认值	说明	环境变量
`chunking.sentence_target_size`	int	512	目标块大小	`CHUNK_SENTENCE_TARGET_SIZE`
`chunking.sentence_max_per_chunk`	int	10	每块最大句子数	`CHUNK_SENTENCE_MAX_PER_CHUNK`

段落策略配置

参数	类型	默认值	说明	环境变量
`chunking.paragraph_target_size`	int	1024	目标块大小	`CHUNK_PARAGRAPH_TARGET_SIZE`
`chunking.paragraph_max_per_chunk`	int	5	每块最大段落数	`CHUNK_PARAGRAPH_MAX_PER_CHUNK`

语义策略配置

参数	类型	默认值	说明	环境变量
`chunking.semantic_similarity_threshold`	float	0.7	相似度阈值	`CHUNK_SEMANTIC_SIMILARITY_THRESHOLD`
`chunking.semantic_min_chunk_size`	int	200	最小块大小	`CHUNK_SEMANTIC_MIN_CHUNK_SIZE`
`chunking.semantic_max_chunk_size`	int	1000	最大块大小	`CHUNK_SEMANTIC_MAX_CHUNK_SIZE`

约束：chunk_overlap 必须小于 chunk_size。

策略详解

FixedSizeChunkingStrategy（固定大小）

字符级分块，尽可能在单词边界处切分。

from ai_service.services.chunking import FixedSizeChunkingStrategy

strategy = FixedSizeChunkingStrategy(
    chunk_size=512,      # 目标块大小
    chunk_overlap=50,    # 重叠字符数
)

特点：

块大小均匀，便于批量处理
词边界感知，避免切断单词
适用于大多数通用文档

SentenceChunkingStrategy（句子级）

使用 NLTK 进行句子切分，按目标大小组合句子。

from ai_service.services.chunking import SentenceChunkingStrategy

strategy = SentenceChunkingStrategy(
    target_size=512,           # 目标块大小
    max_sentences_per_chunk=10,  # 每块最大句子数
)

特点：

保持句子完整性
适合文章、新闻等连贯文本
需要 NLTK 数据（punkt）

ParagraphChunkingStrategy（段落级）

按段落（\n\n）边界切分，按目标大小组合段落。

from ai_service.services.chunking import ParagraphChunkingStrategy

strategy = ParagraphChunkingStrategy(
    target_size=1024,           # 目标块大小
    max_paragraphs_per_chunk=5,  # 每块最大段落数
)

特点：

保持段落完整性
适合结构化文档、技术文档
块通常较大，上下文更完整

RecursiveChunkingStrategy（递归层级）

按层级递归切分：段落 → 句子 → 单词 → 字符。

from ai_service.services.chunking import RecursiveChunkingStrategy

strategy = RecursiveChunkingStrategy(
    chunk_size=512,      # 目标块大小
    chunk_overlap=50,    # 重叠字符数
    separators=["\n\n", "\n", ". ", " ", ""],  # 分隔符层级
)

特点：

优先保持大粒度语义单元
自动降级到更细粒度
综合性能最佳

SemanticChunkingStrategy（语义分割）

基于语义相似度切分（当前为简化实现）。

from ai_service.services.chunking import SemanticChunkingStrategy

strategy = SemanticChunkingStrategy(
    similarity_threshold=0.7,  # 相似度阈值
    min_chunk_size=200,        # 最小块大小
    max_chunk_size=1000,       # 最大块大小
)

特点：

理论上最优的语义连贯性
当前实现为简化版（基于句子长度）
未来版本将集成完整嵌入服务

API 参考

基类：BaseChunkingStrategy

所有分块策略的抽象基类。

class BaseChunkingStrategy(ABC):
    """Abstract base class for document chunking strategies."""

    @abstractmethod
    def chunk_text(
        self,
        text: str,
        document_id: str,
        document_name: str,
        source_id: str,
        extra_metadata: Optional[dict[str, Any]] = None,
    ) -> list[ChunkResult]:
        """Split text into chunks according to the strategy."""
        pass

    @property
    @abstractmethod
    def strategy_name(self) -> str:
        """Return the strategy identifier."""
        pass

工厂函数：get_chunking_strategy

通过策略名称获取策略实例。

def get_chunking_strategy(
    strategy_type: str,
    params: Optional[dict[str, Any]] = None,
) -> BaseChunkingStrategy:
    """Factory function to instantiate a chunking strategy.

    Args:
        strategy_type: Strategy identifier (e.g., 'fixed_size', 'sentence')
        params: Strategy-specific parameters

    Returns:
        Instantiated strategy

    Raises:
        ValueError: If strategy_type is not recognized
    """

支持策略：fixed_size, sentence, paragraph, semantic, recursive

类：ChunkingService（向后兼容）

遗留服务类，包装 FixedSizeChunkingStrategy。

class ChunkingService:
    """Legacy service for splitting documents into chunks.

    Wraps FixedSizeChunkingStrategy for backward compatibility.
    New code should use the strategy classes directly.
    """

    def __init__(
        self,
        *,
        chunk_size: Optional[int] = None,
        chunk_overlap: Optional[int] = None,
    ) -> None:
        """Initialize the chunking service."""

方法：chunk_text

将文本拆分为带重叠的分块并附带元数据。

def chunk_text(
    self,
    text: str,
    document_id: str,
    document_name: str,
    source_id: str,
    extra_metadata: Optional[dict[str, Any]] = None,
) -> list[ChunkResult]:
    """Split text into chunks with overlap.

    Args:
        text: Text content to chunk
        document_id: Document identifier for metadata
        document_name: Document name for metadata
        source_id: Knowledge source ID for metadata
        extra_metadata: Additional metadata to include

    Returns:
        List of ChunkResult objects
    """

返回值：list[ChunkResult]

每个 ChunkResult 包含：

chunk_index：块位置（从 0 开始）
content：分块内容
start_char：起始字符位置
end_char：结束字符位置
metadata：合并后的元数据

方法：estimate_chunk_count

估算给定文本长度的分块数量。

def estimate_chunk_count(self, text_length: int) -> int:
    """Estimate the number of chunks for a given text length.

    Args:
        text_length: Length of the text in characters

    Returns:
        Estimated number of chunks
    """

工厂函数：get_chunking_service

def get_chunking_service() -> ChunkingService:
    """Get or create the singleton chunking service instance."""

使用示例

使用策略工厂

from ai_service.services.chunking import get_chunking_strategy

# 获取句子级分块策略
strategy = get_chunking_strategy(
    strategy_type="sentence",
    params={"target_size": 512, "max_sentences_per_chunk": 10}
)

# 分块
text = "Your long document text here..."
chunks = strategy.chunk_text(
    text,
    document_id="doc-123",
    document_name="example.txt",
    source_id="source-456"
)

# 遍历结果
for chunk in chunks:
    print(f"Chunk {chunk.chunk_index}: {len(chunk.content)} chars")
    print(f"Strategy: {strategy.strategy_name}")

直接使用具体策略

from ai_service.services.chunking import RecursiveChunkingStrategy

strategy = RecursiveChunkingStrategy(
    chunk_size=512,
    chunk_overlap=50,
)

chunks = strategy.chunk_text(
    text="Your document text...",
    document_id="doc-123",
    document_name="example.txt",
    source_id="source-456",
)

基本分块（向后兼容）

from ai_service.services.chunking import get_chunking_service

# 获取服务实例（固定大小策略）
chunking_service = get_chunking_service()

# 分块
text = "Your long document text here..."
chunks = chunking_service.chunk_text(
    text,
    document_id="doc-123",
    document_name="example.txt",
    source_id="source-456"
)

自定义配置

from ai_service.services.chunking import ChunkingService

chunking_service = ChunkingService(
    chunk_size=500,    # 更小的块
    chunk_overlap=100  # 更少重叠
)

chunks = chunking_service.chunk_text("Your text here...")

追加元数据

chunks = chunking_service.chunk_text(
    text,
    document_id="doc-123",
    source_id="source-456",
    extra_metadata={
        "author": "John Doe",
        "category": "technical",
        "language": "en"
    }
)

print(chunks[0].metadata)

估算分块数量

text_length = len(document_text)
estimated_chunks = chunking_service.estimate_chunk_count(text_length)
print(f"Will create approximately {estimated_chunks} chunks")

分块算法

固定大小策略

该服务采用滑动窗口策略：

规范化空白：连续空白折叠为单个空格
计算窗口：起点为 0，窗口长度为 chunk_size
寻找词边界：窗口内尽可能选择最后一个空格作为切分点
生成分块：截取文本并去除首尾空白
窗口推进：移动 chunk_size - chunk_overlap
重复执行：直至处理完文本

词边界切分

# 示例: chunk_size=20, text="The quick brown fox jumps"
# 不处理边界: "The quick brown fox "
# 处理边界:   "The quick brown fox"  # 在空格处截断

若窗口内找不到空格，将在字符边界处截断。

策略选择指南

按文档类型

文档类型	推荐策略	理由
通用文本	`recursive`	综合性能最佳
新闻/文章	`sentence`	保持句子完整
技术文档	`paragraph`	保持段落结构
代码文件	`fixed_size`	均匀分块
法律合同	`sentence`	精确句子边界

按块大小需求

需求	推荐策略	典型参数
小块（<500）	`fixed_size`	`chunk_size=256`
中块（500-1000）	`sentence`	`target_size=512`
大块（>1000）	`paragraph`	`target_size=2048`

最佳实践

分块大小选择

小块 256-512：更高精度，适合短答案场景
中块 512-1024：平衡方案，推荐
大块 1024+：上下文更完整，但精度可能下降

重叠选择

最低重叠：块大小的 10-15%
推荐重叠：块大小的 15-20%
最高重叠：块大小的 25-30%

空文本处理

chunks = strategy.chunk_text("", "", "", "")      # 返回 []
chunks = strategy.chunk_text("   ", "", "", "")   # 返回 []

元数据建议

建议始终带上 document_id 与 source_id
extra_metadata 用于业务字段
避免存放过大的嵌套对象

性能特性

策略	时间复杂度	典型耗时（1 万字）
fixed_size	O(n)	< 10ms
sentence	O(n)	< 50ms
paragraph	O(n)	< 20ms
recursive	O(n)	< 100ms
semantic	O(n)	< 100ms

内存开销：O(n) 归一化文本 + O(k) 分块数量

常见问题

问题：NLTK 数据缺失

现象：LookupError: Resource punkt not found

解决：

import nltk
nltk.download('punkt')
nltk.download('punkt_tab')  # 某些版本需要

问题：分块过小

原因：文本存在大量短行或换行

解决：增大 chunk_size 或选择 paragraph 策略

问题：句子被截断

原因：句子长度超过 chunk_size

解决：使用 sentence 或 recursive 策略

问题：没有分块

原因：文本为空或仅包含空白字符

解决：检查解析结果与输入内容

问题：策略不存在

现象：ValueError: Unknown chunking strategy

解决：检查策略名称拼写，使用以下之一： fixed_size, sentence, paragraph, semantic, recursive

关联文档

架构设计 - 分块在流程中的位置
IngestionService - 摄取流程中使用分块
配置参考 - 分块相关配置