Перехват вызовов инструментов LangChain -- @wrap_tool_call
@wrap_tool_call позволяет реализовать возможности управления, подобные @wrap_model_call, на уровне выполнения инструментов — повторные попытки, кэширование, перезапись параметров и постобработка результатов.
Базовая структура
Структура @wrap_tool_call аналогична @wrap_model_call: принимает два параметра: request и handler:
Пример
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: Содержит имя инструмента и параметры
# request.tool: Сам объект инструмента
# request.state: Текущее состояние агента
# request.runtime: Контекст выполнения
# Инструмент фактически выполняется только при вызове handler(request)
result = handler(request)
# result — это ToolMessage или Command
return result
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: Содержит имя инструмента и параметры
# request.tool: Сам объект инструмента
# request.state: Текущее состояние агента
# request.runtime: Контекст выполнения
# Инструмент фактически выполняется только при вызове handler(request)
result = handler(request)
# result — это ToolMessage или Command
return result
Сценарий 1: Повторная попытка вызова инструмента
Выполнение инструмента может завершиться неудачно из-за нестабильных внешних сервисов; автоматические повторные попытки могут повысить надёжность:
Пример
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Автоматически повторять попытку при сбое вызова инструмента"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Проверить, является ли это результатом ошибки
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Повтор] Инструмент вернул ошибку, повтор #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Повтор успешен] Попытка #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Повтор] Исключение {e}, повтор #{attempt + 1}...")
else:
raise
return last_result
# Смоделировать инструмент, который может дать сбой
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Получить данные курса из EXAMPLE.
Аргументы:
course_id: ID курса
"""
global call_count
call_count += 1
# Имитация сбоя для первых двух попыток, успех с третьей
if call_count < 3:
raise Exception(f"Ошибка сети: не удалось подключиться к службе курсов (попытка #{call_count})")
return f"Курс {course_id}: Базовый учебник по Python3, 30 глав, бесплатно"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="Вы — помощник по курсам для EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Помогите мне найти информацию о курсе python-001")]
})
print(f"\nФинальный ответ: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Автоматически повторять попытку при сбое вызова инструмента"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Проверить, является ли это результатом ошибки
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Повтор] Инструмент вернул ошибку, повтор #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Повтор успешен] Попытка #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Повтор] Исключение {e}, повтор #{attempt + 1}...")
else:
raise
return last_result
# Смоделировать инструмент, который может дать сбой
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Получить данные курса из EXAMPLE.
Аргументы:
course_id: ID курса
"""
global call_count
call_count += 1
# Имитация сбоя для первых двух попыток, успех с третьей
if call_count < 3:
raise Exception(f"Ошибка сети: не удалось подключиться к службе курсов (попытка #{call_count})")
return f"Курс {course_id}: Базовый учебник по Python3, 30 глав, бесплатно"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="Вы — помощник по курсам для EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Помогите мне найти информацию о курсе python-001")]
})
print(f"\nФинальный ответ: {result['messages'][-1].content}")
Вывод:
[重试] 异常 网络错误：无法连接到课程服务（第 1 次尝试），1 次重试... [重试] 异常 网络错误：无法连接到课程服务（第 2 次尝试），2 次重试... [重试成功] 第 3 次尝试 Final reply: Course python-001 is the Python3 basics tutorial, 30 chapters, free of charge.
Сценарий 2: Изменение параметров инструмента
Динамическое изменение параметров перед выполнением инструмента позволяет преобразовывать параметры без изменения кода инструмента:
Пример
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def normalize_city_name(request, handler):
"""Автоматически нормализовать названия городов (преобразовать полноширинные символы в полуширинные, удалить лишние пробелы и т. д.)"""
tool_call = request.tool_call
# Обрабатывать только вызовы инструментов, содержащие параметр city
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Нормализовать название города: удалить пробелы, унифицировать регистр
normalized = city.strip().replace(" ", "") # Удалить полноширинные пробелы
# Заменить параметр
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
@wrap_tool_call
def normalize_city_name(request, handler):
"""Автоматически нормализовать названия городов (преобразовать полноширинные символы в полуширинные, удалить лишние пробелы и т. д.)"""
tool_call = request.tool_call
# Обрабатывать только вызовы инструментов, содержащие параметр city
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Нормализовать название города: удалить пробелы, унифицировать регистр
normalized = city.strip().replace(" ", "") # Удалить полноширинные пробелы
# Заменить параметр
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
Сценарий 3: Кэширование результатов инструментов
Для повторных вызовов инструментов (тот же инструмент + те же параметры) результаты можно кэшировать:
Пример
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Кэширование результатов выполнения инструментов"""
# Создать ключ кэша: имя инструмента + параметры
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Проверить кэш
if cache_key in tool_cache:
print(f"[Попадание в кэш инструмента] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Выполнить инструмент
result = handler(request)
# Сохраняем в кэш
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Запись в кэш инструмента] {tool_name}, сейчас {len(tool_cache)} записей")
return result
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Кэширование результатов выполнения инструментов"""
# Создать ключ кэша: имя инструмента + параметры
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Проверить кэш
if cache_key in tool_cache:
print(f"[Попадание в кэш инструмента] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Выполнить инструмент
result = handler(request)
# Сохраняем в кэш
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Запись в кэш инструмента] {tool_name}, сейчас {len(tool_cache)} записей")
return result
Сценарий 4: Логирование и мониторинг вызовов инструментов
Записывайте подробную информацию обо всех вызовах инструментов:
Пример
import time
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Отслеживать метрики производительности для вызовов инструментов"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Записать время начала
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Записать успешный вызов
print(f"[Monitor] {tool_name}({tool_args}) выполнен успешно, заняло {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Записать неудачный вызов
print(f"[Monitor] {tool_name}({tool_args}) завершился неудачно, заняло {elapsed:.2f}s, ошибка: {e}")
raise
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Отслеживать метрики производительности для вызовов инструментов"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Записать время начала
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Записать успешный вызов
print(f"[Monitor] {tool_name}({tool_args}) выполнен успешно, заняло {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Записать неудачный вызов
print(f"[Monitor] {tool_name}({tool_args}) завершился неудачно, заняло {elapsed:.2f}s, ошибка: {e}")
raise
Сценарий 5: Определение дальнейшего потока на основе результатов
Вы можете решить, продолжать ли цикл Agent, основываясь на результате выполнения инструмента:
Пример
from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Если инструмент возвращает пустой результат, завершите Agent напрямую, не тратя вызовы модели"""
result = handler(request)
# Проверить, был ли возвращён пустой результат
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Непосредственно вернуть Command для обновления состояния
# Добавить AI-сообщение, чтобы объяснить ситуацию
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Извините, не найдено подходящей информации. Попробуйте другое ключевое слово.")
]
})
return result
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Если инструмент возвращает пустой результат, завершите Agent напрямую, не тратя вызовы модели"""
result = handler(request)
# Проверить, был ли возвращён пустой результат
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Непосредственно вернуть Command для обновления состояния
# Добавить AI-сообщение, чтобы объяснить ситуацию
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Извините, не найдено подходящей информации. Попробуйте другое ключевое слово.")
]
})
return result
Когда wrap_tool_call возвращает Command, вы можете изменить состояние Agent через параметр update. Использование Command позволяет напрямую добавить AI-сообщение в список сообщений, и затем цикл Agent естественно завершится.
@wrap_model_call против @wrap_tool_call
| Измерение | @wrap_model_call | @wrap_tool_call |
|---|---|---|
| Цель перехвата | Вызов модели | Выполнение инструментов |
| содержимое запроса | model, messages, tools, system_prompt | tool_call, tool, state, runtime |
| Тип возвращаемого значения | ModelResponse или AIMessage | ToolMessage или Command |
| Применимые сценарии | Повторная попытка модели, резервный вариант, кэширование, изменение промпта | Повторная попытка инструмента, кэширование, перезапись параметров, обработка результатов |