ChatGPT на вашем ноутбуке: бесплатно, анонимно, навсегда

Глава 4.4. Практикум:индексация папки с документами.

В предыдущих главах мы собрали все детали конструктора: embedding-модель превращает текст в векторы, ChromaDB хранит их и ищет. Теперь нам нужен «загрузчик» — функция, которая пройдётся по папке с вашими документами, прочитает файлы, разобьёт их на осмысленные куски, векторизует и сохранит в базу данных. Это и будет индексация — процесс, который выполняется один раз (или периодически) и делает вашу личную базу знаний доступной для вопросов.

Инструменты для чтения разных форматов

Мы хотим поддерживать три самых распространённых типа файлов:

· PDF — контракты, отчёты, сканы (если распознаны). Будем использовать библиотеку PyPDF2.

· DOCX — документы Microsoft Word. Используем python-docx.

· TXT — обычные текстовые файлы (читаются встроенными средствами Python).

Установим недостающие пакеты:

bash

pip install PyPDF2 python-docx langchain-text-splitters

langchain-text-splitters — это лёгкий пакет, который содержит умные разбиватели текста. Он не зависит от всего LangChain, только от langchain-core. Мы будем использовать RecursiveCharacterTextSplitter — он пытается разбить текст по абзацам, потом по предложениям, потом по словам, чтобы сохранить смысловую целостность.

Разбивка на чанки: почему это важно

Если мы скормим в ChromaDB целый 100-страничный PDF как один документ, поиск по нему будет бесполезен — вектор одного огромного текста плохо отражает детали. Нужно разбить текст на фрагменты, каждый из которых содержит законченную мысль.

Оптимальный размер чанка — примерно 500–1000 символов. Слишком маленькие чанки теряют контекст, слишком большие — размывают релевантность и не влезают в контекстное окно LLM. Мы также добавим перекрытие (overlap) — 100–200 символов, которые дублируются с предыдущим чанком. Это помогает сохранить связность: если мысль «переползла» через границу чанка, она будет видна в обоих.

Функция чтения файлов

Создадим модуль document_reader.py:

python

# document_reader.py

import os

from typing import List, Dict

import PyPDF2

from docx import Document

def read_pdf(file_path: str) -> str:

"""Извлекает текст из PDF-файла, страница за страницей."""

text_parts = []

with open(file_path, 'rb') as f:

reader = PyPDF2.PdfReader(f)

for page_num, page in enumerate(reader.pages, start=1):

page_text = page.extract_text()

if page_text:

text_parts.append(f"[Страница {page_num}] {page_text}")

return "\n".join(text_parts)

def read_docx(file_path: str) -> str:

"""Извлекает текст из DOCX-файла."""

doc = Document(file_path)

text_parts = [para.text for para in doc.paragraphs if para.text.strip()]

return "\n".join(text_parts)

def read_txt(file_path: str) -> str:

"""Читает простой текстовый файл с автоопределением кодировки."""

# Пробуем UTF-8, если не получается — cp1251 (Windows-кириллица)

try:

with open(file_path, 'r', encoding='utf-8') as f:

return f.read()

except UnicodeDecodeError:

with open(file_path, 'r', encoding='cp1251') as f:

return f.read()

def read_file(file_path: str) -> str:

"""

Определяет тип файла по расширению и возвращает его полный текст.

"""

ext = os.path.splitext(file_path)[1].lower()

if ext == '.pdf':

return read_pdf(file_path)

elif ext == '.docx':

return read_docx(file_path)

elif ext == '.txt':

return read_txt(file_path)

else:

raise ValueError(f"Неподдерживаемый формат: {ext}")

Пояснения :

· read_pdf нумерует страницы — это пригодится для метаданных и цитирования.

· read_docx извлекает только непустые параграфы, чтобы не засорять базу пустыми строками.

· read_txt обрабатывает распространённую проблему: файлы в кодировке Windows-1251 (часто встречаются в русскоязычных проектах).

Функция индексации папки

Теперь — главный модуль indexer.py, который объединяет всё вместе:

python

# indexer.py

