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

Практические AI-функции

Шаг 1

За первые два модуля наш проект заметно вырос. В нем появились надежный LLMClient, версионируемые промпты, чат с историей, суммаризация, потоковая выдача и evaluation-сценарии. В третьем модуле мы добавили Pydantic-модели, JSON Schema, структурированные ответы и семантическую проверку.

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

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

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

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

Какие роли уже есть в проекте?

Каталог app/llm описывает интеграцию с LLM-провайдером. LLMClient знает адрес ProxyAPI, параметры SDK, форматы ответов и технические ошибки. Он не должен знать, что такое заявка в поддержку или как вести диалог.

Каталог app/prompts хранит инструкции и их версии. app/models описывает типизированные данные и их проверки, а app/schemas готовит JSON Schema для API. Эти каталоги уже имеют ясную роль, поэтому менять их не будем.

Сервисы описывают сценарии приложения. ChatService управляет диалогом, ConversationSummaryService сжимает историю, а SupportService создает резюме обращения. Именно эти классы сейчас лежат вперемешку с остальными файлами app.

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

Чтобы сгруппировать связанные классы и сделать структуру проекта нагляднее, вынесем сервисы в отдельный пакет. Создайте внутри app каталог services, а в нем пустой файл __init__.py. Этот файл позволяет Python воспринимать каталог как пакет, из которого можно импортировать модули с сервисами.

И переместите в новый каталог три файла:

app/chat_service.py
→ app/services/chat_service.py

app/conversation_summary_service.py
→ app/services/conversation_summary_service.py

app/support_service.py
→ app/services/support_service.py

В PyCharm это можно сделать перетаскиванием или через Refactor → Move File. Среда разработки может предложить автоматически обновить импорты. Даже если вы согласились, нужно вручную проверить все места, ведь автоматический рефакторинг не заменяет проверку.

Исправляем импорты в основном приложении

В main.py замените импорты ChatService и ConversationSummaryService:

from app.services.chat_service import ChatService
from app.services.conversation_summary_service import (
    ConversationSummaryService,
)

В app/console.py измените путь к типам чата:

from app.services.chat_service import (
    ChatReply,
    ChatService,
    ChatServiceError,
    ChatStreamCompleted,
    ChatTextDelta,
)

Внутри самого app/services/chat_service.py замените импорт сервиса суммаризации:

from app.services.conversation_summary_service import (
    ConversationSummaryError,
    ConversationSummaryService,
)

Остальные импорты внутри перенесенных файлов остаются прежними. Например ChatService по-прежнему импортирует app.chat_history, app.llm.client и app.prompts.support_chat. Мы изменяем только те пути, в которых файл действительно переехал.

Не забываем про evaluation-код, ведь сервисы использует не только main.py. Во втором модуле мы создали evaluation-запуск, который тоже импортирует ChatService и ConversationSummaryService. Если обновить только основной запуск, чат заработает, но evals позднее упадут при импорте.

В evals/chat_checks.py замените начало импорта:

from app.services.chat_service import (
    ChatReply,
    ChatService,
    ChatServiceError,
)

В evals/run_chat_evals.py замените два импорта:

from app.services.chat_service import ChatService
from app.services.conversation_summary_service import (
    ConversationSummaryService,
)

Сами evaluation-сценарии и логика их проверки не меняются. Мы лишь указали новое место, в котором живут те же классы.

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

Монитор из заказа 7718 пришел с треснутой подставкой. Это мое первое обращение.

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

Как теперь выглядит проект?

app/
├── services/
│   ├── __init__.py
│   ├── chat_service.py
│   ├── conversation_summary_service.py
│   └── support_service.py
├── llm/
├── models/
├── prompts/
├── schemas/
├── __init__.py
├── chat_history.py
├── config.py
└── console.py

Мы не стали делить проект на множество абстрактных слоев. config.py, console.py и chat_history.py пока остались в корне app, ведь у каждого из них понятная роль, а сами файлы пока не образуют большую группу. Хорошая структура не означает, что каждый файл нужно спрятать в свой каталог.

В следующем шаге в services появится новый прикладной сценарий. Перед его созданием мы уточним, какие именно данные должна возвращать практическая AI-функция, и вынесем ее инструкцию из structured_demo.py в версионируемый реестр.

Шаг 2

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

Отдельные части этого сценария у нас уже есть. Мы умеем отправлять модели JSON Schema, преобразовывать полученный JSON в Pydantic-модель и выполнять дополнительную семантическую проверку. Пока все это находится в structured_demo.py и используется как учебный эксперимент.

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

Что такое AI-функция?

AI-функция — это законченная операция приложения, внутри которой используется модель. С точки зрения остального кода она должна выглядеть почти как обычная Python-функция или метод: получить входные данные и вернуть результат.

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

result = analyzer.analyze(user_text)

На этой строке не видно обращения к ProxyAPI, промпта, JSON Schema и Pydantic. Это и есть важное свойство хорошей границы, когда вызывающий код знает, какую задачу выполняет метод, но ему не нужно разбираться во всех технических шагах внутри.

