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

Итоговые сценарии и проверка

Шаг 1

В предыдущих разделах мы собрали техническую основу Document AI Service. Один и тот же сервис уже умеет суммаризировать документы, извлекать ключевые факты, отвечать по технической документации, анализировать таблицы и изображения. Для резюме, договора или внутренней инструкции не требуется создавать отдельный AI-сервис: путь файла через приложение остаётся тем же, а меняются вопрос пользователя и правила конкретного сценария.

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

Мы не будем повторно выполнять эти уже знакомые запросы. Они проходят через проверенный маршрут и только расходуют баланс в ProxyAPI. И перед новым сценарием очистите document_ai_demo.py. Внутри main() удалите все прежние вызовы run_case(), но оставьте загрузку настроек, создание LLMClient, сборку DocumentAIService и блок finally:

    llm_client = LLMClient(settings)

    try:
        service = build_document_ai_service(
            llm_client=llm_client,
            allowed_root=SAMPLE_FILES_ROOT,
        )

        # Здесь будет только один активный сценарий.
    finally:
        llm_client.close()

Функции print_result() и run_case() удалять не нужно. Они продолжат выводить результат, но при одном запуске демонстрационного файла будет выполняться только один AI-сценарий и тем самым мы не будем тратить баланс на лишние запросы, которые уже проверены нами.

У вас может возникнуть вопрос, а почему мы не переносим анализ документов в основной main.py? Дело в том, что document_ai_demo.py это не отдельное приложение, а дополнительная точка запуска того же проекта. Вся логика анализа находится внутри пакета app, а демонстрационный файл только собирает DocumentAIService, передаёт ему запросы и выводит результаты.

Файл main.py пока отвечает за запуск консольного чата, а document_ai_demo.py за проверку работы с документами, таблицами и изображениями. Такое разделение позволяет не перегружать консольный интерфейс временными командами для выбора файлов и параметров анализа. А добавлять в чат команду вроде /document сейчас нет необходимости. Для полноценного анализа пришлось бы вводить путь к файлу, вопрос, лист таблицы, выбранные разделы и дополнительные изображения. Консоль быстро превратилась бы в сложный временный интерфейс. В следующем модуле эти параметры нам будет удобнее принимать через отдельный HTTP эндпоинт FastAPI.

В модели DocumentAnalysis уже есть поле requires_human_review. Оно показывает, можно ли использовать результат автоматически или его сначала должен проверить человек.

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

Но полностью полагаться на решение модели нельзя. Она может не заметить риск и вернуть false, хотя проверка человеком действительно необходима.

Кроме того, в приложении могут существовать обязательные правила, которые не зависят от мнения модели. Например, любой результат анализа договора должен проверять специалист. В таком случае Python-код должен принудительно установить requires_human_review=True, даже если модель решила, что дополнительная проверка не требуется.

Этот договор будет обрабатываться тем же DocumentAIService, что и остальные документы, поэтому создавать отдельный сервис не нужно. Нам достаточно добавить в запрос признак обязательной проверки результата человеком. Перейдём к практике, откройте app/services/document_ai_service.py и добавьте в 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
    additional_image_paths: tuple[Path, ...] = ()
    require_human_review: bool = False

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

Теперь перейдите к методу analyze(). В нём анализаторы сейчас вызываются через несколько ранних return. Замените второй блок try, который выбирает анализатор и обрабатывает его ошибки, следующим кодом:

        try:
            if isinstance(context, PreparedImageBatch):
                result = self._image_analyzer.analyze(
                    question,
                    context,
                )
            elif isinstance(context, PreparedDocumentBatch):
                result = self._large_document_analyzer.analyze(
                    question,
                    context,
                )
            else:
                result = self._content_analyzer.analyze(
                    question,
                    context,
                )
        except DocumentAIContextError as error:
            raise DocumentAIPreparationError(str(error)) from error
        except (
            DocumentContentAnalyzerError,
            ImageContentAnalyzerError,
        ) as error:
            raise DocumentAIModelError(str(error)) from error

        return self._apply_review_policy(
            result,
            required=request.require_human_review,
        )

Маршрутизация не изменилась и пакет изображений получает ImageContentAnalyzer, большой документ — LargeDocumentAnalyzer, остальные контексты — DocumentContentAnalyzer. Мы только сохраняем результат в переменную, чтобы перед возвратом применить единое правило проверки.

После analyze() добавьте новый метод:

    @staticmethod
    def _apply_review_policy(
        result: DocumentAIResult,
        required: bool,
    ) -> DocumentAIResult:
        if not required or result.analysis.requires_human_review:
            return result

        analysis = result.analysis.model_copy(
            update={"requires_human_review": True}
        )
        warnings = tuple(
            dict.fromkeys(
                (
                    *result.warnings,
                    "Проверка человеком обязательна "
                    "по правилу сценария.",
                )
            )
        )

        return replace(
            result,
            analysis=analysis,
            warnings=warnings,
        )

