Как вы обрабатываете смену форматов документов (legacy + новые форматы)?

Краткий тезис

Обработка смены форматов документов требует гибкой архитектуры, основанной на паттерне Strategy. Мы создаём абстрактный парсер с единым интерфейсом, регистрируем конкретные парсеры для каждого формата в реестре (Registry) и используем каскадную конвертацию для приведения к единой схеме (Unified Schema). Версионирование парсеров позволяет поддерживать legacy форматы без поломки новых, а fallback-механизмы обеспечивают отказоустойчивость.

1. Термин: Парсер (Parser) и его роль

Парсер — это компонент, который преобразует документ из исходного формата (PDF, DOCX, Markdown, TXT и т.д.) в структурированное представление (обычно текст + метаданные). В контексте RAG парсер — первый этап пайплайна индексации. От его качества зависит, насколько полно и точно будет извлечена информация.

2. Паттерн Strategy для парсинга

Паттерн Strategy (Стратегия) позволяет определить семейство алгоритмов (парсеров), инкапсулировать каждый из них и сделать их взаимозаменяемыми. Это идеальное решение для сценария, где форматы документов меняются со временем.

Преимущества

• Новый формат = новый класс парсера, без изменения существующего кода.
• Легко тестировать каждый парсер изолированно.
• Можно динамически выбирать стратегию на основе расширения файла или MIME-типа.

3. Абстрактный парсер и интерфейс parse()

Определяем базовый класс (или протокол) с методом parse(), который принимает путь к файлу или байты и возвращает унифицированный словарь.

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

class DocumentParser(ABC):
    @abstractmethod
    def parse(self, file_path: str) -> Dict[str, Any]:
        """
        Возвращает словарь с ключами:
        - 'text': str (извлечённый текст)
        - 'metadata': dict (автор, дата, заголовок и т.д.)
        - 'format_version': str (версия парсера)
        """
        pass

Конкретные реализации переопределяют parse() под свой формат.

4. Registry: реестр парсеров

Registry — это централизованное хранилище, которое сопоставляет идентификатор формата (расширение, MIME-тип) с экземпляром парсера. Реестр может быть реализован как словарь или через декораторы.

class ParserRegistry:
    _parsers: Dict[str, DocumentParser] = {}

    @classmethod
    def register(cls, extension: str, parser: DocumentParser):
        cls._parsers[extension] = parser

    @classmethod
    def get_parser(cls, extension: str) -> Optional[DocumentParser]:
        return cls._parsers.get(extension)

# Регистрация
ParserRegistry.register('.pdf', PDFParser())
ParserRegistry.register('.docx', DOCXParser())

При поступлении документа определяем его расширение и получаем нужный парсер. Если парсера нет — используем fallback.

5. Каскадная конвертация (PDF → текст → чанки)

Каскадная конвертация — это многоэтапное преобразование, где результат одного этапа подаётся на вход следующему. Например:

1. Парсинг формата → извлекаем сырой текст и метаданные.
2. Нормализация → очистка от лишних пробелов, управляющих символов, приведение кодировки.
3. Разбиение на чанки (chunking) → нарезка текста на фрагменты фиксированной длины или по семантическим границам.

Каждый этап может быть реализован как отдельный компонент, что упрощает замену или добавление шагов (например, OCR для сканов).

[PDF] → PDFParser → сырой текст → Normalizer → чистый текст → Chunker → чанки

6. Unified Schema: единая схема для всех форматов

Unified Schema — это стандартизированная структура данных, в которую преобразуются все документы независимо от исходного формата. Она включает:

• text: строка с извлечённым содержимым.
• metadata: словарь с полями source, author, created_at, format, version.
• chunks: список чанков (опционально, если чанкинг выполняется на этом этапе).

schema|Единая схема упрощает последующую обработку (эмбеддинг, индексацию) и позволяет легко менять парсеры без влияния на downstream-компоненты.

7. Версионирование парсеров

Версионирование парсеров необходимо, когда:

• Меняется логика извлечения (например, новый способ обработки таблиц).
• Появляется новый формат, а старые документы должны обрабатываться по-прежнему.
• Нужно откатить изменения при ошибках.

Реализуем через поле version в классе парсера и храним версию в метаданных результата. В реестре можно хранить несколько версий для одного формата и выбирать по дате документа или конфигурации.

