Перейти к содержанию

Большие документы без RAG

Шаг 1

В предыдущем разделе мы собрали первую версию DocumentAIService. Она безопасно проверяет файл, извлекает данные и передаёт модели ограниченный контекст и для небольшого PDF, DOCX или Markdown этого достаточно, ведь весь подготовленный текст помещается в один запрос, а результат возвращается как DocumentAIResult.

Но в DocumentContextPreparer намеренно установлен предел в 24 000 символов. Если подготовленный текст длиннее, сервис не обрезает его, а возвращает DocumentContextTooLargeError. И в этом шаге давайте заменим этот отказ на контролируемую обработку документа по частям.

Почему нельзя просто увеличить лимит?

Напомню, что у модели есть контекстное окно, это максимальный объём данных, который она может учитывать в одном запросе. В него входят не только данные файла, но и developer-инструкция, вопрос пользователя, список ссылок, служебная разметка и место, необходимое для ответа.

Даже если выбранная модель технически поддерживает большой контекст, отправлять любой документ целиком всё равно не следует. Чем больше вход, тем выше стоимость и время ответа. Кроме того, важный фрагмент может потеряться среди большого количества второстепенного текста.

Как заранее оценить размер?

Точное количество токенов зависит от модели и токенизатора. В текущем проекте мы используем более простую предварительную оценку — количество символов. Экстрактор уже сохраняет его в document.metadata.character_count, а DocumentContextPreparer дополнительно измеряет окончательный текст вместе с маркерами [source:...].

Конечно количество символов не равно количеству токенов, но пока нам так проще определить границу. Ведь она позволяет принять решение до обращения к провайдеру:

В предыдущих модулях мы уже использовали TokenCounter, который приблизительно оценивает количество токенов через библиотеку tiktoken. Этот механизм можно переиспользовать и для документов. Однако сейчас оставим ограничение по символам, чтобы сосредоточиться на разделении документа и объединении результатов. Позднее вместо len(content) можно будет вызвать счётчик токенов и сравнить результат с отдельным бюджетом документного контекста. Общая последовательность не изменится, ведь приложение сначала оценивает размер подготовленных данных, а затем выбирает один запрос или обработку по частям.

Какие стратегии существуют?

Для большого документа можно использовать несколько подходов.

  • Явный выбор частей. Пользователь или прикладной сценарий заранее указывает нужные страницы либо разделы. Например, только section:3 и section:7.
  • Последовательная суммаризация. Модель читает части по очереди, а результат каждого шага передаётся в следующий запрос.
  • Map-reduce. Сначала каждая часть анализируется независимо, затем отдельный запрос объединяет промежуточные результаты.
  • RAG. Фрагменты индексируются, а для вопроса извлекаются наиболее подходящие. Этот подход рассмотрим в другом курсе "AI Engineer на Python: RAG, агенты, MCP и production-системы", который на данный момент в разработке.

В этом разделе разберём первые три идеи без векторной базы и поиска по эмбеддингам. В коде реализуем явный выбор частей и map-reduce, а последовательную суммаризацию рассмотрим как альтернативу и сравним её свойства. Если после явного выбора текст небольшой, он по-прежнему пойдёт одним запросом.

И давайте сразу обсудим важный момент, а обязательно ли использовать RAG?

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

В небольшом проекте вполне нормально использовать более простой сценарий, где мы можем разделить документ на части, обработать каждую часть отдельно и затем объединить результаты. Если нужные страницы или разделы известны заранее, можно передать только их. Такой подход не требует векторной базы, эмбеддингов, индексации и отдельной логики поиска. Главное это контролировать размер частей, количество запросов и возможную потерю информации при объединении.

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

Перейдем к практике и первым делом опишем политику разбиения. В каталоге app/services создайте файл document_chunking.py. Добавьте импорты, ошибки и новые объекты:

from dataclasses import dataclass, replace

from app.files.documents.extraction import (
    DocumentPart,
    ExtractedDocument,
)
from app.files.validation import FileKind
from app.services.document_ai_context import (
    DocumentAIContextError,
    PreparedFileContext,
)

class DocumentChunkingError(DocumentAIContextError):
    pass

class DocumentSelectionError(DocumentChunkingError):
    pass

class TooManyDocumentChunksError(DocumentChunkingError):
    pass

@dataclass(frozen=True)
class DocumentChunkingPolicy:
    max_chunk_characters: int = 12_000
    max_chunks: int = 8

    def __post_init__(self) -> None:
        if self.max_chunk_characters <= 0:
            raise ValueError(
                "max_chunk_characters должен быть больше нуля."
            )
        if self.max_chunks <= 0:
            raise ValueError("max_chunks должен быть больше нуля.")

@dataclass(frozen=True)
class PreparedDocumentBatch:
    source_name: str
    source_kind: FileKind
    chunks: tuple[PreparedFileContext, ...]
    references: tuple[str, ...]
    warnings: tuple[str, ...]

DocumentChunkingPolicy задаёт две независимые границы. max_chunk_characters ограничивает размер одной части, передаваемой модели. max_chunks ограничивает количество частей, а значит — число промежуточных запросов, стоимость и время обработки.

Значение 12 000 символов меньше предела контекста целого документа. Это оставляет место для developer-инструкции, вопроса, ссылок и структурированного ответа. Значение 8 для количества частей является учебным ограничением: слишком большой файл лучше остановить явно, чем незаметно создать десятки дорогостоящих запросов.

PreparedDocumentBatch представляет уже подготовленный пакет. Каждая его часть имеет знакомый тип PreparedFileContext, поэтому существующий DocumentContentAnalyzer сможет анализировать её без нового формата входа. На уровне пакета сохраняются имя файла, его тип, все допустимые ссылки и общие предупреждения.

Пока мы только описали контракт и ограничения. В следующем шаге мы реализуем DocumentChunkPreparer и он сможет выбрать части по явным ссылкам, сохранить порядок документа и сформировать фрагменты без потери текста.

Шаг 2

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

В разделе 4.2 «Извлечение текста из документов» мы уже научились разбивать содержимое файла на логические части. Каждый экстрактор возвращает объект ExtractedDocument, в котором хранится полный текст документа и список parts.

Элементы этого списка представлены объектами DocumentPart. Для PDF один такой объект обычно соответствует отдельной странице, а для Markdown и DOCX — разделу, выделенному по заголовку. Каждая часть получает собственную ссылку на исходное место, например page:2 или section:3.

Поэтому сейчас нам не нужно повторно извлекать текст или заново создавать DocumentPart. Мы возьмём готовые части из document.parts и будем объединять соседние страницы или разделы, пока получившийся фрагмент помещается в установленное ограничение. Если одна страница или один раздел окажется слишком большим, только тогда дополнительно разделим его текст на несколько фрагментов.

Кроме разбиения большого документа, мы реализуем возможность заранее выбрать страницы или разделы, которые нужно проанализировать. Это пригодится, если пользователь укажет нужные страницы в интерфейсе или приложение само определит их по правилам сценария.

Каждый объект DocumentPart уже содержит ссылку на своё место в документе. Для PDF используются ссылки page:2, page:3, а для Markdown и DOCX — section:4, section:7.

Наша логика будет работать следующим образом:

Если ссылки не переданы, приложение продолжит работать со всем документом. Если хотя бы одной ссылки в документе нет, обработка завершится ошибкой до обращения к ProxyAPI. Модель здесь ничего не ищет и не выбирает: это обычная проверка точных значений в Python, а не семантический поиск.

