Dify中的文本分词处理技术详解

- 引言
- 核心架构概览
- - 索引处理器工厂
- 文本分词技术详解
- - 基础分词器
  - 增强型递归字符分词器
  - 固定分隔符文本分词器
  - 递归分割算法
- 索引处理器中的分词应用
- 特殊索引处理器的分词特点
- - 问答索引处理器
  - 父子索引处理器
- 分词技术的应用场景
- 技术亮点与优势
- 总结

引言

在现代RAG（检索增强生成）系统中，文本分词（Text Splitting）是一个至关重要的环节。它直接影响到检索的精度和生成内容的质量。本文将深入解析Dify项目中的文本分词处理技术，探讨其实现原理、核心算法和应用场景。

核心架构概览

Dify采用了工厂模式和策略模式来实现灵活的文本处理流程。整个文本处理架构主要包含两个核心部分：

索引处理器（Index Processor）：负责文档的提取、转换和加载
文本分词器（Text Splitter）：负责将长文本切分成适合处理的小块

索引处理器工厂

索引处理器工厂（IndexProcessorFactory）是创建不同类型索引处理器的核心类，它通过工厂模式实现了对不同索引处理策略的封装和创建：

class IndexProcessorFactory:"""IndexProcessorInit."""def __init__(self, index_type: str | None):self._index_type = index_typedef init_index_processor(self) -> BaseIndexProcessor:"""Init index processor."""if not self._index_type:raise ValueError("Index type must be specified.")if self._index_type == IndexType.PARAGRAPH_INDEX:return ParagraphIndexProcessor()elif self._index_type == IndexType.QA_INDEX:return QAIndexProcessor()elif self._index_type == IndexType.PARENT_CHILD_INDEX:return ParentChildIndexProcessor()else:raise ValueError(f"Index type {self._index_type} is not supported.")

该工厂类支持三种索引处理器：

段落索引处理器（ParagraphIndexProcessor）：将文档分割成段落级别的块
问答索引处理器（QAIndexProcessor）：专门处理问答格式的文本
父子索引处理器（ParentChildIndexProcessor）：创建层次化的文档结构

文本分词技术详解

基础分词器

Dify的分词系统建立在抽象基类TextSplitter之上，它定义了分词的基本接口：

@abstractmethod
def split_text(self, text: str) -> list[str]:"""Split text into multiple components."""

所有具体的分词器都必须实现这个方法，以提供特定的分词逻辑。

增强型递归字符分词器

EnhanceRecursiveCharacterTextSplitter是一个关键的分词器实现，它通过递归方式处理文本，并支持使用不同的编码器计算token数量：

class EnhanceRecursiveCharacterTextSplitter(RecursiveCharacterTextSplitter):"""This class is used to implement from_gpt2_encoder, to prevent using of tiktoken"""@classmethoddef from_encoder(cls: type[TS],embedding_model_instance: Optional[ModelInstance],allowed_special: Union[Literal["all"], Set[str]] = set(),disallowed_special: Union[Literal["all"], Collection[str]] = "all",**kwargs: Any,):def _token_encoder(texts: list[str]) -> list[int]:if not texts:return []if embedding_model_instance:return embedding_model_instance.get_text_embedding_num_tokens(texts=texts)else:return [GPT2Tokenizer.get_num_tokens(text) for text in texts]# ... 其他代码 ...return cls(length_function=_token_encoder, **kwargs)

这个分词器的特点是可以使用嵌入模型的tokenizer或默认的GPT2 tokenizer来计算文本长度，避免了对tiktoken的依赖。

固定分隔符文本分词器

FixedRecursiveCharacterTextSplitter是一个更为专业的分词器，它在增强型递归分词器的基础上，增加了对固定分隔符的支持：

class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter):def __init__(self, fixed_separator: str = "\n\n", separators: Optional[list[str]] = None, **kwargs: Any):"""Create a new TextSplitter."""super().__init__(**kwargs)self._fixed_separator = fixed_separatorself._separators = separators or ["\n\n", "\n", " ", ""]def split_text(self, text: str) -> list[str]:"""Split incoming text and return chunks."""if self._fixed_separator:chunks = text.split(self._fixed_separator)else:chunks = [text]final_chunks = []chunks_lengths = self._length_function(chunks)for chunk, chunk_length in zip(chunks, chunks_lengths):if chunk_length > self._chunk_size:final_chunks.extend(self.recursive_split_text(chunk))else:final_chunks.append(chunk)return final_chunks

这个分词器的工作流程如下：

首先使用固定分隔符（默认为\n\n）将文本分割成初步的块
对每个块计算token长度
如果块的长度超过了设定的最大长度（_chunk_size），则调用recursive_split_text方法进一步分割
否则直接将块添加到最终结果中

