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])# ... 處理子節點 ...