Промежуточное ПО LangChain
Middleware — самая мощная функция LangChain. Она позволяет вставлять пользовательскую логику на различных этапах выполнения агента для реализации повторных попыток, деградации, кэширования, фильтрации содержимого, логирования и других функций — без изменения собственного кода агента.
Что такое Middleware
Middleware состоит изхуков (Hook)в потоке выполнения агента. Каждый хук позволяет выполнять пользовательский код в определённые моменты времени:
Пример
# Интуитивное понимание Middleware:
# Предположим, поток выполнения агента выглядит так:
# 1. Ввод пользователя → 2. Размышление модели → 3. Может вызывать инструменты → 4. Модель снова размышляет → 5. Вывод результата
# Middleware позволяет вставлять пользовательскую логику между этими 5 этапами:
# 1. Ввод пользователя
# ↓ [хук before_agent: логирование, проверки разрешений]
# 2. Мышление модели
# ↓ [хук before_model: предобработка сообщений]
# ↓ [хук wrap_model_call: повторная попытка, деградация, кэширование]
# ↓ [хук after_model: модерация контента]
# 3. Выполнение инструментов
# ↓ [хук wrap_tool_call: повторная попытка вызова инструмента]
# 4. Возврат к мышлению модели (цикл до завершения)
# ↓ [хук after_agent: форматирование результатов, статистический анализ]
# 5. Вывод результата
# Предположим, поток выполнения агента выглядит так:
# 1. Ввод пользователя → 2. Размышление модели → 3. Может вызывать инструменты → 4. Модель снова размышляет → 5. Вывод результата
# Middleware позволяет вставлять пользовательскую логику между этими 5 этапами:
# 1. Ввод пользователя
# ↓ [хук before_agent: логирование, проверки разрешений]
# 2. Мышление модели
# ↓ [хук before_model: предобработка сообщений]
# ↓ [хук wrap_model_call: повторная попытка, деградация, кэширование]
# ↓ [хук after_model: модерация контента]
# 3. Выполнение инструментов
# ↓ [хук wrap_tool_call: повторная попытка вызова инструмента]
# 4. Возврат к мышлению модели (цикл до завершения)
# ↓ [хук after_agent: форматирование результатов, статистический анализ]
# 5. Вывод результата
Шесть точек хуков
Middleware в LangChain предоставляет 6 хуков, разделённых на две категории по времени выполнения:
| Хук | Частота выполнения | Позиция выполнения | Основное назначение |
|---|---|---|---|
| before_agent | Один раз | Перед запуском агента | Инициализация, проверка разрешений, предварительная обработка ввода |
| before_model | Каждая итерация цикла | Перед вызовом модели | Предварительная обработка сообщений, динамическое внедрение контекста |
| wrap_model_call | Каждая итерация цикла | Оборачивает вызов модели | Повторные попытки, деградация, кэширование, перезапись запроса |
| after_model | Каждая итерация цикла | После вызова модели | Модерация контента, фильтрация ответов, логирование |
| wrap_tool_call | Каждый вызов инструмента | Оборачивает выполнение инструмента | Повторные попытки инструмента, кэширование результатов, перезапись параметров |
| after_agent | Один раз | После завершения Agent | Форматирование вывода, статистика, очистка ресурсов |
Два способа использования
Промежуточное ПО можно использовать двумя способами: наследование классов или декораторы.
Способ 1: Декоратор (рекомендуется)
Пример
from langchain.agents.middleware import before_model, after_model
# Подход с декоратором: просто и интуитивно понятно
@before_model
def log_before(state, runtime):
"""Логирование перед каждым вызовом модели"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Текущее количество сообщений: {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Логирование после каждого вызова модели"""
last_msg = state["messages"][-1] if state.get("messages") else None
if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
print(f"[after_model] Модель запросила вызов инструмента")
return None
# Подход с декоратором: просто и интуитивно понятно
@before_model
def log_before(state, runtime):
"""Логирование перед каждым вызовом модели"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Текущее количество сообщений: {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Логирование после каждого вызова модели"""
last_msg = state["messages"][-1] if state.get("messages") else None
if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
print(f"[after_model] Модель запросила вызов инструмента")
return None
Способ 2: Наследование классов (подходит для сложной логики)
Пример
from langchain.agents.middleware import AgentMiddleware
class LoggingMiddleware(AgentMiddleware):
"""Пользовательское промежуточное ПО для логирования"""
@property
def name(self) -> str:
# Пользовательское имя промежуточного ПО (по умолчанию — имя класса)
return "logging"
def before_agent(self, state, runtime):
"""Логика перед запуском Agent"""
print("[Логирование] Агент начал выполнение")
return None
def before_model(self, state, runtime):
"""Логика перед вызовом модели"""
msg_count = len(state.get("messages", []))
print(f"[Логирование] Подготовка к вызову модели, сейчас {msg_count} сообщений")
return None
def after_model(self, state, runtime):
"""Логика после вызова модели"""
print("[Логирование] Вызов модели завершён")
return None
def after_agent(self, state, runtime):
"""Логика после завершения агента"""
print("[Логирование] Выполнение агента завершено")
return None
class LoggingMiddleware(AgentMiddleware):
"""Пользовательское промежуточное ПО для логирования"""
@property
def name(self) -> str:
# Пользовательское имя промежуточного ПО (по умолчанию — имя класса)
return "logging"
def before_agent(self, state, runtime):
"""Логика перед запуском Agent"""
print("[Логирование] Агент начал выполнение")
return None
def before_model(self, state, runtime):
"""Логика перед вызовом модели"""
msg_count = len(state.get("messages", []))
print(f"[Логирование] Подготовка к вызову модели, сейчас {msg_count} сообщений")
return None
def after_model(self, state, runtime):
"""Логика после вызова модели"""
print("[Логирование] Вызов модели завершён")
return None
def after_agent(self, state, runtime):
"""Логика после завершения агента"""
print("[Логирование] Выполнение агента завершено")
return None
Полный пример жизненного цикла
Пример
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_agent, after_agent,
before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@before_agent
def start_log(state, runtime):
"""Перед запуском агента"""
print(">>> [before_agent] Агент запущен <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Перед каждым вызовом модели"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Сообщение {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""После каждого вызова модели"""
last = state["messages"][-1] if state.get("messages") else None
if hasattr(last, 'tool_calls') and last.tool_calls:
tools = [tc['name'] for tc in last.tool_calls]
print(f" <- [after_model] Запрошенный инструмент: {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Прямой ответ: {content}...")
return None
@after_agent
def end_log(state, runtime):
"""После завершения агента"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] Агент завершил работу, всего сообщений: {total_msgs} <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Запрос погоды"""
return f"{city}: Солнечно, 25°C"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[start_log, pre_model, post_model, end_log],
system_prompt="Вы — ассистент.",
)
print("\n========== Первый вопрос (требуется инструмент) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Какая погода в Ханчжоу?")]
})
print(f"\nФинальный ответ: {result['messages'][-1].content}")
print("\n========== Второй вопрос (инструмент не нужен) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hello")]
})
print(f"\nФинальный ответ: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_agent, after_agent,
before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@before_agent
def start_log(state, runtime):
"""Перед запуском агента"""
print(">>> [before_agent] Агент запущен <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Перед каждым вызовом модели"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Сообщение {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""После каждого вызова модели"""
last = state["messages"][-1] if state.get("messages") else None
if hasattr(last, 'tool_calls') and last.tool_calls:
tools = [tc['name'] for tc in last.tool_calls]
print(f" <- [after_model] Запрошенный инструмент: {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Прямой ответ: {content}...")
return None
@after_agent
def end_log(state, runtime):
"""После завершения агента"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] Агент завершил работу, всего сообщений: {total_msgs} <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Запрос погоды"""
return f"{city}: Солнечно, 25°C"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[start_log, pre_model, post_model, end_log],
system_prompt="Вы — ассистент.",
)
print("\n========== Первый вопрос (требуется инструмент) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Какая погода в Ханчжоу?")]
})
print(f"\nФинальный ответ: {result['messages'][-1].content}")
print("\n========== Второй вопрос (инструмент не нужен) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hello")]
})
print(f"\nФинальный ответ: {result['messages'][-1].content}")
Вывод:
========== 第一个问题（需要工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 2 条消息 <- [after_model] 请求工具: ['get_weather'] -> [before_model] 第 3 条消息 <- [after_model] 直接回复: 杭州今天晴，气温25°C。... <<< [after_agent] Agent 结束，共 4 条消息 <<< Final reply: 杭州今天晴，气温25°C. ========== 第二个问题（无需工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 1 条消息 <- [after_model] 直接回复: 你好！有什么可以帮你的？... <<< [after_agent] Agent 结束，共 2 条消息 <<< Final reply: Hello! How can I help you?
Из вывода видно:
- before_agent и after_agent: выполняются только один раз на вопрос
- before_model и after_model: выполняются при каждом вызове модели (первый вопрос вызвал модель дважды, поэтому каждый выполнился дважды)
Возвращаемое значение Middleware
Возвращаемое значение Middleware определяет, изменять ли состояние Agent или управлять потоком:
| Возвращаемое значение | Эффект | Пример |
|---|---|---|
| Отсутствует | Не изменяет никакое состояние; продолжает обычный поток | Только логирование |
| dict | Обновляет состояние Agent (объединяется с текущим состоянием) | Возвращает {"custom_field": "value"} |
| Словарь, содержащий jump_to | Перейти к указанному узлу | Вернуть {"jump_to": "end"} |
Другие расширенияВозвращаемый словарь объединяется через редьюсер состояния Agent. Для поля messages используется редьюсер add_messages, поэтому возвращаемые сообщения добавляются, а не перезаписываются.