Обработка ошибок при вызове инструментов

Таксономия ошибок: что может пойти не так

Прежде чем писать обработчики, важно понять: ошибки при вызове инструментов разных типов требуют разных стратегий. Повторная попытка при ошибке валидации — пустая трата токенов. Немедленный отказ при сетевом таймауте — упущенный шанс дождаться ответа. Все ошибки делятся на три категории по стратегии реакции.

100%

колёсико — масштаб · зажать и тянуть — перемещение

Логика разделения простая: транзиентные ошибки возникают из-за временных проблем — следующая попытка может пройти. Детерминированные ошибки (аргументы, логика приложения) не исчезнут от повторного вызова — нужно сообщить LLM и дать ему изменить стратегию.

⚠️ Ретраить валидационные ошибки — бессмысленно

Если LLM вызвал search_web с пустым query, повторный вызов с теми же аргументами даст тот же результат. Единственное, что поможет — сообщить LLM об ошибке и дать ему скорректировать запрос.

Протокол is_error: как правильно сообщить LLM об ошибке

Большинство разработчиков при первом знакомстве с tool calling делают одну и ту же ошибку: возвращают строку с ошибкой как обычный контент. LLM не понимает, что это ошибка — он видит просто текст и начинает его интерпретировать как результат работы инструмента.

Anthropic API предоставляет специальный флаг is_error: true в блоке tool_result. Это прямой сигнал модели: «инструмент не выполнился успешно, вот описание того, что пошло не так». Claude реагирует на это принципиально иначе.

Как is_error меняет поведение LLM

Без is_error — ошибка становится «данными»

tool_result (ошибка как контент)

{
type: "tool_result",
tool_use_id: "toolu_01",

content: "ConnectionTimeout after 5s"
← LLM думает: это результат поиска

}

ASSISTANT →
«Согласно результатам поиска: ConnectionTimeout after 5s...»

С is_error — LLM знает что делать

tool_result (is_error: true)

{
type: "tool_result",
tool_use_id: "toolu_01",

content: "Таймаут 5с. Попробуй другой источник.",
is_error: true

}

ASSISTANT →
«Поиск недоступен. Расскажу по своим данным или попробую read_file...»

Разница критическая. Без is_error LLM продолжает работу как будто всё в порядке и галлюцинирует на основе строки с ошибкой. С is_error: true он понимает, что инструмент не сработал, и ищет альтернативный путь к цели.

Как это выглядит в коде

python

def make_tool_result(tool_use_id: str, content: str, is_error: bool = False) -> dict:
    """Создаёт блок tool_result для передачи в API."""
    result = {
        "type": "tool_result",
        "tool_use_id": tool_use_id,
        "content": content,
    }
    if is_error:
        result["is_error"] = True
    return result


# ── Использование в цикле агента ──────────────────────

tool_results = []
for block in response.content:
    if block.type != "tool_use":
        continue

    handler = tool_handlers.get(block.name)

    if handler is None:
        # Инструмент не существует — ошибка аргументов
        tool_results.append(make_tool_result(
            block.id,
            f"Инструмент {block.name!r} не найден. Доступные: {list(tool_handlers)}",
            is_error=True,
        ))
        continue

    try:
        result = handler(**block.input)
        tool_results.append(make_tool_result(block.id, str(result)))

    except Exception as exc:
        tool_results.append(make_tool_result(
            block.id,
            f"Ошибка при выполнении {block.name!r}: {type(exc).__name__}: {exc}",
            is_error=True,
        ))

messages.append({"role": "user", "content": tool_results})

ℹ️ Содержимое is_error — инструкция для LLM

Текст в content при is_error: true — это не технический лог, а объяснение для модели. Пишите его как сообщение для разумного собеседника: что пошло не так и что стоит попробовать дальше. «httpx.ConnectTimeout» — плохо. «Сервис поиска не ответил, попробуй read_file или ответь по своим данным» — хорошо.

