86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
# chunker.py
|
||
"""
|
||
Модуль для разбиения больших текстов на небольшие части (чанки)
|
||
с помощью RecursiveCharacterTextSplitter из LangChain.
|
||
|
||
Используется в процессе загрузки документов в Qdrant:
|
||
1. Читаем исходный файл.
|
||
2. Разбиваем его на чанки, чтобы каждый чанк не превышал заданную длину.
|
||
3. Возвращаем список строк (чанков) и метаданные для каждого чанка
|
||
(например, номер чанка и оригинальный заголовок).
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from typing import List, Tuple
|
||
|
||
from langchain_text_splitters import RecursiveCharacterTextSplitter
|
||
|
||
|
||
def split_document(
|
||
file_path: str | Path,
|
||
chunk_size: int = 1000,
|
||
chunk_overlap: int = 200,
|
||
) -> List[Tuple[str, dict]]:
|
||
"""
|
||
Разбивает содержимое файла на чанки.
|
||
|
||
Parameters
|
||
----------
|
||
file_path : str | Path
|
||
Путь к текстовому файлу (может быть .txt, .md и т.п.).
|
||
chunk_size : int, default=1000
|
||
Максимальная длина одного чанка в символах.
|
||
chunk_overlap : int, default=200
|
||
Количество пересекающихся символов между соседними чанками.
|
||
|
||
Returns
|
||
-------
|
||
List[Tuple[str, dict]]
|
||
Список кортежей (чанк, метаданные). Метаданные включают:
|
||
- `source`: путь к файлу,
|
||
- `chunk_index`: порядковый номер чанка в документе.
|
||
"""
|
||
# Читаем файл
|
||
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,
|
||
length_function=len,
|
||
)
|
||
|
||
# Получаем чанки
|
||
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") |