Мидлварь: как менять поведение агента

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

#Зачем нужен отдельный механизм

Цикл агента внутри create_agent фиксирован: модель, инструменты, снова модель. Как только появляются требования уровня продакшена (ограничить число вызовов, вырезать персональные данные, переключиться на резервную модель, сжать историю), возникает соблазн отказаться от create_agent и собрать граф вручную.

Мидлварь закрывает этот разрыв. Это объекты, которые встраиваются в известные точки цикла и получают доступ к состоянию до и после каждого шага. Агент остаётся стандартным, а поведение меняется.

#Точки встраивания

Хуков шесть, и делятся они на две группы по способу работы.

Узловые хуки выполняются в конкретной точке и возвращают обновление состояния либо None, если менять нечего. before_agent срабатывает один раз в начале запуска, before_model перед каждым обращением к модели, after_model после каждого ответа, after_agent один раз в конце.

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

#Простейшая мидлварь на декораторах

Для разовых задач достаточно функции с декоратором. Сигнатура одинаковая: состояние агента и рантайм на входе, словарь с обновлением состояния или None на выходе:

from typing import Any
from langchain.agents.middleware import before_model, after_model, AgentState
from langgraph.runtime import Runtime

@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"Отправляем модели {len(state['messages'])} сообщений")
    return None

@after_model
def log_response(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"Модель ответила: {state['messages'][-1].text}")
    return None

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search],
    middleware=[log_before_model, log_response],
)

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

#Досрочный выход

Мидлварь может прервать цикл, вернув jump_to. Допустимые цели: "end" завершает работу агента, "model" отправляет на новый вызов модели, "tools" переводит к выполнению инструментов. Намерение прыгать объявляется заранее через hook_config:

from langchain.agents.middleware import after_model, hook_config, AgentState
from langchain.messages import AIMessage

@after_model
@hook_config(can_jump_to=["end"])
def block_unsafe_answer(state: AgentState, runtime) -> dict[str, Any] | None:
    last_message = state["messages"][-1]
    if "BLOCKED" in last_message.text:
        return {
            "messages": [AIMessage("Не могу ответить на этот запрос.")],
            "jump_to": "end",
        }
    return None

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

#Мидлварь классом

Когда нужно переиспользование, собственные поля состояния или дополнительные инструменты, мидлварь оформляют классом:

from typing import Any
from langchain.agents.middleware import AgentMiddleware, AgentState
from langgraph.runtime import Runtime

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"Отправляем модели {len(state['messages'])} сообщений")
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"Модель ответила: {state['messages'][-1].text}")
        return None

У класса три полезных атрибута. state_schema расширяет состояние агента собственными полями, tools добавляет инструменты вместе с мидлварью, transformers меняет то, что уходит в поток. Асинхронные версии хуков объявляются с префиксом a (abefore_model), и агент выбирает нужную по режиму вызова.

#Собственные поля состояния

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

from typing import Any
from typing_extensions import NotRequired
from langchain.agents.middleware import AgentState, before_model, after_model

class CustomState(AgentState):
    model_call_count: NotRequired[int]

@before_model(state_schema=CustomState, can_jump_to=["end"])
def check_call_limit(state: CustomState, runtime) -> dict[str, Any] | None:
    if state.get("model_call_count", 0) > 10:
        return {"jump_to": "end"}
    return None

@after_model(state_schema=CustomState)
def increment_counter(state: CustomState, runtime) -> dict[str, Any] | None:
    return {"model_call_count": state.get("model_call_count", 0) + 1}

Счётчик в состоянии переживает прерывания и восстанавливается вместе с контрольной точкой, чего не даст переменная в объекте.

#Порядок выполнения

Список мидлварей обрабатывается не как простая очередь. Хуки before_* идут в порядке списка, after_* — в обратном, а wrap_* вкладываются друг в друга подобно вызовам функций: первая мидлварь оборачивает все последующие.

Для списка [m1, m2, m3] порядок такой: m1.before_model, m2.before_model, m3.before_model, затем m1.wrap_model_call снаружи, внутри неё m2.wrap_model_call, внутри той m3.wrap_model_call и сам вызов модели, а после ответа m3.after_model, m2.after_model, m1.after_model.

Отсюда правило порядка: то, что должно видеть финальную картину и иметь последнее слово, ставят первым в списке. Например, мидлварь безопасности разумно поставить в начало: её after_model отработает последним и сможет заблокировать ответ, уже изменённый остальными.

#Готовые мидлвари

Значительная часть типовых задач уже решена. Все перечисленные классы импортируются из langchain.agents.middleware.

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

Ограничения и надёжность: ModelCallLimitMiddleware и ToolCallLimitMiddleware не дают агенту крутиться бесконечно, ModelRetryMiddleware и ToolRetryMiddleware повторяют неудачные вызовы с нарастающей задержкой, ToolErrorMiddleware превращает исключения инструментов в сообщения, на которые модель может отреагировать, ModelFallbackMiddleware переключается на резервные модели при отказе основной.

Работа с инструментами и данными: LLMToolSelectorMiddleware оставляет модели только релевантные инструменты из большого набора, PIIMiddleware находит и обрабатывает персональные данные, LLMToolEmulator подменяет реальные инструменты имитацией для тестов, TodoListMiddleware даёт агенту инструмент планирования задач, HumanInTheLoopMiddleware ставит выполнение на паузу для подтверждения человеком.

Подключаются они списком, как обычные объекты:

from langchain.agents.middleware import (
    ModelCallLimitMiddleware,
    SummarizationMiddleware,
    ToolRetryMiddleware,
)

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search, fetch_page],
    middleware=[
        ModelCallLimitMiddleware(thread_limit=20, run_limit=8),
        SummarizationMiddleware(model="openai:gpt-4o-mini"),
        ToolRetryMiddleware(max_retries=3),
    ],
)

Разница между thread_limit и run_limit важна на практике: первый считает вызовы за всё время жизни диалога, второй — за один запуск. Диалог из сотни реплик упрётся в thread_limit, зациклившийся агент в рамках одного запроса — в run_limit.

#Подмена модели на лету

wrap_model_call получает запрос и обработчик. Внутри можно изменить запрос, вызвать обработчик и даже вызвать его повторно с другими параметрами. Типичный сценарий: длинные диалоги отправлять модели с большим контекстным окном, короткие — дешёвой:

from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model

cheap_model = init_chat_model("openai:gpt-4o-mini")
strong_model = init_chat_model("openai:gpt-5.5")

@wrap_model_call
def pick_model_by_length(request, handler):
    model = strong_model if len(request.state["messages"]) > 30 else cheap_model
    return handler(request.override(model=model))

override не меняет исходный запрос, а возвращает новый: объект ModelRequest неизменяем. Тем же способом подменяют системное сообщение (request.override(system_message=...)) или набор инструментов для конкретного шага.

Так же оборачиваются инструменты. Перехват исключения в wrap_tool_call превращает падение в сообщение, которое модель прочитает и попробует исправить ввод:

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:
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Ошибка инструмента, проверьте аргументы и попробуйте снова. ({e})",
            tool_call_id=request.tool_call["id"],
        )

#Чего делать не стоит

Не храните изменяемое состояние в полях объекта мидлвари: она общая для всех запусков и потоков, а состояние должно жить в схеме агента.

Не делайте тяжёлые операции в before_model: этот хук вызывается перед каждым обращением к модели, и лишний запрос к базе умножается на число шагов.

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

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

Устранение неисправностей