Багатошарові спроби для промислових AI-агентів

Багаторівневі спроби повторного запуску для промислових AI-агентів: затримки викликів LLM, відновлення після помилок інструментів, контрольні точки та стійкі робочі процеси, щоб нестабільні API та збої не зупиняли роботу.

Видавець: REACTAPP.TOPДодано на сайт 26 вер. 2026 р.10 хв читання2377 слів

Джерела: Building Retries in Agents: How to Build AI Agents That Survive Failures

Агенти, які повторюють спроби після справжніх збоїв

Іграшковий агент із двома інструментами та циклом while здається нормальним, поки не виникають перші обмеження швидкості, нестабільні API чи збої під час виконання плану. Агенти у продакшені потребують багатошарової стійкості — а не одного блоку try/except навколо LLM.

Збої не є однаковими

Класифікуйте їх перед повторною спробою:

Тимчасові помилки LLM/API (429, 503, таймаути)
Помилки інструментів, які модель може виправити (неправильні параметри, відсутні поля)
Логічні глухі кути (неправильний план), що вимагають перепланування
Збої процесів/інфраструктури, що призводять до втрати даних у пам’яті

Сліпе повторення спроб для неправильних категорій — це марна трата грошей та посилення негативних наслідків.

Рівень 1 — Повторна спроба виклику LLM

Запити на завершення роботи, які є ідемпотентними, отримують експоненційне затримування із варіаціями:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import anthropic

llm_retry = retry(
    stop=stop_after_attempt(2),           # 2 total attempts (1 retry)
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type((
        anthropic.RateLimitError,          # 429
        anthropic.APITimeoutError,         # Timeout
        anthropic.APIConnectionError,      # Network errors
        anthropic.InternalServerError,     # 500s (often transient)
    )),
)

@llm_retry
async def call_llm(messages, tools):
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        messages=messages,
        tools=tools,
    )
    return response

Обмежте кількість спроб; відображайте формалізовану помилку, коли бюджет вичерпаний.

Рівень 2 — Дозвольте моделі вирішувати проблеми інструментів

Відображайте збої інструментів як спостереження, а не повні зупинки роботи, щоб агент міг виправити параметри:

# Inside the agentic loop
for block in response.content:
    if block.type == "tool_use":
        try:
            result = run_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result),
            })
        except Exception as e:
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": f"Error: {str(e)}",
                "is_error": True,
            })

Цикли у стилі Claude Code розглядають stderr/stdout як контекст для наступної думки.

@agent.tool
async def process_document(ctx, file_path: str) -> dict:
    """Process a document through the enrichment pipeline."""
    try:
        result = await run_enrichment(file_path)

        if result.status == "error":
            raise RuntimeError(f"Processing failed: {result.error}")

        return {
            "status": "success",
            "records_processed": result.count,
            "output_path": result.output_path,
        }
    except Exception as e:
        return {
            "status": "error",
            "error": str(e),
            "traceback": traceback.format_exc(),
        }

Рівень 3 — стійкість на рівні агента

Поступове створення контрольних точок

Зберігайте кроки плану, результати роботи інструментів та фрагменти розмови, щоб після перезапуску можна було продовжити роботу, а не повторювати вже виконані дії:

async def execute_with_resilience(agent, prompt, deps, max_retries=3):
    """Execute agent with checkpoint-based retry."""
    message_history = []

    for attempt in range(max_retries):
        try:
            result = await agent.run(
                prompt,
                deps=deps,
                message_history=message_history,  # Resume from last checkpoint
            )
            return result

        except (RateLimitError, APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise  # Final attempt failed

            # Checkpoint: save what we have so far
            # How you access partial messages depends on your framework.
            # PydanticAI exposes all_messages() on the result; other frameworks
            # may store it differently. The key is: grab whatever history exists.
            if hasattr(e, '_partial_messages'):
                message_history = e._partial_messages

            # Also save context that tools have accumulated
            save_checkpoint({
                "message_history": message_history,
                "context_fields": deps.get_context_snapshot(),
            })

            wait_time = min(2 ** attempt * 5, 60)
            logger.warning(
                f"Agent run failed (attempt {attempt + 1}/{max_retries}), "
                f"retrying in {wait_time}s with {len(message_history)} messages"
            )
            await asyncio.sleep(wait_time)

Особливо створюйте контрольну точку після успішного виконання дій інструментом.

Рівень 4 — збої інфраструктури та Temporal

Надійні робочі процеси є корисними, коли кроки мають продовжувати працювати навіть після зупинки виконавця:

# After each successful tool call, save a checkpoint
async def on_tool_complete(tool_name, result, current_messages):
    """Called after each tool completes successfully."""
    save_checkpoint({
        "message_history": current_messages,
        "last_completed_tool": tool_name,
        "context_fields": get_current_context(),
    })
    logger.info(f"Checkpoint saved after {tool_name}")

Де допомагає Temporal

Довготривалі процеси схвалення, участь людини, багатогодинні ланцюги інструментів.

Де виникають складнощі

Прості чат-агенти можуть не потребувати двигуна робочих процесів; витрати на їх управління можуть перевищувати ризик.

Коли його використовувати

Використовуйте Temporal (або подібні інструменти), коли побічні ефекти мають виникати рівно один раз та існувати довше, ніж процес. Уникайте їх для коротких агентів запитів/відповідей із дешевими спробами повторного виконання.

Легший шаблон для продакшну

Багато команд поєднують: спроби повторного виконання LLM + фіксацію помилок інструментів + контрольні точки у Redis/DB + систему керування, яка відновлює роботу з останнього завершеного кроку — без повної організації робочого процесу з самого початку.

from temporalio.common import RetryPolicy

TOOL_RETRY_POLICY = RetryPolicy(
    maximum_attempts=2,
    initial_interval=timedelta(seconds=5),
    backoff_coefficient=2.0,
)

Чек-лист

SandboxRestrictions.default.with_passthrough_modules(
    "anthropic", "httpx", "httpcore", "os", "anyio",
    "anyio._backends", "anyio._backends._asyncio", "threading",
    "pydantic_ai", "pydantic_ai.agent",
    "pydantic_ai.durable_exec", "pydantic_ai.durable_exec.temporal",
    "pydantic_graph", "pydantic_graph.beta", "fitz",  # ... and more
)

Відстрочка спроб після тимчасових помилок LLM
Помилки інструментів як повідомлення, а не мовчазне завершення роботи
Створення контрольних точок після змін інструментів
Таймаути та обмеження кількості кроків
Ключі ідемпотентності під час запису даних
Метрики: частота спроб повторного виконання, затримка кроків, частота відмов

Наступне

Додайте тести на стабільність: завершуйте процес під час роботи з інструментом, вводьте помилки 429 та перевіряйте поведінку відновлення. Стійкість — це функція продукту для агентів, а не щось другорядне.

Максимальна кількість кроків та максимальна сума у доларах за один запуск; перемикачі вимикання розташовані поруч із кнопками повторної спроби.

← Усі статті Надіслати бриф