Чтение PDF, DOCX, TXT, CSV и других форматов документов для RAG-системы
RAG-система настолько хороша, насколько хороши документы, которые в неё загружены.
В реальных RAG-системах документы приходят в разных форматах: PDF-отчёты, DOCX-договоры, CSV-таблицы, HTML-страницы, простые 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 — сложный формат: текст может быть в колонках, таблицах, с изображениями. Для извлечения текста есть несколько библиотек.
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-сканы (изображения) не обрабатывает, сложные макеты с колонками разбирает плохо.
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.
При загрузке полезно сохранять метаданные — номер страницы, источник, автора. Это поможет при фильтрации и цитировании:
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 — формат 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 — табличные данные. Для чтения используем 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 — веб-страницы. Используем 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)Можно загрузить документы из интернета через 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)} файлов")pypdf для простых случаев, PyMuPDF для сложных макетовpython-docx для извлечения текста, таблиц, заголовковpandas для чтения, превращение строк в текстовые документыBeautifulSoup для очистки от скриптов и извлечения текстаВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.