Сначала метод проверяет, нужно ли вообще вмешиваться. Если обязательного правила нет или модель уже вернула true, объект остаётся прежним. В остальных случаях создаётся копия Pydantic-модели с изменённым флагом, а затем копия замороженного DocumentAIResult.

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

Для наглядной проверки создайте в sample_files учебный файл service_contract.md:

# Учебный договор обслуживания

## Срок

Договор действует с 1 сентября 2026 года по 31 августа 2027 года.

## Стоимость

Стоимость обслуживания составляет 120 000 рублей в год.

## Ответственность

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

## Расторжение

Каждая сторона может расторгнуть договор, уведомив другую
сторону не менее чем за 30 календарных дней.

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

В document_ai_demo.py после создания service добавьте единственный активный вызов:

        run_case(
            service,
            "Анализ учебного договора",
            DocumentAIRequest(
                source_path=(
                    SAMPLE_FILES_ROOT / "service_contract.md"
                ),
                question=(
                    "Перечисли срок действия, стоимость, условия "
                    "приостановки обслуживания и расторжения. "
                    "Не давай юридическую оценку."
                ),
                require_human_review=True,
            ),
        )

При запуске модель может сама вернуть true. Если она вернёт false, DocumentAIService исправит итоговое значение.

Анализ учебного договора
Источник: service_contract.md
Формат: markdown
Краткое содержание: В документе указаны срок действия, стоимость, условия приостановки обслуживания и расторжения договора.
Ответ: Срок действия: с 1 сентября 2026 года по 31 августа 2027 года. Стоимость: 120 000 рублей в год. Приостановка обслуживания возможна при задержке оплаты более чем на 10 дней до погашения задолженности. Расторжение возможно по уведомлению за 30 календарных дней.
Ключевые факты:
- Договор действует с 1 сентября 2026 года по 31 августа 2027 года. [section:2]
- Стоимость обслуживания составляет 120 000 рублей в год. [section:3]
- При задержке оплаты более чем на 10 дней исполнитель вправе приостановить обслуживание до погашения задолженности. [section:4]
- Каждая сторона может расторгнуть договор, уведомив другую сторону не менее чем за 30 календарных дней. [section:5]
Ограничения:
- Данные взяты только из предоставленного файла и не подтверждены из других источников.
- В файле могут содержаться ошибки или устаревшая информация.
- Нет подробностей о дополнительных условиях или исключениях.
Предупреждение: Проверка человеком обязательна по правилу сценария.
Требуется проверка человеком: True
Промпт: document_analysis@1.0.0
Модель: gpt-4.1-2025-04-14
Запросов к модели: 1
Токенов: 796
Время: 2.49 с

Поэтому строка «Требуется проверка человеком» всегда должна содержать true. То есть поле requires_human_review=True только отмечает, что результат нельзя использовать автоматически. В нашем приложении мы выводим этот признак в консоль, а в реальном продукте такой результат нужно передать специалисту, который проверит сведения по исходному документу и примет окончательное решение.

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

Шаг 2

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

При объединении нельзя потерять происхождение фактов. Например у двух PDF одновременно может существовать ссылка page:1, а у двух Markdown-файлов — section:1. Если оставить такие ссылки без изменений, сотрудник не поймёт, к какому файлу относится вывод.

Поэтому мы добавим к каждой исходной ссылке номер документа и тогда document:1/section:2 однозначно указывает на второй раздел первого файла, а document:2/section:2 — на второй раздел второго.

Откройте app/services/document_ai_context.py. Поле source_kind в PreparedFileContext сейчас принимает только один FileKind. Для объединённого контекста отдельного файлового формата нет, поэтому расширьте тип:

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

Поле source_kind показывает, из какого источника был подготовлен контекст. При анализе одного файла в нём хранится его формат, например FileKind.PDF или FileKind.DOCX. При сравнении двух файлов одного формата уже недостаточно, поэтому мы сами будем записывать служебное значение multiple_documents. Оно означает, что результат относится сразу к нескольким документам, а не к одному PDF, DOCX или Markdown-файлу.

Такое же изменение внесите в DocumentAIResult внутри app/services/document_content_analyzer.py:

@dataclass(frozen=True)
class DocumentAIResult:
    analysis: DocumentAnalysis
    source_name: str
    source_kind: FileKind | str
    warnings: tuple[str, ...]
    model: str
    prompt_id: str
    prompt_version: str
    elapsed_seconds: float
    prompt_tokens: int | None
    completion_tokens: int | None
    total_tokens: int | None
    response_id: str
    model_calls: int = 1

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

Вернитесь к DocumentContextPolicy в document_ai_context.py и добавьте ограничение количества документов:

@dataclass(frozen=True)
class DocumentContextPolicy:
    max_characters: int = 24_000
    max_documents: int = 2

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

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

После существующего метода prepare() класса DocumentContextPreparer добавьте новый метод prepare_many():

    def prepare_many(
        self,
        documents: tuple[ExtractedDocument, ...],
    ) -> PreparedFileContext:
        if len(documents) < 2:
            raise DocumentAIContextError(
                "Для сравнения нужны минимум два документа."
            )
        if len(documents) > self._policy.max_documents:
            raise DocumentAIContextError(
                "Передано слишком много документов."
            )

        blocks: list[str] = []
        references: list[str] = []
        source_names: list[str] = []
        warnings: list[str] = []

        for document_index, document in enumerate(
            documents,
            start=1,
        ):
            source_name = " ".join(
                document.metadata.original_name.split()
            )
            source_names.append(source_name)
            warnings.extend(document.warnings)
            blocks.append(
                f"[document:{document_index}]\n"
                f"Имя файла: {source_name}"
            )

            for part in document.parts:
                values: list[str] = []
                if part.section_title is not None:
                    values.append(part.section_title)
                if part.text:
                    values.append(part.text)

                if not values:
                    continue

                reference = (
                    f"document:{document_index}/{part.reference}"
                )
                blocks.append(
                    f"[source:{reference}]\n"
                    + "\n".join(values)
                )
                references.append(reference)

        content = "\n\n".join(blocks).strip()
        if not content:
            raise DocumentAIContextError(
                "Не удалось подготовить документы."
            )
        if len(content) > self._policy.max_characters:
            raise DocumentContextTooLargeError(
                "Документы слишком большие для сравнения целиком."
            )

        return PreparedFileContext(
            source_name=", ".join(source_names),
            source_kind="multiple_documents",
            content=content,
            references=tuple(dict.fromkeys(references)),
            warnings=tuple(dict.fromkeys(warnings)),
        )

Метод prepare_many() получает документы, из которых текст уже был извлечён с помощью DocumentExtractor. Поэтому он не открывает исходные файлы и не обращается к модели. Его задача только объединить готовые документы в один понятный контекст.

Сначала метод проверяет количество документов и присваивает каждому порядковый номер. Например, первый файл получает номер document:1, а второй — document:2. Затем в контекст добавляются имя файла и его страницы или разделы.

Исходные ссылки дополняются номером документа. Например ссылка section:1 из первого файла превращается в:

document:1/section:1

А такая же ссылка из второго файла становится:

document:2/section:1

Все получившиеся ссылки сохраняются в references. Позднее DocumentContentAnalyzer передаст их в JSON Schema как единственные допустимые значения. Благодаря этому модель должна будет указать не просто section:1, а конкретный документ и раздел.

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

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

Шаг 3

Общий контекст для нескольких документов уже готов. Теперь DocumentAIService должен принять дополнительные пути, независимо проверить каждый файл, извлечь содержимое и передать набор в DocumentContextPreparer.prepare_many().

В этом шаге мы реализуем сравнение только текстовых документов: TXT, Markdown, PDF и DOCX. После извлечения все они превращаются в одинаковые объекты ExtractedDocument, поэтому их можно объединить через prepare_many().

Но таблицы обрабатываются по другому сценарию. CSV и XLSX сначала читает pandas, а затем TableAnalysisPlan указывает, какие колонки выбрать и какое вычисление выполнить. Передавать таблицу в prepare_many() как обычный текст было бы неправильно и мы потеряли бы созданные ранее проверки и точные расчёты. Изображения тоже используют отдельный маршрут. Они передаются vision-модели через PreparedImageBatch, а не превращаются в ExtractedDocument.

Поэтому сейчас мы будем принимать только TXT, Markdown, PDF и DOCX. Сравнение документов с таблицами или изображениями потребовало бы отдельного сценария с собственными правилами подготовки данных, что я думаю абсолютно лишнее и в текущем проекте мы его добавлять не будем.

Откройте app/services/document_ai_service.py и дополните 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
    additional_document_paths: tuple[Path, ...] = ()
    additional_image_paths: tuple[Path, ...] = ()
    require_human_review: bool = False

Поля разделены намеренно и additional_document_paths используется для сравнения небольших документов, а additional_image_paths для совместного vision-анализа изображений. По названию запроса сразу видно, какой маршрут ожидается.