递归分割算法

recursive_split_text方法是固定分隔符分词器的核心，它实现了复杂的递归分割逻辑：

def recursive_split_text(self, text: str) -> list[str]:"""Split incoming text and return chunks."""final_chunks = []separator = self._separators[-1]new_separators = []# 寻找最合适的分隔符for i, _s in enumerate(self._separators):if _s == "":separator = _sbreakif _s in text:separator = _snew_separators = self._separators[i + 1 :]break# 使用找到的分隔符分割文本if separator:if separator == " ":splits = text.split()else:splits = text.split(separator)else:splits = list(text)splits = [s for s in splits if (s not in {"", "\n"})]# ... 处理分割后的文本块 ...

该算法的精妙之处在于：

它会按照优先级顺序尝试不同的分隔符（如\n\n, \n, , ""）
一旦找到文本中存在的分隔符，就使用它进行分割
如果当前分隔符分割后的块仍然过大，会使用下一级别的分隔符继续分割
最终确保所有文本块都不超过指定的最大token数量

索引处理器中的分词应用

在实际应用中，索引处理器会根据处理规则选择合适的分词器。以ParagraphIndexProcessor为例，它在transform方法中使用分词器处理文档：

def transform(self, documents: list[Document], **kwargs) -> list[Document]:# ... 其他代码 ...splitter = self._get_splitter(processing_rule_mode=process_rule.get("mode"),max_tokens=rules.segmentation.max_tokens,chunk_overlap=rules.segmentation.chunk_overlap,separator=rules.segmentation.separator,embedding_model_instance=kwargs.get("embedding_model_instance"),)all_documents = []for document in documents:# 文档清洗document_text = CleanProcessor.clean(document.page_content, kwargs.get("process_rule", {}))document.page_content = document_text# 解析文档为节点document_nodes = splitter.split_documents([document])# ... 处理分割后的节点 ...

分词器的选择逻辑在_get_splitter方法中实现：

def _get_splitter(self,processing_rule_mode: str,max_tokens: int,chunk_overlap: int,separator: str,embedding_model_instance: Optional[ModelInstance],) -> TextSplitter:"""Get the NodeParser object according to the processing rule."""if processing_rule_mode in ["custom", "hierarchical"]:# 用户自定义分割规则# ... 参数验证 ...character_splitter = FixedRecursiveCharacterTextSplitter.from_encoder(chunk_size=max_tokens,chunk_overlap=chunk_overlap,fixed_separator=separator,separators=["\n\n", "。", ". ", " ", ""],embedding_model_instance=embedding_model_instance,)else:# 自动分割character_splitter = EnhanceRecursiveCharacterTextSplitter.from_encoder(chunk_size=DatasetProcessRule.AUTOMATIC_RULES["segmentation"]["max_tokens"],chunk_overlap=DatasetProcessRule.AUTOMATIC_RULES["segmentation"]["chunk_overlap"],separators=["\n\n", "。", ". ", " ", ""],embedding_model_instance=embedding_model_instance,)return character_splitter

这里的逻辑很清晰：

对于自定义或层次化处理模式，使用FixedRecursiveCharacterTextSplitter，允许指定固定分隔符
对于自动处理模式，使用EnhanceRecursiveCharacterTextSplitter，采用预设的参数

特殊索引处理器的分词特点

问答索引处理器

QAIndexProcessor针对问答格式的文本有特殊的处理逻辑：

def _format_split_text(self, text):regex = r"Q\d+:\s*(.*?)\s*A\d+:\s*([\s\S]*?)(?=Q\d+:|$)"matches = re.findall(regex, text, re.UNICODE)return [{"question": q, "answer": re.sub(r"\n\s*", "\n", a.strip())} for q, a in matches if q and a]

它使用正则表达式识别问题和答案的模式，将文本转换为结构化的问答对。

父子索引处理器

ParentChildIndexProcessor实现了层次化的文档处理，它会先将文档分割成父节点，然后对每个父节点进一步分割成子节点：

def _split_child_nodes(self,document_node: Document,rules: Rule,process_rule_mode: str,embedding_model_instance: Optional[ModelInstance],) -> list[ChildDocument]:# ... 获取子块分割规则 ...child_splitter = self._get_splitter(processing_rule_mode=process_rule_mode,max_tokens=rules.subchunk_segmentation.max_tokens,chunk_overlap=rules.subchunk_segmentation.chunk_overlap,separator=rules.subchunk_segmentation.separator,embedding_model_instance=embedding_model_instance,)# 解析文档为子节点child_nodes = []child_documents = child_splitter.split_documents([document_node])# ... 处理子节点 ...