Перед тем как переходить к коду, разберём логику, которую нам предстоит реализовать. Объекты DocumentPart у нас уже есть и экстракторы создают их для отдельных страниц и разделов. Теперь между DocumentPart и готовым PreparedFileContext появится промежуточный уровень — внутренние блоки. Так мы будем называть небольшие фрагменты текста вместе со ссылкой [source:...]. Отдельный класс для блока создавать не будем: в коде он будет представлен кортежем из текста и ссылок.

Именно такую последовательность мы сейчас реализуем. Будущий метод _split_part() будет превращать каждый DocumentPart в один или несколько внутренних блоков. После этого метод prepare() будет объединять соседние блоки, пока они помещаются в установленное ограничение, и создавать из них объекты PreparedFileContext. В конце готовые контексты будут собраны в один PreparedDocumentBatch.

Перейдем к практике и продолжайте писать в app/services/document_chunking.py. После PreparedDocumentBatch добавьте класс:

class DocumentChunkPreparer:
    def __init__(
        self,
        policy: DocumentChunkingPolicy | None = None,
    ) -> None:
        self._policy = policy or DocumentChunkingPolicy()

    def select(
        self,
        document: ExtractedDocument,
        references: tuple[str, ...] | None,
    ) -> ExtractedDocument:
        if references is None:
            return document

        requested = tuple(
            dict.fromkeys(value.strip() for value in references)
        )
        if not requested or any(not value for value in requested):
            raise DocumentSelectionError(
                "Список ссылок документа не должен быть пустым."
            )

        available = {part.reference for part in document.parts}
        unknown = sorted(set(requested) - available)
        if unknown:
            raise DocumentSelectionError(
                "В документе нет указанных частей: "
                + ", ".join(unknown)
            )

        requested_set = set(requested)
        selected = tuple(
            part
            for part in document.parts
            if part.reference in requested_set
        )
        text = "\n\n".join(
            self._render_part_text(part) for part in selected
        ).strip()
        warnings = document.warnings
        if len(selected) < len(document.parts):
            warnings += (
                "Для анализа выбраны не все части документа.",
            )

        return replace(
            document,
            text=text,
            parts=selected,
            metadata=replace(
                document.metadata,
                part_count=len(selected),
                character_count=len(text),
            ),
            warnings=tuple(dict.fromkeys(warnings)),
        )

    def prepare(
        self,
        document: ExtractedDocument,
    ) -> PreparedDocumentBatch:
        blocks: list[tuple[str, tuple[str, ...]]] = []

        for part in document.parts:
            blocks.extend(self._split_part(part))

        chunks: list[PreparedFileContext] = []
        current_blocks: list[str] = []
        current_references: list[str] = []

        def flush_chunk() -> None:
            if not current_blocks:
                return

            chunks.append(
                PreparedFileContext(
                    source_name=document.metadata.original_name,
                    source_kind=document.metadata.kind,
                    content="\n\n".join(current_blocks),
                    references=tuple(
                        dict.fromkeys(current_references)
                    ),
                    warnings=(),
                )
            )
            current_blocks.clear()
            current_references.clear()

        for block, references in blocks:
            candidate = "\n\n".join((*current_blocks, block))
            if (
                current_blocks
                and len(candidate) > self._policy.max_chunk_characters
            ):
                flush_chunk()

            current_blocks.append(block)
            current_references.extend(references)

        flush_chunk()

        if not chunks:
            raise DocumentChunkingError(
                "Не удалось подготовить части документа."
            )
        if len(chunks) > self._policy.max_chunks:
            raise TooManyDocumentChunksError(
                "Документ требует слишком много запросов к модели."
            )

        references = tuple(
            dict.fromkeys(
                reference
                for chunk in chunks
                for reference in chunk.references
            )
        )
        warnings = document.warnings + (
            f"Документ обработан по частям: {len(chunks)}.",
        )

        return PreparedDocumentBatch(
            source_name=document.metadata.original_name,
            source_kind=document.metadata.kind,
            chunks=tuple(chunks),
            references=references,
            warnings=tuple(dict.fromkeys(warnings)),
        )

    def _split_part(
        self,
        part: DocumentPart,
    ) -> list[tuple[str, tuple[str, ...]]]:
        prefix = f"[source:{part.reference}]\n"
        text = self._render_part_text(part)
        available = self._policy.max_chunk_characters - len(prefix)
        if available <= 0:
            raise DocumentChunkingError(
                "Лимит части слишком мал для ссылки на источник."
            )

        pieces = self._split_text(text, available)
        return [
            (prefix + piece, (part.reference,))
            for piece in pieces
        ]

    @staticmethod
    def _render_part_text(part: DocumentPart) -> str:
        values: list[str] = []
        if part.section_title is not None:
            values.append(part.section_title)
        if part.text:
            values.append(part.text)
        return "\n".join(values).strip()

    @staticmethod
    def _split_text(text: str, limit: int) -> list[str]:
        pieces: list[str] = []
        remaining = text.strip()

        while len(remaining) > limit:
            cut = remaining.rfind("\n\n", 0, limit + 1)
            if cut < limit // 2:
                cut = remaining.rfind("\n", 0, limit + 1)
            if cut < limit // 2:
                cut = remaining.rfind(" ", 0, limit + 1)
            if cut <= 0:
                cut = limit

            pieces.append(remaining[:cut].strip())
            remaining = remaining[cut:].strip()

        if remaining:
            pieces.append(remaining)

        return pieces

Данный конструктор сохраняет переданную политику разбиения, а если её нет — создаёт DocumentChunkingPolicy со стандартными ограничениями.

Метод select() отвечает только за явный выбор страниц или разделов. При references=None он возвращает весь документ. В остальных случаях ссылки очищаются от пробелов, повторы удаляются, а неизвестные значения приводят к DocumentSelectionError. Выбранные части собираются в порядке исходного документа.

С помощью replace() метод создаёт новую версию ExtractedDocument с выбранными parts, заново собранным текстом и обновлёнными метаданными. Исходный объект при этом не изменяется.

Метод prepare() выполняет само разбиение. Сначала _split_part() превращает каждый DocumentPart в один или несколько внутренних блоков. Затем соседние блоки объединяются в готовые контексты, пока их общая длина не превышает max_chunk_characters.

current_blocks и current_references хранят содержимое текущего фрагмента. Переменная candidate показывает его размер после добавления очередного блока. Если лимит будет превышен, flush_chunk() сохраняет накопленные данные в PreparedFileContext, очищает временные списки, а новый блок начинает следующий фрагмент.

После цикла flush_chunk() сохраняет последний фрагмент. Затем метод проверяет, что результат не пуст и количество контекстов не превышает max_chunks. Это ограничение не позволяет одному файлу неожиданно вызвать слишком много обращений к модели. Готовые контексты, ссылки и предупреждения возвращаются внутри PreparedDocumentBatch.

Метод _split_part() добавляет к тексту маркер [source:...]. Если страница или раздел вместе с этим маркером не помещается в лимит, _split_text() делит текст на несколько блоков. Граница ищется сначала между абзацами, затем между строками и словами. Только если подходящего места нет, используется точная позиция лимита.

один большой section:5
→ блок 1 [source:section:5]
→ блок 2 [source:section:5]
→ блок 3 [source:section:5]

Остаток текста не удаляется, а переносится в следующий блок. Все блоки сохраняют ссылку исходного DocumentPart. При этом очень длинный абзац всё равно может быть разделён не по смысловой границе — это ограничение выбранного подхода.

Теперь большой документ можно превратить в контролируемый пакет контекстов, однако запросы к модели пока не выполняются. В следующем шаге мы сравним последовательную суммаризацию и map-reduce, а затем начнём анализировать каждую часть с помощью уже существующего DocumentContentAnalyzer.

Шаг 3

