from typing import Literal from transformers import AutoTokenizer from langchain_text_splitters import RecursiveCharacterTextSplitter, NLTKTextSplitter from langchain_experimental.text_splitter import SemanticChunker from langchain_huggingface.embeddings import HuggingFaceEmbeddings class Splitter: """ Класс описывает функционал разделения текста на чанки тремя способами на выбор: - рекурсивно разбивая чанки различными разделителями в порядке возрастания "жесткости" их эффекта; - объединяя выделенные с помощью библиотеки NLTK предложения в чанки определенного размера и с наложением; - разбивая текст на семантически связанные блоки с помощью векторных представлений текстов; """ def __init__( self, mode: Literal["recursive", "nltk", "semantic"], model_name: str = "deepvk/USER-bge-m3", chunk_size: int = 256, chunk_overlap: int = 64, **splitter_kwargs, ): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.tokenizer = AutoTokenizer.from_pretrained(model_name) match mode: case "recursive": self.splitter = RecursiveCharacterTextSplitter( separators=[ "\n### ", "\n## ", "\n# ", "\n\n", "\n", "!", "?", ". ", ";", ",", ")", " ", "", ], keep_separator="end", chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=lambda x: len(self.tokenizer.encode(x, add_special_tokens=False)), **splitter_kwargs, ) self.split_fn = self._recursive_split case "nltk": self.splitter = NLTKTextSplitter( language="russian", **splitter_kwargs, ) self.split_fn = self._nltk_split case "semantic": self.splitter = SemanticChunker( HuggingFaceEmbeddings( model_name=model_name, encode_kwargs={"normalize_embeddings": True}, ), **splitter_kwargs, ) self.split_fn = self._semantic_split def split_text(self, text: str) -> list[str]: """ Доступная пользователю функция разделения текста на чанки """ return self.split_fn(text) def _recursive_split(self, text: str) -> list[str]: """ Функция разделения текста на чанки при self.splitter == RecursiveCharacterTextSplitter """ return [ chunk for chunk in self.splitter.split_text(text) if any(ch.isalpha() for ch in set(chunk)) ] def _nltk_split(self, text: str) -> list[str]: """ Функция разделения текста на чанки при self.splitter == NLTKTextSplitter """ sentences = self.splitter.split_text(text)[0].split("\n\n") sent_sizes = [ len(self.tokenizer.encode(sent, add_special_tokens=False)) for sent in sentences ] chunks = [] i, n = 0, len(sentences) while i < n: cur_len, cur_texts = 0, [] # --- Собираем строки в чанк --- j = i while (j < n) and (cur_len + sent_sizes[j] <= self.chunk_size): cur_texts.append(sentences[j]) cur_len += sent_sizes[j] j += 1 chunks.append(cur_texts) # --- Сдвигаем окно с overlap --- if j >= n: break # Держим overlap в токенах, но не превышая его overlap_len, k = 0, j - 1 while (k >= i) and (overlap_len + sent_sizes[k] <= self.chunk_overlap): overlap_len += sent_sizes[k] k -= 1 # идём назад от конца чанка # Следующий старт = k+1 i = k + 1 return chunks def _semantic_split(self, text: str) -> list[str]: """ Функция разделения текста на чанки при self.splitter == SemanticChunker """ return self.splitter.split_text(text)