LangChain @before_agent и @after_agent

before_agent и after_agent — это хуки уровня агента, которые выполняются один раз до и после запуска агента соответственно. Они подходят для инициализации, предварительной обработки, постобработки и статистического анализа.

before_agent -- Подготовка перед запуском агента

before_agent выполняется перед тем, как агент официально начинает выполнение, только один раз. Здесь можно выполнить предварительную обработку входных данных, проверку информации о пользователе, инициализацию ресурсов и т. д.

Сценарий 1: предварительная обработка ввода -- Автоисправление пользовательского ввода

Пример

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import before_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool

@before_agent
def preprocess_input(state, runtime):
"""Обработать пользовательский ввод до запуска агента"""
messages = state.get("messages", [])
if not messages:
return None

# Получить последнее сообщение пользователя
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

# Автоматически добавлять вежливые фразы (если пользователь спрашивает напрямую)
greetings = ["Hi", "Hello", "hi", "hello", "Hey"]
if content and not any(content.lower().startswith(g) for g in greetings):
# Не изменять, вернуть напрямую
pass

return None

@tool
def search_course(keyword: str) -> str:
"""Поиск курсов на Example"""
courses = {
"python": "Базовый учебник по Python3 (бесплатно, 30 глав)",
"html": "Базовый учебник по HTML (бесплатно, 25 глав)",
}
return courses.get(keyword.lower(), f"Не найдено курсов, связанных с {keyword}")

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="Вы консультант по курсам для Example.",
)

result = agent.invoke({
"messages": [HumanMessage(content="курсы Python")]
})
print(f"Ответ: {result['messages'][-1].content}")

Сценарий 2: Управление доступом -- Проверка разрешений

Пример

from langchain.agents.middleware import before_agent

@before_agent
def access_control(state, runtime):
"""Проверить, имеет ли пользователь разрешение на использование Agent"""
# Получить информацию о пользователе из runtime.context
context = runtime.context
if context is None:
return None

user_role = context.get("user_role", "guest")

# Пользователи-гости могут использовать только ограниченный набор функций
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# Проверить, затрагивает ли это ограниченные функции
restricted_keywords = ["delete", "manage", "configure", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content="У вас недостаточно прав для выполнения этой операции. Пожалуйста, войдите в систему и попробуйте снова."
)]
}

return None

after_agent -- Обработка после завершения работы Агента

after_agent выполняется после завершения всей обработки Агентом (только один раз). Здесь можно отформатировать итоговый вывод, записать статистическую информацию, очистить ресурсы и т. д.

Сценарий 3: Статистический анализ -- Запись данных диалога

Пример

from langchain.agents.middleware import after_agent

@after_agent
def conversation_stats(state, runtime):
"""Подсчитывает информацию о диалоге и добавляет её к результату"""
messages = state.get("messages", [])

# Статистические данные
model_calls = 0
tool_calls = 0
total_chars = 0

for msg in messages:
if msg.type == "ai":
model_calls += 1
if hasattr(msg, 'tool_calls') and msg.tool_calls:
tool_calls += len(msg.tool_calls)
if hasattr(msg, 'content') and msg.content:
total_chars += len(str(msg.content))

# Отправить статистическую информацию через пользовательский поток
runtime.stream_writer({
"type": "stats",
"model_calls": model_calls,
"tool_calls": tool_calls,
"total_messages": len(messages),
"total_chars": total_chars,
})

return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[conversation_stats],
system_prompt="Вы консультант по курсам для Example.",
)

# Используйте stream_mode=["updates", "custom"], чтобы получать пользовательские события
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Найти курсы Python")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"Статистика: {chunk}")

Вывод:

统计信息: {'type': 'stats', 'model_calls': 2, 'tool_calls': 1,
            'total_messages': 4, 'total_chars': 127}

Сценарий 4: Форматирование вывода -- Унификация стиля ответа

Пример

from langchain.agents.middleware import after_agent
from langchain.messages import AIMessage