В предыдущем шаге мы научились превращать большой документ в несколько объектов PreparedFileContext. Каждый такой объект содержит текст отдельного фрагмента и ссылки на его страницы или разделы. Однако пользователю нужен не набор отдельных ответов, а один результат по всему документу.

Значит, теперь необходимо решить две задачи: отдельно проанализировать каждый подготовленный фрагмент, а затем объединить найденные сведения в общий ответ. Сначала сравним два возможных способа, после чего реализуем выбранный вариант.

Первый способ — последовательная суммаризация. Модель получает первый фрагмент и составляет его краткое резюме. Затем это резюме отправляется вместе со вторым фрагментом, после чего модель создаёт новую сокращённую версию. Так процесс продолжается до конца документа.

Такой подход экономит место, ведь на каждом шаге хранится только текущее резюме. Он также учитывает порядок документа. Но у него есть важный недостаток — каждая следующая версия строится уже не по исходным предыдущим частям, а по их сокращённому описанию. Факт, потерянный на раннем шаге, нельзя восстановить в конце.

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

Но для нашего сценария выберем подход MapReduce. На этапе Map каждый фрагмент документа анализируется отдельно, но с учетом одного и того же вопроса пользователя. В результате для каждого фрагмента формируется собственный структурированный DocumentAnalysis. Затем наступает этап Reduce, это когда модель получает все промежуточные результаты, объединяет найденную информацию и на их основе формирует один итоговый ответ.

Каждая map-операция видит исходный текст своей части, поэтому раннее резюме не влияет на анализ поздних частей. Отдельные операции логически независимы. В текущем синхронном приложении мы вызываем их по очереди, но в будущем такую работу можно распараллелить с отдельным ограничением конкурентности.

Цена данного подхода, это большее количество запросов. Для документа из трёх частей потребуется три map-запроса и ещё один reduce-запрос. Поэтому ранее мы ограничили количество частей через max_chunks.

Перед реализацией map-этапа усилим контракт структурированного ответа. Сейчас поле DocumentFact.reference имеет обычный тип str. Pydantic проверяет, что модель вернула строку, а метод _parse_result() уже после ответа сравнивает её со списком доступных ссылок.

Для небольшого контекста этого часто достаточно, но при обработке нескольких фрагментов модель может перепутать номера разделов. Например, во втором фрагменте доступны section:7–section:12, а модель способна заново пронумеровать их и вернуть section:1–section:6. Python правильно отклонит такой результат, но весь map-reduce завершится ошибкой.

Сделаем ограничение частью JSON Schema самого запроса. Для каждого PreparedFileContext будем добавлять в поле reference свой enum — список единственных значений, которые разрешено вернуть модели.

Откройте app/services/document_content_analyzer.py и добавьте JsonSchema в существующий импорт из LLM-клиента:

from app.llm.client import LLMClient, LLMResult, Message, JsonSchema

Строку со статической схемой теперь удалите:

DOCUMENT_ANALYSIS_SCHEMA = DocumentAnalysis.model_json_schema()

Она больше не подходит, потому что разные запросы содержат разные ссылки. Вместо одной общей схемы будем создавать новую непосредственно перед каждым обращением к модели.

В методе analyze() найдите вызов generate_structured() и замените значение аргумента schema:

llm_result = self._llm_client.generate_structured(
    messages=messages,
    schema_name=DOCUMENT_ANALYSIS_SCHEMA_NAME,
    schema=self._build_schema(context.references),
)

Теперь перед _parse_result() добавьте новый статический метод:

    @staticmethod
    def _build_schema(
        available_references: tuple[str, ...],
    ) -> JsonSchema:
        schema = DocumentAnalysis.model_json_schema()
        fact_schema = schema["$defs"]["DocumentFact"]
        reference_schema = fact_schema["properties"]["reference"]
        reference_schema["enum"] = list(available_references)
        return schema

Pydantic создаёт новую JSON Schema, а код находит в $defs описание вложенной модели DocumentFact. В правила поля reference добавляется enum со ссылками текущего контекста. Для каждого map-запроса этот список будет своим, а для reduce-запроса в него войдут ссылки выбранного документа.

Проверку в _parse_result() не удаляем. Ограничение в структурированных ответах помогают модели сформировать правильный результат, а Python повторно защищает границу приложения на случай ошибки или неполной поддержки схемы провайдером.

Теперь перейдём к следующим изменениям в проекте. И сначала научим результат показывать реальное количество обращений к модели. Для небольшого документа один DocumentAIResult по-прежнему соответствует одному вызову. При map-reduce вызовов будет несколько, поэтому откройте app/services/document_content_analyzer.py и в конец DocumentAIResult, после response_id, добавьте поле:

    model_calls: int = 1

Значение по умолчанию сохраняет прежнее поведение: обычный анализатор создаёт результат одного вызова и его код менять не нужно. Позднее LargeDocumentAnalyzer заменит это значение общим количеством map- и reduce-запросов.

Теперь создайте файл app/services/large_document_analyzer.py. В нём будет находиться отдельный сервис, управляющий обоими этапами анализа большого документа. Добавьте следующий код:

from collections.abc import Iterable
from dataclasses import dataclass, replace
from time import perf_counter

from app.services.document_ai_context import (
    DocumentAIContextError,
    PreparedFileContext,
)
from app.services.document_chunking import PreparedDocumentBatch
from app.services.document_content_analyzer import (
    DocumentAIResult,
    DocumentContentAnalyzer,
)

@dataclass(frozen=True)
class LargeDocumentAnalysisPolicy:
    max_reduce_characters: int = 32_000

    def __post_init__(self) -> None:
        if self.max_reduce_characters <= 0:
            raise ValueError(
                "max_reduce_characters должен быть больше нуля."
            )

class LargeDocumentAnalyzer:
    def __init__(
        self,
        content_analyzer: DocumentContentAnalyzer,
        policy: LargeDocumentAnalysisPolicy | None = None,
    ) -> None:
        self._content_analyzer = content_analyzer
        self._policy = policy or LargeDocumentAnalysisPolicy()

    def analyze(
        self,
        question: str,
        batch: PreparedDocumentBatch,
    ) -> DocumentAIResult:
        started_at = perf_counter()
        partial_results = tuple(
            self._content_analyzer.analyze(question, chunk)
            for chunk in batch.chunks
        )
        reduce_context = self._build_reduce_context(
            batch,
            partial_results,
        )
        final_result = self._content_analyzer.analyze(
            question,
            reduce_context,
        )

        all_results = (*partial_results, final_result)
        warnings = tuple(
            dict.fromkeys(
                (
                    *batch.warnings,
                    *(
                        warning
                        for result in all_results
                        for warning in result.warnings
                    ),
                )
            )
        )

        return replace(
            final_result,
            source_name=batch.source_name,
            source_kind=batch.source_kind,
            warnings=warnings,
            elapsed_seconds=perf_counter() - started_at,
            prompt_tokens=self._sum_optional(
                result.prompt_tokens for result in all_results
            ),
            completion_tokens=self._sum_optional(
                result.completion_tokens for result in all_results
            ),
            total_tokens=self._sum_optional(
                result.total_tokens for result in all_results
            ),
            model_calls=len(all_results),
        )

LargeDocumentAnalysisPolicy хранит ограничение для финального этапа. Здесь мы измеряем не исходный документ и не отдельный фрагмент, а общий текст промежуточных ответов, который будет передан reduce-запросу. Если эти ответы окажутся слишком объёмными, приложение должно остановиться, а не отправлять модели неконтролируемый контекст.

Метод analyze() управляет всей последовательностью. Сначала он запоминает время начала работы. Затем генератор проходит по batch.chunks и передаёт каждый фрагмент существующему DocumentContentAnalyzer. Поэтому нам не приходится повторять код обращения к модели, то есть каждый промежуточный ответ проходит структурированные ответы, проверку Pydantic и проверку ссылок относительно своего фрагмента.