Валидация аргументов до исполнения

LLM иногда ошибается в аргументах: пропускает обязательные поля, передаёт число вместо строки или выходит за допустимый диапазон. JSON Schema в описании инструмента задаёт ожидаемую структуру, но не защищает код инструмента — проверка происходит только на стороне модели, а не на стороне исполнения.

Правило простое: валидируйте на границе системы — там, где данные приходят извне (от LLM). Это предотвращает side effects от невалидных данных и даёт LLM чёткое сообщение о том, что именно не так.

Валидация через Pydantic

Pydantic V2 идеально подходит для этой задачи: он проверяет типы, применяет конвертации (строка → int, если это безопасно) и возвращает детальные ошибки. Каждый инструмент получает свою модель аргументов.

python

from pydantic import BaseModel, field_validator, ValidationError
from typing import Literal


# ── Схемы аргументов для каждого инструмента ─────────

class SearchInput(BaseModel):
    query: str
    max_results: int = 10
    language: str = "ru"

    @field_validator("query")
    @classmethod
    def query_not_empty(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("Поисковый запрос не может быть пустым")
        return v

    @field_validator("max_results")
    @classmethod
    def max_results_in_range(cls, v: int) -> int:
        if not 1 <= v <= 50:
            raise ValueError(f"Ожидалось от 1 до 50, получено: {v}")
        return v


class ReadFileInput(BaseModel):
    path: str
    encoding: str = "utf-8"

    @field_validator("path")
    @classmethod
    def path_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Путь к файлу не может быть пустым")
        return v


class WriteFileInput(BaseModel):
    path: str
    content: str
    mode: Literal["write", "append"] = "write"


# ── Функция валидации ─────────────────────────────────

TOOL_SCHEMAS: dict[str, type[BaseModel]] = {
    "search_web": SearchInput,
    "read_file":  ReadFileInput,
    "write_file": WriteFileInput,
}


def validate_input(tool_name: str, tool_input: dict) -> tuple[BaseModel | None, str | None]:
    """
    Валидирует аргументы инструмента.
    Возвращает (validated_args, None) при успехе
    или (None, error_message) при ошибке.
    """
    schema_cls = TOOL_SCHEMAS.get(tool_name)
    if schema_cls is None:
        return None, f"Инструмент {tool_name!r} не зарегистрирован"

    try:
        return schema_cls(**tool_input), None
    except ValidationError as e:
        # Собираем человекочитаемые сообщения об ошибках
        errors = []
        for err in e.errors():
            field = " → ".join(str(loc) for loc in err["loc"]) if err["loc"] else "?"
            errors.append(f"{field}: {err['msg']}")
        return None, f"Невалидные аргументы для {tool_name!r}: {'; '.join(errors)}"

✅ Валидация сохраняет токены и предотвращает side effects

Если LLM передал path="" в инструмент записи файла — без валидации ваш код попытается записать файл с пустым именем. Pydantic останавливает это до исполнения и возвращает конкретную ошибку: «path: Путь к файлу не может быть пустым». LLM исправит вызов в следующей итерации.

Обработка runtime ошибок

Даже при валидных аргументах инструмент может упасть: сеть недоступна, внешний API вернул 500, файл удалили между проверкой и чтением. Ловить все исключения одним широким except Exception — это антипаттерн: вы теряете информацию о природе ошибки и не можете принять правильное решение (ретраить или нет).

Стратегия — ловить конкретные исключения в правильном порядке (от частного к общему) и для каждого формировать осмысленное сообщение и решение о retry.

Категории ошибок и их обработка

python

import asyncio
import httpx
import logging

logger = logging.getLogger(__name__)


async def execute_tool_safe(
    tool_name: str,
    handler,
    validated_input: dict,
    timeout: float = 10.0,
) -> tuple[str, bool]:
    """
    Выполняет инструмент и возвращает (content, is_error).
    Никогда не поднимает исключений наружу — все ошибки
    превращаются в is_error=True с описанием для LLM.
    """
    try:
        result = await asyncio.wait_for(
            handler(**validated_input),
            timeout=timeout,
        )
        return str(result), False

    # ── Таймаут ──────────────────────────────────────────
    except asyncio.TimeoutError:
        msg = f"Инструмент {tool_name!r} не ответил за {timeout:.0f}с. Попробуй позже."
        logger.warning("Tool timeout: %s (%.1fs)", tool_name, timeout)
        return msg, True  # транзиентная → ретраим

    # ── HTTP ошибки ───────────────────────────────────────
    except httpx.HTTPStatusError as exc:
        code = exc.response.status_code
        if code == 429:
            retry_after = exc.response.headers.get("Retry-After", "60")
            msg = f"Превышен лимит запросов API. Повторите через {retry_after}с."
            logger.warning("Rate limit in %s: 429", tool_name)
            return msg, True  # транзиентная → ретраим

        if code >= 500:
            msg = f"Сервер {tool_name!r} вернул {code}. Временная ошибка."
            logger.error("Server error in %s: %d", tool_name, code)
            return msg, True  # транзиентная → ретраим

        if code == 404:
            msg = f"Ресурс не найден (404): {exc.request.url}"
            return msg, True  # постоянная → не ретраим

        if code in (401, 403):
            msg = f"Ошибка аутентификации ({code}). Проверь API-ключ."
            logger.error("Auth error in %s: %d", tool_name, code)
            return msg, True  # постоянная → не ретраим

        msg = f"HTTP {code}: {exc.response.text[:200]}"
        return msg, True

    # ── Сетевые ошибки ────────────────────────────────────
    except httpx.ConnectError as exc:
        msg = f"Не удалось подключиться к серверу {tool_name!r}: {exc}"
        logger.warning("Connect error in %s: %s", tool_name, exc)
        return msg, True  # транзиентная → ретраим

    # ── Файловые ошибки ───────────────────────────────────
    except FileNotFoundError as exc:
        msg = f"Файл не найден: {exc.filename}"
        return msg, True  # постоянная → не ретраим

    except PermissionError as exc:
        msg = f"Нет прав доступа: {exc}"
        logger.error("Permission error in %s: %s", tool_name, exc)
        return msg, True  # постоянная → не ретраим

    # ── Всё остальное ─────────────────────────────────────
    except Exception as exc:
        msg = f"Непредвиденная ошибка в {tool_name!r}: {type(exc).__name__}: {exc}"
        logger.exception("Unhandled exception in tool %s", tool_name)
        return msg, True

Обратите внимание на asyncio.wait_for() — без него инструмент может зависнуть на неограниченное время. Всегда задавайте явный таймаут для сетевых операций.

Стратегия retry: что и когда повторять

Retry — это инструмент для транзиентных ошибок. Главное правило: ретраить только то, что имеет шанс пройти на следующей попытке. Повтор детерминированных ошибок тратит токены, время и деньги без какой-либо пользы.

Тип ошибки	Ретраить?	Причина
asyncio.TimeoutError	✓ Да	Транзиентная, сервер мог освободиться
HTTP 429 (Rate Limit)	⏱ Да, с задержкой	Лимит временный, нужен Retry-After
HTTP 500, 502, 503, 504	✓ Да	Временная недоступность сервера
httpx.ConnectError	✓ Да	Транзиентная сетевая проблема
HTTP 400 (Bad Request)	✗ Нет	Наши параметры запроса неверны
HTTP 401, 403 (Auth)	✗ Нет	Проблема с ключом, не исчезнет
HTTP 404 (Not Found)	✗ Нет	Ресурс не существует
ValidationError (Pydantic)	✗ Нет	LLM дал неверные аргументы, нужен is_error
FileNotFoundError	✗ Нет	Файл не появится от повтора
PermissionError	✗ Нет	Права не изменятся между попытками

Exponential backoff своими руками

Формула задержки: delay = min(base × 2ⁿ + jitter, max_delay). Jitter (случайный сдвиг) — обязателен: без него все параллельные агенты будут повторять попытку одновременно, усугубляя rate limit.

python

import asyncio
import random
import httpx
from dataclasses import dataclass, field


@dataclass
class RetryConfig:
    max_attempts: int = 3
    base_delay: float = 1.0    # секунды
    max_delay: float = 30.0
    jitter_factor: float = 0.25  # ±25% к задержке


# Ошибки, при которых ретраим
RETRYABLE_EXCEPTIONS = (
    asyncio.TimeoutError,
    httpx.ConnectError,
    httpx.TimeoutException,
)
RETRYABLE_HTTP_CODES = {429, 500, 502, 503, 504}


def is_retryable(exc: Exception) -> bool:
    """True, если ошибка транзиентная и имеет смысл повторить."""
    if isinstance(exc, RETRYABLE_EXCEPTIONS):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in RETRYABLE_HTTP_CODES
    return False


async def execute_with_retry(
    tool_name: str,
    handler,
    tool_input: dict,
    config: RetryConfig = RetryConfig(),
    timeout: float = 10.0,
) -> tuple[str, bool]:
    """
    Запускает инструмент с retry. Возвращает (content, is_error).
    """
    last_exc: Exception | None = None

    for attempt in range(config.max_attempts):
        try:
            result = await asyncio.wait_for(handler(**tool_input), timeout=timeout)
            return str(result), False

        except Exception as exc:
            last_exc = exc

            if not is_retryable(exc):
                # Детерминированная ошибка — повторять бесполезно
                break

            if attempt < config.max_attempts - 1:
                # Вычисляем задержку с jitter
                base = min(config.base_delay * (2 ** attempt), config.max_delay)
                jitter = random.uniform(-config.jitter_factor, config.jitter_factor) * base
                delay = max(0.1, base + jitter)

                # Особый случай: rate limit может дать нам Retry-After
                if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code == 429:
                    retry_after = exc.response.headers.get("Retry-After")
                    if retry_after:
                        delay = float(retry_after)

                logger.warning(
                    "Tool %s: попытка %d/%d упала (%s), повтор через %.1fс",
                    tool_name, attempt + 1, config.max_attempts,
                    type(exc).__name__, delay,
                )
                await asyncio.sleep(delay)

    # Все попытки исчерпаны
    return (
        f"Инструмент {tool_name!r} не ответил после {config.max_attempts} попыток: {last_exc}",
        True,
    )

💡 Альтернатива: библиотека tenacity

Если не хотите писать retry-логику вручную — pip install tenacity. Декоратор @retry(stop=stop_after_attempt(3), wait=wait_exponential(...)) делает то же самое в 2 строки. Но понимать, как это работает внутри, всё равно важно — иначе сложно отлаживать.

Полный обработчик: production-ready паттерн

Соберём все части вместе: реестр инструментов, валидацию, retry и is_error протокол. Этот код — минимальная production-ready база, которую можно расширять под конкретный проект.

python

"""
production_tool_handler.py
Полный обработчик инструментов с валидацией, retry и is_error.
"""
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Callable, Type, Any

from pydantic import BaseModel, ValidationError

logger = logging.getLogger(__name__)


# ── Описание инструмента ──────────────────────────────

@dataclass
class ToolDef:
    handler: Callable                       # async функция-инструмент
    input_schema: Type[BaseModel]           # Pydantic-схема аргументов
    timeout: float = 10.0                   # таймаут в секундах
    retry: RetryConfig = field(default_factory=RetryConfig)


# ── Реестр инструментов ───────────────────────────────
# Регистрируйте здесь все инструменты агента

REGISTRY: dict[str, ToolDef] = {}

def register(name: str, schema: Type[BaseModel], timeout: float = 10.0, max_attempts: int = 3):
    """Декоратор для регистрации инструмента."""
    def decorator(fn: Callable) -> Callable:
        REGISTRY[name] = ToolDef(
            handler=fn,
            input_schema=schema,
            timeout=timeout,
            retry=RetryConfig(max_attempts=max_attempts),
        )
        return fn
    return decorator


# ── Пример регистрации инструментов ──────────────────

@register("search_web", SearchInput, timeout=10.0, max_attempts=3)
async def search_web(query: str, max_results: int = 10) -> str:
    """Поиск в интернете — здесь реальная логика."""
    ...

@register("read_file", ReadFileInput, timeout=5.0, max_attempts=1)
async def read_file(path: str, encoding: str = "utf-8") -> str:
    """Читает файл — файловые операции не ретраим."""
    with open(path, encoding=encoding) as f:
        return f.read()


# ── Центральный обработчик ────────────────────────────

async def handle_tool_call(
    tool_name: str,
    tool_use_id: str,
    tool_input: dict,
) -> dict:
    """
    Обрабатывает один tool_use блок. Возвращает tool_result.
    Никогда не поднимает исключений — все ошибки идут в is_error.
    """

    # 1. Проверяем, что инструмент существует
    defn = REGISTRY.get(tool_name)
    if defn is None:
        available = list(REGISTRY)
        logger.warning("LLM запросил незарегистрированный инструмент: %s", tool_name)
        return _error_result(
            tool_use_id,
            f"Инструмент {tool_name!r} не существует. "
            f"Доступные инструменты: {available}",
        )

    # 2. Валидируем аргументы
    try:
        validated = defn.input_schema(**tool_input)
    except ValidationError as e:
        errors = "; ".join(
            f"{'.'.join(str(l) for l in err['loc']) or '?'}: {err['msg']}"
            for err in e.errors()
        )
        return _error_result(
            tool_use_id,
            f"Невалидные аргументы для {tool_name!r}: {errors}",
        )

    # 3. Выполняем с retry
    content, is_err = await execute_with_retry(
        tool_name=tool_name,
        handler=defn.handler,
        tool_input=validated.model_dump(),
        config=defn.retry,
        timeout=defn.timeout,
    )

    logger.debug("Tool %s → is_error=%s, content[:80]=%r", tool_name, is_err, content[:80])

    return {
        "type": "tool_result",
        "tool_use_id": tool_use_id,
        "content": content,
        **({"is_error": True} if is_err else {}),
    }


def _error_result(tool_use_id: str, message: str) -> dict:
    return {
        "type": "tool_result",
        "tool_use_id": tool_use_id,
        "content": message,
        "is_error": True,
    }


async def handle_all_tool_calls(response_content: list) -> list[dict]:
    """
    Обрабатывает все tool_use блоки из ответа модели.
    Параллельно — если блоков несколько.
    """
    tasks = [
        handle_tool_call(block.name, block.id, block.input)
        for block in response_content
        if block.type == "tool_use"
    ]
    return await asyncio.gather(*tasks)

Интеграция в цикл агента

python

import anthropic

client = anthropic.AsyncAnthropic()

async def run_agent(task: str, max_iterations: int = 20) -> str:
    messages = [{"role": "user", "content": task}]

    for _ in range(max_iterations):
        response = await client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            tools=TOOLS_JSON_SCHEMA,   # список JSON Schema инструментов
            messages=messages,
        )

        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return ""

        # Обрабатываем все tool_use — с валидацией, retry и is_error
        tool_results = await handle_all_tool_calls(response.content)
        messages.append({"role": "user", "content": tool_results})

    return "Достигнут лимит итераций"