Конечно AI-функция отличается от обычного математического вычисления. Ее результат создается моделью, поэтому он может зависеть от формулировки запроса, версии промпта и выбранной модели. Именно поэтому внутри такой функции требуется больше проверок. Она должна не просто вызвать LLM, а убедиться, что ответ завершен, соответствует ожидаемой структуре и не противоречит исходному тексту.

Пользователь или другой код приложения передаст анализатору текст обращения. Анализатор соберет сообщения и попросит LLMClient выполнить структурированную генерацию. Клиент отправит запрос через ProxyAPI и вернет технический результат вызова.

После этого анализатор проверит результат. Сначала JSON будет преобразован в Pydantic-модель. Затем отдельная семантическая проверка сопоставит полученные данные с исходным обращением. Только если все этапы завершатся успешно, вызывающий код получит готовый объект анализа.

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

Какую задачу будет решать наша AI-функция?

На вход она получит исходный текст клиента, например:

Вчера получил заказ 8041 с кофемашиной. Устройство не включается, проверил две розетки.

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

Чтобы сформировать такой объект, модель выполнит несколько связанных операций.

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

Извлечение данных означает поиск конкретных фактов в исходном тексте. В нашем примере модель должна найти номер заказа 8041 и товар «кофемашина». Если номера заказа нет, поле order_id должно содержать None. Придумывать отсутствующее значение нельзя.

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

Определение недостающих сведений показывает, каких данных не хватает для дальнейшей обработки. Если клиент не назвал товар, значение product будет равно None, а в missing_fields появится product. Благодаря этому приложение сможет понять, какой уточняющий вопрос следует задать.

И перед тем как перейти к коду, давайте обсудим что мы сделаем дальше. Сам SupportRequestAnalyzer появится только в следующем шаге, но сейчас мы подготовим его контракт и инструкцию:

  1. добавим в Pydantic-модель описание проблемы и оценку уверенности
  2. дополним проверки, связывающие значения с missing_fields
  3. вынесем developer-инструкцию из structured_demo.py в отдельный версионируемый промпт

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

Начнем с Pydantic-модели и дополним ее. Для этого откройте app/models/support_request.py. В перечислении MissingField уже есть значение PROBLEM_DESCRIPTION, но в самой модели SupportRequestAnalysis пока нет поля с описанием проблемы.

Из-за этого модель может определить категорию и товар, но не может вернуть короткую формулировку того, что именно произошло. Добавьте после поля product:

    problem_description: str | None = Field(
        description="Краткое описание проблемы или null,если суть обращения непонятна.")

problem_description не должно содержать копию всего обращения. Его задача только сохранить краткую суть проблемы, например «кофемашина не включается» или «подставка монитора повреждена». Если из текста вообще невозможно понять, что произошло, значение будет равно None.

И чуть забежим вперед, и после missing_fields добавим оценку уверенности:

    confidence: float = Field(ge=0, le=1,
                              description="Оценка уверенности модели в анализе от 0 до 1.")

Параметры ge=0 и le=1 задают допустимый диапазон. Pydantic отклонит, например, -0.2 или 1.4. При этом он проверяет только границы числа и не может определить, насколько честно модель оценила свою уверенность. Что это за поле и за что оно отвечает мы поговорим в последнем шаге данного раздела.

Модель может вернуть текст с пробелами по краям или повторяющимися пробелами между словами. Это не смысловая ошибка, поэтому удобнее исправить ее автоматически при создании Pydantic-объекта. Поэтому после validate_order_id() добавьте валидатор:

    @field_validator("product", "problem_description")
    @classmethod
    def normalize_optional_text(cls, value: str | None) -> str | None:
        if value is None:
            return None

        normalized = " ".join(value.split())
        if not normalized:
            raise ValueError(
                "Текстовое поле не должно быть пустым."
            )
        return normalized

Один валидатор применяется сразу к product и problem_description. Если значение равно None, оно остается без изменений. Для строки метод split() разделяет текст по пробельным символам, а " ".join(...) собирает его обратно с одинарными пробелами.

Например, строка " кофемашина не включается " превратится в "кофемашина не включается". Если строка состояла только из пробелов, валидатор выбросит ошибку. Пустая строка и отсутствие значения — это разные состояния, поэтому вместо пустой строки модель должна вернуть None.

Теперь дополним семантическую проверку. Ведь Pydantic проверит, что problem_description содержит строку или None, но он не проверяет связь этого поля со списком missing_fields.

А мы договорились о следующем правиле:

  • если значение поля отсутствует, соответствующее поле должно находиться в missing_fields
  • если значение заполнено, его не должно быть в missing_fields

Например если product=None, то в списке должен находиться MissingField.PRODUCT. Если product содержит "кофемашина", добавлять MissingField.PRODUCT уже нельзя.

Такое правило уже применяется к order_id и product. Теперь оно должно работать и для problem_description. Можно было бы скопировать ещё две проверки, но тогда один и тот же код повторялся бы три раза. При добавлении новых полей количество повторений продолжало бы расти.