Все промежуточные ответы сохраняются в partial_results. Затем будущий метод _build_reduce_context() превратит их в один ограниченный контекст, и тот же DocumentContentAnalyzer выполнит финальный reduce-запрос. После этого replace() берёт готовый результат и заменяет метаданные, которые теперь относятся не к одному вызову, а ко всей операции:

  • предупреждения собираются без повторов
  • время измеряется для всей операции
  • токены суммируются по всем ответам провайдера
  • model_calls содержит общее число вызовов

Пока класс ещё не закончен, ведь в нём отсутствуют методы подготовки reduce-контекста и суммирования необязательных значений. Поэтому запускать новый сценарий рано. В следующем шаге добавим эти методы и подробно разберём, какие данные можно безопасно передать на этап объединения.

Шаг 4

После map-этапа у нас есть несколько объектов DocumentAIResult — по одному для каждого фрагмента документа. Показывать их пользователю как окончательный результат нельзя, ведь каждый ответ знает только свою часть, сведения могут повторяться, а общий вывод ещё не сформирован.

Поэтому добавим reduce-этап. Он соберёт важные данные из промежуточных ответов, подготовит из них новый ограниченный контекст и ещё раз обратится к модели, чтобы получить один результат по всему документу.

Исходные фрагменты повторно отправлять нельзя: если снова объединить весь текст, мы вернёмся к проблеме слишком большого контекста. В финальный запрос передадим не исходный документ, а компактное представление каждого промежуточного результата:

  • краткое содержание части
  • ответ на вопрос по этой части
  • ключевые факты вместе со ссылками
  • ограничения
  • признак необходимости проверки человеком

Получается двухуровневое сокращение:

Промежуточный результат создала модель, поэтому он не становится новым доказанным источником. Он только переносит найденные сведения на следующий этап. Именно поэтому ссылки нельзя удалять, ведь итоговый факт должен по-прежнему указывать на исходную страницу или секцию, а не на абстрактный partial:2.

Перейдём к реализации. Продолжайте писать класс LargeDocumentAnalyzer в файле app/services/large_document_analyzer.py. Сразу после метода analyze() добавьте три вспомогательных метода:

    def _build_reduce_context(
        self,
        batch: PreparedDocumentBatch,
        results: tuple[DocumentAIResult, ...],
    ) -> PreparedFileContext:
        blocks = [
            self._render_partial(index, result)
            for index, result in enumerate(results, start=1)
        ]
        content = "\n\n".join(blocks)

        if len(content) > self._policy.max_reduce_characters:
            raise DocumentAIContextError(
                "Промежуточные результаты слишком велики "
                "для объединения."
            )

        return PreparedFileContext(
            source_name=batch.source_name,
            source_kind=batch.source_kind,
            content=content,
            references=batch.references,
            warnings=batch.warnings,
        )

    @staticmethod
    def _render_partial(
        index: int,
        result: DocumentAIResult,
    ) -> str:
        analysis = result.analysis
        fact_lines = "\n".join(
            f"- {fact.text} [{fact.reference}]"
            for fact in analysis.key_facts
        ) or "- нет"
        limitation_lines = "\n".join(
            f"- {limitation}"
            for limitation in analysis.limitations
        ) or "- нет"

        return (
            f"[partial:{index}]\n"
            f"Краткое содержание: {analysis.summary}\n"
            f"Ответ по части: {analysis.answer}\n"
            f"Ключевые факты:\n{fact_lines}\n"
            f"Ограничения:\n{limitation_lines}\n"
            "Требуется проверка человеком: "
            f"{analysis.requires_human_review}"
        )

    @staticmethod
    def _sum_optional(
        values: Iterable[int | None],
    ) -> int | None:
        known = tuple(
            value
            for value in values
            if isinstance(value, int)
        )
        return sum(known) if known else None

Разберём код по порядку его выполнения. Метод _build_reduce_context() получает пакет исходных фрагментов и все результаты map-этапа. Для каждого результата он вызывает _render_partial(), а затем объединяет получившиеся блоки через пустую строку.

_render_partial() берёт типизированный DocumentAIResult и превращает только нужные поля в компактный текст: краткое содержание, ответ по фрагменту, ключевые факты, ограничения и признак ручной проверки. Маркер [partial:...] помогает отделить один промежуточный ответ от другого, но не считается ссылкой на источник. Настоящими ссылками остаются только значения page:... и section:..., полученные из исходного документа.

После объединения метод сравнивает длину content с max_reduce_characters. Если промежуточные ответы сами оказались слишком длинными, приложение не обрезает их молча, потому что при обрезке можно потерять важный факт или ссылку. Вместо этого возникает понятная DocumentAIContextError. В более сложной системе здесь можно добавить многоуровневое объединение, но для нашего учебного проекта достаточно одной контролируемой границы.

В references передаётся полный список ссылок исходного документа из batch.references. Мы не ограничиваем его только ссылками из промежуточных key_facts, потому что модель может упомянуть важное сведение в summary или answer. При этом произвольную ссылку вернуть нельзя: все значения по-прежнему проверяются по списку частей исходного документа.

Готовый PreparedFileContext снова передаётся обычному DocumentContentAnalyzer. Он строит те же developer- и user-сообщения, запрашивает структурированные ответы, проверяет Pydantic-модель и отклоняет неизвестные ссылки. Таким образом, финальный ответ проходит ту же границу, что и ответ по небольшому документу.

Последний вспомогательный метод _sum_optional() нужен для общих метрик. Провайдер иногда не возвращает статистику токенов, поэтому значения имеют тип int | None. Метод отбрасывает None и складывает только известные числа. Если статистики нет ни у одного вызова, он возвращает None, а не ноль, ведь ноль означал бы, что токены точно не расходовались, что было бы неверно.

После двух этапов мы знаем следующее:

  • каждый запрос получил ограниченный объём текста
  • каждый промежуточный ответ соответствовал Pydantic-схеме
  • ссылки промежуточных фактов существовали в соответствующем чанке
  • ссылки итоговых фактов присутствовали в промежуточных результатах
  • общее количество запросов и токенов сохранено в результате

Но это не доказывает, что ни один факт не был потерян или искажён. Map-этап может не включить важную деталь в key_facts, а reduce-этап может слишком сильно обобщить несколько формулировок. Смысловая проверка важных выводов человеком остаётся обязательной.

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

Шаг 5

Мы уже создали все компоненты для обработки большого документа и теперь у нас DocumentChunkPreparer формирует ограниченные фрагменты, а LargeDocumentAnalyzer выполняет map- и reduce-запросы. Но основной DocumentAIService пока о них не знает. Он по-прежнему пытается подготовить весь документ через DocumentContextPreparer и при превышении 24 000 символов возвращает ошибку.

Теперь свяжем готовые компоненты в один сценарий. Пользователь не должен самостоятельно решать, сколько запросов потребуется. Сервис сначала попробует подготовить документ целиком. Если текст помещается, всё продолжит работать одним вызовом. Только конкретный DocumentContextTooLargeError станет сигналом разделить документ и запустить map-reduce.

Перейдём к коду. Откройте app/services/document_ai_service.py. Сервису понадобятся тип ошибки, по которому он распознает слишком большой документ, объект подготовленного пакета и два новых компонента. Обновите импорты следующим образом:

from app.services.document_ai_context import (
    DocumentAIContextError,
    DocumentContextPreparer,
    DocumentContextTooLargeError,
    PreparedFileContext,
    TableAnalysisPlan,
    TableContextPreparer,
)
from app.services.document_chunking import (
    DocumentChunkPreparer,
    PreparedDocumentBatch,
)
from app.services.document_content_analyzer import (
    DocumentAIResult,
    DocumentContentAnalyzer,
    DocumentContentAnalyzerError,
)
from app.services.large_document_analyzer import LargeDocumentAnalyzer