class PDFParser(DocumentParser):
    VERSION = "2.0"

    def parse(self, file_path: str) -> Dict[str, Any]:
        # ...
        return {"text": text, "metadata": {"format_version": self.VERSION}}

При индексации legacy-документов можно явно указать, какую версию парсера использовать (например, через конфиг legacy_parsers).

8. Обработка ошибок и fallback

Документы могут быть повреждены, иметь неожиданную структуру или требовать OCR. Архитектура должна предусматривать:

• Try-except внутри каждого парсера с логированием ошибки.
• Fallback-парсер (например, TXT-парсер, который читает файл как text|plain text).
• Graceful degradation: если парсер не справился, документ помечается как partially_parsed и отправляется на ручную обработку или повторную попытку с другой версией.

def parse_document(file_path: str) -> Dict[str, Any]:
    ext = Path(file_path).suffix
    parser = ParserRegistry.get_parser(ext)
    if not parser:
        parser = FallbackParser()  # читает как текст
    try:
        return parser.parse(file_path)
    except Exception as e:
        logger.error(f"Parse failed for {file_path}: {e}")
        return {"text": "", "metadata": {"error": str(e)}, "format_version": "fallback"}

9. Пример реализации на Python

Ниже — минимальный рабочий пример системы парсинга с реестром и каскадной конвертацией.

from pathlib import Path
from typing import Dict, Any

class TXTParser(DocumentParser):
    def parse(self, file_path: str) -> Dict[str, Any]:
        with open(file_path, 'r', encoding='utf-8') as f:
            text = f.read()
        return {"text": text, "metadata": {"format": "txt"}, "format_version": "1.0"}

class PDFParser(DocumentParser):
    def parse(self, file_path: str) -> Dict[str, Any]:
        # Имитация парсинга PDF
        text = "PDF content placeholder"
        return {"text": text, "metadata": {"format": "pdf"}, "format_version": "2.0"}

# Регистрация
ParserRegistry.register('.txt', TXTParser())
ParserRegistry.register('.pdf', PDFParser())

# Использование
def process_document(file_path: str) -> Dict[str, Any]:
    ext = Path(file_path).suffix
    parser = ParserRegistry.get_parser(ext)
    if not parser:
        raise ValueError(f"No parser for {ext}")
    raw = parser.parse(file_path)
    # Каскад: нормализация + чанкинг (упрощённо)
    normalized_text = raw['text'].strip()
    return {"text": normalized_text, "metadata": raw['metadata']}

10. Интеграция с RAG-пайплайном

В RAG-системе парсеры встраиваются в этап Document Loader или Ingestion Pipeline. После парсинга и унификации данные отправляются в сплиттер (chunker), затем в эмбеддер и векторную БД.

Рекомендации

• Парсеры должны быть stateless (не хранить состояние между вызовами) для горизонтального масштабирования.
• Метаданные о версии парсера сохраняются в векторной БД, чтобы при переиндексации можно было перезапустить только документы, обработанные старой версией.
• Для потоковой обработки (real-time) парсеры должны быть легковесными.

11. Тестирование и мониторинг

• Юнит-тесты для каждого парсера с эталонными файлами.
• Интеграционные тесты на каскадную конвертацию.
• Мониторинг доли успешных парсингов, времени обработки, размера извлечённого текста.
• Alerting при падении процента успешных парсингов ниже порога.

Пет-проект для закрепления

Задача Разработать систему парсинга документов для RAG, поддерживающую PDF, DOCX, Markdown и legacy TXT.

Инструменты Python, PyMuPDF (fitz), python-docx, markdown, pytest.

Шаги:

1. Создать абстрактный класс DocumentParser с методом parse().
2. Реализовать парсеры для PDF, DOCX, Markdown, TXT.
3. Реализовать ParserRegistry с декоратором @register('.pdf').
4. Добавить каскадную нормализацию (удаление лишних пробелов, приведение к UTF-8).
5. Написать функцию process_document(), которая выбирает парсер, парсит, нормализует и возвращает унифицированный словарь.
6. Добавить версионирование: каждому парсеру присвоить версию, сохранять её в метаданных.
7. Написать тесты для каждого парсера с образцами файлов.

Ожидаемый результат Унифицированный вывод для любого формата; возможность добавить новый парсер (например, для HTML) без изменения существующего кода; логирование ошибок и fallback на TXT-парсер.

Связь с другими вопросами

Навигация

• Предыдущий: 84
• Следующий: 86
• Индекс: 00. Индекс разборов