Вместо этого опишем проверяемые поля в одном кортеже, а общее правило применим к ним в цикле.

Откройте app/models/support_request_validation.py и полностью замените существующую функцию validate_support_request():

def validate_support_request(analysis: SupportRequestAnalysis, user_text: str) -> None:
    issues: list[str] = []

    fields_to_check = (
        (
            "order_id",
            analysis.order_id,
            MissingField.ORDER_ID,
        ),
        (
            "product",
            analysis.product,
            MissingField.PRODUCT,
        ),
        (
            "problem_description",
            analysis.problem_description,
            MissingField.PROBLEM_DESCRIPTION,
        ),
    )

    for field_name, value, missing_field in fields_to_check:
        is_marked_missing = (
                missing_field in analysis.missing_fields
        )

        if value is None and not is_marked_missing:
            issues.append(
                f"{field_name} равен null, "
                "но не указан в missing_fields."
            )

        if value is not None and is_marked_missing:
            issues.append(
                f"{field_name} заполнен, "
                "но указан в missing_fields."
            )

    if analysis.order_id is not None:
        order_ids = set(
            re.findall(r"\b\d+\b", user_text)
        )
        if analysis.order_id not in order_ids:
            issues.append(
                "order_id отсутствует в исходном обращении."
            )

    if issues:
        raise SupportRequestSemanticError(tuple(issues))

Теперь функция выполняет две группы проверок.

Сначала она проверяет согласованность order_id, product и problem_description со списком missing_fields. Для каждого поля действует одно правило:

  • если значение равно None, поле должно быть отмечено как отсутствующее
  • если значение заполнено, отметки об отсутствии быть не должно

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

Затем отдельно проверяется order_id. Если модель вернула номер заказа, функция ищет такой же номер среди последовательностей цифр в исходном обращении. Эта проверка относится только к номеру заказа, поэтому она не входит в общий цикл.

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

И вот после этих правок можно полноценно сказать что контракт данных у нас подготовлен. И теперь нам нужно вынести developer-инструкцию из демонстрационного файла. И так как данный промпт влияет на категории, извлеченные значения и confidence, поэтому его нужно хранить и версионировать так же, как инструкции чата и суммаризации. Тогда изменение текста не уничтожит предыдущий вариант, а evaluation-наборы смогут сравнивать поведение разных версий.

Создайте каталог app/prompts/support_request_analysis. Внутри создайте пустой __init__.py и файл v1_0_0.py следующего содержания:

from app.prompts.template import PromptTemplate

SUPPORT_REQUEST_ANALYSIS_PROMPT_V1_0_0 = PromptTemplate(
    prompt_id="support_request_analysis",
    version="1.0.0",
    body="""
Ты анализируешь обращения клиентов службы поддержки.

Задачи:
1. Выбери одну категорию из разрешенных схемой.
2. Извлеки номер заказа и товар.
3. Кратко сформулируй описание проблемы.
4. Определи срочность и недостающие сведения.
5. Оцени уверенность в анализе числом от 0 до 1.

Правила:
- Используй только факты из обращения.
- Не придумывай отсутствующие данные.
- Если факт не указан, верни null и добавь поле в missing_fields.
- Нормализуй категорию до значения из схемы.
- Снижай confidence, если текст неоднозначен или данных мало.
""".strip(),
    required_variables=frozenset(),
)

Теперь создадим реестр версий, для этого в том же каталоге создайте registry.py:

from app.prompts.support_request_analysis.v1_0_0 import (
    SUPPORT_REQUEST_ANALYSIS_PROMPT_V1_0_0,
)
from app.prompts.template import PromptTemplate, PromptTemplateError

ACTIVE_SUPPORT_REQUEST_ANALYSIS_VERSION = "1.0.0"

_PROMPTS: dict[str, PromptTemplate] = {
    SUPPORT_REQUEST_ANALYSIS_PROMPT_V1_0_0.version: (
        SUPPORT_REQUEST_ANALYSIS_PROMPT_V1_0_0
    ),
}

def get_support_request_analysis_prompt(
    version: str | None = None,
) -> PromptTemplate:
    selected = (
        ACTIVE_SUPPORT_REQUEST_ANALYSIS_VERSION
        if version is None
        else version.strip()
    )

    try:
        return _PROMPTS[selected]
    except KeyError as error:
        available = ", ".join(_PROMPTS)
        raise PromptTemplateError(
            f"Неизвестная версия: {selected}. "
            f"Доступны: {available}."
        ) from error

def list_support_request_analysis_versions() -> tuple[str, ...]:
    return tuple(_PROMPTS)

Теперь откройте app/prompts/support_request_analysis/__init__.py и добавьте публичные импорты:

from app.prompts.support_request_analysis.registry import (
    ACTIVE_SUPPORT_REQUEST_ANALYSIS_VERSION,
    get_support_request_analysis_prompt,
    list_support_request_analysis_versions,
)

SUPPORT_REQUEST_ANALYSIS_PROMPT = get_support_request_analysis_prompt()

