Перехват вызовов модели LangChain -- @wrap_model_call
@wrap_model_call — самый мощный хук в middleware.
В отличие от before/after, @wrap_model_call не просто наблюдает; он можетполностью управлять процессом выполнения модели— повторять попытки, деградировать, кэшировать или даже пропустить модель и напрямую использовать заранее заданный ответ.
Понимание callback-функции handler
В основе @wrap_model_call лежитфункция обратного вызова handlerТолько вызов handler(request) фактически выполняет модель; если её не вызвать, модель пропускается.
Пример
# Базовая структура wrap_model_call
# request: содержит всю информацию, такую как model, messages, tools, и т. д.
# handler: вызываемый объект; только его выполнение фактически вызывает модель
@wrap_model_call
def my_middleware(request, handler):
# Вы можете сделать что угодно перед вызовом модели
print("Сейчас будет вызвана модель...")
# Только вызов handler(request) действительно выполняет модель
response = handler(request)
# Вы можете сделать что угодно после вызова модели
print("Вызов модели завершён")
return response
# request: содержит всю информацию, такую как model, messages, tools, и т. д.
# handler: вызываемый объект; только его выполнение фактически вызывает модель
@wrap_model_call
def my_middleware(request, handler):
# Вы можете сделать что угодно перед вызовом модели
print("Сейчас будет вызвана модель...")
# Только вызов handler(request) действительно выполняет модель
response = handler(request)
# Вы можете сделать что угодно после вызова модели
print("Вызов модели завершён")
return response
Сценарий 1: Механизм повторных попыток
Это самый распространённый сценарий — вызовы модели могут завершаться неудачно из-за проблем с сетью, а автоматические повторные попытки могут повысить надёжность:
Пример
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Автоматически повторять при сбое вызова модели, до 3 раз"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Повторная попытка успешна] Попытка {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Инкрементное ожидание: 2s, 4s
print(f" [Повтор] Попытка {attempt + 1} не удалась, повтор через {wait_time} секунд...")
import time
time.sleep(wait_time)
# Все повторные попытки не удались
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Вы — ассистент EXAMPLE Tutorial.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Представь руководство по Python")]
})
print(f"\nОтвет: {result['messages'][-1].content[:100]}...")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Автоматически повторять при сбое вызова модели, до 3 раз"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Повторная попытка успешна] Попытка {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Инкрементное ожидание: 2s, 4s
print(f" [Повтор] Попытка {attempt + 1} не удалась, повтор через {wait_time} секунд...")
import time
time.sleep(wait_time)
# Все повторные попытки не удались
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Вы — ассистент EXAMPLE Tutorial.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Представь руководство по Python")]
})
print(f"\nОтвет: {result['messages'][-1].content[:100]}...")
Сценарий 2: Деградация модели / Переключение при сбое
Когда основная модель недоступна, автоматически переключиться на резервную модель:
Пример
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# Основная модель и резервная модель
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # Ограничить количество токенов для деградированной модели
@wrap_model_call
def fallback_on_error(request, handler):
"""Автоматическое переключение на резервную модель при сбое основной модели"""
try:
# Попытка использовать основную модель
return handler(request)
except Exception as e:
print(f"[Деградация] Основная модель дала сбой: {e}, переключение на резервную модель...")
# Переопределить модель в request, переключиться на резервную модель
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Деградация] Резервная модель также не сработала: {e2}")
# Обе модели не сработали, вернуть дружелюбное сообщение
return AIMessage(
content="Извините, сервис временно недоступен, пожалуйста, попробуйте позже."
)
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# Основная модель и резервная модель
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # Ограничить количество токенов для деградированной модели
@wrap_model_call
def fallback_on_error(request, handler):
"""Автоматическое переключение на резервную модель при сбое основной модели"""
try:
# Попытка использовать основную модель
return handler(request)
except Exception as e:
print(f"[Деградация] Основная модель дала сбой: {e}, переключение на резервную модель...")
# Переопределить модель в request, переключиться на резервную модель
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Деградация] Резервная модель также не сработала: {e2}")
# Обе модели не сработали, вернуть дружелюбное сообщение
return AIMessage(
content="Извините, сервис временно недоступен, пожалуйста, попробуйте позже."
)
request.override() — это неизменяемый метод: он возвращает новую копию request, не изменяя исходный объект. Это гарантирует, что каждый вызов независим и безопасен.
Сценарий 3: Кэширование ответов модели
Для повторяющихся запросов можно кэшировать ответы модели, чтобы снизить затраты на вызовы API:
Пример
from langchain.agents.middleware import wrap_model_call
from langchain.messages import AIMessage
# Простой кэш в памяти
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Кэшировать ответы модели, не вызывать повторно для одного и того же вопроса"""
# Использовать содержимое последнего сообщения пользователя в качестве ключа кэша
messages = request.messages
if not messages:
return handler(request)
# Сгенерировать ключ кэша
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Обрезать слишком длинное содержимое
# Проверить кэш
if cache_key in cache:
print(f"[Попадание в кэш] Непосредственно вернуть кэшированный результат")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(из кэша)*")
# Промах кэша, вызвать модель
result = handler(request)
# Содержимое AIMessage может быть смешано в списке, напрямую взять первый элемент
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Промах кэша] Сохранено в кэше, текущее количество записей: {len(cache)}")
elif hasattr(result, 'model_response'):
# Случай ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Вы — ассистент EXAMPLE Tutorial, и вы должны отвечать кратко.",
)
# Первый запрос (промах кэша)
result = agent.invoke({
"messages": [{"role": "user", "content": "Что такое Python?"}]
})
print(f"Первый раз: {result['messages'][-1].content[:80]}...\n")
# Второй запрос с тем же вопросом (попадание в кэш)
result = agent.invoke({
"messages": [{"role": "user", "content": "Что такое Python?"}]
})
print(f"Второй раз: {result['messages'][-1].content[:80]}...")
from langchain.messages import AIMessage
# Простой кэш в памяти
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Кэшировать ответы модели, не вызывать повторно для одного и того же вопроса"""
# Использовать содержимое последнего сообщения пользователя в качестве ключа кэша
messages = request.messages
if not messages:
return handler(request)
# Сгенерировать ключ кэша
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Обрезать слишком длинное содержимое
# Проверить кэш
if cache_key in cache:
print(f"[Попадание в кэш] Непосредственно вернуть кэшированный результат")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(из кэша)*")
# Промах кэша, вызвать модель
result = handler(request)
# Содержимое AIMessage может быть смешано в списке, напрямую взять первый элемент
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Промах кэша] Сохранено в кэше, текущее количество записей: {len(cache)}")
elif hasattr(result, 'model_response'):
# Случай ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Вы — ассистент EXAMPLE Tutorial, и вы должны отвечать кратко.",
)
# Первый запрос (промах кэша)
result = agent.invoke({
"messages": [{"role": "user", "content": "Что такое Python?"}]
})
print(f"Первый раз: {result['messages'][-1].content[:80]}...\n")
# Второй запрос с тем же вопросом (попадание в кэш)
result = agent.invoke({
"messages": [{"role": "user", "content": "Что такое Python?"}]
})
print(f"Второй раз: {result['messages'][-1].content[:80]}...")
Результат выполнения:
[缓存未命中] 已存入缓存，当前 1 条 第一次: Python 是一种高级编程语言，以简洁易读的语法著称... [缓存命中] 直接返回缓存结果 第二次: Python 是一种高级编程语言，以简洁易读的语法著称... *(来自缓存)*
Сценарий 4: Изменение запроса — динамическое внедрение системных сообщений
Пример
from datetime import datetime
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Внедрить информацию о текущем времени перед каждым вызовом модели"""
now = datetime.now()
time_context = (
f"Текущее время: {now.strftime('%Y год %m месяц %d день %H:%M')}."
f"Сегодня {['понедельник','вторник','среда','четверг','пятница','суббота','воскресенье'][now.weekday()]}."
)
# Добавить информацию о времени к существующему system_message
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Использовать override для создания нового запроса
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Внедрить информацию о текущем времени перед каждым вызовом модели"""
now = datetime.now()
time_context = (
f"Текущее время: {now.strftime('%Y год %m месяц %d день %H:%M')}."
f"Сегодня {['понедельник','вторник','среда','четверг','пятница','суббота','воскресенье'][now.weekday()]}."
)
# Добавить информацию о времени к существующему system_message
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Использовать override для создания нового запроса
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
Сценарий 5: Комбинирование нескольких wrap_model_call
Несколько middleware wrap_model_call автоматически объединяются по порядку — первый находится на самом внешнем слое:
Пример
from langchain.agents.middleware import wrap_model_call
@wrap_model_call
def outer_middleware(request, handler):
"""Самый внешний middleware"""
print("[Внешний] Старт")
result = handler(request) # Здесь происходит вход в inner_middleware
print("[Внешний] Конец")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Внутренний middleware"""
print(" [Внутренний] Старт")
result = handler(request) # Здесь фактически вызывается модель
print(" [Внутренний] Конец")
return result
# Порядок выполнения:
# [Внешний] Начало
# [Внутренний] Начало
# → Собственно вызов модели
# [Внутренний] Конец
# [Внешний] Конец
@wrap_model_call
def outer_middleware(request, handler):
"""Самый внешний middleware"""
print("[Внешний] Старт")
result = handler(request) # Здесь происходит вход в inner_middleware
print("[Внешний] Конец")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Внутренний middleware"""
print(" [Внутренний] Старт")
result = handler(request) # Здесь фактически вызывается модель
print(" [Внутренний] Конец")
return result
# Порядок выполнения:
# [Внешний] Начало
# [Внутренний] Начало
# → Собственно вызов модели
# [Внутренний] Конец
# [Внешний] Конец
Другие расширенияНесколько вызовов wrap_model_call оборачивают слой за слоем, как луковица. Самый внешний слой выполняется первым и возвращает результат последним. Это позволяет объединять несколько независимых функций — например, внешний слой выполняет проверку кэша, внутренний слой выполняет повторные попытки, не мешая друг другу.