Теперь расширим данные запроса и зависимости сервиса. Поле document_references позволит интерфейсу передать точные ссылки на выбранные страницы или разделы. По умолчанию оно равно None, поэтому существующие вызовы продолжат анализировать весь документ.

Сам DocumentAIService не должен создавать компоненты внутри своих методов. Он получит готовые DocumentChunkPreparer и LargeDocumentAnalyzer через конструктор, как и остальные зависимости. Замените определения DocumentAIRequest и конструктора сервиса следующим кодом:

@dataclass(frozen=True)
class DocumentAIRequest:
    source_path: Path
    question: str
    sheet_name: str | None = None
    table_plan: TableAnalysisPlan | None = None
    document_references: tuple[str, ...] | None = None

class DocumentAIService:
    def __init__(
        self,
        file_policy: FileValidationPolicy,
        validator: FileValidator,
        storage: TemporaryFileStorage,
        document_extractor: DocumentExtractor,
        table_extractor: TableExtractor,
        document_context_preparer: DocumentContextPreparer,
        table_context_preparer: TableContextPreparer,
        content_analyzer: DocumentContentAnalyzer,
        document_chunk_preparer: DocumentChunkPreparer,
        large_document_analyzer: LargeDocumentAnalyzer,
    ) -> None:
        self._file_policy = file_policy
        self._validator = validator
        self._storage = storage
        self._document_extractor = document_extractor
        self._table_extractor = table_extractor
        self._document_context_preparer = document_context_preparer
        self._table_context_preparer = table_context_preparer
        self._content_analyzer = content_analyzer
        self._document_chunk_preparer = document_chunk_preparer
        self._large_document_analyzer = large_document_analyzer

document_references=None означает анализ всего документа. Если передан кортеж, DocumentChunkPreparer.select() сначала проверит ссылки и оставит только указанные части. Для таблиц этот параметр не имеет смысла и будет отклонён.

После подготовки метод analyze() теперь может получить два разных результата. PreparedFileContext означает, что данные помещаются в один запрос. PreparedDocumentBatch означает, что документ разделён и требует map-reduce. Замените текущий метод analyze() целиком:

    def analyze(
        self,
        request: DocumentAIRequest,
    ) -> DocumentAIResult:
        question = request.question.strip()
        if not question:
            raise DocumentAIInputError(
                "Вопрос не должен быть пустым."
            )

        try:
            context = self._prepare_from_file(request)
        except DocumentAIServiceError:
            raise
        except (
            FileValidationError,
            TemporaryFileStorageError,
        ) as error:
            raise DocumentAIInputError(str(error)) from error
        except (
            DocumentExtractionError,
            TableExtractionError,
            TableAnalysisError,
            DocumentAIContextError,
        ) as error:
            raise DocumentAIPreparationError(str(error)) from error

        try:
            if isinstance(context, PreparedDocumentBatch):
                return self._large_document_analyzer.analyze(
                    question,
                    context,
                )

            return self._content_analyzer.analyze(question, context)
        except DocumentAIContextError as error:
            raise DocumentAIPreparationError(str(error)) from error
        except DocumentContentAnalyzerError as error:
            raise DocumentAIModelError(str(error)) from error

После вызова _prepare_from_file() проверка isinstance(context, PreparedDocumentBatch) выбирает способ анализа. Для пакета вызывается LargeDocumentAnalyzer, а обычный PreparedFileContext передаётся прежнему DocumentContentAnalyzer. Эта проверка не определяет формат файла: к этому моменту формат уже распознан. Она выбирает только один запрос или map-reduce.

DocumentAIContextError может появиться и после map-этапа, например если промежуточные результаты не помещаются в reduce-контекст. Такая ситуация преобразуется в DocumentAIPreparationError. Ошибки обращения к модели по-прежнему превращаются в DocumentAIModelError.

Чтобы такая развилка стала возможной, методы подготовки тоже должны уметь возвращать либо один контекст, либо пакет. В том же файле полностью замените методы _prepare_from_file() и _prepare_context() следующим кодом:

    def _prepare_from_file(
        self,
        request: DocumentAIRequest,
    ) -> PreparedFileContext | PreparedDocumentBatch:
        validated = self._validator.validate(request.source_path)

        with self._storage.stage(validated) as temporary_path:
            staged_policy = replace(
                self._file_policy,
                allowed_root=temporary_path.parent,
            )
            staged_validator = FileValidator(staged_policy)
            staged_file = staged_validator.validate(
                temporary_path,
                declared_media_type=validated.media_type,
            )
            staged_file = replace(
                staged_file,
                original_name=validated.original_name,
            )

            return self._prepare_context(staged_file, request)

    def _prepare_context(
        self,
        file: ValidatedFile,
        request: DocumentAIRequest,
    ) -> PreparedFileContext | PreparedDocumentBatch:
        if file.kind in DOCUMENT_KINDS:
            if request.sheet_name is not None:
                raise DocumentAIInputError(
                    "Для документа нельзя выбирать лист."
                )
            if request.table_plan is not None:
                raise DocumentAIInputError(
                    "Для документа не нужен табличный план."
                )

            document = self._document_extractor.extract(file)
            document = self._document_chunk_preparer.select(
                document,
                request.document_references,
            )

            try:
                return self._document_context_preparer.prepare(document)
            except DocumentContextTooLargeError:
                return self._document_chunk_preparer.prepare(document)

        if file.kind in TABLE_KINDS:
            if request.document_references is not None:
                raise DocumentAIInputError(
                    "Для таблицы нельзя выбирать части документа."
                )
            if request.table_plan is None:
                raise DocumentAIInputError(
                    "Для таблицы нужен TableAnalysisPlan."
                )

            table = self._table_extractor.extract(
                file,
                sheet_name=request.sheet_name,
            )
            return self._table_context_preparer.prepare(
                table,
                request.table_plan,
            )

        raise DocumentAIPreparationError(
            f"AI-анализ формата {file.kind} пока не поддерживается."
        )

Проследим документную ветку по шагам. Сначала экстрактор создаёт ExtractedDocument. Затем select() оставляет все части или только явно указанные ссылки. После этого прежний DocumentContextPreparer пытается собрать единый контекст. Если он помещается в лимит, метод сразу возвращает PreparedFileContext, и модель будет вызвана один раз.

Если возникает именно DocumentContextTooLargeError, сервис передаёт тот же документ в DocumentChunkPreparer.prepare() и получает PreparedDocumentBatch. Другие ошибки не запускают map-reduce. Например, пустой документ и неизвестная ссылка означают неправильные входные данные, а не большой размер, поэтому их нужно показать пользователю как ошибки.

Важна и последовательность относительно временного файла. Внутри TemporaryFileStorage.stage() выполняются проверка, извлечение и подготовка текстовых chunks. Затем with завершается и временный файл удаляется. Только после этого analyze() обращается к модели. Map-reduce работает уже с данными в памяти и не продлевает жизнь временной копии.

Логика сервиса готова, но при сборке приложения ему необходимо передать новые зависимости. Откройте app/application.py и добавьте импорты:

from app.services.document_ai_context import (
    DocumentContextPreparer,
    TableContextPreparer,
)
from app.services.document_ai_service import DocumentAIService
from app.services.document_chunking import (
    DocumentChunkingPolicy,
    DocumentChunkPreparer,
)
from app.services.document_content_analyzer import (
    DocumentContentAnalyzer,
)
from app.services.large_document_analyzer import (
    LargeDocumentAnalysisPolicy,
    LargeDocumentAnalyzer,
)