__all__ = [
    "ACTIVE_SUPPORT_REQUEST_ANALYSIS_VERSION",
    "SUPPORT_REQUEST_ANALYSIS_PROMPT",
    "get_support_request_analysis_prompt",
    "list_support_request_analysis_versions",
]

Публичное имя SUPPORT_REQUEST_ANALYSIS_PROMPT указывает на активную версию. Напомню, что будущий анализатор сможет импортировать одну константу и не будет зависеть от имени файла v1_0_0.py. А при создании новой версии мы добавим ее в реестр и изменим активный номер, не переписывая сервис.

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

Шаг 3

В предыдущем шаге мы подготовили контракт AI-функции. Теперь Pydantic-модель описывает все нужные поля, семантическая проверка ищет противоречия, а developer-инструкция хранится в отдельном версионируемом промпте.

Однако эти части пока не объединены в одну операцию. Файл structured_demo.py одновременно читает настройки, собирает сообщения, вызывает модель, проверяет результат, выполняет repair и печатает данные. Поэтому в этом шаге мы создадим SupportRequestAnalyzer — отдельный сервис, который спрячет весь сценарий за одним методом analyze().

Что изменится после появления сервиса?

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

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

Создайте app/services/support_request_analyzer.py и добавьте импорты:

from dataclasses import dataclass
from time import perf_counter

from pydantic import ValidationError

from app.llm.client import LLMClient, LLMResult, Message
from app.llm.errors import LLMClientError
from app.models.support_request import SupportRequestAnalysis
from app.models.support_request_validation import (
    SupportRequestSemanticError,
    validate_support_request,
)
from app.prompts.support_request_analysis import (
    SUPPORT_REQUEST_ANALYSIS_PROMPT,
)
from app.schemas.support_request import (
    SUPPORT_REQUEST_SCHEMA,
    SUPPORT_REQUEST_SCHEMA_NAME,
)

Ниже импортов добавьте:

DEVELOPER_INSTRUCTION = SUPPORT_REQUEST_ANALYSIS_PROMPT.render()
MAX_REPAIR_ATTEMPTS = 1

Напомню, что метод render() получает текст активной версии промпта, а значение MAX_REPAIR_ATTEMPTS = 1 разрешает только один дополнительный запрос. Если модель снова вернет противоречивые данные, сервис остановится с ошибкой, а не начнет бесконечно повторять генерацию.

Далее добавьте иерархию прикладных ошибок:

class SupportRequestAnalyzerError(RuntimeError):
    pass

class SupportRequestRefusalError(SupportRequestAnalyzerError):
    pass

class IncompleteSupportRequestError(SupportRequestAnalyzerError):
    pass

class InvalidSupportRequestError(SupportRequestAnalyzerError):
    pass

Затем опишите объект успешного результата:

@dataclass(frozen=True)
class SupportRequestResult:
    analysis: SupportRequestAnalysis
    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
    repair_attempts: int

Обратите внимание, тут главное поле analysis, которое будет содержать проверенный Pydantic-объект. Остальные поля помогают понять, как был получен результат, а именно какая модель и версия промпта использовались, сколько занял анализ, сколько токенов потрачено и понадобился ли repair.

Далее создайте класс и сохраните переданный клиент:

class SupportRequestAnalyzer:
    def __init__(self, llm_client: LLMClient) -> None:
        self._llm_client = llm_client

Как и в нашем экспериментальном файле по структурированным ответам, сервис не читает настройки и не создает LLMClient самостоятельно. Он получает уже настроенный объект снаружи.

Внутри класса добавьте первоначальную сборку сообщений:

    @staticmethod
    def _build_messages(user_text: str) -> list[Message]:
        return [
            {
                "role": "developer",
                "content": DEVELOPER_INSTRUCTION,
            },
            {
                "role": "user",
                "content": (
                    "<customer_request>\n"
                    f"{user_text}\n"
                    "</customer_request>"
                ),
            },
        ]

Постоянные правила находятся в developer message, а исходное обращение — отдельно в сообщении user.

Следом добавьте сборку исправляющего запроса:

    @classmethod
    def _build_repair_messages(cls, user_text: str, issues: tuple[str, ...]) -> list[Message]:
        messages = cls._build_messages(user_text)
        issue_text = "\n".join(f"- {issue}" for issue in issues)

        messages[0] = {
            "role": "developer",
            "content": (
                f"{DEVELOPER_INSTRUCTION}\n\n"
                "Предыдущий результат не прошел проверку.\n"
                "Исправь перечисленные противоречия:\n"
                f"{issue_text}"
            ),
        }
        return messages

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

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

    @staticmethod
    def _parse_result(result: LLMResult) -> SupportRequestAnalysis:
        if result.refusal is not None:
            raise SupportRequestRefusalError(result.refusal)

        if result.finish_reason == "length":
            raise IncompleteSupportRequestError(
                "Модели не хватило выходных токенов."
            )

        if result.finish_reason != "stop":
            raise InvalidSupportRequestError(
                "Неожиданная причина завершения: "
                f"{result.finish_reason}."
            )

        if result.text is None:
            raise InvalidSupportRequestError(
                "Модель не вернула данные."
            )

        try:
            return SupportRequestAnalysis.model_validate_json(result.text)
        except ValidationError as error:
            raise InvalidSupportRequestError(
                "Ответ модели не соответствует контракту."
            ) from error

