Как вы делаете агента "отказоустойчивым" (graceful degradation)?

Краткий тезис

Отказоустойчивость агента — это способность системы продолжать выполнение задач (или корректно завершать их) при сбоях отдельных компонентов: инструментов, LLM, внешних API. Ключевые приёмы: try-catch на каждый вызов, fallback действия, ретраи с экспоненциальной задержкой, health check внешних сервисов, human-in-the-loop для критических ошибок и сохранение состояния для восстановления. В production важно не просто ловить ошибки, а проектировать поток выполнения так, чтобы сбой любого звена не обрушивал всего агента, а приводил к предсказуемому поведению.

1. Термин: Graceful degradation (отказоустойчивое ухудшение)

Это подход, при котором в случае сбоя система не падает полностью, а продолжает работать с пониженным качеством или функциональностью, сообщая пользователю о возникших ограничениях. Для агента это означает, что если, например, инструмент поиска недоступен, агент может перейти к запросу уточнения у пользователя или использовать кэшированные данные, а не завершаться с ошибкой.

Пример: Представьте агента поддержки клиентов, который вызывает API погоды, API базы знаний и API службы доставки. Если API погоды временно не отвечает, агент должен всё равно ответить на вопрос о статусе заказа, просто без погодного контекста, а не возвращать 500-ю ошибку.

2. Try-catch на каждый вызов инструмента

Базовая защита: каждый вызов любого инструмента должен быть обёрнут в try-catch (или аналогичную конструкцию). Это позволяет перехватить исключение (network timeout, HTTP 5xx, некорректный ответ) и принять решение, а не допустить аварийного завершения всего агента.

Пример на Python:

import logging
from typing import Any

async def call_tool_safe(tool_func, *args, **kwargs) -> dict:
    try:
        result = await tool_func(*args, **kwargs)
        return {"success": True, "data": result}
    except Exception as e:
        logging.error(f"Tool call failed: {e}")
        return {"success": False, "error": str(e)}

После получения success: False агент может выполнить fallback (см. раздел 3).

Важно Ловить нужно конкретные исключения (например, aiohttp.ClientError, TimeoutError), а не все подряд, чтобы не маскировать ошибки в логике самого агента.

3. Fallback действия

Если инструмент не сработал, у агента должен быть заранее определённый fallback (запасной вариант). Варианты:

Пример реализации в коде агента:

async def execute_action(action: str, context: dict) -> str:
    # Попытка основного инструмента
    result = await call_tool_safe(tool_search, action)
    if result["success"]:
        return result["data"]
    # Fallback: спросить пользователя
    user_feedback = await ask_user("Инструмент поиска недоступен. Можете уточнить запрос?")
    if user_feedback:
        return await call_tool_safe(tool_search, user_feedback)
    # Если и это не помогло — возвращаем заглушку
    return "Извините, я не могу сейчас найти ответ. Попробуйте позже."

4. Ретраи с экспоненциальной задержкой (3 попытки)

Для временных сбоев (network hiccup, rate limit) эффективен retry с экспоненциальной задержкой (backoff|exponential backoff). Обычно делают 3 попытки с задержками: 1 сек, 2 сек, 4 сек (или с добавлением случайного jitter).

Пример с использованием tenacity:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def call_api_with_retry(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=5) as resp:
            resp.raise_for_status()
            return await resp.json()

Когда не стоит ретраить

• Код ошибки 4xx (кроме 429 — rate limit) — ошибка на стороне клиента, повтор не поможет.
• Ошибки валидации — нужно исправить запрос.
• После третьей попытки нужно перейти к fallback.

5. Health check для внешних API

Чтобы агент не тратил время на попытки вызова заведомо недоступного сервиса, полезно делать health check (проверка работоспособности) перед вызовом. Health check можно выполнять периодически (например, раз в 30 секунд) или lazily перед каждым вызовом.

Пример реализации:

class ServiceRegistry:
    def __init__(self):
        self._healthy = {}

    async def check_health(self, service_name: str, health_url: str) -> bool:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(health_url, timeout=3) as resp:
                    healthy = resp.status == 200
                    self._healthy[service_name] = healthy
                    return healthy
        except Exception:
            self._healthy[service_name] = False
            return False

    def is_healthy(self, service_name: str) -> bool:
        return self._healthy.get(service_name, False)