import os

import uuid

from typing import List, Dict

from langchain_text_splitters import RecursiveCharacterTextSplitter

from document_reader import read_file

from embedder import Embedder

from vector_store import VectorStore

def index_folder(

folder_path: str,

store: VectorStore,

embedder: Embedder,

chunk_size: int = 800,

chunk_overlap: int = 150,

file_extensions: List[str] = ['.pdf', '.docx', '.txt']

) -> int:

"""

Рекурсивно обходит папку, читает все поддерживаемые файлы,

разбивает на чанки, векторизует и сохраняет в ChromaDB.

Возвращает общее количество добавленных чанков.

"""

# Создаём разбиватель текста

splitter = RecursiveCharacterTextSplitter(

chunk_size=chunk_size,

chunk_overlap=chunk_overlap,

separators=["\n\n", "\n", ". ", " ", ""] # порядок: от крупных разделителей к мелким

)

total_chunks = 0

# Обходим все файлы в папке и подпапках

for root, dirs, files in os.walk(folder_path):

for file_name in files:

ext = os.path.splitext(file_name)[1].lower()

if ext not in file_extensions:

continue # пропускаем неподдерживаемые форматы

file_path = os.path.join(root, file_name)

print(f"Обработка: {file_path}")

try:

# Читаем полный текст документа

full_text = read_file(file_path)

if not full_text.strip():

print(f" -> Пустой файл, пропущен.")

continue

# Разбиваем на чанки

chunks = splitter.split_text(full_text)

print(f" -> Получено чанков: {len(chunks)}")

# Готовим данные для пакетной вставки

ids = []

embeddings_list = []

documents = []

metadatas = []

for i, chunk in enumerate(chunks):

chunk_id = f"{file_path}_{i}_{uuid.uuid4().hex[:8]}"

ids.append(chunk_id)

documents.append(chunk)

metadatas.append({

"source": file_path,

"file_name": file_name,

"chunk_index": i

})

# Векторизуем сразу все чанки (batch-обработка)

embeddings_list = embedder.embed_documents(chunks)

# Сохраняем в ChromaDB

store.add_documents(

ids=ids,

embeddings=embeddings_list,

documents=documents,

metadatas=metadatas

)

total_chunks += len(chunks)

except Exception as e:

print(f" -> Ошибка: {e}")

continue

return total_chunks

Ключевые моменты :

· Мы используем RecursiveCharacterTextSplitter, который пробует разбивать текст по двойным переводам строки (абзацы), затем по одному, затем по точкам и пробелам. Так чанки получаются более «умными».

· Метаданные (source, file_name, chunk_index) позволят нам при ответе указать, из какого файла и какого места взят фрагмент.

· Векторизуем чанки пакетом: embedder.embed_documents(chunks). Это намного быстрее, чем векторизовать по одному, потому что модель обрабатывает список за один проход.

· Идентификатор чанка содержит путь к файлу и случайную часть, чтобы избежать коллизий.

Тестирование индексации

Создадим в проекте папку sample_docs/ с тестовыми файлами:

· test.txt — пара абзацев.

· test.pdf (можно создать из Word и сохранить как PDF).

· test.docx — любой документ.

Напишем тестовый скрипт:

python

# test_indexer.py

from embedder import Embedder

from vector_store import VectorStore

from indexer import index_folder

# 1. Инициализируем компоненты

print("Загрузка embedding-модели...")

emb = Embedder()

store = VectorStore(collection_name="test_docs")

# 2. Индексируем папку

print("Начинаем индексацию...")

count = index_folder(

folder_path="./sample_docs",

store=store,

embedder=emb,

chunk_size=500,

chunk_overlap=50

)

print(f"\nИндексация завершена. Всего чанков: {count}")

# 3. Поищем что-нибудь

query = "ваш тестовый запрос"

query_vec = emb.embed_query(query)

results = store.search(query_vec, n_results=3)

print("\nРезультаты поиска:")

for doc, meta in zip(results["documents"][0], results["metadatas"][0]):

print(f"- [{meta['file_name']}] {doc[:100]}...")