Теперь добавьте основной внутренний сценарий:

    def _request_analysis(
        self,
        user_text: str,
    ) -> tuple[SupportRequestAnalysis, LLMResult, int]:
        messages = self._build_messages(user_text)

        for repair_attempts in range(
            MAX_REPAIR_ATTEMPTS + 1
        ):
            result = self._llm_client.generate_structured(
                messages=messages,
                schema_name=SUPPORT_REQUEST_SCHEMA_NAME,
                schema=SUPPORT_REQUEST_SCHEMA,
            )
            analysis = self._parse_result(result)

            try:
                validate_support_request(analysis, user_text)
            except SupportRequestSemanticError as error:
                if repair_attempts == MAX_REPAIR_ATTEMPTS:
                    issue_text = " ".join(error.issues)
                    raise InvalidSupportRequestError(
                        "Результат не прошел смысловую "
                        f"проверку: {issue_text}"
                    ) from error

                messages = self._build_repair_messages(
                    user_text,
                    error.issues,
                )
                continue

            return analysis, result, repair_attempts

        raise RuntimeError("Недостижимое состояние.")

На первом проходе отправляется обычный запрос. Ответ проходит техническую, Pydantic и семантическую проверки. Если противоречий нет, метод сразу возвращает данные.

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

Все внутренние шаги готовы. Теперь добавьте в SupportRequestAnalyzer после __init__() публичный метод:

    def analyze(self, user_text: str) -> SupportRequestResult:
        text = user_text.strip()
        if not text:
            raise SupportRequestAnalyzerError(
                "Обращение не должно быть пустым."
            )

        started_at = perf_counter()

        try:
            analysis, llm_result, repair_attempts = self._request_analysis(text)
        except LLMClientError as error:
            raise SupportRequestAnalyzerError(
                f"Не удалось получить анализ обращения: {error}"
            ) from error

        return SupportRequestResult(
            analysis=analysis,
            model=llm_result.model,
            prompt_id=SUPPORT_REQUEST_ANALYSIS_PROMPT.prompt_id,
            prompt_version=SUPPORT_REQUEST_ANALYSIS_PROMPT.version,
            elapsed_seconds=perf_counter() - started_at,
            prompt_tokens=llm_result.prompt_tokens,
            completion_tokens=llm_result.completion_tokens,
            total_tokens=llm_result.total_tokens,
            response_id=llm_result.response_id,
            repair_attempts=repair_attempts,
        )

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

И чтобы на данном этапе проверить работу сервиса, мы можем упростить наш structured_demo.py и запустить его. Ведь логика анализа теперь находится в сервисе, поэтому полностью замените его содержимое:

import json

from pydantic import ValidationError

from app.config import Settings
from app.llm.client import LLMClient
from app.services.support_request_analyzer import (
    SupportRequestAnalyzer,
    SupportRequestAnalyzerError,
    SupportRequestResult,
)

def print_result(result: SupportRequestResult) -> None:
    print("\nПроверенный результат:")
    print(
        json.dumps(
            result.analysis.model_dump(mode="json"),
            ensure_ascii=False,
            indent=2,
        )
    )
    print(f"\nМодель: {result.model}")
    print(f"Промпт: {result.prompt_id}@{result.prompt_version}")
    print(f"Всего токенов: {result.total_tokens}")
    print(f"ID ответа: {result.response_id}")
    print(f"Исправляющих запросов: {result.repair_attempts}")

def main() -> None:
    try:
        settings = Settings()
    except ValidationError as error:
        print("Ошибка конфигурации:")
        for issue in error.errors():
            field = ".".join(str(part) for part in issue["loc"])
            print(f"- {field}: {issue['msg']}")
        return

    llm_client = LLMClient(settings)
    analyzer = SupportRequestAnalyzer(llm_client)
    user_text = input("Введите обращение клиента: ")

    try:
        result = analyzer.analyze(user_text)
    except SupportRequestAnalyzerError as error:
        print(f"Не удалось выполнить анализ: {error}")
        return
    finally:
        llm_client.close()

    print_result(result)

if __name__ == "__main__":
    main()

Теперь этот файл отвечает только за запуск: читает настройки и ввод, создает зависимости, вызывает analyze() и печатает результат. Сборки сообщений, проверки JSON и repair-цикла в нем больше нет.

Давайте проверим его, для этого запустите этот файл и отправьте полное обращение:

Вчера получил заказ 8041 с кофемашиной. Устройство не включается, проверил две розетки.

В первом случае проверьте извлеченные факты, они должны быть корректны:

{
  "category": "technical_issue",
  "order_id": "8041",
  "product": "кофемашина",
  "problem_description": "Кофемашина не включается, проверено на двух розетках.",
  "urgency": "normal",
  "missing_fields": [],
  "confidence": 1.0
}

Затем повторите запуск с неполным текстом:

Не могу включить устройство. Помогите.

Во втором модель не должна придумывать номер заказа или точное название товара: отсутствующие сведения должны отражаться через None и missing_fields.

{
  "category": "technical_issue",
  "order_id": null,
  "product": null,
  "problem_description": "Клиент не может включить устройство.",
  "urgency": "normal",
  "missing_fields": [
    "order_id",
    "product"
  ],
  "confidence": 0.7
}

Мы видим что все работает корректно, а значит в следующем шаге можно подключать этот сервис и сам анализатор к main.py и проверим новый сценарий вместе с уже существующим чатом.

Шаг 4

В предыдущем шаге мы создали SupportRequestAnalyzer. Он принимает текст, обращается к модели, создает Pydantic-объект, выполняет смысловую проверку и при необходимости делает один исправляющий запрос. При успешном вызове сервис возвращает SupportRequestResult. Его структура проверена, а поля согласованы с известными правилами. Но это еще не означает, что любой результат можно сразу использовать без участия человека.

Почему успешной валидации недостаточно?

Рассмотрим обращение:

Хочу вернуть деньги: товар перестал включаться, а завтра заканчивается срок возврата.

Модель может выбрать return_request, потому что клиент просит вернуть деньги, или technical_issue, потому что товар неисправен. И ведь оба значения разрешены схемой. Pydantic подтвердит правильный тип, а семантическая проверка может не найти противоречий. Тем не менее выбор категории остается неоднозначным.

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

Что означает confidence?

confidence — это число от 0 до 1, которым модель оценивает собственную уверенность в результате анализа. Чем полнее и однозначнее обращение, тем выше обычно будет это значение.

Однако confidence=0.9 не означает, что результат верен с вероятностью 90%. Эту оценку создаёт та же модель, которая анализировала обращение. Она может сомневаться в правильном результате или, наоборот, уверенно ошибиться. Поэтому использовать confidence как единственное основание для автоматического решения нельзя.

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

  • confidence ниже установленного порога;
  • обращение имеет высокую срочность;
  • модель не смогла выбрать известную категорию и вернула other;
  • первоначальный ответ модели содержал противоречие и потребовал repair-запроса.

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

Такую проверку результата человеком обычно называют human review. Сотрудник может подтвердить результат модели, исправить категорию или извлечённые данные либо запросить у клиента дополнительную информацию.

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

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

Создайте app/services/support_request_review.py:

from dataclasses import dataclass
from enum import StrEnum

from app.models.support_request import SupportCategory, Urgency
from app.services.support_request_analyzer import (
    SupportRequestResult,
)

MIN_AUTOMATIC_CONFIDENCE = 0.75

class ReviewReason(StrEnum):
    LOW_CONFIDENCE = "low_confidence"
    HIGH_URGENCY = "high_urgency"
    UNKNOWN_CATEGORY = "unknown_category"
    REPAIRED_RESULT = "repaired_result"

@dataclass(frozen=True)
class SupportRequestDecision:
    result: SupportRequestResult
    requires_human_review: bool
    review_reasons: tuple[ReviewReason, ...]

Константа MIN_AUTOMATIC_CONFIDENCE задает учебный порог. В настоящем проекте его подбирают по результатам evaluation-набора, когда сравнивают количество ошибок и долю обращений, переданных человеку. А ReviewReason хранит стабильные технические коды причин. Их можно выводить, сохранять и подсчитывать. SupportRequestDecision объединяет исходный анализ с решением: нужен ли человек и почему.

Создаем политику проверки, для этого ниже добавьте:

def decide_review(result: SupportRequestResult) -> SupportRequestDecision:
    analysis = result.analysis
    reasons: list[ReviewReason] = []

    if analysis.confidence < MIN_AUTOMATIC_CONFIDENCE:
        reasons.append(ReviewReason.LOW_CONFIDENCE)

    if analysis.urgency is Urgency.HIGH:
        reasons.append(ReviewReason.HIGH_URGENCY)

    if analysis.category is SupportCategory.OTHER:
        reasons.append(ReviewReason.UNKNOWN_CATEGORY)

    if result.repair_attempts > 0:
        reasons.append(ReviewReason.REPAIRED_RESULT)

    return SupportRequestDecision(
        result=result,
        requires_human_review=bool(reasons),
        review_reasons=tuple(reasons),
    )

Функция проверяет все условия и собирает каждую сработавшую причину. Низкий confidence указывает на сомнение модели. Высокая срочность требует внимания независимо от уверенности. Категория other означает отсутствие подходящего известного маршрута. А repair показывает, что первый результат содержал противоречие.

bool(reasons) равен False для пустого списка и True, если причина хотя бы одна. Поэтому флаг не может разойтись с перечнем причин.

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

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

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