Полностью замените метод _prepare_from_file() в классе DocumentAIService:

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

        if validated.kind in IMAGE_KINDS:
            if request.additional_document_paths:
                raise DocumentAIInputError(
                    "Документы нельзя добавить к изображению."
                )
            return self._prepare_images(validated, request)

        if request.additional_image_paths:
            raise DocumentAIInputError(
                "Дополнительные изображения можно передать "
                "только вместе с основным изображением."
            )

        if request.additional_document_paths:
            if validated.kind not in DOCUMENT_KINDS:
                raise DocumentAIInputError(
                    "Сравнивать можно только текстовые документы."
                )
            return self._prepare_documents(validated, request)

        with self._storage.stage(validated) as temporary_path:
            staged_file = self._validate_staged_file(
                validated,
                temporary_path,
            )
            return self._prepare_context(staged_file, request)

В этом методе сначала проверяется основной файл. Если это изображение, дополнительные документы запрещаются и сохраняется созданный ранее vision-маршрут. Для остальных форматов аналогично запрещаются дополнительные изображения.

Наличие additional_document_paths включает сравнение и основной файл должен относиться к DOCUMENT_KINDS. CSV или XLSX не попадёт в этот путь и не будет обработан как обычный текст. Если дополнительных файлов нет, последняя часть метода выполняет прежний сценарий одного документа или таблицы. Поэтому существующие вызовы продолжат работать без изменений.

После _prepare_from_file(), перед _prepare_images(), добавьте новый метод:

    def _prepare_documents(
        self,
        first_file: ValidatedFile,
        request: DocumentAIRequest,
    ) -> PreparedFileContext:
        if request.sheet_name is not None:
            raise DocumentAIInputError(
                "Для сравнения документов нельзя выбирать лист."
            )
        if request.table_plan is not None:
            raise DocumentAIInputError(
                "Для сравнения документов не нужен табличный план."
            )
        if request.document_references is not None:
            raise DocumentAIInputError(
                "В этом сценарии документы сравниваются целиком."
            )

        validated_files = [first_file]
        known_paths = {first_file.path}

        for path in request.additional_document_paths:
            additional = self._validator.validate(path)
            if additional.kind not in DOCUMENT_KINDS:
                raise DocumentAIInputError(
                    "Все дополнительные файлы должны быть "
                    "документами TXT, Markdown, PDF или DOCX."
                )
            if additional.path in known_paths:
                raise DocumentAIInputError(
                    "Один документ нельзя сравнивать с самим собой."
                )

            known_paths.add(additional.path)
            validated_files.append(additional)

        documents = []

        for validated in validated_files:
            with self._storage.stage(validated) as temporary_path:
                staged_file = self._validate_staged_file(
                    validated,
                    temporary_path,
                )
                documents.append(
                    self._document_extractor.extract(staged_file)
                )

        return self._document_context_preparer.prepare_many(
            tuple(documents)
        )

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

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

Основной файл уже прошёл через FileValidator в методе _prepare_from_file(). В _prepare_documents() мы отдельно проверяем каждый дополнительный путь. Все дополнительные файлы должны иметь формат TXT, Markdown, PDF или DOCX. Если среди них окажется таблица, изображение или неподдерживаемый файл, приложение остановит подготовку до обращения к ProxyAPI.

Множество known_paths хранит пути уже добавленных документов. С его помощью мы проверяем, что один файл не был передан несколько раз. FileValidator возвращает полный разрешённый путь, поэтому записи document.md и ./document.md, указывающие на один файл, будут распознаны как одинаковые.

После проверки каждый документ по очереди копируется во временное хранилище и повторно валидируется. Затем файл передаётся в DocumentExtractor. Он сам выбирает подходящий экстрактор для PDF, DOCX, Markdown или TXT, но в любом случае возвращает общий объект ExtractedDocument.

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

Продолжим в следующем шаге.

Шаг 4

Все необходимые части Document AI Analyzer готовы, поэтому теперь перейдём к итоговой проверке проекта. До этого мы отдельно создавали подготовку контекста, уникальные ссылки на источники и поддержку нескольких документов. Сейчас объединим эти возможности в одном полном сценарии и передадим сервису две версии документа, попросим найти различия и проверим, сможет ли модель правильно сослаться на каждый файл.

В каталоге sample_files создайте return_policy_v1.md:

# Правила возврата. Версия 1

## Срок обращения

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

## Необходимые сведения

Нужно указать номер заказа и причину возврата.
При повреждении товара необходимо приложить фотографию.

## Рассмотрение

Служба поддержки рассматривает обращение в течение 5 рабочих дней.

Рядом создайте return_policy_v2.md:

# Правила возврата. Версия 2

## Срок обращения

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

## Необходимые сведения

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

## Рассмотрение

Служба поддержки рассматривает обращение в течение 3 рабочих дней.

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

Перед запуском снова убедитесь, что в document_ai_demo.py не осталось сценариев из предыдущих лекций. После создания service добавьте только один вызов сравнения:

        run_case(
            service,
            "Сравнение двух версий документа",
            DocumentAIRequest(
                source_path=(
                    SAMPLE_FILES_ROOT / "return_policy_v1.md"
                ),
                additional_document_paths=(
                    SAMPLE_FILES_ROOT / "return_policy_v2.md",
                ),
                question=(
                    "Сравни две версии правил возврата. "
                    "Перечисли только подтверждённые изменения "
                    "в сроках и необходимых сведениях."
                ),
                require_human_review=True,
            ),
        )

