Как вы делаете agent robustness к missing API (когда инструмент временно недоступен)?

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

Agent robustness (устойчивость агента) к недоступности инструментов достигается многоуровневой системой откатов: первичный API → резервный API → симулятор на основе истории → кэш. Ключевые приёмы — fallback chain, prompt conditioning (динамическое изменение промпта агента), graceful degradation (плавное снижение качества) и human-in-the-loop (передача управления человеку). Система должна логировать каждую ошибку и иметь timeout/retry механизмы.

1. Термин: Agent robustness (устойчивость агента)

Agent robustness — способность AI-агента корректно функционировать и выдавать результат, даже когда один или несколько внешних инструментов (API) временно недоступны, возвращают ошибки или задерживаются.

Почему это важно:

• В production API|внешние API могут падать, иметь rate limits, сетевые проблемы.
• Агент, который при ошибке просто возвращает «не могу», неприемлем для бизнес-сценариев.
• Пользователь ожидает хотя бы частичного ответа или осмысленной альтернативы.

Ключевые принципы обеспечения robustness:

• Fail-fast (быстрое обнаружение) — определить ошибку за минимальное время.
• Graceful degradation — не падать, а перейти в режим пониженной функциональности.
• Observability — логировать каждый сбой для анализа.

2. Fallback chain (цепочка откатов)

Fallback chain — последовательность из нескольких вариантов выполнения запроса, упорядоченных по убыванию качества/скорости. Если первый вариант недоступен, агент переходит ко второму, и так далее.

Типичная цепочка для поискового инструмента:

Реализация на Python (упрощённая):

async def robust_search(query: str, state: AgentState) -> SearchResult:
    for level, api_name in enumerate(["primary", "secondary", "simulator", "cache"]):
        try:
            if api_name == "primary":
                return await call_primary_api(query)
            elif api_name == "secondary":
                return await call_secondary_api(query)
            elif api_name == "simulator":
                return await simulate_from_history(query, state)
            else:
                return await cache_lookup(query)
        except (HTTPError, TimeoutError, ConnectionError) as e:
            logger.warning(f"Fallback {api_name} failed: {e}")
            continue
    return SearchResult(error="All sources unavailable")

Важно: каждый fallback должен иметь свой timeout (первичный – 2с, вторичный – 5с, симулятор – 10с).

3. Prompt conditioning (условное управление промптом)

Prompt conditioning — динамическое изменение system prompt агента в зависимости от доступности инструментов.

Если primary API недоступен, агент получает инструкцию:

«API поиска временно недоступен. Используй свой внутренний knowledge base или задай пользователю уточняющий вопрос, чтобы получить больше контекста.»

Пример в коде:

def build_prompt(available_tools: set, user_query: str) -> str:
    base = "Ты — ассистент. Отвечай на основе предоставленных инструментов.\n"
    if "search" not in available_tools:
        base += "[ВНИМАНИЕ] Инструмент поиска недоступен. Используй свои знания или попроси пользователя уточнить.\n"
    if "calculator" not in available_tools:
        base += "[ВНИМАНИЕ] Калькулятор недоступен. Проверяй вычисления вручную (можно ошибиться).\n"
    return base

Термин system prompt — начальная инструкция для LLM, задающая роль и правила поведения.

4. Graceful degradation (плавная деградация)

Graceful degradation — стратегия, при которой агент не прекращает работу, а переключается на менее точные, но работающие методы.

Пример: если API погоды недоступен, агент может:

• Использовать среднестатистические данные за прошлые годы.
• Сообщить пользователю: «Точные данные недоступны, вот приблизительная оценка».
• Перейти в offline mode: использовать только внутреннюю документацию и базу знаний.

Offline mode подразумевает отключение всех внешних вызовов и ответы исключительно на основе эмбеддингового поиска по корпоративной вики.

5. Human-in-the-loop (человек в контуре)

Когда все автоматические fallback исчерпаны, агент передаёт управление человеку. Это может быть:

• Запрос на ручной ввод данных.
• Эскалация к модератору.
• Открытие формы для заполнения.

Пример:

async def handle_api_failure(query: str, state: AgentState):
    if state.retry_count >= 3:
        return escalate_to_human(query, reason="API недоступен после трёх попыток")
    # ... попытка fallback

Термин human-in-the-loop — подход, при котором критическое решение принимает человек, а агент лишь подготавливает контекст.

6. Мониторинг и логирование ошибок API

Для анализа robustness необходимо логировать каждый сбой:

• Какой инструмент упал.
• Тип ошибки (timeout, 5xx, 429).
• Время отказа.
• Какой fallback сработал.

Хорошей практикой является circuit breaker pattern (шаблон «автоматический выключатель»): если API отвечает ошибками N раз подряд, он временно исключается из ротации на T секунд.

class CircuitBreaker:
    def __init__(self, failure_threshold=3, reset_timeout=30):
        self.failure_count = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time = 0

    async def call(self, api_func, *args):
        if self.failure_count >= self.threshold:
            if time.time() - self.last_failure_time < self.reset_timeout:
                raise CircuitBreakerOpen("API temporarily disabled")
            else:
                self.failure_count = 0  # попробовать сбросить
        try:
            result = await api_func(*args)
            self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = time.time()
            raise

7. Timeout и retry стратегии

Timeout — максимальное время ожидания ответа от API. Должны быть разные таймауты для разных API и fallback'ов.

Retry strategy (стратегия повторных попыток):

• Exponential backoff — увеличение задержки между попытками (1с, 2с, 4с, 8с…).
• Jitter — добавление случайной задержки, чтобы не синхронизироваться с другими агентами.

import asyncio, random

async def call_with_retry(api_func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await asyncio.wait_for(api_func(), timeout=3)
        except asyncio.TimeoutError:
            wait = 2 ** attempt + random.uniform(0, 1)
            await asyncio.sleep(wait)
    raise RuntimeError("All retries exhausted")

8. Эвристики для определения доступности API

Не всегда ошибка очевидна. Используются эвристики:

• Health check endpoint — отдельный лёгкий вызов для проверки статуса.
• Ping / heartbeat — агент периодически проверяет доступность API в фоне.
• Fallback based on time — в ночные часы, когда API может давать сбои, агент сразу использует симулятор.

9. Тестирование устойчивости (chaos engineering)

Для проверки robustness проводят chaos testing:

• Искусственно отключают primary API.
• Повышают latency до 10+ секунд.
• Возвращают случайные 500 ошибки.

Ожидаемый результат: агент корректно переключается на fallback, а human-in-the-loop срабатывает при полном отказе всех источников.

10. Сравнение стратегий

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

Задача: Создать простого AI-агента для ответов на вопросы, который использует внешний API (например, имитация OpenWeatherMap). Реализовать fallback chain: primary API → кэш (простой dict) → генерация ответа на основе LLM (без данных).

Инструменты: Python, asyncio, httpx, transformers (или OpenAI API), pytest.

Шаги:

1. Написать класс Agent с методами call_weather_api(query) (выбрасывает исключение с вероятностью 50%).
2. Реализовать robust_call с цепочкой: API → кэш → LLM simulator.
3. Написать тест, в котором API всегда падает, и проверить, что агент возвращает ответ (из кэша или от LLM).
4. Добавить логирование и circuit breaker.
5. Измерить, какой процент запросов покрыт без падения.

Ожидаемый результат: Агент ни при каких обстоятельствах не возвращает ошибку, а даёт либо точный ответ (из кэша), либо приблизительный (от LLM), либо запрашивает уточнение.

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

Навигация

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