Для запуска анализа добавим служебную команду /analyze. Консоль перехватит её до отправки сообщения в ChatService, отдельно попросит ввести текст обращения и передаст его анализатору. Полученный результат пройдёт правила проверки и будет выведен в консоль.

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

Откройте app/console.py и добавьте импорты:

from app.services.support_request_analyzer import (
    SupportRequestAnalyzer,
    SupportRequestAnalyzerError,
)
from app.services.support_request_review import (
    SupportRequestDecision,
    decide_review,
)

Перед run_console() добавьте функцию вывода:

def print_analysis(
    decision: SupportRequestDecision,
) -> None:
    result = decision.result

    print("\nАнализ обращения:")
    print(result.analysis.model_dump_json(indent=2))
    print(
        "Нужна проверка человеком: "
        f"{'да' if decision.requires_human_review else 'нет'}"
    )

    if decision.review_reasons:
        reasons = ", ".join(
            reason.value for reason in decision.review_reasons
        )
        print(f"Причины проверки: {reasons}")

    print(f"Промпт: {result.prompt_id}@{result.prompt_version}")
    print(f"Исправляющих запросов: {result.repair_attempts}")
    print(f"Время анализа: {result.elapsed_seconds:.2f} с")
    print(f"Всего токенов: {result.total_tokens}")
    print(f"ID ответа: {result.response_id}")

Анализатор не использует print() и не знает, как выглядит консоль. Он возвращает данные, а отдельная функция решает, как их показать. Так обязанности остаются разделенными.

Добавьте анализатор в параметры run_console():

def run_console(
    service: ChatService,
    analyzer: SupportRequestAnalyzer,
    app_env: str,
    configured_model: str,
) -> None:

Строку с командами замените:

    print("Команды: /new — новый диалог, /analyze — анализ обращения, /exit — выход.")

В цикле while после обработки /new, но до создания stream, добавьте:

        if command == "/analyze":
            request_text = input("Обращение: ")

            try:
                analysis_result = analyzer.analyze(request_text)
            except SupportRequestAnalyzerError as error:
                print(f"Ошибка анализа: {error}")
                continue

            decision = decide_review(analysis_result)
            print_analysis(decision)
            continue

Сначала выполняется AI-функция, затем обычная Python-функция применяет правила, после чего консоль показывает результат. Последний continue не позволяет команде попасть ниже в stream_reply(), поэтому анализ не добавляется в историю чата.

И последним делом обновляем main.py, и к импортам добавьте:

from app.services.support_request_analyzer import SupportRequestAnalyzer

После создания chat_service создайте анализатор:

    request_analyzer = SupportRequestAnalyzer(llm_client)

Оба сервиса получают один настроенный LLMClient. Они не создают и не закрывают его самостоятельно: за жизненный цикл общей зависимости по-прежнему отвечает main.py.

Передайте анализатор в run_console() вторым аргументом:

        run_console(
            chat_service,
            request_analyzer,
            app_env=settings.app_env,
            configured_model=settings.model,
        )

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

Вы: Монитор пришел с треснутой подставкой.

Вы: Номер заказа 7718.

Так мы убедимся, что чат по-прежнему отвечает потоком и сохраняет историю.

Затем введите /analyze и передайте:

Вчера получил заказ 8041 с кофемашиной. Устройство не включается, проверил две розетки.

Консоль должна показать JSON с постоянными полями, версию промпта, токены и решение о необходимости проверки человеком:

Для проверки высокой срочности выполните /analyze еще раз:

После включения удлинителя появился запах гари и дым. Заказ 9912, товар — удлинитель.

Мы получим следующий ответ:

Обратите внимание модель определила urgency как high, среди причин появится high_urgency независимо от значения confidence.

Давайте подводить итоги. Мы превратили учебный эксперимент в полноценную AI-функцию нашего приложения AI Support Desk. Сначала улучшили структуру проекта, затем определили контракт анализа, создали версионируемый промпт и перенесли сценарий в отдельный сервис. И вот structured_demo.py можно удалить, он нам теперь больше не нужен, ведь логика полностью переехала в наш проект.

И теперь надежность обеспечивают несколько последовательных уровней. Структурированные ответы ограничивают форму ответа. Pydantic проверяет типы и правила полей. Семантическая проверка ищет противоречия. Ограниченный repair дает модели одну возможность исправиться. Обычный Python-код определяет, нужна ли проверка человеком.

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

Шаг 5

Что лучше всего описывает AI-функцию в этом модуле?

Правильный ответ
  • ✅ Узкая операция с понятным входом, проверенным результатом и явными ошибками

Шаг 6

Зачем анализатору ограниченная исправляющая попытка после смысловой ошибки?

Правильный ответ
  • ✅ Передать модели конкретные противоречия и один раз запросить исправленный результат

Шаг 7

Выберите все причины, по которым пример из лекции направляет анализ на проверку человеком.

Правильный ответ
  • ✅ Результат был исправлен дополнительным запросом
  • ✅ Неизвестная категория
  • ✅ Низкая уверенность модели
  • ✅ Высокая срочность

Шаг 8

Что означает confidence = 0.92 в структурированном анализе?