Запустим:

bash

python test_indexer.py

Вывод:

text

Загрузка embedding-модели intfloat/multilingual-e5-base...

Модель загружена, размерность векторов: 768

Начинаем индексацию...

Обработка: ./sample_docs/test.txt

-> Получено чанков: 2

Обработка: ./sample_docs/test.docx

-> Получено чанков: 3

Обработка: ./sample_docs/test.pdf

-> Получено чанков: 5

Индексация завершена. Всего чанков: 10

Результаты поиска:

- [test.pdf] ... текст, релевантный запросу...

Переиндексация и обновление данных

Что делать, когда документы меняются? Пока что мы просто добавляем новые чанки, но старые версии тех же файлов остаются в базе. Для начала этого достаточно: при повторном запуске index_folder мы создадим дубликаты. Чтобы решить проблему, перед индексацией можно полностью очищать коллекцию:

python

store.clear()

Или, как вариант, хранить время последней индексации и добавлять только новые/изменённые файлы. Для персонального использования очистка и переиндексация с нуля — самый надёжный вариант, ведь это занимает минуты на обычном ноутбуке.

Возможные проблемы прииндексации документов

PDF содержит только изображения. Если документ представляет собой скан, а не текст, библиотека PyPDF2 не сможет извлечь из него слова — она работает только с текстовым слоем. Для таких файлов нужно предварительное распознавание через OCR, например, с помощью Tesseract. Это выходит за рамки нашей книги, но вы можете добавить поддержку сканов позже, установив pytesseract и модифицировав функцию read_pdf.

Очень большой PDF на сотни страниц. Загрузка всего документа в память целиком может привести к её исчерпанию. Решение — читать PDF постранично с помощью PyPDF2.PdfReader и обрабатывать страницы порциями, а не все сразу. Наш код уже делает это: мы проходим по страницам в цикле, извлекая текст по одной, что экономит память.

Ошибка кодировки в текстовом файле. Не все TXT-файлы сохранены в UTF-8. В России до сих пор распространена кодировка Windows-1251, особенно в старых документах. В функции read_txt мы предусмотрели fallback: если чтение в UTF-8 не удалось, пробуем cp1251. Если вы работаете с файлами в других кодировках, можно добавить библиотеку chardet для автоматического определения кодировки.

Разрыв слов при разбивке на чанки. RecursiveCharacterTextSplitter старается разрезать текст по естественным границам — абзацам, предложениям, пробелам. Но с кириллицей он иногда ошибается, и слово может быть разорвано на границе чанка. Чтобы уменьшить вероятность разрывов, увеличьте параметр chunk_overlap — например, со 150 до 250 символов. Это создаст более широкую зону перекрытия между соседними чанками, и мысль не потеряется на стыке.

Что дальше

Мы подготовили базу знаний. Теперь осталось связать поиск по документам с нашей языковой моделью. В главе 4.5 «Реализация функции "Спросить по документам" с цитированием» мы напишем метод ask_documents(), который примет вопрос, найдёт релевантные чанки, вставит их в промпт и вернёт ответ со ссылками на источники. Это станет «вишенкой» нашего локального RAG.

Глава 4.5. Реализацияфункции «Спросить по документам» с цитированием

Мы подошли к моменту, когда разрозненные детали — embedding-модель, векторная база данных, индексатор — складываются в работающий механизм. В этой главе мы напишем функцию ask_documents(), которая примет вопрос, найдёт релевантные фрагменты в вашей личной коллекции документов, передаст их языковой модели и вернёт осмысленный ответ с указанием источников.

Это не просто выдача списка найденных кусков. Модель прочитает их и синтезирует ответ: объяснит, сопоставит факты, сделает вывод. А цитаты подтвердят, что информация взята из ваших файлов, а не выдумана.

Архитектура: что происходит внутри ask_documents()

Функция выполняет четыре шага:

1. Векторизация вопроса. Мы берём вопрос пользователя и прогоняем через ту же embedding-модель, что использовали для индексации документов.

