Индексация документов для RAG (PDF, DOCX, HTML, Markdown)

Индексация документов для RAG (PDF, DOCX, HTML, Markdown)

Индексация документов — первый и критически важный этап RAG-системы. Качество поиска зависит от качества парсинга: потерянные таблицы, слитый текст из колонок, неправильно распознанные заголовки — всё это деградирует релевантность ответов.

Парсинг различных форматов

from pathlib import Path
from dataclasses import dataclass

@dataclass
class ParsedDocument:
    text: str
    metadata: dict
    source_format: str
    page_count: int = None

class DocumentParser:
    def parse(self, file_path: str) -> ParsedDocument:
        path = Path(file_path)
        ext = path.suffix.lower()

        if ext == '.pdf':
            return self._parse_pdf(file_path)
        elif ext in ['.docx', '.doc']:
            return self._parse_docx(file_path)
        elif ext in ['.html', '.htm']:
            return self._parse_html(file_path)
        elif ext in ['.md', '.markdown']:
            return self._parse_markdown(file_path)
        else:
            raise ValueError(f"Unsupported format: {ext}")

    def _parse_pdf(self, path: str) -> ParsedDocument:
        # Для сложных PDF (с таблицами, колонками) — pdfplumber
        import pdfplumber
        with pdfplumber.open(path) as pdf:
            pages_text = []
            for page in pdf.pages:
                # Сохранение таблиц как markdown
                tables = page.extract_tables()
                text = page.extract_text() or ""

                for table in tables:
                    table_md = self._table_to_markdown(table)
                    text += f"\n\n{table_md}\n\n"

                pages_text.append(text)

        full_text = "\n\n---PAGE BREAK---\n\n".join(pages_text)
        return ParsedDocument(
            text=full_text,
            metadata={"source": path, "pages": len(pdf.pages)},
            source_format="pdf",
            page_count=len(pdf.pages)
        )

    def _parse_docx(self, path: str) -> ParsedDocument:
        from docx import Document
        doc = Document(path)

        elements = []
        for element in doc.element.body:
            if element.tag.endswith('p'):  # Параграф
                para = element
                style = para.style.name if hasattr(para, 'style') else ''
                text = element.text_content()
                if style.startswith('Heading'):
                    level = int(style.split()[-1]) if style[-1].isdigit() else 1
                    elements.append('#' * level + ' ' + text)
                elif text.strip():
                    elements.append(text)
            elif element.tag.endswith('tbl'):  # Таблица
                table = self._extract_table_from_docx(element)
                elements.append(table)

        return ParsedDocument(
            text='\n\n'.join(elements),
            metadata={"source": path},
            source_format="docx"
        )

    def _parse_html(self, path: str) -> ParsedDocument:
        from bs4 import BeautifulSoup
        with open(path, 'r', encoding='utf-8') as f:
            soup = BeautifulSoup(f.read(), 'html.parser')

        # Удаление скриптов и стилей
        for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
            tag.decompose()

        # Извлечение структурированного текста
        from markdownify import markdownify
        text = markdownify(str(soup), heading_style="ATX")

        return ParsedDocument(
            text=text,
            metadata={"source": path, "title": soup.title.string if soup.title else ""},
            source_format="html"
        )

Структурированное извлечение метаданных

class MetadataExtractor:
    def extract(self, doc: ParsedDocument) -> dict:
        metadata = doc.metadata.copy()

        # Извлечение заголовков для навигации
        headers = re.findall(r'^#{1,3}\s+(.+)$', doc.text, re.MULTILINE)
        metadata['headers'] = headers[:20]  # Первые 20 заголовков

        # Извлечение дат
        date_pattern = r'\b\d{1,2}[./]\d{1,2}[./]\d{2,4}\b'
        dates = re.findall(date_pattern, doc.text)
        if dates:
            metadata['dates_mentioned'] = dates[:5]

        # Язык документа
        from langdetect import detect
        try:
            metadata['language'] = detect(doc.text[:1000])
        except Exception:
            metadata['language'] = 'unknown'

        return metadata

Подготовка к индексации

После парсинга документы чанкируются (разбиваются на фрагменты), эмбеддируются и загружаются в векторную БД. Ключевой момент: сохранение структурных маркеров (заголовки, номера страниц) в метаданных чанков для обеспечения атрибуции источника в ответах RAG.

Для 1000-страничного PDF полный цикл (парсинг → чанкинг → эмбеддинг → индексация): 5-15 минут при использовании OpenAI Embeddings API.