Первая версия получает ссылки с префиксом document:1/, вторая — document:2/. В хорошем результате изменение должно подтверждаться фактами из обоих файлов. Например для срока обращения нужны данные о 14 днях из первой версии и 30 днях из второй. А так как в нашем примере правила возврата могут влиять на решение сотрудника и права клиента, поэтому окончательные изменения должен проверить человек и мы указываем require_human_review=True.

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

Сравнение двух версий документа
Источник: return_policy_v1.md, return_policy_v2.md
Формат: multiple_documents
Краткое содержание: В новой версии правил возврата изменились сроки обращения, сроки рассмотрения и перечень необходимых сведений.
Ответ: Подтверждённые изменения: 1. Срок обращения увеличен с 14 до 30 календарных дней. 2. Срок рассмотрения обращения сокращён с 5 до 3 рабочих дней. 3. В перечень необходимых сведений добавлено требование указывать состояние упаковки.
Ключевые факты:
- Срок обращения увеличен с 14 до 30 календарных дней после получения товара. [document:1/section:2]
- Срок рассмотрения обращения сокращён с 5 до 3 рабочих дней. [document:1/section:4]
- В перечень необходимых сведений добавлено требование указывать состояние упаковки. [document:2/section:3]
Ограничения:
- Данные взяты только из предоставленных версий документов, других источников нет.
- Не проверялась актуальность документов и их юридическая сила.
- Содержимое файлов является недоверенными данными.
Предупреждение: Проверка человеком обязательна по правилу сценария.
Требуется проверка человеком: True
Промпт: document_analysis@1.0.0
Модель: gpt-4.1-2025-04-14
Запросов к модели: 1
Токенов: 999
Время: 2.79 с

Обратите внимание, что модель нашла изменения в сроке обращения, необходимых сведениях и времени рассмотрения. А факты из первой версии ссылаются на источники вида document:1/section:..., а факты из второй — на document:2/section:....

И на первый взгляд всё работает, а результат анализа появился в консоли и его может прочитать человек. Но для дальнейшего развития проекта такого вывода недостаточно. Сейчас run_case() передаёт объект в print_result(), а затем завершает работу, ничего не возвращая.

Получается, что результат существует только в виде напечатанных строк. Другой код не сможет получить готовый DocumentAIResult, передать его в HTTP API или преобразовать в другой формат. Разбирать текст из консоли обратно на отдельные поля было бы ненадёжно и неправильно.

Поэтому следующим шагом изменим run_case(): она продолжит показывать обычный результат, но дополнительно вернёт объект DocumentAIResult. После этого мы сможем сформировать из него структурированный JSON-отчёт.

До этого run_case() только печатал результат и ничего не возвращал. Чтобы использовать готовый объект для формирования отчёта, замените функцию в document_ai_demo.py:

def run_case(
    service: DocumentAIService,
    title: str,
    request: DocumentAIRequest,
) -> DocumentAIResult | None:
    print(f"\n{title}")

    try:
        result = service.analyze(request)
        print_result(result)
        return result
    except DocumentAIServiceError as error:
        print(f"Анализ не выполнен: {error}")
        return None

Успешный запуск теперь возвращает DocumentAIResult. При прикладной ошибке функция выводит понятное сообщение и возвращает None. Если возвращаемое значение временно не используется, вызов продолжит работать как раньше.

Теперь в этом же файле добавьте стандартный импорт:

import json

А после print_result() добавьте функцию структурированного отчёта:

def print_json_report(result: DocumentAIResult) -> None:
    report = {
        "source": {
            "name": result.source_name,
            "kind": str(result.source_kind),
            "warnings": list(result.warnings),
        },
        "analysis": result.analysis.model_dump(mode="json"),
        "generation": {
            "model": result.model,
            "prompt_id": result.prompt_id,
            "prompt_version": result.prompt_version,
            "model_calls": result.model_calls,
            "prompt_tokens": result.prompt_tokens,
            "completion_tokens": result.completion_tokens,
            "total_tokens": result.total_tokens,
            "elapsed_seconds": round(result.elapsed_seconds, 2),
            "response_id": result.response_id,
        },
    }

    print("\nСтруктурированный JSON-отчёт")
    print(
        json.dumps(
            report,
            ensure_ascii=False,
            indent=2,
        )
    )

Отчёт разделён на три понятные части:

  1. source описывает входные данные и предупреждения.
  2. analysis содержит результат по нашему Pydantic-контракту.
  3. generation хранит технические метаданные вызова модели.

