Агент с RAG‑памятью: chunker.py

solutions/6a02e23da6fe2e4ac16acf65_Агент_с_RAG_памятью/chunker.py

@@ -1,41 +1,86 @@
-
 # chunker.py
 """
-Модуль для разбиения больших текстов на небольшие чанки.
+Модуль для разбиения больших текстов на небольшие части (чанки)
-Использует RecursiveCharacterTextSplitter из LangChain.
+с помощью RecursiveCharacterTextSplitter из LangChain.
 
-Публичный API:
+Используется в процессе загрузки документов в Qdrant:
-    split_text(text: str, chunk_size: int = 1000, chunk_overlap: int = 200) -> List[str]
+1. Читаем исходный файл.
+2. Разбиваем его на чанки, чтобы каждый чанк не превышал заданную длину.
+3. Возвращаем список строк (чанков) и метаданные для каждого чанка
+   (например, номер чанка и оригинальный заголовок).
 """
 
-from typing import List
+from pathlib import Path
-from langchain.text_splitter import RecursiveCharacterTextSplitter
+from typing import List, Tuple
+
+from langchain_text_splitters import RecursiveCharacterTextSplitter
 
 
-def split_text(
+def split_document(
-    text: str,
+    file_path: str | Path,
-    *,
     chunk_size: int = 1000,
     chunk_overlap: int = 200,
-) -> List[str]:
+) -> List[Tuple[str, dict]]:
     """
-    Разбивает входной текст на чанки заданного размера с перекрытием.
+    Разбивает содержимое файла на чанки.
 
-    Args:
+    Parameters
-        text: Текст, который нужно разбить.
+    ----------
-        chunk_size: Максимальная длина одного чанкa (по символам).
+    file_path : str | Path
-        chunk_overlap: Количество символов, которые будут пересекаться между
+        Путь к текстовому файлу (может быть .txt, .md и т.п.).
-            соседними чанками. Позволяет сохранить контекст при поиске.
+    chunk_size : int, default=1000
+        Максимальная длина одного чанка в символах.
+    chunk_overlap : int, default=200
+        Количество пересекающихся символов между соседними чанками.
 
-    Returns:
+    Returns
-        Список строк – чанков текста.
+    -------
+    List[Tuple[str, dict]]
+        Список кортежей (чанк, метаданные). Метаданные включают:
+            - `source`: путь к файлу,
+            - `chunk_index`: порядковый номер чанка в документе.
     """
-    if not isinstance(text, str):
+    # Читаем файл
-        raise TypeError("text must be a string")
+    path = Path(file_path)
+    if not path.is_file():
+        raise FileNotFoundError(f"Файл не найден: {file_path}")
 
+    text = path.read_text(encoding="utf-8")
+
+    # Создаём splitter
     splitter = RecursiveCharacterTextSplitter(
         chunk_size=chunk_size,
         chunk_overlap=chunk_overlap,
-        separators=["\n\n", "\n", " ", ""],
+        length_function=len,
     )
-    return splitter.split_text(text)
+
+    # Получаем чанки
+    chunks = splitter.split_text(text)
+
+    # Формируем список с метаданными
+    result: List[Tuple[str, dict]] = []
+    for idx, chunk in enumerate(chunks):
+        metadata = {
+            "source": str(path.resolve()),
+            "chunk_index": idx,
+        }
+        result.append((chunk, metadata))
+
+    return result
+
+
+# Пример использования (можно закомментировать при импорте)
+if __name__ == "__main__":
+    import sys
+
+    if len(sys.argv) < 2:
+        print("Usage: python chunker.py <file_path> [chunk_size] [chunk_overlap]")
+        sys.exit(1)
+
+    file = sys.argv[1]
+    size = int(sys.argv[2]) if len(sys.argv) > 2 else 1000
+    overlap = int(sys.argv[3]) if len(sys.argv) > 3 else 200
+
+    for chunk, meta in split_document(file, size, overlap):
+        print(f"--- Chunk {meta['chunk_index']} ({len(chunk)} chars) ---")
+        print(chunk[:200] + "...\n")