После импортов обновите фабрику build_document_ai_service(). Именно здесь мы создадим политики с конкретными ограничениями, соберём новые компоненты и передадим их в DocumentAIService:

def build_document_ai_service(
    llm_client: LLMClient,
    allowed_root: Path,
) -> DocumentAIService:
    file_policy = FileValidationPolicy(
        allowed_root=allowed_root,
        max_size_bytes=10 * 1_024 * 1_024,
    )
    validator = FileValidator(file_policy)
    storage = TemporaryFileStorage(
        max_size_bytes=file_policy.max_size_bytes
    )

    document_extractor = build_default_document_extractor()

    table_policy = TableExtractionPolicy(
        max_rows=5_000,
        max_columns=100,
    )
    table_extractor = build_default_table_extractor(table_policy)
    table_analyzer = TableAnalyzer()
    content_analyzer = DocumentContentAnalyzer(llm_client)
    document_chunk_preparer = DocumentChunkPreparer(
        DocumentChunkingPolicy(
            max_chunk_characters=12_000,
            max_chunks=8,
        )
    )
    large_document_analyzer = LargeDocumentAnalyzer(
        content_analyzer=content_analyzer,
        policy=LargeDocumentAnalysisPolicy(
            max_reduce_characters=32_000,
        ),
    )

    return DocumentAIService(
        file_policy=file_policy,
        validator=validator,
        storage=storage,
        document_extractor=document_extractor,
        table_extractor=table_extractor,
        document_context_preparer=DocumentContextPreparer(),
        table_context_preparer=TableContextPreparer(
            table_analyzer
        ),
        content_analyzer=content_analyzer,
        document_chunk_preparer=document_chunk_preparer,
        large_document_analyzer=large_document_analyzer,
    )

Обратите внимание, что создаётся только один DocumentContentAnalyzer. Он используется и для обычного документа, и внутри LargeDocumentAnalyzer. Поэтому все map- и reduce-запросы проходят одну версию промпта, одну Pydantic-схему и один настроенный LLMClient.

Политики собираются в фабрике, а не внутри сервисов. Здесь явно видно:

  • целый документ допускает до 24 000 символов по политике DocumentContextPreparer
  • один чанков ограничен 12 000 символов
  • документ может создать не более восьми чанков
  • reduce-контекст ограничен 32 000 символов

DocumentAIService не знает, почему выбраны именно такие числа. Если модель, требования или бюджет изменятся, политики можно заменить в фабрике без изменения последовательности сценария.

Адаптивный маршрут готов. В следующем шаге мы подключим большой Markdown-файл и выведем количество запросов и проверим как полный map-reduce, так и явный выбор отдельных разделов.

Правильный ответ
  • ✅ Только когда подготовка целого документа завершилась DocumentContextTooLargeError

Шаг 6

Все части сценария теперь соединены. Небольшой документ должен по-прежнему анализироваться одним запросом, а большой — автоматически разделяться и проходить map-reduce. В этом шаге мы запустим наше приложение и наглядно проверим оба пути, число вызовов модели, ссылки и предупреждения.

Для проверки мы будем использовать файл large_support_handbook.md, который вам нужно скачать и поместить в папку sample_files. Это учебный регламент службы поддержки размером более 28 000 символов. В нём есть Markdown-заголовки, поэтому знакомый TextDocumentExtractor создаст отдельные DocumentPart со ссылками section:1, section:2 и далее.

Перед запуском дополним демонстрационный вывод. Откройте document_ai_demo.py и полностью замените функцию print_result(). Кроме самого ответа она теперь покажет model_calls — общее количество обращений к модели:

def print_result(result: DocumentAIResult) -> None:
    analysis = result.analysis
    print(f"Источник: {result.source_name}")
    print(f"Формат: {result.source_kind}")
    print(f"Краткое содержание: {analysis.summary}")
    print(f"Ответ: {analysis.answer}")
    print("Ключевые факты:")
    for fact in analysis.key_facts:
        print(f"- {fact.text} [{fact.reference}]")

    if analysis.limitations:
        print("Ограничения:")
        for limitation in analysis.limitations:
            print(f"- {limitation}")

    for warning in result.warnings:
        print(f"Предупреждение: {warning}")

    print(
        "Требуется проверка человеком: "
        f"{analysis.requires_human_review}"
    )
    print(
        f"Промпт: {result.prompt_id}@{result.prompt_version}"
    )
    print(f"Модель: {result.model}")
    print(f"Запросов к модели: {result.model_calls}")
    print(f"Токенов: {result.total_tokens}")
    print(f"Время: {result.elapsed_seconds:.2f} с")

Для небольшого PDF и таблицы будет выведено значение 1. Для большого документа число зависит от количества chunks: к map-запросам добавляется один reduce-запрос.

Теперь добавим отдельный сценарий для большого файла. Внутри main(), после существующих вызовов run_case(), добавьте:

    run_case(
        service,
        "Анализ большого документа",
        DocumentAIRequest(
            source_path=(
                SAMPLE_FILES_ROOT
                / "large_support_handbook.md"
            ),
            question=(
                "Когда обращение передают на второй уровень "
                "и когда ему назначают критический приоритет?"
            ),
        ),
    )

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

Код готов к проверке. Сначала убедитесь, что в .env модели разрешено вернуть достаточно подробный структурированный ответ:

MAX_OUTPUT_TOKENS=1000

И давайте наконец запустим файл document_ai_demo.py. В результате мы получим следующий ответ:

Анализ большого документа
Источник: large_support_handbook.md
Формат: markdown
Краткое содержание: Обращение передают на второй уровень, если проблема повторяется после стандартного решения, затрагивает несколько заказов, содержит признаки технического сбоя или не может быть закрыта в установленный для процесса срок (от 12 до 42 часов в зависимости от типа обращения). Критический приоритет назначается только при риске утечки данных, массовой недоступности сервиса, двойном списании или угрозе безопасности клиента, и такое обращение немедленно передаётся дежурному руководителю, минуя обычную очередь.
Ответ: Обращение передают на второй уровень при повторении проблемы после стандартного решения, если оно затрагивает несколько заказов, содержит признаки технического сбоя или не может быть закрыто в установленный срок (от 12 до 42 часов в зависимости от процесса). Критический приоритет назначается только при риске утечки данных, массовой недоступности сервиса, двойном списании или угрозе безопасности клиента — в этом случае обращение немедленно передаётся дежурному руководителю и не ждёт обычной очереди.
Ключевые факты:
- Передача на второй уровень выполняется, если проблема повторилась после стандартного решения, затрагивает несколько заказов, содержит признаки технического сбоя или не может быть закрыта в течение установленного срока (от 18 до 42 часов в зависимости от процесса). [section:2]
- Критический приоритет назначается только при риске утечки данных, массовой недоступности сервиса, двойном списании или угрозе безопасности клиента. Такое обращение немедленно передаётся дежурному руководителю и не ожидает обычной очереди. [section:2]
- Сроки передачи на второй уровень: Доставка — 18 часов, Возвраты — 24 часа, Оплата — 30 часов, Гарантия — 36 часов, Аккаунт — 42 часа. [section:2]
- Обычное недовольство сроком ответа само по себе не является критическим инцидентом. [section:2]
- Передача на второй уровень выполняется, если проблема повторилась после стандартного решения, затрагивает несколько заказов, содержит признаки технического сбоя или не может быть закрыта в течение установленного срока (срок зависит от процесса). [section:7]
- Сроки передачи на второй уровень различаются в зависимости от процесса: 12 часов (Партнёры), 18 часов (Доставка), 24 часа (Возвраты), 30 часов (Оплата), 36 часов (Гарантия), 42 часа (Аккаунт). [section:7]
Ограничения:
- Данные взяты только из предоставленного файла и не могут считаться полностью достоверными без проверки.
- Возможны различия в процедурах, если есть обновления вне этого документа.
- Нет сведений о возможных исключениях или специальных случаях.
- Нет информации о деталях процессов вне описанных очередей (Партнёры, Доставка, Возвраты, Оплата, Гарантия, Аккаунт).
Предупреждение: Документ обработан по частям: 3.
Требуется проверка человеком: True
Промпт: document_analysis@1.0.0
Модель: gpt-4.1-2025-04-14
Запросов к модели: 4
Токенов: 11828
Время: 19.20 с

