Загрузка документов

RAG-система настолько хороша, насколько хороши документы, которые в неё загружены.

#Зачем нужна загрузка документов

В реальных RAG-системах документы приходят в разных форматах: PDF-отчёты, DOCX-договоры, CSV-таблицы, HTML-страницы, простые TXT-файлы. Каждый формат имеет свою структуру и особенности. Неправильная загрузка приведёт к потере информации, битому тексту и плохому качеству ответов.

В этой теме мы разберём, как правильно читать каждый формат и извлекать текст для последующей обработки.

#Загрузка TXT-файлов

Самый простой формат — plain text. Но даже здесь есть нюансы с кодировкой.

# Простой случай
with open("document.txt", "r", encoding="utf-8") as f:
    text = f.read()

print(f"Длина текста: {len(text)} символов")

Кодировка — частая проблема. Если файл в Windows-1251, а вы читаете как UTF-8, получите «кракозябры». Решение — указывать правильную кодировку или использовать автоопределение:

import chardet

# Определяем кодировку
with open("document.txt", "rb") as f:
    raw_data = f.read()
    result = chardet.detect(raw_data)
    encoding = result["encoding"]

print(f"Определённая кодировка: {encoding}")

# Читаем с правильной кодировкой
text = raw_data.decode(encoding)

Библиотека chardet анализирует байты и угадывает кодировку. Это не идеально — для маленьких файлов может ошибиться — но для большинства случаев работает.

#Загрузка PDF-файлов

PDF — сложный формат: текст может быть в колонках, таблицах, с изображениями. Для извлечения текста есть несколько библиотек.

#pypdf

from pypdf import PdfReader

reader = PdfReader("report.pdf")

# Количество страниц
print(f"Страниц: {len(reader.pages)}")

# Извлекаем текст со всех страниц
all_text = []
for page in reader.pages:
    text = page.extract_text()
    if text:  # extract_text может вернуть None
        all_text.append(text)

full_text = "\n--- page break ---\n".join(all_text)

pypdf бесплатен и работает хорошо для текстовых PDF. Но у него есть ограничения: PDF-сканы (изображения) не обрабатывает, сложные макеты с колонками разбирает плохо.

#PyMuPDF (fitz)

import fitz  # PyMuPDF

doc = fitz.open("report.pdf")

all_text = []
for page_num in range(len(doc)):
    page = doc[page_num]
    text = page.get_text()
    all_text.append(text)

full_text = "\n--- page break ---\n".join(all_text)

# Можно получить метаданные
metadata = doc.metadata
print(f"Автор: {metadata.get('author')}")
print(f"Дата создания: {metadata.get('creationDate')}")

PyMuPDF быстрее и точнее pypdf, лучше работает со сложными макетами. Требует установки: pip install PyMuPDF.

#Метаданные из PDF

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

from pypdf import PdfReader

reader = PdfReader("report.pdf")
documents_with_metadata = []

for page_num, page in enumerate(reader.pages):
    text = page.extract_text()
    if text:
        documents_with_metadata.append({
            "text": text,
            "source": "report.pdf",
            "page": page_num + 1,
            "author": reader.metadata.get("/Author", "Unknown")
        })

print(f"Загружено {len(documents_with_metadata)} страниц")

#Загрузка DOCX-файлов

DOCX — формат Microsoft Word. Для чтения используем python-docx:

from docx import Document

doc = Document("contract.docx")

# Извлекаем текст из всех параграфов
paragraphs = []
for para in doc.paragraphs:
    if para.text.strip():  # Пропускаем пустые
        paragraphs.append(para.text)

full_text = "\n".join(paragraphs)
print(f"Параграфов: {len(paragraphs)}")

python-docx также даёт доступ к таблицам, заголовкам, колонтитулам:

# Таблицы
for table in doc.tables:
    for row in table.rows:
        cells = [cell.text for cell in row.cells]
        print(" | ".join(cells))

# Заголовки (стили Heading)
headings = [p.text for p in doc.paragraphs if p.style.name.startswith("Heading")]
print(f"Заголовки: {headings}")

Извлечение из таблиц особенно важно для RAG — контракты и отчёты часто содержат данные в таблицах.

#Загрузка CSV-файлов

CSV — табличные данные. Для чтения используем pandas:

import pandas as pd

df = pd.read_csv("data.csv", encoding="utf-8")

print(f"Строк: {len(df)}, Колонок: {len(df.columns)}")
print(f"Колонки: {list(df.columns)}")

# Просматриваем первые строки
print(df.head())

Для RAG важно превратить строки CSV в текстовые фрагменты. Вот типичный подход:

# Превращаем каждую строку в текстовый документ
documents = []
for idx, row in df.iterrows():
    text_parts = []
    for col in df.columns:
        text_parts.append(f"{col}: {row[col]}")
    doc_text = ", ".join(text_parts)
    documents.append({
        "text": doc_text,
        "source": "data.csv",
        "row": idx
    })