Агент перед вызовом инструмента проверяет is_healthy и, если сервис нездоров, сразу переходит к fallback.

6. Human-in-the-loop при критических ошибках

Если сбой происходит на этапе, где неправильное решение может привести к серьёзным последствиям (финансовые транзакции, медицинские рекомендации), нужно human-in-the-loop (HITL). Принцип: агент останавливает выполнение, сохраняет контекст и запрашивает approval от человека.

Пример:

class HumanInLoopError(Exception):
    pass

async def execute_with_hitl(state):
    # Выполняем действие
    result = await call_tool_safe(tool_payment, state)
    if not result["success"]:
        # Логируем
        logging.critical(f"Payment tool failed: {result['error']}")
        # Запрашиваем вмешательство человека
        decision = await request_human_approval(
            "Payment tool failed. Approve manual retry or cancel?",
            options=["retry", "cancel"]
        )
        if decision == "retry":
            return await call_tool_safe(tool_payment, state)
        else:
            raise HumanInLoopError("Cancelled by operator")
    return result["data"]

Когда использовать HITL при ошибках в финансовых операциях, при нераспознавании пользовательского ввода (если агент готовится принять неверное решение), при сбоях в системах безопасности.

7. Сохранение состояния перед каждым действием (checkpointing)

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

Пример схемы checkpoint:

@dataclass
class AgentState:
    session_id: str
    current_step: int
    pending_tool: str
    tool_args: dict
    partial_results: list
    timestamp: float

async def save_checkpoint(state: AgentState):
    await redis.set(f"agent_state:{state.session_id}", json.dumps(asdict(state)))

async def load_checkpoint(session_id: str) -> AgentState:
    data = await redis.get(f"agent_state:{session_id}")
    if data:
        return AgentState(**json.loads(data))
    return None

При старте агента после падения он проверяет наличие checkpoint, и если есть — восстанавливает состояние и продолжает с прерванного шага.

Важно состояние должно быть устойчивым к частичной записи — используйте атомарные операции или транзакции.

8. Мониторинг и алертинг

Отказоустойчивость невозможна без наблюдения. Нужно логировать:

• каждый вызов инструмента и его успешность;
• ретраи (сколько, какие ошибки);
• fallback-срабатывания;
• обращения к human-in-the-loop.

Настроить метрики (Prometheus) и алерты (если частота ошибок > порога).

9. Тестирование отказоустойчивости

Используйте chaos engineering для агентов: намеренно отключайте внешние API, задерживайте ответы, возвращайте ошибки в тестовом окружении и проверяйте, что agent корректно отрабатывает fallback'и.

Простой пример с pytest:

@pytest.mark.asyncio
async def test_agent_downgrade_when_api_down():
    # Мокаем инструмент поиска, чтобы он всегда падал
    with patch('my_agent.tool_search', side_effect=ConnectionError("API down")):
        response = await agent.run("Найди информацию по теме X")
        assert "Извините" in response  # агент должен вернуть fallback

Пет-проект для закрепления

Задача Создать простого агента-помощника, который возвращает погоду и курс валют, используя два публичных API. Реализовать отказоустойчивость: если один API недоступен, агент возвращает только данные второго и сообщение о недоступности.

Инструменты Python, aiohttp, tenacity, pytest-asyncio.

Шаги:

1. Реализовать асинхронные функции get_weather(city) и get_currency(base, target).
2. Обернуть каждый вызов в try-except и ретраи (3 попытки с exponential backoff через tenacity).
3. Написать агента с оркестрацией: последовательно вызывает оба инструмента.
4. Если get_weather упал – вернуть {"currency": ..., "weather": "unavailable"}, и наоборот.
5. Сохранять состояние перед каждым вызовом в Redis (можно использовать fakeredis для тестов).
6. Написать юнит-тесты, где один из API замокан на ошибку, и проверить fallback.

Ожидаемый результат Вы получите агента, который:

• при нормальной работе возвращает полный ответ;
• при падении любого API возвращает частичный ответ без краха;
• логирует все сбои;
• проходит тесты на graceful degradation.

Связь с другими вопросами

Навигация

• Предыдущий: 55
• Следующий: 57
• Индекс: 00. Индекс разборов