Warstwowe próby ponowne dla agentów AI produkcyjnych

Warstwowe próby ponowne dla agentów AI produkcyjnych: opóźnienia w wywoływaniu modeli LLM, przywracanie po błędach narzędzi, punkty kontrolne oraz trwałe procesy pracy, dzięki czemu niestabilne API i awarie nie przerywają działania.

Wydawca: REACTAPP.TOPDodane na tej stronie 26 wrz 202610 min czytania2377 słów

Źródła: Building Retries in Agents: How to Build AI Agents That Survive Failures

Agenty, które próbują ponownie po prawdziwych awariach

Zabawczy agent z dwoma narzędziami i pętlą czasową wydaje się w porządku, dopóki nie wystąpi pierwszy limit szybkości, niestabilna API lub awaria w trakcie realizacji planu. Agenty produkcyjne wymagają wielowarstwowej odporności — a nie tylko jednego bloku try/except wokół LLM.

Awaria nie są takie same

Klasyfikuj przed ponowną próbą:

Tymczasowe błędy LLM/API (429, 503, przekroczenie czasu oczekiwania)
Błędy narzędzi, które model może naprawić (błędne argumenty, brakujące pola)
Patyny logiczne (błędny plan), które wymagają ponownego planowania
Awarie procesu/infrastruktury powodujące utratę stanu w pamięci

Słupe ponownych prób bez klasyfikacji marnują pieniądze i pogłębiają negatywne skutki.

Warstwa 1 — Ponowna próba wywołania LLM

Zapytania o uzupełnienie danych, które są idempotentne, korzystają z eksponencjalnego opóźnienia z elementem jittera:

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

Ogranicz liczba prób; wyświetl błąd typowany, gdy wyczerpie się budżet prób.

Warstwa 2 — Pozwól modelowi radzić sobie z błędami narzędzi

Zwracaj błędy narzędzi jako obserwacje, a nie poważne awarie, aby agent mógł skorygować argumenty:

# 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,
            })

Pętle w stylu Claude Code traktują stderr/stdout jako kontekst dla następnej myśli.

@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(),
        }

Szczegół 3 — Odporność na poziomie agenta

Kontrolowane przechowywanie stanu

Zachowuj kroki planu, wyniki narzędzi oraz fragmenty rozmowy, aby po ponownym uruchomieniu praca mogła być kontynuowana, zamiast robić ją od nowa:

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)

Kontrolowane przechowywanie stanu szczególnie po udanych efektach ubocznych narzędzi.

Szczegół 4 — Awarie infrastruktury i Temporal

Trwałe procesy pracy sprawdzają się, gdy kroki muszą przetrwać śmierć procesora:

# 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}")

Gdzie przydaje się Temporal

Długotrwałe procedury zatwierdzania, udział człowieka w procesie, wielogodzinne łańcuchy narzędziowe.

Gdzie staje się to skomplikowane

Proste agenty do czatowania mogą nie potrzebować silnika procesów pracy; koszty operacyjne mogą przewyższyć ryzyko.

Kiedy go używać

Użyj Temporal (lub podobnego narzędzia), gdy efekty uboczne muszą wystąpić dokładnie raz i przetrwać proces. Pomiń to w przypadku krótkich agentów obsługujących żądania i odpowiedzi z tanimi możliwościami ponawiania prób.

Lekszy wzorzec produkcyjny

Wiele zespołów łączy: ponawianie prób przez LLM + obserwację błędów narzędzi + punkty kontrolne w Redis/DB + nadzorcę, który wznowia działanie od ostatniego ukończonego kroku — bez pełnej orkiestracji przepływów pracy od samego początku.

from temporalio.common import RetryPolicy

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

List kontrolny

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
)

Odkładanie prób przy tymczasowych błędach LLM
Błędy narzędzi jako komunikaty, a nie ciche zamykanie procesów
Punkty kontrolne po modyfikacji narzędzi
Czas wygaśnięcia i maksymalna liczba kroków
Klucze idempotencji przy zapisach
Metryki: częstotliwość ponawiania prób, opóźnienie kroków, wskaźnik porzucenia

Następny krok

Dodaj testy chaosowe: zatrzymaj proces w trakcie działania narzędzia, wprowadź błędy 429 i sprawdź zachowanie przy wznowieniu działania. Odporność to funkcja produktu dla agentów, a nie coś dodanego później.

Maksymalna liczba kroków i maksymalna kwota za jedno wykonywanie w ramach budżetu; przyciski wyłączania znajdują się obok pokręteł do prób ponownych.

Pozycje pokrewne

Wzmacnianie agencyjnych grafów: umowy stanu, izolacja i bramy HITL — Niezawodność w produkcji dla LangGraph: prosty stan typu TypedDict, węzły do jednego celu, krawędzie powtórek/błędów, ludzka weryfikacja oraz możliwość obserwacji.
Strukturyzacja zapytania systemu agentów AI przed tym, nim stanie się drugą bazą danych — Podział tożsamości, zachowań, narzędzi, zasad oraz ograniczeń, aby agenci w produkcji byli łatwi do utrzymania, a logika deterministyczna znajdowała się w kodzie aplikacji, a nie w zapytaniu.

← Wszystkie artykuły Wyślij brief