print(f"Создано {len(documents)} документов из CSV")

Каждая строка CSV становится отдельным документом с метаданными. Это удобно для поиска: «найди все записи, где статус = активный».

#Загрузка HTML-файлов

HTML — веб-страницы. Используем BeautifulSoup для извлечения текста из разметки:

from bs4 import BeautifulSoup

with open("page.html", "r", encoding="utf-8") as f:
    html_content = f.read()

soup = BeautifulSoup(html_content, "html.parser")

# Удаляем скрипты и стили
for script in soup(["script", "style"]):
    script.decompose()

# Получаем чистый текст
text = soup.get_text(separator="\n", strip=True)

print(f"Длина текста: {len(text)} символов")

separator="\n" добавляет переносы между элементами, strip=True убирает лишние пробелы. Без удаления script и style в текст попадёт JavaScript-код и CSS-стили.

#Извлечение конкретных элементов

Иногда нужен не весь текст, а определённые элементы:

# Только основной контент
main_content = soup.find("main") or soup.find("article") or soup.find("body")

# Заголовки
headings = soup.find_all(["h1", "h2", "h3"])
for h in headings:
    print(f"{h.name}: {h.get_text()}")

# Параграфы
paragraphs = main_content.find_all("p")
text = "\n".join(p.get_text() for p in paragraphs)

#Загрузка с OpenRouter API

Можно загрузить документы из интернета через HTTP:

import requests
from bs4 import BeautifulSoup

response = requests.get("https://example.com/article.html")
response.raise_for_status()  # Ошибка при неудачном запросе

soup = BeautifulSoup(response.text, "html.parser")
for script in soup(["script", "style"]):
    script.decompose()

text = soup.get_text(separator="\n", strip=True)
print(f"Загружено {len(text)} символов с URL")

#Универсальный загрузчик документов

На практике удобно создать единый интерфейс, который определяет формат по расширению:

from pathlib import Path
from pypdf import PdfReader
from docx import Document
import pandas as pd
from bs4 import BeautifulSoup

def load_document(filepath: str) -> list[dict]:
    """
    Загружает документ и возвращает список фрагментов с метаданными.
    Каждый фрагмент: {"text": str, "source": str, "metadata": dict}
    """
    path = Path(filepath)
    ext = path.suffix.lower()
    
    if ext == ".txt":
        with open(filepath, "r", encoding="utf-8") as f:
            text = f.read()
        return [{"text": text, "source": filepath, "metadata": {}}]
    
    elif ext == ".pdf":
        reader = PdfReader(filepath)
        documents = []
        for page_num, page in enumerate(reader.pages):
            text = page.extract_text()
            if text:
                documents.append({
                    "text": text,
                    "source": filepath,
                    "metadata": {"page": page_num + 1}
                })
        return documents
    
    elif ext == ".docx":
        doc = Document(filepath)
        text = "\n".join(p.text for p in doc.paragraphs if p.text.strip())
        return [{"text": text, "source": filepath, "metadata": {}}]
    
    elif ext == ".csv":
        df = pd.read_csv(filepath)
        documents = []
        for idx, row in df.iterrows():
            text_parts = [f"{col}: {row[col]}" for col in df.columns]
            documents.append({
                "text": ", ".join(text_parts),
                "source": filepath,
                "metadata": {"row": idx}
            })
        return documents
    
    elif ext in (".html", ".htm"):
        with open(filepath, "r", encoding="utf-8") as f:
            soup = BeautifulSoup(f.read(), "html.parser")
        for script in soup(["script", "style"]):
            script.decompose()
        text = soup.get_text(separator="\n", strip=True)
        return [{"text": text, "source": filepath, "metadata": {}}]
    
    else:
        raise ValueError(f"Неподдерживаемый формат: {ext}")

# Использование
docs = load_document("report.pdf")
print(f"Загружено {len(docs)} фрагментов из {docs[0]['source']}")

Такой унифицированный загрузчик — основа RAG-пайплайна. Он принимает любой файл и отдаёт структурированные фрагменты, готовые для chunking и индексации.

#Обработка ошибок

При загрузке реальных документов ошибки неизбежны: битые файлы, недоступные URL, неправильные кодировки. Обработка ошибок — важная часть:

def safe_load(filepath: str) -> list[dict] | None:
    """Загружает документ, возвращает None при ошибке."""
    try:
        return load_document(filepath)
    except FileNotFoundError:
        print(f"Файл не найден: {filepath}")
        return None
    except Exception as e:
        print(f"Ошибка загрузки {filepath}: {e}")
        return None

# Загрузка нескольких файлов
files = ["doc1.pdf", "doc2.docx", "missing.txt", "doc3.csv"]
all_docs = []

for f in files:
    result = safe_load(f)
    if result:
        all_docs.extend(result)

print(f"Успешно загружено {len(all_docs)} фрагментов из {len(files)} файлов")