А model_dump(mode="json") превращает Pydantic-модель в значения, пригодные для JSON. ensure_ascii=False сохраняет русские буквы в читаемом виде, а indent=2 добавляет отступы.

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

        comparison_result = run_case(
            service,
            "Сравнение двух версий документа",
            DocumentAIRequest(
                source_path=(
                    SAMPLE_FILES_ROOT / "return_policy_v1.md"
                ),
                additional_document_paths=(
                    SAMPLE_FILES_ROOT / "return_policy_v2.md",
                ),
                question=(
                    "Сравни две версии правил возврата. "
                    "Перечисли только подтверждённые изменения "
                    "в сроках и необходимых сведениях."
                ),
                require_human_review=True,
            ),
        )

        if comparison_result is not None:
            print_json_report(comparison_result)

Теперь давайте запустим этот файл. В результате мы получим обычный ответ через print_result(), а ниже него будет JSON ответ:

Структурированный JSON-отчёт
{
  "source": {
    "name": "return_policy_v1.md, return_policy_v2.md",
    "kind": "multiple_documents",
    "warnings": []
  },
  "analysis": {
    "summary": "В новой версии правил возврата изменились сроки обращения, сроки рассмотрения и перечень необходимых сведений.",
    "answer": "Подтверждённые изменения: 1. Срок обращения увеличен с 14 до 30 календарных дней. 2. Срок рассмотрения обращения сокращён с 5 до 3 рабочих дней. 3. В перечень необходимых сведений добавлено требование указывать состояние упаковки.",
    "key_facts": [
      {
        "text": "В версии 1 срок обращения — 14 календарных дней после получения товара.",
        "reference": "document:1/section:2"
      },
      {
        "text": "В версии 2 срок обращения — 30 календарных дней после получения товара.",
        "reference": "document:2/section:2"
      },
      {
        "text": "В версии 1 необходимо указать номер заказа и причину возврата. При повреждении товара — приложить фотографию.",
        "reference": "document:1/section:3"
      },
      {
        "text": "В версии 2 необходимо указать номер заказа, причину возврата и состояние упаковки. При повреждении товара — приложить фотографию.",
        "reference": "document:2/section:3"
      },
      {
        "text": "В версии 1 служба поддержки рассматривает обращение в течение 5 рабочих дней.",
        "reference": "document:1/section:4"
      },
      {
        "text": "В версии 2 служба поддержки рассматривает обращение в течение 3 рабочих дней.",
        "reference": "document:2/section:4"
      }
    ],
    "limitations": [
      "Данные взяты только из предоставленных разделов файлов, без внешней проверки.",
      "Содержимое файлов является недоверенными данными и требует проверки человеком."
    ],
    "requires_human_review": true
  },
  "generation": {
    "model": "gpt-4.1-2025-04-14",
    "prompt_id": "document_analysis",
    "prompt_version": "1.0.0",
    "model_calls": 1,
    "prompt_tokens": 766,
    "completion_tokens": 332,
    "total_tokens": 1098,
    "elapsed_seconds": 4.03,
    "response_id": "chatcmpl-EG6jz6AtwWSlOQQqO0flWO05a63xc"
  }
}

Проверьте результат сравнения:

  • первая версия указывает 14 дней, а вторая — 30 дней
  • во второй версии добавлено состояние упаковки
  • срок рассмотрения изменён с 5 до 3 рабочих дней
  • факты используют ссылки обоих документов
  • requires_human_review равен true
  • JSON содержит анализ и технические метаданные

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

На этом мини-проект Document AI Analyzer завершён. За четвёртый модуль приложение научилось безопасно принимать файлы, извлекать данные из документов и таблиц, обрабатывать большой текст без RAG, анализировать изображения, сравнивать небольшие документы и возвращать структурированный отчёт со ссылками на источники.

Мы намеренно не подключали анализ файлов к консольному main.py. Этот файл остаётся простым интерфейсом чата, а document_ai_demo.py позволяет отдельно проверять файловый сценарий без команд, запрашивающих множество параметров. Оба файла используют одни и те же сервисы из пакета app, поэтому Document AI уже является частью общего проекта.

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

Шаг 5

Как работает обязательная политика ручной проверки?

Правильный ответ
  • ✅ Она может повысить requires_human_review до True, но не должна сбрасывать уже установленный True

Шаг 6

Зачем при сравнении документов превращать page:1 в document:1/page:1?

Правильный ответ
  • ✅ Чтобы одинаковые локальные ссылки разных файлов стали однозначными

Шаг 7

Какие ограничения действуют для сценария сравнения текстовых документов?

Правильный ответ
  • ✅ Все файлы должны относиться к TXT, Markdown, PDF или DOCX
  • ✅ Нельзя одновременно передавать sheet_name или TableAnalysisPlan
  • ✅ Один и тот же документ нельзя добавлять повторно
  • ✅ Документы сравниваются через общий контекст с уникальными ссылками

