Структурированный ответ агента и middleware — AI агенты на Python: шаг за шагом

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

Вторая проблема — контроль. Агент может зациклиться на вызовах инструментов, а инструмент — упасть с исключением и оборвать весь запуск. В первом модуле вы решали это вручную через max_steps и try/except. В этом уроке вы научитесь получать от агента ответ строго по схеме Pydantic и подключать middleware, которые ограничивают число шагов и перехватывают ошибки инструментов.

Что уже умеем

В уроке 1.5 вы получали ответ модели по схеме Pydantic, в уроке 1.6 ограничивали цикл агента параметром max_steps. В модуле 2 вы собрали агента на create_agent с инструментами, системным промптом и памятью, а в прошлом уроке написали первый middleware с @before_model. Сегодня объединим эти навыки.

Что вы узнаете в рамках статьи

как получить ответ агента по схеме Pydantic через response_format;
где искать структурированный ответ в результате агента;
как ограничить число вызовов модели с помощью ModelCallLimitMiddleware;
как ограничить вызовы инструментов через ToolCallLimitMiddleware;
как перехватывать ошибки инструментов через @wrap_tool_call.

Ключевые термины

response_format — параметр create_agent, в который передаётся схема финального ответа агента.
structured_response — ключ результата агента, где лежит проверенный объект схемы.
Встроенный middleware — готовый класс LangChain, решающий типовую задачу: лимиты, повторы, суммаризация.
Хук — момент цикла агента, в который выполняется middleware: перед моделью, после модели, вокруг вызова инструмента.
@wrap_tool_call — декоратор middleware, который оборачивает каждый вызов инструмента и может обработать его результат или ошибку.

Ответ по схеме: response_format

Опишите нужный ответ моделью Pydantic, как в уроке 1.5, и передайте её в параметр response_format. Агент, как и раньше, может вызывать инструменты сколько нужно, но в конце обязан вернуть данные по схеме. LangChain проверит их и положит готовый объект в result["structured_response"].

from typing import Literal
from pydantic import BaseModel, Field
from langchain.agents import create_agent

class Ticket(BaseModel):
    """Разобранное обращение клиента."""
    category: Literal["доставка", "оплата", "возврат", "другое"]
    order_id: int | None = Field(description="Номер заказа, если он указан")
    urgent: bool = Field(description="Нужна ли срочная реакция")
    summary: str = Field(description="Суть обращения в одном предложении")

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[],
    system_prompt="Ты разбираешь обращения клиентов интернет-магазина.",
    response_format=Ticket,
)

result = agent.invoke({"messages": [{"role": "user", "content":
    "Заказ 1077 должен был прийти вчера, а его нет! Завтра улетаю, срочно!"}]})

ticket = result["structured_response"]
print(ticket)
print(ticket.category, ticket.urgent)

Что увидим в консоли:

category='доставка' order_id=1077 urgent=True summary='Клиент не получил заказ 1077 вовремя и просит срочно решить вопрос.'
доставка True

ticket — настоящий объект Ticket: поля имеют правильные типы, category принимает только одно из четырёх значений, а order_id — число. Такой результат можно сразу сохранить в базу без ручного разбора.

Для OpenAI LangChain использует встроенный механизм structured outputs провайдера, который вы видели в уроке 1.5. Для моделей без такой поддержки схема передаётся как специальный инструмент: этот способ можно включить явно, обернув схему в ToolStrategy из langchain.agents.structured_output. В обоих случаях код чтения ответа одинаковый.

Структурированный ответ вместе с инструментами

response_format не мешает агенту работать с инструментами. Модель сначала собирает данные, а затем оформляет итог по схеме.

from langchain.tools import tool

ORDERS = {1042: "доставлен", 1077: "в пути, задержка 2 дня"}

@tool
def order_status(order_id: int) -> str:
    """Возвращает статус заказа по его номеру."""
    return ORDERS.get(order_id, f"Ошибка: заказ {order_id} не найден")

class Answer(BaseModel):
    order_id: int
    status: str = Field(description="Статус заказа из инструмента")
    reply: str = Field(description="Ответ клиенту, не длиннее двух предложений")

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[order_status],
    response_format=Answer,
)

result = agent.invoke({"messages": [{"role": "user", "content": "Где мой заказ 1077?"}]})
answer = result["structured_response"]
print(answer.status, "|", answer.reply)

