Зміцнення агента Python LangChain за допомогою семи вбудованих проміжників
Дізнайтеся, як проміжний сервіс LangChain 1.0 додає функції узагальнення, обмежень кількості викликів, повторних спроб, альтернативних моделей, приховування персональних даних та людського схвалення до агента Gemini, не впливаючи на його основну логіку.
Щоб агент LangChain відповідав на запитання в ноутбуці, потрібно кілька хвилин. Однак створити агента, якому можна довіряти у продакшені, складніше: він не повинен бути у безкінечному циклі, який виснажує бюджет API, передавати номер картки клієнта постачальнику моделей чи надсилати електронні листи, які ніхто не схвалив. LangChain 1.0 вирішує ці проблеми за допомогою мідлверу — шару хуків навколо циклу агента, який можна налаштувати у вигляді простого списку. У цьому посібнику спочатку описується концепція, потім застосовуються сім мідлверів до Python-агента на базі Gemini, а на завершення — створюється власний мідлвер, щоб ви могли точно бачити, що змінює кожен хук та коли його використовувати.
Чекпоїнти навколо циклу агента
Уявіть собі аеропорт. Метою є переліт з одного міста в інше, проте навколо цієї основної дії існує низка контрольних пунктів: реєстрація підтверджує особу, безпекові сканери перевіряють багаж, термінал верифікує проїзні та після посадки бере на себе збір багажу. Жоден з них не керує літаком, а пілот не перевіряє багаж. Кожен етап виконує свою функцію до чи після основної дії.
Мідлверк застосовує ту саму ідею до агентів. Основою агента є цикл: виклик моделі, дозвіл їй обрати інструменти, їх виконання та повторення процесу до тих пір, доки модель не надасть остаточну відповідь. Мідлверк додає контрольні пункти навколо цього циклу, не змінюючи його:
- Видалення номерів карток перед тим, як текст потрапить до LLM, — це безпековий сканер.
- Вимога до людини схвалити відправлену електронну пошту — це термінал посадки.
- Зупинка після десяти викликів моделі для обмеження витрат — це аварійний вимикач.
Якщо ви працюєте з 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)
Модель формує свою відповідь, так і не отримавши справжню адресу чи повний номер картки. Ви також можете зареєструвати власний тип PII із власним регулярним виразом, наприклад, щоб заблокувати будь-який текст, схожий на внутрішній API-ключ:
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")
Виявлення на основі шаблонів фіксує правильно сформовані значення, а не кожну варіацію написання, тому слід розглядати його як один із рівнів захисту, а не гарантію дотримання правил.
6. HumanInTheLoopMiddleware для контрольних точок схвалення
Деякі дії мають занадто серйозні наслідки для повної автономії: надсилання електронних листів, видалення записів чи здійснення оплат. HumanInTheLoopMiddleware негайно зупиняє агента перед запуском конфіденційного інструменту, чекає на рішення людини, а потім продовжує роботу.
Цей механізм відрізняється від попередніх мідлварів. Зупинений агент не завершується. Checkpointer зберігає його повний стан, а ідентифікатор потоку використовується як ключ для знаходження та продовження його роботи пізніше. Оглядач може схвалити запит, змінити його параметри чи відхилити його.
Імпорти включають мідлвейр, контролер стану в пам’яті від 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.