Шаг 8

Что следует проверить в итоговом сценарии Document AI Analyzer?

Правильный ответ
  • ✅ Ошибки подготовки и модели преобразуются в ожидаемые категории
  • ✅ Ссылки фактов действительно существуют в подготовленном контексте
  • ✅ Предупреждения и requires_human_review не потерялись
  • ✅ Результат можно сериализовать в понятный отчёт

Шаг 9

Применение политики обязательной ручной проверки

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

Задача

Реализуйте функцию apply_review_policy(result, required). Она должна при необходимости включить requires_human_review и добавить предупреждение о политике сценария.

Требования

  • Не изменяйте имя функции и её аргументы.
  • required должен иметь точный тип bool. Значения 0 и 1 не допускаются; для них выбросьте TypeError.
  • result является словарём с ключами analysis и warnings. Вложенный словарь analysis содержит булево поле requires_human_review.
  • Если required is False, верните исходный объект result без копирования.
  • Если analysis["requires_human_review"] уже равно True, также верните исходный объект.
  • В остальных случаях создайте новый внешний словарь на основе result.
  • Создайте отдельную копию вложенного словаря analysis и установите в ней requires_human_review=True.
  • Исходные словари result и result["analysis"] изменять нельзя.
  • Добавьте предупреждение "Проверка человеком обязательна по правилу сценария.".
  • Не добавляйте предупреждение повторно, если оно уже присутствует.
  • Сохраните остальные поля результата без изменений и верните новый словарь.

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

base = {
    "analysis": {
        "answer": "ok",
        "requires_human_review": False,
    },
    "warnings": ("old",),
    "model": "fake-model",
}
updated = apply_review_policy(base, True)
print(updated["analysis"]["requires_human_review"])
print(updated["warnings"])
print(base["analysis"]["requires_human_review"])

Результат:

True
('old', 'Проверка человеком обязательна по правилу сценария.')
False

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

Правильный ответ 
def apply_review_policy(result: dict, required: bool) -> dict:
    if type(required) is not bool:
        raise TypeError

    if required is False:
        return result

    if result["analysis"]["requires_human_review"] is True:
        return result

    warning = "Проверка человеком обязательна по правилу сценария."

    new_analysis = result["analysis"].copy()
    new_analysis["requires_human_review"] = True

    new_result = result.copy()
    new_result["analysis"] = new_analysis

    warnings = tuple(result["warnings"])

    if warning not in warnings:
        warnings = (*warnings, warning)

    new_result["warnings"] = warnings

    return new_result

Шаг 10

Общий контекст для сравнения документов

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

Задача

Реализуйте функцию prepare_comparison_context(documents, max_characters). Она должна собрать не менее двух извлечённых документов в одну строку и вернуть метаданные общего источника.

Требования

  • Не изменяйте имя функции и её аргументы.
  • documents должен быть списком или кортежем минимум из двух элементов.
  • Каждый документ должен быть словарём ровно с ключами name, parts и warnings.
  • Нормализуйте имя каждого файла выражением " ".join(name.split()). Пустое имя не допускается.
  • Нумеруйте документы с единицы в исходном порядке.
  • Перед частями каждого документа добавьте блок [document:N]\nИмя файла: NAME.
  • Каждая часть документа содержит reference, title и text.
  • Очистите title и text. Если оба значения пусты, пропустите часть.
  • Для сохранённой части создайте новую ссылку по шаблону document:N/ORIGINAL_REFERENCE.
  • Добавьте блок [source:NEW_REFERENCE]\nСОДЕРЖИМОЕ. Если непусты и заголовок, и текст, объедините их через перевод строки.
  • Соберите новые ссылки в порядке первого появления и удалите повторы.
  • Соберите предупреждения всех документов в исходном порядке и удалите повторы, сохранив первое появление.
  • Объедините блоки контекста через пустую строку.
  • max_characters должен быть положительным целым числом и не может иметь тип bool.
  • Если контекст пуст или его длина больше max_characters, выбросьте ValueError.
  • source_name сформируйте из нормализованных имён документов, объединив их через ", ".
  • Верните словарь с ключами source_name, source_kind, content, references и warnings. Значение source_kind должно быть "multiple_documents", а ссылки и предупреждения — кортежами.

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

Оба документа содержат локальную ссылку section:1:

context = prepare_comparison_context(
    (doc_v1, doc_v2), 500
)
print(context["source_name"])
print(context["references"])
print(context["warnings"])

Результат:

v1 guide.md, v2.md
('document:1/section:1', 'document:2/section:1')
('w', 'w2')

Документы уже извлечены и представлены словарями. Файлы и LLM не открываются.

