Хуки middleware LangChain -- @before_model и @after_model

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

@before_model — перехват перед вызовом модели

@before_model выполняется перед каждым вызовом модели. Здесь можно изменять сообщения, внедрять контекстные условия или полностью пропускать вызов модели.

Сценарий 1: предобработка сообщений — ограничение длины диалога

Пример

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@before_model
def limit_context(state, runtime):
    """Ограничить длину истории сообщений, чтобы предотвратить слишком длинный контекст"""
    messages = state.get("messages", [])

    # Сохранить системное сообщение + последние 6 сообщений
    MAX_MESSAGES = 6
    if len(messages) > MAX_MESSAGES:
        # Обрезать до последних сообщений
        trimmed = messages[-MAX_MESSAGES:]
        # Убедиться, что первое сообщение — от пользователя
        if trimmed and trimmed[0].type != "human":
            trimmed = trimmed[1:]
        return {"messages": trimmed}

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[limit_context],
    system_prompt="Вы — ассистент учебника EXAMPLE.",
)

# Симуляция многоходового диалога
result = agent.invoke({
    "messages": [
        HumanMessage(content="Раунд 1"),
        HumanMessage(content="Раунд 2"),
        HumanMessage(content="Раунд 3"),
        HumanMessage(content="Раунд 4"),
        HumanMessage(content="Раунд 5"),
        HumanMessage(content="Раунд 6"),
        HumanMessage(content="Раунд 7"),
    ]
})
print(f"Количество сообщений: {len(result['messages'])}")
print(f"Ответ: {result['messages'][-1].content}")

Вывод:

Message count: 8
Reply: Hello! It looks like you are testing multi-turn conversation. How can I help you?

Сценарий 2: Фильтрация содержимого — блокировка чувствительных слов

Пример

from langchain.agents.middleware import before_model


SENSITIVE_WORDS = ["пароль", "номер банковской карты", "номер ID-карты"]


@before_model
def content_filter(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 ""

    for word in SENSITIVE_WORDS:
        if word in content:
            print(f"[Заблокировано] Обнаружено чувствительное слово: {word}")
            # jump_to="end" завершает выполнение напрямую, предотвращая ответ модели
            return {
                "jump_to": "end",
                "messages": [
                    HumanMessage(content=f"Извините, по соображениям безопасности запросы, содержащие '{word}', не могут быть обработаны.")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[content_filter],
    system_prompt="Вы — ассистент учебника EXAMPLE.",
)

# Обычный вопрос
result = agent.invoke({
    "messages": [HumanMessage(content="Как начать работать с Python?")]
})
print(f"Обычный вопрос: {result['messages'][-1].content[:80]}")

# Конфиденциальный вопрос
result = agent.invoke({
    "messages": [HumanMessage(content="Какой номер моей банковской карты? Можете помочь проверить его?")]
})
print(f"\nКонфиденциальный вопрос: {result['messages'][-1].content}")

Вывод:

[拦截] 检测到敏感词: 银行卡号
Normal question: Python 入门可以从安装 Python 环境开始...

Sensitive question: 抱歉，为了安全，不能处理包含「银行卡号」的请求。

@after_model — Обработка после вызова модели

@after_model выполняется после ответа модели, подходит для проверки вывода модели, извлечения ключевой информации, добавления последующих инструкций и т. д.

Сценарий 3: Проверка содержимого ответа

Пример

from langchain.agents.middleware import after_model


FORBIDDEN_TOPICS = ["политика", "насилие", "порнография"]


@after_model
def response_audit(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 ""

    for topic in FORBIDDEN_TOPICS:
        if topic in content:
            runtime.stream_writer({
                "type": "warning",
                "message": f"Ответ распознан как содержащий контент, связанный с '{topic}', и был заменён"
            })
            # Вернуть перезаписывающее сообщение (через add_messages reducer)
            from langchain.messages import AIMessage
            return {
                "messages": [
                    AIMessage(content="Извините, я не могу ответить на этот вопрос."
                                      "Пожалуйста, спрашивайте о контенте, связанном с изучением программирования.")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[response_audit],
    system_prompt="Вы — ассистент учебника EXAMPLE.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Какие учебные ресурсы доступны для Python?")]
})
print(f"Обычный ответ: {result['messages'][-1].content}")

Сценарий 4: Автоматическое добавление информации к промпту

Пример

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


@after_model
def append_disclaimer(state, runtime):
    """Автоматически добавлять дисклеймер после каждого ответа модели"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]

    # Добавлять только тогда, когда это финальный ответ ИИ (без tool_calls)
    if (last_msg.type == "ai"
        and last_msg.content
        and not (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls)):
        # Заменить последнее ИИ-сообщение, добавив дисклеймер
        return {
            "messages": [
                AIMessage(
                    content=(
                        last_msg.content
                        + "\n\n---\n*Приведённый выше контент был сгенерирован ИИ-ассистентом EXAMPLE и предназначен только для справки.*
                    )
                )
            ]
        }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[append_disclaimer],
    system_prompt="Вы — ассистент EXAMPLE.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Легко ли выучить Python?")]
})
print(result["messages"][-1].content)

Вывод:

Python 是一门非常适合初学者的编程语言，它的语法简洁、贴近自然语言...
---
*The above content was generated by the Example AI assistant and is for reference only.*

can_jump_to — Управление переходами потока

В before_model и after_model вы можете управлять потоком агента с помощью параметра can_jump_to и состояния jump_to:

Пример

from langchain.agents.middleware import before_model


@before_model(can_jump_to=["end"])  # Объявите цели, к которым можно перейти
def conditional_exit(state, runtime):
    """Завершить агента непосредственно при определённых условиях"""
    messages = state.get("messages", [])
    if not messages:
        return None

    # Если пользователь говорит "goodbye", завершить напрямую
    last_content = str(messages[-1].content)
    if last_content.strip() in ["Goodbye", "Bye", "bye"]:
        return {
            "jump_to": "end",  # Завершить агента напрямую
            "messages": [{"role": "assistant", "content": "До свидания! С нетерпением ждём возможности обслужить вас в следующий раз."}]
        }

    return None
can_jump_to ЗначениеЗначениеПрименимые сценарии
["end"]Может перейти к endУсловный выход, перехват в целях безопасности
["model"]Может вернуться к modelКогда model требуется повторная обработка
["tools"]Может перейти к узлу инструментаПропустить model и выполнить инструмент напрямую
["model", "end"]Может перейти к model или endНесколько условных ветвей

Если цель не объявлена в can_jump_to, jump_to будет проигнорирован. Это механизм безопасности, предотвращающий случайный переход промежуточного ПО к недопустимым узлам.

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