Schichtweise Wiederholungsversuche für Produktions-AI-Agenten

Mehrschichtige Wiederholungsversuche für Produktions-AI-Agenten: Verzögerung bei LLM-Aufrufen, Wiederherstellung bei Tool-Fehlern, Checkpoints sowie zuverlässige Arbeitsabläufe, damit instabile APIs und Abstürze die Ausführungen nicht beeinträchtigen.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 26. Sept. 202610 Min. Lesezeit2377 Wörter

Quellen: Building Retries in Agents: How to Build AI Agents That Survive Failures

Agenten, die nach echten Fehlern weiterarbeiten

Ein einfacher Agent mit zwei Werkzeugen und einer Schleife scheint in Ordnung zu sein – bis zum ersten Rate-Limit, einer unzuverlässigen API oder einem Absturz mitten im Plan. Produktive Agenten benötigen mehrschichtige Widerstandsfähigkeit – nicht nur einen try/except-Block um den LLM herum.

Fehler sind nicht alle gleich

Klassifizieren Sie vor dem erneuten Versuch:

Vorübergehende Fehler des LLMs/APIs (429, 503, Zeitüberschreitungen)
Werkzeugfehler, die das Modell beheben kann (falsche Argumente, fehlende Felder)
Logikblockaden (falscher Plan), die eine Neuplanung erfordern
Prozess- oder Infrastrukturabstürze, die den In-Memory-Zustand verlieren

Blinder Neversuche bei der falschen Fehlerklasse verschwenden Geld und verstärken schlechte Abläufe.

Ebene 1 – Der LLM-Aufruf wird erneut versucht

Idempotente Vervollständigungsanfragen nutzen exponentielles Backoff mit Jitter:

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

Begrenzen Sie die Anzahl der Versuche; geben Sie einen typisierten Fehler an, wenn das Budget aufgebraucht ist.

Ebene 2 – Das Modell kümmert sich um Werkzeugfehler

Geben Sie Tool-Fehler als Beobachtungen statt als totale Abstürze zurück, damit der Agent die Argumente korrigieren kann:

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

Loop-Strukturen im Stil von Claude Code behandeln stderr/stdout als Kontext für den nächsten Gedanken.

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

Schicht 3 – Widerstandsfähigkeit auf Agentenebene

Progressives Checkpointing

Speichern Sie Plan-Schritte, Tool-Ergebnisse sowie Teile der Konversation, damit ein Neustart fortgesetzt werden kann, anstatt die bereits erledigte Arbeit nochmals durchzuführen:

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)

Vor allem nach erfolgreichen Nebeneffekten der Tools sollte ein Checkpoint erstellt werden.

Schicht 4 – Infrastrukturabstürze und Temporal

Robuste Workflows sind besonders nützlich, wenn Schritte auch nach dem Ausfall eines Prozessors weiterlaufen müssen:

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

Wo Temporal hilft

Lange Laufzeiten bei Genehmigungsprozessen, menschliche Einmischung sowie mehrstündige Tool-Ketten.

Wo es schwierig wird

Einfache Chat-Agenten benötigen möglicherweise keinen Workflow-Engine; die Betriebskosten können das Risiko übersteigen.

Wann man sie verwenden sollte

Verwenden Sie Temporal (oder ähnliche Tools), wenn Nebeneffekte genau einmal auftreten müssen und länger bestehen als ein Prozess. Vermeiden Sie es bei kurzen Anfrage/Antwort-Agenten mit günstigen Wiederholungsversuchen.

Ein leichteres Produktionsmuster

Viele Teams kombinieren: Wiederholungsversuche von LLMs + Beobachtungen von Tool-Fehlern + Checkpoints in Redis/DB + einen Supervisor, der vom letzten abgeschlossenen Schritt aus fortfährt – ohne von Anfang an eine vollständige Workflow-Orchestrierung.

from temporalio.common import RetryPolicy

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

Checkliste

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
)

Backoff bei vorübergehenden LLM-Fehlern
Tool-Fehler als Nachrichten, nicht stille Abbrüche
Checkpoints nach Änderungen der Tools
Timeouts und maximale Schrittanzahl
Idempotenzschlüssel bei Schreibvorgängen
Metriken: Wiederholungsrate, Schwellenzeit pro Schritt, Abbruchrate

Fügen Sie Chaos-Tests hinzu: Beenden Sie den Prozess mitten in einem Tool, injizieren Sie 429-Fehler und überprüfen Sie das Wiederanfangsverhalten. Resilienz ist eine Produktfunktion für Agenten – kein Nachgedanke.

Budget für maximale Schritte und maximalen Betrag pro Lauf; Abschaltschalter sollten neben den Wiederholungsknöpfen angeordnet sein.

← Alle Artikel Brief senden