Что увидим: в пути, задержка 2 дня | Ваш заказ 1077 в пути, доставка задерживается на 2 дня. Приносим извинения. Программа получила и данные для системы (status), и готовый текст для клиента (reply). Ищите ответ именно в structured_response: последнее сообщение в result["messages"] при этом может быть служебным.

Ограничение шагов: ModelCallLimitMiddleware

Помните max_steps из урока 1.6? В LangChain его роль играет встроенный middleware. ModelCallLimitMiddleware считает обращения к модели и останавливает агента при достижении лимита. Параметр run_limit ограничивает обращения за один вызов invoke, а thread_limit — за весь диалог, но он работает только с checkpointer. ToolCallLimitMiddleware аналогично ограничивает вызовы инструментов — всех сразу или одного, указанного в tool_name.

from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[order_status],
    middleware=[
        ModelCallLimitMiddleware(run_limit=5, exit_behavior="end"),
        ToolCallLimitMiddleware(tool_name="order_status", run_limit=3),
    ],
)

Что увидим: обычные вопросы агент решает за 2–3 шага, и лимиты незаметны. Если модель зациклится, на пятом обращении агент остановится. При exit_behavior="end" он завершит работу спокойно, без исключения, а при "error" выбросит исключение, которое вы обработаете в своём коде. Лимит на order_status не даёт агенту перебирать номера заказов: четвёртый вызов будет заблокирован, и модель получит сообщение об этом вместо результата.

Ошибки инструментов: @wrap_tool_call

В уроке 2.2 вы возвращали текст ошибки из самого инструмента. Но неожиданные исключения всё равно случаются: сеть упала, сторонняя библиотека выбросила ошибку. Чтобы не оборачивать в try/except каждый инструмент, напишите один middleware с @wrap_tool_call. Он получает запрос на вызов и функцию handler, которая выполняет инструмент.

from collections.abc import Callable
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest

@wrap_tool_call
def handle_tool_errors(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
    name = request.tool_call["name"]
    print(f"Вызов {name} с аргументами {request.tool_call['args']}")
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Инструмент {name} временно недоступен ({e}). Сообщи клиенту.",
            tool_call_id=request.tool_call["id"],
        )

agent = create_agent(
    model="openai:gpt-6-luna",
    tools=[order_status],
    middleware=[handle_tool_errors, ModelCallLimitMiddleware(run_limit=5)],
)

Что увидим: при каждом вызове в консоли появится строка с именем инструмента и аргументами — это простое логирование. Если инструмент выбросит исключение, агент не упадёт: модель получит ToolMessage с понятным текстом и скажет клиенту, что сервис временно недоступен. Обратите внимание на tool_call_id: он связывает ответ с конкретным запросом модели, как вы делали вручную в уроке 1.6.

Middleware передаются списком и могут сочетаться: обработка ошибок, лимиты, обрезка истории из прошлого урока. Каждый решает одну задачу, а вместе они делают агента надёжным.

Частые ошибки

Искать структурированный ответ в последнем сообщении. Проверенный объект лежит в result["structured_response"], а не в result["messages"][-1].content.
Делать все поля схемы обязательными. Если номера заказа в обращении нет, модель будет вынуждена его выдумать. Необязательные поля объявляйте как int | None.
Запускать агента без лимита шагов. Без ModelCallLimitMiddleware зациклившийся агент тратит деньги, пока не сработает внутренний предел фреймворка.
Отдавать модели сырой текст исключения. В нём бывают пути к файлам и детали инфраструктуры. Пишите в ToolMessage короткое понятное сообщение.

Наиболее частые вопросы

Можно ли использовать response_format без инструментов? — Да, тогда агент работает как надёжный извлекатель данных: один вызов модели и проверенный объект на выходе.
Что будет, если модель вернёт данные не по схеме? — LangChain проверит ответ и при ошибке попросит модель исправиться. Поведение настраивается параметром handle_errors у ToolStrategy.
В каком порядке выполняются middleware? — В порядке списка. Если результат зависит от порядка, держите middleware как можно более независимыми друг от друга.
Чем run_limit отличается от thread_limit? — run_limit считает обращения за один вызов invoke, thread_limit — за весь диалог в потоке и требует checkpointer.

Итоги

Поздравляем с завершением второго модуля! Теперь ваш агент отдаёт программе проверенные данные по схеме, не уходит в бесконечный цикл и не падает от ошибки инструмента. Вы освоили LangChain 1.x целиком: модели и сообщения, инструменты, системный промпт, память и middleware — а значит, можете собрать надёжного агента для реальной задачи.

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