@after_agent
def format_output(state, runtime):
"""Добавить форматированную сводную информацию к результату"""
messages = state.get("messages", [])
if not messages:
return None

# Найти последнее AI-сообщение (финальный ответ)
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break

if last_ai:
# Статистическая информация
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)

footer = (
f"\n\n---\n"
f"> В этом диалоге было {len(messages)} сообщений,"
f"Инструмент был вызван {tool_count} раз.\n"
f"> Работает на базе Example AI Assistant."
)

# Добавить к итоговому ответу
return {
"messages": [
AIMessage(content=last_ai.content + footer)
]
}

return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[format_output],
system_prompt="Вы консультант по курсам для Example.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Какие курсы есть у Python?")]
})
print(result["messages"][-1].content)

Вывод:

Example offers the Python3 basics tutorial in 30 chapters, completely free and perfect for Python beginners.

---
> 本次对话共进行 4 条消息，调用了 1 次工具。
> 由Example AI 助手提供支持。

Полный пример совместной работы четырёх хуков

Пример

from langchain.agents.middleware import (
before_agent, after_agent, before_model, after_model
)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool

# ----- Определить все хуки -----

@before_agent
def init_session(state, runtime):
"""Начало: инициализация сессии"""
print(">>> Сессия запущена")
return None

@before_model
def pre_model_check(state, runtime):
"""Перед каждым вызовом модели"""
msg_count = len(state.get("messages", []))
print(f" [Перед вызовом модели] Количество сообщений: {msg_count}")
return None

@after_model
def post_model_check(state, runtime):
"""После каждого вызова модели"""
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [После модели] Требуется вызов инструмента")
return None

@after_agent
def finish_session(state, runtime):
"""Конец: очистка ресурсов"""
total = len(state.get("messages", []))
print(f"<<< Сессия завершена, всего сообщений: {total}")
return None

# ----- Создать агента -----

@tool
def get_weather(city: str) -> str:
"""Запрос погоды"""
return f"{city}: Солнечно"

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[init_session, pre_model_check, post_model_check, finish_session],
system_prompt="Вы — ассистент.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Какая погода в Ханчжоу?")]
})
print(f"\n"Окончательный ответ: {result['messages'][-1].content}")

Вывод:

>>> 会话开始
  [model前] 消息数: 2
  [model后] 需要工具调用
  [model前] 消息数: 3
<<< 会话结束，共 4 条消息

Final reply: Hangzhou is sunny today, great for going out.

Сводка по хукам промежуточного ПО

Хук	Количество выполнений	Когда использовать	Ключевые возможности
before_agent	Один раз	Проверка разрешений, предварительная обработка ввода, инициализация ресурсов	Можно использовать jump_to="end" для досрочного завершения
before_model	Каждый цикл	Усечение сообщений, фильтрация содержимого, внедрение контекста	Можно управлять потоком с помощью jump_to
wrap_model_call	Каждый цикл	Повторные попытки, резервный вариант, кэширование, изменение промпта	Полный контроль над выполнением модели
after_model	Каждый цикл	Проверка ответа, добавление содержимого, логирование	Может заменять вывод модели
wrap_tool_call	Каждый вызов инструмента	Повторные попытки вызова инструмента, кэширование, переписывание параметров	Полный контроль над выполнением инструментов
after_agent	Один раз	Форматирование вывода, статистический анализ, очистка	Изменение конечного состояния

Другие расширения

Agent Конфигурация разработки

Справочное руководство

LangChain @before_agent и @after_agent

before_agent -- Подготовка перед запуском агента

Сценарий 1: предварительная обработка ввода -- Автоисправление пользовательского ввода

Пример

Сценарий 2: Управление доступом -- Проверка разрешений

Пример

after_agent -- Обработка после завершения работы Агента

Сценарий 3: Статистический анализ -- Запись данных диалога

Пример

Сценарий 4: Форматирование вывода -- Унификация стиля ответа

Пример

Полный пример совместной работы четырёх хуков

Пример

Сводка по хукам промежуточного ПО