2. Поиск в ChromaDB. Векторная база возвращает, скажем, 4–5 самых близких по смыслу чанков.

3. Сборка промпта. Мы формируем системное сообщение, вставляем найденные чанки с указанием источников и задаём вопрос пользователя.

4. Генерация ответа. Модель получает промпт и генерирует ответ, содержащий факты из документов и ссылки на источники.

Всё это происходит локально, в одном процессе Python, без единого внешнего запроса.

Системный промпт для RAG

Правильно составленный системный промпт — ключ к качественному ответу. Он должен заставить модель:

· Опираться только на предоставленные документы.

· Честно признаваться, если ответа в них нет.

· Цитировать источники.

Вот промпт, который мы будем использовать:

text

Ты — эксперт-аналитик, который отвечает на вопросы, основываясь исключительно на предоставленных фрагментах документов.

Правила:

1. Если ответ можно найти в документах, дай его чётко и по делу.

2. После каждого факта указывай в скобках источник в формате (файл: имя_файла).

3. Если информация из разных фрагментов противоречит друг другу, укажи на противоречие.

4. Если в документах нет ответа на вопрос, скажи: "В предоставленных документах информация не найдена."

5. Не придумывай ничего, чего нет в документах.

Код : метод ask_documents в классе LocalModel

Добавим новый метод в наш класс LocalModel. Откройте local_model.py и дополните его:

python

# local_model.py (дополнение к существующему классу)

from vector_store import VectorStore

from embedder import Embedder

class LocalModel:

# ... (предыдущие методы остаются без изменений) ...

def ask_documents(

self,

question: str,

store: VectorStore,

embedder: Embedder,

n_results: int = 4,

temperature: float = 0.3,

max_tokens: int = 512

) -> str:

"""

Задаёт вопрос по проиндексированным документам.

:param question: вопрос пользователя

:param store: объект VectorStore с загруженными документами

:param embedder: объект Embedder для векторизации вопроса

:param n_results: сколько релевантных чанков извлечь из базы

:param temperature: креативность ответа (низкая — строже по документам)

:param max_tokens: максимальная длина ответа

:return: ответ модели со ссылками на источники

"""

# Шаг 1: векторизуем вопрос

query_embedding = embedder.embed_query(question)

# Шаг 2: ищем релевантные чанки

results = store.search(query_embedding, n_results=n_results)

if not results["documents"] or not results["documents"][0]:

return "В предоставленных документах информация не найдена."

# Шаг 3: собираем фрагменты в промпт

context_parts = []

for doc, meta in zip(results["documents"][0], results["metadatas"][0]):

file_name = meta.get("file_name", "неизвестный файл")

context_parts.append(f"--- Фрагмент из файла: {file_name} ---\n{doc}\n")

context = "\n".join(context_parts)

system_prompt = (

"Ты — эксперт-аналитик, который отвечает на вопросы, основываясь исключительно на "

"предоставленных фрагментах документов. Правила:\n"

"1. Если ответ можно найти в документах, дай его чётко и по делу.\n"

"2. После каждого факта указывай в скобках источник в формате (файл: имя_файла).\n"

"3. Если информация из разных фрагментов противоречит друг другу, укажи на противоречие.\n"

"4. Если в документах нет ответа на вопрос, скажи: \"В предоставленных документах информация не найдена.\"\n"

"5. Не придумывай ничего, чего нет в документах."

)

user_message = (

f"Вопрос пользователя: {question}\n\n"

f"Фрагменты документов для анализа:\n\n{context}\n"

)

# Шаг 4: генерируем ответ

response = self.chat(

messages=[{"role": "user", "content": user_message}],

system_prompt=system_prompt,

temperature=temperature,

max_tokens=max_tokens

)

return response

Обратите внимание:

· temperature=0.3 — низкая температура делает модель более «послушной» и менее склонной к фантазиям. Для фактологического поиска это важно.

· Мы формируем промпт так, чтобы модель не получала голый вопрос, а видела перед собой задание аналитика с конкретными данными.

Полный тестовый сценарий

Создадим скрипт test_rag.py, который демонстрирует весь пайплайн: от индексации до вопросов.