Правильный ответ
  • ✅ Это сигнал уверенности модели, который приложение использует вместе с другими правилами

Шаг 9

Формирование причин для ручной проверки

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

Задача

Реализуйте функцию get_review_reasons(analysis, repair_attempts, min_confidence=0.75). Она должна проверить четыре независимых условия и вернуть кортеж кодов причин в установленном порядке.

Требования

  • Не изменяйте имя функции, её аргументы и значение min_confidence по умолчанию.
  • analysis является словарём с ключами confidence, urgency и category.
  • Создайте пустой список причин.
  • Если analysis["confidence"] < min_confidence, добавьте строку "low_confidence".
  • Равенство порогу не считается низкой уверенностью.
  • Если analysis["urgency"] == "high", добавьте "high_urgency".
  • Если analysis["category"] == "other", добавьте "unknown_category".
  • Если repair_attempts > 0, добавьте "repaired_result".
  • Проверяйте условия именно в указанном порядке, чтобы порядок причин был стабильным.
  • Верните причины как tuple. Если ни одно условие не выполнено, верните пустой кортеж ().

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

reasons = get_review_reasons(
    {
        "confidence": 0.6,
        "urgency": "high",
        "category": "other",
    },
    repair_attempts=1,
)
print(reasons)

Результат:

('low_confidence', 'high_urgency', 'unknown_category', 'repaired_result')

При confidence=0.75, обычной срочности, известной категории и нуле исправлений результатом должен быть ().

Заготовка функции уже находится в редакторе. Дополнительные импорты не нужны.

Правильный ответ 
def get_review_reasons(analysis: dict, repair_attempts: int, min_confidence: float = 0.75) -> tuple[str, ...]:
    reasons = []

    if analysis["confidence"] < min_confidence:
        reasons.append("low_confidence")

    if analysis["urgency"] == "high":
        reasons.append("high_urgency")

    if analysis["category"] == "other":
        reasons.append("unknown_category")

    if repair_attempts > 0:
        reasons.append("repaired_result")

    return tuple(reasons)

Шаг 10

AI-функция анализа обращения с фейковым клиентом

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

Задача

Реализуйте функцию analyze_request(client, instruction, user_text, schema_name, schema). Она должна выполнить один структурированный запрос и вернуть разобранный анализ вместе с техническими метаданными ответа.

Результат client.generate_structured() содержит поля refusal, finish_reason, text, model и response_id.

Требования

  • Не изменяйте имя функции и её аргументы.
  • user_text должен быть строкой с хотя бы одним непробельным символом. Для другого типа или пустого значения выбросьте ValueError.
  • Удалите внешние пробелы из user_text.
  • Создайте developer-сообщение с content=instruction.strip().
  • Создайте user-сообщение. Его content должен состоять из трёх строк: <customer_request>, очищенный текст обращения и </customer_request>.
  • Вызовите client.generate_structured() ровно один раз с именованными аргументами messages, schema_name=schema_name и schema=schema.
  • Если result.refusal не равен None, выбросьте ValueError.
  • Если result.finish_reason != "stop", выбросьте ValueError.
  • result.text должен быть непустой строкой. Иначе выбросьте ValueError.
  • Разберите result.text через json.loads(). Некорректный JSON должен приводить к ValueError.
  • Результат разбора должен быть словарём. Для другого типа верхнего уровня выбросьте ValueError.
  • Верните словарь с тремя ключами: analysis содержит разобранный JSON, model — result.model, response_id — result.response_id.

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

Фейковый клиент возвращает текст '{"order_id": "8041"}', модель fake-model и идентификатор resp-1:

result = analyze_request(
    fake_client,
    "Извлеки факты.",
    " Заказ 8041 ",
    "support",
    schema,
)
print(result)

Результат:

{'analysis': {'order_id': '8041'}, 'model': 'fake-model', 'response_id': 'resp-1'}

Импорт json, фейковый клиент и объект ответа предоставляет проверяющая система. Настоящий сетевой запрос не выполняется.

Правильный ответ 
import json

def analyze_request(client, instruction, user_text, schema_name, schema) -> dict:
    if not isinstance(user_text, str):
        raise ValueError

    user_text = user_text.strip()

    if not user_text:
        raise ValueError

    messages = [
        {
            "role": "developer",
            "content": instruction.strip(),
        },
        {
            "role": "user",
            "content": (
                f"<customer_request>\n"
                f"{user_text}\n"
                f"</customer_request>"
            ),
        },
    ]

    result = client.generate_structured(
        messages=messages,
        schema_name=schema_name,
        schema=schema,
    )

    if (
        result.refusal is not None
        or result.finish_reason != "stop"
        or not isinstance(result.text, str)
        or not result.text.strip()
    ):
        raise ValueError

    try:
        analysis = json.loads(result.text)
    except json.JSONDecodeError:
        raise ValueError

    if not isinstance(analysis, dict):
        raise ValueError

    return {
        "analysis": analysis,
        "model": result.model,
        "response_id": result.response_id,
    }