⚠️ Защита от бесконечного цикла retry на уровне агента

Если один инструмент постоянно возвращает is_error: True, LLM может зациклиться: повторять вызов снова и снова. Помогает max_iterations, но лучше добавить счётчик ошибок для каждого инструмента и после N неудач менять стратегию в сообщении: «Инструмент X недоступен уже 3 раза — используй другой подход».

Типичные ошибки

1. Возвращать ошибку без is_error: True

Самая распространённая ошибка. LLM видит строку «ConnectionError: ...» и интерпретирует её как результат работы инструмента. Начинаются галлюцинации на основе текста ошибки.

✓ Всегда добавляйте "is_error": True при любой ошибке исполнения.

2. Глотать исключения с пустым контентом

except Exception: return "" — LLM получает пустой результат и не знает, что пошло не так. Он либо галлюцинирует, либо бесконечно переспрашивает.

✓ В content всегда кладите объяснение: что произошло и что стоит попробовать вместо этого.

3. Ретраить всё подряд

Retry при ValidationError, FileNotFoundError или 401 — трата токенов и времени. Эти ошибки не исчезнут от повторной попытки с теми же параметрами.

✓ Ретраите только транзиентные ошибки: таймауты, 5xx, 429, ConnectError.

4. Нет таймаутов у сетевых инструментов