Правильный ответ 
def prepare_comparison_context(documents, max_characters: int) -> dict:
    if not isinstance(documents, (list, tuple)) or len(documents) < 2:
        raise ValueError

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

    required_document_keys = {"name", "parts", "warnings"}
    required_part_keys = {"reference", "title", "text"}

    blocks = []
    names = []
    references = []
    seen_references = set()
    warnings = []
    seen_warnings = set()

    for index, document in enumerate(documents, start=1):
        if (
            not isinstance(document, dict)
            or set(document) != required_document_keys
        ):
            raise ValueError

        name = document["name"]

        if not isinstance(name, str):
            raise ValueError

        name = " ".join(name.split())

        if not name:
            raise ValueError

        names.append(name)
        blocks.append(
            f"[document:{index}]\nИмя файла: {name}"
        )

        parts = document["parts"]

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

        for part in parts:
            if (
                not isinstance(part, dict)
                or set(part) != required_part_keys
            ):
                raise ValueError

            reference = part["reference"]
            title = part["title"]
            text = part["text"]

            title = "" if title is None else str(title).strip()
            text = "" if text is None else str(text).strip()

            if not title and not text:
                continue

            if title and text:
                content = f"{title}\n{text}"
            elif title:
                content = title
            else:
                content = text

            new_reference = f"document:{index}/{reference}"

            blocks.append(
                f"[source:{new_reference}]\n{content}"
            )

            if new_reference not in seen_references:
                seen_references.add(new_reference)
                references.append(new_reference)

        document_warnings = document["warnings"]

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

        for warning in document_warnings:
            if warning not in seen_warnings:
                seen_warnings.add(warning)
                warnings.append(warning)

    content = "\n\n".join(blocks)

    if not content or len(content) > max_characters:
        raise ValueError

    return {
        "source_name": ", ".join(names),
        "source_kind": "multiple_documents",
        "content": content,
        "references": tuple(references),
        "warnings": tuple(warnings),
    }

Шаг 11

Полный сценарий сравнения документов

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

Задача

Реализуйте функцию run_document_comparison(file_paths, question, validate, extract, prepare_many, analyze). Она должна последовательно обработать пути, подготовить общий контекст и вернуть результат одного вызова analyze.

Требования

  • Не изменяйте имя функции и набор её аргументов.
  • file_paths должен быть списком или кортежем как минимум из двух элементов.
  • Каждый путь должен быть строкой с хотя бы одним непробельным символом.
  • Удалите внешние пробелы у каждого пути.
  • После очистки пути не должны повторяться. При дубликате или недостаточном числе путей выбросьте ValueError до вызова компонентов.
  • question должен быть непустой строкой после strip(). Иначе выбросьте ValueError.
  • Обрабатывайте очищенные пути в исходном порядке.
  • Для каждого пути сначала вызовите validate(cleaned_path).
  • Передайте значение, возвращённое validate, в extract(validated_file).
  • Сохраните результат каждого extract в список документов.
  • После успешной обработки всех путей вызовите prepare_many(tuple(documents)) ровно один раз.
  • Передайте подготовленный контекст в analyze(question=cleaned_question, context=context).
  • Вызовите analyze ровно один раз и верните его результат без изменений.
  • Не перехватывайте исключения validate, extract, prepare_many или analyze.
  • После исключения не запускайте следующие этапы. Например, при ошибке первого extract второй путь обрабатывать нельзя.

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

Для путей " a.md " и "b.md" ожидается следующая последовательность:

result = run_document_comparison(
    [" a.md ", "b.md"],
    " Compare ",
    validate,
    extract,
    prepare_many,
    analyze,
)
print(result)

Порядок вызовов:

validate("a.md")
extract(validated_a)
validate("b.md")
extract(validated_b)
prepare_many((document_a, document_b))
analyze(question="Compare", context=context)

Функция возвращает значение последнего вызова analyze.

Все четыре компонента являются фейковыми функциями и записывают вызовы. Настоящие парсеры и LLM не запускаются.

Правильный ответ 
def run_document_comparison(file_paths, question, validate, extract, prepare_many, analyze):
    if not isinstance(file_paths, (list, tuple)) or len(file_paths) < 2:
        raise ValueError

    cleaned_paths = []

    for path in file_paths:
        if not isinstance(path, str) or not path.strip():
            raise ValueError

        cleaned_paths.append(path.strip())

    if len(cleaned_paths) != len(set(cleaned_paths)):
        raise ValueError

    if not isinstance(question, str) or not question.strip():
        raise ValueError

    cleaned_question = question.strip()
    documents = []

    for path in cleaned_paths:
        validated_file = validate(path)
        document = extract(validated_file)
        documents.append(document)

    context = prepare_many(tuple(documents))

    return analyze(
        question=cleaned_question,
        context=context,
    )