Усиление агента LangChain на Python с семью встроенными промежуточными компонентами
Узнайте, как промежуточный слой LangChain 1.0 добавляет функции краткого изложения, ограничений на количество вызовов, повторных попыток, использования резервных моделей, удаления персональных данных и утверждения человеком в агент Gemini, не затрагивая его основную логику.
Заставить агента LangChain отвечать на вопросы в ноутбуке занимает несколько минут. Однако создать агента, которому можно доверять в производственных условиях, гораздо сложнее: он не должен бесконечно циклировать до исчерпания лимита API, передавать номер карты клиента поставщику моделей или отправлять электронные письма без одобрения. LangChain 1.0 решает эти проблемы с помощью мидлвэра — слоя хуков, окружающего цикл агента и настраиваемого в виде простого списка. В этом руководстве сначала описывается базовая концепция, затем применяются семь мидлвэров к Python-агенту на базе Gemini, и в заключение показан собственный мидлвэр, чтобы вы могли точно понимать, что изменяет каждый хук и когда его использовать.
Чекпоинты вокруг цикла агента
Представьте аэропорт. Цель — долететь из одного города в другой, но вокруг этого основного действия существует ряд контрольных точек: регистрация подтверждает личность, досмотр багажа обеспечивается с помощью сканеров, на выходе проверяются билеты, а после посадки — зона выдачи багажа. Никто из них не управляет самолётом, и пилот не проверяет багаж. Каждый этап выполняет свою функцию до или после основного действия.
Мидлвэрт применяет ту же идею к агентам. Основой агента является цикл: вызов модели, выбор инструментов, их выполнение и повторение процесса до тех пор, пока модель не даст окончательный ответ. Мидлвэрт вставляет контрольные точки в этот цикл, не изменяя его самого:
- Удаление номеров карт перед тем, как текст попадёт в большую языковую модель, — это функция сканера безопасности.
- Требование к человеку утвердить отправляемое письмо — это выход на посадку.
- Остановка после десяти вызовов модели для ограничения расходов — это защитный предохранитель.
Если вы работаете с TypeScript, сопутствующая статья LangChain guardrails and middleware рассматривает те же идеи с точки зрения JavaScript; данный гид посвящён исключительно Python и сосредоточен на конкретных встроенных классах.
Хуки, которые может использовать middleware
Цикл предоставляет хуки на каждом этапе, и middleware может быть привязан к одному или нескольким из них:
before_agentиafter_agentвыполняются по одному разу — в самом начале и в самом конце вызова.before_modelиafter_modelсрабатывают каждый раз, когда цикл собирается вызвать модель или только что её вызвал.wrap_model_callиwrap_tool_callоборачивают сам вызов, что позволяет повторно его выполнить, заменить или получить из кэша.
Вот и весь модель мышления. Мидлвэры подключаются путем передачи списка в функцию create_agent, как показано в этом примере, где объединяются функции маскировки электронных писем, ограничения количества звонков и повторные попытки работы инструментов:
agent = create_agent(
model=model,
tools=[my_tool],
middleware=[
PIIMiddleware("email", strategy="redact"),
ModelCallLimitMiddleware(run_limit=5),
ToolRetryMiddleware(max_retries=3),
],
)
Порядок элементов в списке имеет значение. Мидлвэры применяются по порядку, подобно слоям лука, обернутым вокруг агента, поэтому элемент списка, указанный первым, обрабатывает исходные данные раньше всех остальных.
Настройка и базовый агент
Для работы мидлвэров требуется LangChain 1.0 или новее, поэтому установите его с флагом -U для обновления старой версии. В примерах используется бесплатный тариф Gemini; ключ можно создать в Google AI Studio.
!pip install -qU langchain langchain-google-genai
Импортируйте модуль os и класс чат-модели Gemini:
import os
from langchain_google_genai import ChatGoogleGenerativeAI
Затем настраивается модель. В этом фрагменте ключ задан вручную исключительно в целях демонстрации; в реальном коде следует экспортировать GOOGLE_API_KEY в среду разработки или загружать его из менеджера секретов, вместо того чтобы хранить его прямо в исходном коде. Значение температуры, равное нулю, обеспечивает воспроизводимость результатов:
os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)
Объектом для каждого эксперимента является минимальный агент без промежуточного программного обеспечения. Для его работы необходимы функция create_agent и декоратор tool:
from langchain.agents import create_agent
from langchain_core.tools import tool
У этого агента есть один фиктивный инструмент для получения погодных данных, который всегда сообщает о солнечной погоде, и он вызывается с одним сообщением от пользователя:
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)
Каждый следующий раздел добавляет ещё одну точку контроля для этого агента.
1. SummarizationMiddleware для ограниченной памяти
Во время длительных разговоров история сообщений постоянно увеличивается, пока не превысит размер окна контекста, и за каждый дополнительный токен взимается плата. SummarizationMiddleware проверяет размер этой истории в функции before_model; когда он превышает установленный порог, более старые сообщения сжимаются в краткое содержание, а только самые свежие сохраняются без изменений.
from langchain.agents.middleware import SummarizationMiddleware
В приведённой ниже конфигурации используется тот же модель для формирования кратких изложений, активация происходит при 10 сообщениях, причём последние 4 остаются нетронутыми. В тесте создаётся фиктивная история из шести городов (двенадцать сообщений), после чего задаётся вопрос о том, какой город появился первым:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
SummarizationMiddleware(
model=model, # which LLM writes the summary
trigger=("messages", 10), # summarize when history hits 10 messages
keep=("messages", 4), # keep the 4 most recent messages intact
),
],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history)) # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"])) # 6
Приходит тринадцать сообщений, а после этого остаётся шесть: краткое изложение, четыре сохранённых сообщения и новый ответ. Модель по-прежнему отвечает «Дели», потому что эта информация попала в краткое изложение. Подсчёт сообщений позволяет легко наблюдать такое поведение в демо-версии, но в производственных условиях триггеры, основанные на количестве токенов, такие как ("tokens", 3000), или доля контекстного окна, например ("fraction", 0.8), гораздо лучше отслеживают реальные затраты и устанавливают ограничения. Имейте в виду, что краткие изложения являются сжатыми: точные цифры или идентификаторы, упомянутые в начале разговора, могут не сохраниться.
2. Ограничения количества вызовов как защитные механизмы
Самой дорогостоящей причиной сбоев агента является бесконечный цикл, при котором модель и инструменты продолжают вызывать друг друга, тратя деньги в течение нескольких минут, прежде чем кто-либо это заметит. Два промежуточных компонента устанавливают строгий лимит для этого:
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
Здесь модель ограничена тремя вызовами за одну попытку выполнения, причем используется exit_behavior="end", чтобы агент корректно остановился, вместо того чтобы вызвать исключение; кроме того, количество вызовов инструментов ограничено двумя. В запросе намеренно просится информация о шести городах по одной:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
# "end" = stop gracefully instead of raising an error
ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
ToolCallLimitMiddleware(run_limit=2),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)
Агенту требуется шесть поисков, но он достигает своего лимита и завершает работу с частичными результатами, которые у него есть. Также существует параметр thread_limit, позволяющий ограничить количество вызовов во всем потоке общения, а не только в одной попытке выполнения. Эти два механизма являются простым резервом; устанавливайте лимиты значительно выше, чем требуется для законных запросов, чтобы они срабатывали только в случае реальных сбоев.
3. ToolRetryMiddleware для ненадежных зависимостей
Реальные инструменты могут выходить из строя: HTTP-запросы таймаутят, а соединения прерываются. ToolRetryMiddleware использует функцию wrap_tool_call, чтобы обнаруживать сбои и повторять попытки с экспоненциальным увеличением интервалов.
from langchain.agents.middleware import ToolRetryMiddleware
Для демонстрации этого инструмент для получения котировок акций считает количество попыток и вызывает ConnectionError после первых двух неудачных попыток, прежде чем добиться успеха. Мидлвэр предусматривает до трех попыток повтора, начиная с односекундной задержки, которая удваивается при каждой следующей попытке:
attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
"""Get the current stock price for a ticker symbol."""
attempt_counter["count"] += 1
print(f" [tool called — attempt #{attempt_counter['count']}]")
if attempt_counter["count"] < 3:
raise ConnectionError("API timeout — please retry")
return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
model=model,
tools=[flaky_stock_price],
middleware=[
ToolRetryMiddleware(
max_retries=3, # retry a failed tool up to 3 times
initial_delay=1.0, # wait 1s before first retry
backoff_factor=2.0, # double the wait each time: 1s, 2s, 4s
),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)
Инструмент дважды выходит из строя, мидлвэр ждет и пытается снова, и на третьей попытке операция успешна. С точки зрения агента ничего не пошло не так. Повторные попытки безопасны только для идемпотентных операций, таких как чтение данных; повторная попытка использования инструмента, который берет деньги с карты или отправляет сообщение, может привести к повторению побочных эффектов.
4. ModelFallbackMiddleware для ситуаций простоя поставщиков
Та же самая концепция устойчивости может защитить вызов модели. Когда основная модель терпит неудачу из-за ограничения скорости или простоя, например, этот промежуточный компонент повторяет запрос к резервным моделям в том порядке, в котором они указаны:
from langchain.agents.middleware import ModelFallbackMiddleware
В примере в качестве основной используется более легкая модель Gemini, а в качестве резерва добавляется ещё одна модель Gemini:
backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
model=model, # primary: gemini-3.5-flash-lite
tools=[get_weather],
middleware=[ModelFallbackMiddleware(backup_model)],
)
Пока основная модель работает нормально, вы ничего не заметите, что и предусмотрено. Проблемы проявляются только тогда, когда у поставщика возникают трудности. Для более надежной защиты рассмотрите возможность использования резерва от другого поставщика, поскольку простой часто влияет на все модели, работающие через один и тот же API.
5. PIIMiddleware для конфиденциальных данных
Часто вы вовсе не хотите, чтобы электронные адреса, номера карт или IP-адреса отправлялись поставщику модели. PIIMiddleware сканирует текст в методе before_model, до того как модель его увидит, и применяет одну из четырех стратегий: redact, mask, hash или block.
from langchain.agents.middleware import PIIMiddleware
У этого агента нет специальных инструментов. Он полностью заменяет электронные адреса на местоопределители и маскирует номера карт так, чтобы оставались только последние четыре цифры; оба правила применяются к вводимым пользователями данным:
agent = create_agent(
model=model,
tools=[],
middleware=[
# Replace emails entirely with [REDACTED_EMAIL]
PIIMiddleware("email", strategy="redact", apply_to_input=True),
# Mask credit cards — keeps last 4 digits
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Draft a support reply to priya.sharma@example.com confirming her card "
"4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)
Модель формирует свой ответ, так и не получая настоящего адреса или полного номера карты. Вы также можете зарегистрировать собственный тип персональных данных с использованием собственного регулярного выражения, например для блокировки любого текста, похожего на внутренний API-ключ:
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")
Обнаружение на основе шаблонов позволяет выявлять корректно сформированные значения, но не каждую креативную форму написания, поэтому следует рассматривать его как один из уровней защиты, а не гарантию соблюдения правил.
6. HumanInTheLoopMiddleware для контрольных точек утверждения
Некоторые действия слишком важны, чтобы выполняться полностью автономно: отправка электронных писем, удаление записей или осуществление платежей. HumanInTheLoopMiddleware немедленно останавливает агента перед запуском чувствительного инструмента, ждет решения человека, а затем продолжает работу.
Этот механизм отличается от предыдущих промежуточных компонентов. Агент не прерывается полностью; контрольный механизм сохраняет его полное состояние, а идентификатор потока служит ключом для последующего поиска и возобновления его работы. Проверяющий может утвердить запрос, изменить его параметры или отклонить его.
Импорты включают мидлвэр, внутренний контроллер состояния от LangGraph и тип Command, используемый для возобновления работы:
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
У приведённого ниже агента есть инструмент send_email; он отмечается как подлежащий прерыванию, а состояние паузы сохраняется в InMemorySaver. При первом вызове на потоке demo-1 агенту задаётся задача отправить письмо менеджеру:
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to the given recipient."""
return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
model=model,
tools=[send_email],
middleware=[
# Pause and ask a human whenever the agent wants to call send_email
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
],
checkpointer=InMemorySaver(), # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
{"messages": [{"role": "user", "content":
"Send an email to boss@company.com saying the report is ready."}]},
config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])
Выполнение останавливается до запуска инструмента. Запись __interrupt__ в результатах показывает задержанную отправку с указанием получателя, темы и содержимого; письмо так и не было отправлено. Одобрение передаётся в виде Command на том же потоке, что приводит к возобновлению приостановленной работы:
# Step 2: approve and resume
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config, # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)
Вместо approve можно отправить reject с причиной или edit с изменёнными аргументами. В реальном приложении именно здесь следует отображать экран для утверждения. Два практических замечания: InMemorySaver теряет состояние при перезапуске процесса, поэтому в продакшн-системах требуется постоянный указатель состояния; кроме того, формат данных для возобновления работы изменился между версиями LangChain, поэтому уточните его в документации по middleware для используемой вами версии.
7. Создание собственного middleware с использованием декоратора
Когда встроенные решения не подходят, собственный middleware остаётся небольшим, поскольку у каждого хука существует соответствующий декоратор. Необходимыми импортами являются декоратор before_model и тип AgentState:
from langchain.agents.middleware import before_model, AgentState
В этом примере фиксируется количество сообщений, которые собираются быть отправлены при каждом вызове модели, и они добавляются в список так же, как и любой готовый промежуточный компонент:
@before_model
def log_before_model(state: AgentState, runtime) -> None:
print(f" [middleware] Calling model with {len(state['messages'])} messages")
# Returning None = observe only.
# Returning a dict would UPDATE the agent's state (e.g., trim messages)
return Noneagent = create_agent(
model=model,
tools=[get_weather],
middleware=[log_before_model], # plugs in like any prebuilt middleware
)
Значение, возвращаемое функцией, является важным элементом проектирования. Возврат None означает, что промежуточный компонент лишь наблюдает за процессом. Возврат словаря позволяет обновлять состояние агента, что используется для фильтрации сообщений, вставки контекста или применения пользовательских ограничений. Для других моментов взаимодействия существуют декораторы: @before_agent, @after_model, @wrap_model_call, @wrap_tool_call, а также @dynamic_prompt для создания системных промптов во время выполнения.
Основные выводы
- Промежуточные компоненты разделяют операционные аспекты на логику агента: цикл остается прежним, в то время как контрольные точки добавляются к нему в виде обычного списка, передаваемого функции
create_agent.
TodoListMiddleware, LLMToolSelectorMiddleware и ContextEditingMiddleware, описан в официальном руководстве.Связанная литература
- От чат-бота с одним узлом к агенту, работающему на основе MCP в LangGraph — Создавайте приложения LangGraph постепенно, слой за слоем: состояния и редюсеры, ребра, циклы с инструментами, потоки с сохранением состояния, три режима стриминга и инструменты, предоставляемые через MCP.
- Гибкая память агента: объединение методов BM25 и векторного поиска с использованием RRF на Python — Узнайте, почему чистый векторный поиск не подходит в качестве памяти агента, как метод Reciprocal Rank Fusion объединяет результаты BM25 и плотных поисков в Python, и когда помогают резюме, сгенерированные с использованием GraphRAG.