Инструмент без таймаута может завис на минуты, блокируя весь агент. В asyncio это особенно опасно: зависший корутин держит event loop.

✓ Всегда оборачивайте сетевые вызовы в asyncio.wait_for(..., timeout=N).

5. Технические сообщения об ошибках вместо объяснений для LLM

«httpx.ConnectTimeout: HTTPSConnectionPool(host='api.example.com', port=443)» — LLM этот стектрейс не поможет принять правильное решение.

✓ Переводите технические ошибки в объяснения: «Сервис поиска не ответил за 10с. Попробуй read_file или ответь по своим данным.»

Шпаргалка

Протокол is_error:

Всегда добавляйте "is_error": True при любой ошибке выполнения инструмента
content при ошибке — объяснение для LLM, не технический лог
Без is_error LLM будет считать строку с ошибкой валидным результатом

Три категории ошибок:

Ошибки аргументов (нет инструмента, невалидные параметры) → is_error немедленно, не ретраить
Транзиентные (таймаут, 429, 5xx, ConnectError) → retry + backoff, потом is_error
Постоянные (404, 401/403, файл не найден) → is_error немедленно, не ретраить

Паттерн надёжного обработчика:

Проверить, что инструмент зарегистрирован
Валидировать аргументы через Pydantic (ValidationError → is_error)
Выполнить с явным таймаутом (asyncio.wait_for)
Поймать конкретные исключения в нужном порядке
Транзиентные → retry с exponential backoff + jitter
Все остальные → is_error: True с человекочитаемым объяснением

Exponential backoff: delay = min(base × 2ⁿ + jitter, max_delay). Jitter обязателен — без него параллельные агенты создают thundering herd.

Практическое задание

Задача 1. Возьмите любой агент из предыдущих уроков и добавьте в его цикл обработку is_error: замените голый except Exception на структурированный обработчик с конкретными типами исключений. Убедитесь, что при сетевой ошибке LLM получает осмысленное сообщение, а не пустую строку или стектрейс.

Задача 2. Создайте Pydantic-схемы для 2–3 инструментов вашего агента. Напишите тест, который проверяет: (а) валидные аргументы проходят без ошибок, (б) невалидные аргументы возвращают понятное сообщение об ошибке, (в) сообщение об ошибке содержит имя конкретного поля, а не просто «ValidationError».

Задача 3 (продвинутая). Реализуйте функцию execute_with_retry и напишите к ней тесты через unittest.mock: (а) при первом вызове бросается asyncio.TimeoutError, второй проходит — проверь, что функция возвращает успех, (б) при трёх TimeoutError подряд функция возвращает is_error=True с упоминанием числа попыток, (в) при FileNotFoundError функция возвращает is_error=True сразу, без retry.