Формулировки модели могут различаться, но для большого документа проверьте следующие свойства:

  • анализ завершается без DocumentContextTooLargeError
  • в ключевых фактах используются существующие ссылки section:...
  • появляется предупреждение Документ обработан по частям: 3.
  • поле Запросов к модели равно четырём, то есть у нас три map-запроса и один reduce-запрос
  • количество токенов отражает сумму всех четырёх вызовов

После проверки полного документа посмотрим, как работает явный выбор разделов. Временно добавьте в DocumentAIRequest большого документа параметр:

document_references=(
    "section:2",
    "section:8",
),

Код сначала убедится, что обе ссылки существуют, а затем оставит только эти разделы. Их общий текст помещается в обычный лимит, поэтому map-reduce не понадобится и model_calls станет равен единице. В предупреждениях появится сообщение о том, что для анализа выбраны не все части документа.

Анализ большого документа
Источник: large_support_handbook.md
Формат: markdown
Краткое содержание: Передача обращения на второй уровень происходит при повторении проблемы после стандартного решения, затрагивании нескольких заказов, наличии признаков технического сбоя или невозможности закрытия обращения в течение 18 часов. Критический приоритет назначается только при риске утечки данных, массовой недоступности сервиса, двойном списании или угрозе безопасности клиента.
Ответ: Обращение передают на второй уровень, если проблема повторилась после стандартного решения, затрагивает несколько заказов, содержит признаки технического сбоя или не может быть закрыта в течение 18 часов. Критический приоритет назначается только при риске утечки данных, массовой недоступности сервиса, двойном списании или угрозе безопасности клиента.
Ключевые факты:
- Передача на второй уровень выполняется, если проблема повторилась после стандартного решения, затрагивает несколько заказов, содержит признаки технического сбоя или не может быть закрыта в течение 18 часов. [section:2]
- Критический приоритет назначается только при риске утечки данных, массовой недоступности сервиса, двойном списании или угрозе безопасности клиента. [section:2]
Ограничения:
- Информация основана только на предоставленных разделах 2 и 8, которые дублируют друг друга.
- Нет сведений о других возможных критериях передачи или назначения приоритета вне указанных случаев.
- Данные являются недоверенными и требуют проверки человеком при принятии важных решений.
Предупреждение: Для анализа выбраны не все части документа.
Требуется проверка человеком: True
Промпт: document_analysis@1.0.0
Модель: gpt-4.1-2025-04-14
Запросов к модели: 1
Токенов: 1555
Время: 5.73 с

Если указать section:999, запрос завершится до обращения к модели. Это показывает отличие от RAG, то есть части выбираются не по смысловой близости, а по точным метаданным, известным прикладному сценарию.

После запуска важно правильно понимать результат. MapReduce позволяет обработать документ, который не помещается в один запрос, но не гарантирует сохранение каждой детали. На map-этапе модель выбирает относящиеся к вопросу факты из каждого фрагмента. Если важное сведение не попало в промежуточный ответ, reduce-этап уже не увидит исходный текст и не сможет его восстановить. Затем финальный запрос ещё раз обобщает собранные данные.

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

Текущая реализация подходит, когда:

  • нужно один раз проанализировать один документ умеренного размера
  • количество частей заранее ограничено
  • важно охватить документ целиком
  • допустимы несколько последовательных запросов к модели
  • результат можно проверить по сохранённым ссылкам

MapReduce обрабатывает все части, поэтому его стоимость растёт вместе с размером документа. Он не создаёт индекс и не ускоряет повторные вопросы по одному и тому же файлу. И вот тут переход к RAG становится оправданным, если документов много, они часто используются повторно или для каждого вопроса требуется лишь небольшая часть общей базы. Тогда фрагменты заранее индексируются, а перед запросом выполняется поиск релевантных частей.

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

Подведём итог раздела. Мы сохранили простой путь для небольших документов и добавили отдельную стратегию для больших. Приложение измеряет подготовленный текст до обращения к модели, по возможности сохраняет страницы и разделы целиком, ограничивает размер и количество фрагментов, а затем объединяет промежуточные структурированные ответы. Ссылки на исходные места, число запросов, расход токенов и предупреждения остаются видимыми в итоговом результате.

При этом DocumentAIService по-прежнему не умеет анализировать PNG и JPEG. Валидатор может подтвердить их формат, но подходящей AI-стратегии ещё нет. В следующем разделе добавим изображения, сравним OCR с прямым анализом vision-моделью и подключим новый маршрут к существующему сервису.

Правильный ответ
  • ✅ Делить слишком большую часть на меньшие куски
  • ✅ Сохранять ссылки исходных DocumentPart
  • ✅ Ограничивать общее число chunks
  • ✅ Соблюдать max_chunk_characters

Шаг 7

Когда Document AI Service должен перейти к обработке документа по частям?

Правильный ответ
  • ✅ Map отдельно анализирует каждый chunk, reduce объединяет промежуточные результаты

Шаг 8

Какие правила важны при подготовке фрагментов большого документа?

Правильный ответ
  • ✅ Краткие содержания, ответы по частям, ключевые факты со ссылками и ограничения

Шаг 9

Что происходит на этапах map и reduce?

Правильный ответ
  • ✅ Выбрать только существующие части документа, а неизвестные ссылки отклонить

Шаг 10

Что следует передавать в reduce-запрос?

Правильный ответ 
def split_document_text(text: str, limit: int) -> tuple[str, ...]:
    if not isinstance(text, str):
        raise TypeError

    text = text.strip()

    if not text:
        raise ValueError

    if (
        isinstance(limit, bool)
        or not isinstance(limit, int)
        or limit <= 0
    ):
        raise ValueError

    chunks = []
    remaining = text
    min_boundary = limit / 2

    while remaining:
        if len(remaining) <= limit:
            chunk = remaining.strip()

            if chunk:
                chunks.append(chunk)

            break

        cut_position = None

        for separator in ("\n\n", "\n", " "):
            position = remaining.rfind(separator, 0, limit + 1)

            if position >= min_boundary:
                cut_position = position
                break

        if cut_position is None:
            cut_position = limit

        chunk = remaining[:cut_position].strip()

        if chunk:
            chunks.append(chunk)

        remaining = remaining[cut_position:].strip()

    return tuple(chunks)

Шаг 11

Перед разбиением большого документа выбираются части по ссылкам:

selected_parts = []
for reference in document_references:
    # Здесь нужна проверка reference
    selected_parts.append(parts_by_reference[reference])

chunks = make_chunks(selected_parts)

Что нужно сделать на месте комментария до подготовки chunks?