python

# test_rag.py

from model_loader import download_model

from local_model import LocalModel

from embedder import Embedder

from vector_store import VectorStore

from indexer import index_folder

# 1. Подготовка: загружаем модель, embedding и хранилище

print("Загрузка языковой модели...")

model_path = download_model()

llm = LocalModel(model_path=model_path, n_threads=4, n_gpu_layers=0)

print("Загрузка embedding-модели...")

emb = Embedder()

print("Подготовка векторного хранилища...")

store = VectorStore(collection_name="my_docs")

# 2. Индексация (выполняется один раз или после добавления файлов)

# Если коллекция уже существует — очистим для чистоты эксперимента

store.clear()

count = index_folder(

folder_path="./sample_docs",

store=store,

embedder=emb,

chunk_size=500,

chunk_overlap=50

)

print(f"Проиндексировано чанков: {count}\n")

# 3. Задаём вопросы

questions = [

"Какие гарантийные обязательства у исполнителя?",

"В каких случаях договор может быть расторгнут досрочно?",

"Какой размер штрафа за просрочку оплаты?",

"Есть ли в документах упоминание о конфиденциальности?"

]

for q in questions:

print(f"\n{'='*60}")

print(f"Вопрос: {q}")

print(f"{'='*60}")

answer = llm.ask_documents(

question=q,

store=store,

embedder=emb,

n_results=3,

temperature=0.2

)

print(f"Ответ: {answer}")

Запустим:

bash

python test_rag.py

Пример вывода:

text

Загрузка языковой модели...

Модель загружена.

Загрузка embedding-модели...

Модель загружена, размерность векторов: 768

Подготовка векторного хранилища...

Обработка: ./sample_docs/contract.txt

-> Получено чанков: 4

Проиндексировано чанков: 4

============================================================

Вопрос: Какие гарантийные обязательства у исполнителя?

============================================================

Ответ: Согласно контракту (файл: contract.txt), исполнитель гарантирует устранение

недостатков в течение 30 календарных дней с момента получения уведомления.

Гарантийный срок на результаты работ составляет 12 месяцев (файл: contract.txt).

============================================================

Вопрос: В каких случаях договор может быть расторгнут досрочно?

============================================================

Ответ: Досрочное расторжение возможно при существенном нарушении условий одной из

сторон (файл: contract.txt). Также договор может быть расторгнут по взаимному

соглашению сторон с письменным уведомлением за 30 дней (файл: contract.txt).

============================================================

Вопрос: Какой размер штрафа за просрочку оплаты?

============================================================

Ответ: За каждый день просрочки оплаты начисляется пеня в размере 0.1% от

неоплаченной суммы (файл: contract.txt).

============================================================

Вопрос: Есть ли в документах упоминание о конфиденциальности?

============================================================

Ответ: В предоставленных документах информация не найдена.

Модель корректно извлекла факты и честно призналась, когда информации нет. Это именно то поведение, которое нам нужно.

Потоковая версия метода

Для будущего GUI нам понадобится потоковая версия ask_documents(), чтобы ответ появлялся на экране постепенно, как в чате. Добавим её:

python

def ask_documents_stream(

self,

question: str,

store: VectorStore,

embedder: Embedder,

n_results: int = 4,

temperature: float = 0.3,

max_tokens: int = 512

):

"""Потоковая версия ask_documents()."""

query_embedding = embedder.embed_query(question)

results = store.search(query_embedding, n_results=n_results)

if not results["documents"] or not results["documents"][0]:

yield "В предоставленных документах информация не найдена."

return

context_parts = []

for doc, meta in zip(results["documents"][0], results["metadatas"][0]):

file_name = meta.get("file_name", "неизвестный файл")

context_parts.append(f"--- Фрагмент из файла: {file_name} ---\n{doc}\n")

context = "\n".join(context_parts)

system_prompt = (

"Ты — эксперт-аналитик... Правила: ..." # тот же промпт

)

user_message = f"Вопрос пользователя: {question}\n\nФрагменты документов:\n\n{context}"