Правильный ответ 
def pack_document_blocks(blocks, max_characters: int, max_chunks: int) -> tuple[dict, ...]:
    if not isinstance(blocks, (list, tuple)) or not blocks:
        raise ValueError

    if (
        isinstance(max_characters, bool)
        or not isinstance(max_characters, int)
        or max_characters <= 0
    ):
        raise ValueError

    if (
        isinstance(max_chunks, bool)
        or not isinstance(max_chunks, int)
        or max_chunks <= 0
    ):
        raise ValueError

    chunks = []
    current_texts = []
    current_references = []
    seen_references = set()

    def save_chunk():
        if not current_texts:
            return

        chunks.append({
            "content": "\n\n".join(current_texts),
            "references": tuple(current_references),
        })

    for block in blocks:
        if not isinstance(block, (list, tuple)) or len(block) != 2:
            raise ValueError

        text, references = block

        if not isinstance(text, str):
            raise ValueError

        text = text.strip()

        if not text:
            raise ValueError

        if len(text) > max_characters:
            raise ValueError

        if not isinstance(references, (list, tuple)):
            raise ValueError

        for reference in references:
            if not isinstance(reference, str) or not reference.strip():
                raise ValueError

        if current_texts:
            candidate = "\n\n".join([*current_texts, text])

            if len(candidate) > max_characters:
                save_chunk()

                if len(chunks) >= max_chunks:
                    raise ValueError

                current_texts = []
                current_references = []
                seen_references = set()

        current_texts.append(text)

        for reference in references:
            reference = reference.strip()

            if reference not in seen_references:
                seen_references.add(reference)
                current_references.append(reference)

    save_chunk()

    if len(chunks) > max_chunks:
        raise ValueError

    return tuple(chunks)

Шаг 12

Разбиение текста по естественным границам

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

Задача

Реализуйте функцию split_document_text(text, limit). Она должна последовательно отделять фрагменты длиной не больше limit и вернуть их в исходном порядке.

Требования

  • Не изменяйте имя функции и её аргументы.
  • text должен быть строкой. Для другого типа выбросьте TypeError.
  • Удалите внешние пробелы из текста. Если после очистки строка пуста, выбросьте ValueError.
  • limit должен быть положительным целым числом. Значение типа bool, ноль, отрицательное число или другой тип должны приводить к ValueError.
  • Если оставшийся текст не длиннее limit, добавьте его как последний фрагмент.
  • Если остаток длиннее лимита, сначала найдите последнюю границу абзаца \n\n на участке не дальше limit.
  • Используйте найденную границу, только если она находится не раньше половины лимита.
  • Если подходящей границы абзаца нет, таким же образом попробуйте последнюю границу строки \n.
  • Если и она находится слишком рано или отсутствует, попробуйте последний обычный пробел.
  • Если подходящей естественной границы нет, разделите текст ровно по позиции limit.
  • Каждый полученный фрагмент очистите через strip(). Пустые фрагменты не добавляйте.
  • Верните фрагменты как tuple. Длина каждого элемента не должна превышать limit.

Пример результата

print(split_document_text(
    "alpha beta gamma",
    10,
))
print(split_document_text("A" * 25, 10))

Результат:

('alpha beta', 'gamma')
('AAAAAAAAAA', 'AAAAAAAAAA', 'AAAAA')

Алгоритм соответствует изученному DocumentChunkPreparer. Дополнительные библиотеки не нужны.

Правильный ответ 
def run_document_map_reduce(client, question, chunks) -> dict:
    if not isinstance(question, str):
        raise ValueError

    question = question.strip()

    if not question:
        raise ValueError

    if not isinstance(chunks, (list, tuple)) or not chunks:
        raise ValueError

    partial_results = []

    for index, chunk in enumerate(chunks, start=1):
        partial = client.analyze_chunk(
            question=question,
            chunk=chunk,
            index=index,
        )
        partial_results.append(partial)

    partials = tuple(partial_results)

    result = client.reduce(
        question=question,
        partials=partials,
    )

    return {
        "result": result,
        "partials": partials,
        "model_calls": len(chunks) + 1,
    }

Шаг 13

Объединение блоков в ограниченные chunks

После подготовки документа имеются готовые текстовые блоки со ссылками на источники. Соседние блоки нужно последовательно объединять, пока общий текст помещается в лимит. При этом порядок текста и первое появление каждой ссылки должны сохраняться.

Задача

Реализуйте функцию pack_document_blocks(blocks, max_characters, max_chunks). Она должна упаковать входные блоки в ограниченное количество словарей с ключами content и references.

Требования

  • Не изменяйте имя функции и её аргументы.
  • blocks должен быть непустым списком или кортежем.
  • Каждый блок должен быть парой (text, references).
  • text должен быть непустой строкой после strip().
  • references должен быть списком или кортежем непустых строк.
  • max_characters и max_chunks должны быть положительными целыми числами и не могут иметь тип bool.
  • Если текст отдельного блока длиннее max_characters, выбросьте ValueError. Повторно делить блоки в этой задаче не нужно.
  • Обрабатывайте блоки в исходном порядке.
  • Попробуйте добавить очередной текст в текущий chunk через разделитель \n\n.
  • Если объединённая длина не превышает max_characters, оставьте блок в текущем chunk.
  • Если лимит будет превышен, завершите текущий chunk и начните новый с очередного блока.
  • Для каждого chunk соберите ссылки из вошедших блоков. Удалите повторы с сохранением порядка первого появления.
  • Каждый chunk верните как словарь {"content": объединённый текст, "references": tuple ссылок}.
  • Если число сформированных chunks больше max_chunks, выбросьте ValueError.
  • Верните все chunks как tuple.

Пример результата

blocks = [
    ("one", ("p1",)),
    ("two", ("p1", "p2")),
    ("123456", ("p3",)),
]
print(pack_document_blocks(blocks, 9, 3))

Результат:

({'content': 'one\n\ntwo', 'references': ('p1', 'p2')}, {'content': '123456', 'references': ('p3',)})

Блоки уже содержат нужные префиксы источников. Дополнительные библиотеки не нужны.

Шаг 14

Последовательный map-reduce по фрагментам документа

Большой документ анализируется в два этапа. На этапе map каждый chunk обрабатывается отдельно и даёт частичный результат. После успешной обработки всех частей один reduce-вызов объединяет частичные результаты в окончательный ответ.

Задача

Реализуйте функцию run_document_map_reduce(client, question, chunks). Она должна последовательно вызвать фейковый анализатор для всех chunks, затем один раз вызвать объединение и вернуть результат с метаданными.

Требования

  • Не изменяйте имя функции и её аргументы.
  • question должен быть строкой с хотя бы одним непробельным символом. Очистите его через strip(). При нарушении выбросьте ValueError.
  • chunks должен быть непустым списком или кортежем. Иначе выбросьте ValueError.
  • Обрабатывайте chunks последовательно в исходном порядке.
  • Для каждого элемента вызовите client.analyze_chunk() с именованными аргументами question=cleaned_question, chunk=chunk и index=index.
  • Индексы должны начинаться с 1 и увеличиваться на единицу.
  • Сохраняйте каждый возвращённый map-результат в список частичных результатов.
  • После успешной обработки всех chunks преобразуйте частичные результаты в tuple.
  • Вызовите client.reduce(question=cleaned_question, partials=partials) ровно один раз.
  • Верните словарь с ключами result, partials и model_calls.
  • В result сохраните значение reduce-вызова, в partials — кортеж map-результатов, в model_calls — число chunks плюс один.
  • Не перехватывайте исключения клиента.
  • Если очередной map-вызов завершился с ошибкой, следующие chunks и reduce запускать нельзя.

Пример результата

Фейковый клиент возвращает p1 и p2 для двух chunks, а reduce возвращает final:

result = run_document_map_reduce(
    fake_client, " Найди правила ", ["a", "b"]
)
print(result)

Результат:

{'result': 'final', 'partials': ('p1', 'p2'), 'model_calls': 3}

Фейковый клиент только записывает вызовы. API-ключ и сетевой доступ не нужны.