Ich Baue Zuverlässige Agenten Aus Unzuverlässigen Teilen: Meine Entwicklungsstrategie

Hey Leute, Leo hier von agntdev.com! Heute möchte ich über etwas sprechen, was mir in letzter Zeit oft durch den Kopf geht, besonders da ich mit ein paar neuen Agentenprojekten kämpfe. Es geht um die „Dev“-Seite der Agentenentwicklung, insbesondere darum, wie wir zuverlässige Agenten aus unzuverlässigen Teilen bauen. Ja, ihr habt richtig gehört. Denn seien wir ehrlich, das ist die Realität für die meisten von uns, oder?

Wir arbeiten normalerweise nicht mit perfekt kuratierten, enterprise-grade APIs und Dienstleistungen. Vielmehr fügen wir häufig Open-Source-Modelle, Drittanbieter-APIs mit fragwürdigen Ratenlimits und vielleicht sogar ein paar selbstgebauten Mikrodiensten zusammen, die, sagen wir mal, eine Persönlichkeit haben. Und trotzdem wird immer erwartet, dass unsere Agenten einfach… funktionieren. Konsistent. Zuverlässig. Selbst wenn die zugrunde liegenden Komponenten einen Anfall haben.

Ich bin diesen Weg schon so oft gegangen. Ich erinnere mich an ein Projekt letztes Jahr, bei dem ich einen Agenten gebaut habe, um meine Open-Source-Beiträge zu verwalten. Er musste mit der GitHub-API, einem auf einem kostenlosen Plan gehosteten Sentiment-Analyse-Modell und einem benutzerdefinierten Benachrichtigungsdienst interagieren, den ich an einem Wochenende zusammengebastelt hatte. Jeder dieser Teile hatte seine Eigenheiten. GitHub hat mich manchmal unerwartet limitiert, das Sentiment-Modell hat gelegentlich die Zeitüberschreitung überschritten und mein Benachrichtigungsdienst… nun, sagen wir einfach, es hatte die Gewohnheit, nach einer Stunde Betriebszeit seine Manieren zu vergessen. Hätte ich nicht einige ernsthafte Sicherheitsvorkehrungen getroffen, wäre das ganze Ding sofort wie ein Kartenhaus zusammengebrochen.

Deshalb möchte ich heute ein paar praktische Strategien und Muster teilen, die ich angenommen habe, um meine Agenten widerstandsfähiger zu machen, selbst wenn die Teile, aus denen sie gebaut sind, alles andere als das sind.

Die unvermeidliche Wahrheit: Dinge werden kaputtgehen

Zuerst einmal sollten wir das als Evangelium akzeptieren. Keine API ist 100% verfügbar. Kein Modell ist 100% genau. Kein Netzwerk ist 100% stabil. Sobald du das akzeptierst, kannst du beginnen, für das Versagen zu entwerfen, was paradoxerweise dazu führt, dass dein Agent erfolgreicher wird.

Das Problem, das ich oft sehe, insbesondere bei neuen Entwicklern, die sich mit der Arbeit an Agenten beschäftigen, ist, dass sie bei jedem externen Aufruf den Erfolg annehmen. Sie schreiben Code wie diesen:

response = external_api.call_method(data)
# Gehe davon aus, dass die Antwort immer perfekt ist und fahre fort
processed_data = process_response(response)

Und dann, wenn external_api.call_method einen Verbindungsfehler auslöst oder 500 zurückgibt oder einfach mal das fehlerhafte JSON zurücksendet, kommt der gesamte Agent zum Stillstand. Wir können das besser machen.

Strategie 1: Solide Wiederholungen mit Zurückhaltung

Das ist wahrscheinlich die grundlegendste Technik, und doch wird sie oft schlecht oder gar nicht implementiert. Einfach sofort nach einem Fehler erneut zu versuchen, ist normalerweise keine gute Idee. Wenn der externe Dienst nicht verfügbar ist, schlägst du nur darauf ein, was die Situation möglicherweise verschlechtert oder dich selbst limitiert.

Der Schlüssel ist exponentielles Backoff. Das bedeutet, dass man zwischen den Wiederholungen progressiv längere Zeiträume wartet. Es gibt dem externen Dienst die Chance sich zu erholen und verringert die Last, die du ihm aufbürdest.

Beispiel: Python mit Tenacity

Für Python ist meine bevorzugte Bibliothek dafür Tenacity. Sie macht es unglaublich einfach, die Wiederholungslogik hinzuzufügen.

import random
import logging
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ExternalServiceError(Exception):
 """Benutzerdefinierte Ausnahme für externe Dienstfehler."""
 pass

# Simuliere einen unzuverlässigen externen API-Aufruf
def call_unreliable_api(data):
 if random.random() < 0.6: # 60% Wahrscheinlichkeit für einen Fehler
 logger.warning(f"API-Aufruf fehlgeschlagen für die Daten: {data}")
 raise ExternalServiceError("Simulierter API-Fehler oder Zeitüberschreitung")
 logger.info(f"API-Aufruf erfolgreich für die Daten: {data}")
 return {"status": "success", "result": f"processed_{data}"}

@retry(wait=wait_exponential(multiplier=1, min=4, max=10),
 stop=stop_after_attempt(5),
 retry=retry_if_exception_type(ExternalServiceError))
def get_processed_data_with_retries(input_data):
 logger.info(f"Versuche, die API für: {input_data} aufzurufen")
 return call_unreliable_api(input_data)

if __name__ == "__main__":
 try:
 result = get_processed_data_with_retries("some_important_item")
 print(f"Endergebnis: {result}")
 except ExternalServiceError as e:
 print(f"Nach mehreren Wiederholungen fehlgeschlagen: {e}")
 except Exception as e:
 print(f"Ein unerwarteter Fehler ist aufgetreten: {e}")

In diesem Snippet:

• wait_exponential sorgt dafür, dass die Wartezeiten mit jeder Wiederholung länger werden (4s, dann ~8s, dann ~16s usw., maximal 10s).
• stop_after_attempt(5) bedeutet, dass es maximal 5 Mal versucht wird.
• retry_if_exception_type(ExternalServiceError) stellt sicher, dass nur bei bestimmten Fehlern wiederholt wird, nicht z.B. bei einem KeyboardInterrupt.

Dieses Muster ist ein Lebensretter. Ich benutze es für Datenbankverbindungen, HTTP-Anfragen und sogar für die interne Kommunikation zwischen Agentenmodulen, wenn ich weiß, dass eines temporär überlastet sein könnte.

Strategie 2: Schutzschalter, um kaskadierende Fehler zu verhindern

Wiederholungen sind großartig bei transienten Fehlern. Aber was ist, wenn der Dienst komplett ausgefallen ist? Mehrmalige Versuche werden nur deine Ressourcen erschöpfen und das Problem möglicherweise für den externen Dienst verschlimmern, wenn er Schwierigkeiten hat, sich zu erholen. Hier kommt das Schutzschalter-Muster ins Spiel.

Denke daran wie an einen elektrischen Schutzschalter in deinem Haus. Wenn ein Fehler auftritt (zu viele Fehlschläge), „schaltet er sich aus“, wodurch weiterer Stromfluss verhindert wird, um das System zu schützen. Nach einer Weile kann er zurückgesetzt werden, aber er wird nicht weiterhin versuchen, Strom durch einen kurzgeschlossenen Draht zu senden.

Für Agenten überwacht ein Schutzschalter die Aufrufe an einen externen Dienst. Wenn die Fehlerquote innerhalb eines bestimmten Zeitfensters einen bestimmten Schwellenwert überschreitet, „öffnet sich“ der Schutzschalter. Wenn er offen ist, schlagen alle nachfolgenden Aufrufe an diesen Dienst sofort fehl, ohne den Aufruf überhaupt zu versuchen. Nach einer konfigurierbaren „Auszeit“-Periode wechselt der Schutzschalter in einen „halboffenen“ Zustand, der eine begrenzte Anzahl von Testaufrufen zulässt, um zu sehen, ob sich der Dienst erholt hat. Wenn diese erfolgreich sind, wird er geschlossen; wenn sie fehlschlagen, öffnet er sich wieder.

Warum das für Agenten wichtig ist:

• Ressourcenschutz: Dein Agent verschwendet keine Zeit und Ressourcen damit, einen abgestorbenen Dienst anzurufen.
• Schnellerer Fehler: Anstatt auf eine Zeitüberschreitung zu warten, erhält dein Agent ein sofortiges Fehlersignal, das es ihm ermöglicht, die Situation zu bewältigen (z.B. einen Fallback zu verwenden, das Problem zu protokollieren, einen Operator zu benachrichtigen).
• Schutz externer Dienste: Verhindert, dass dein Agent einen kämpfenden Dienst DDOS-t.

Ich implementiere dies normalerweise mit Bibliotheken. Für Python ist Pybreaker ausgezeichnet.

import time
import random
import logging
from pybreaker import CircuitBreaker, CircuitBreakerError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ExternalAPIClient:
 def __init__(self):
 # Konfiguriere den Schutzschalter:
 # 3 Fehlschläge in Folge innerhalb von 60 Sekunden öffnen den Schalter.
 # Er bleibt 5 Sekunden lang offen.
 self.breaker = CircuitBreaker(fail_max=3, reset_timeout=5, exclude=[TypeError]) # Breche nicht bei TypeErrors

 def _unreliable_call(self, data):
 if random.random() < 0.7: # 70% Wahrscheinlichkeit für einen Fehler
 logger.warning(f"Simuliere internen API-Fehler für die Daten: {data}")
 raise ConnectionError("Dienst nicht erreichbar")
 logger.info(f"API-Aufruf erfolgreich für die Daten: {data}")
 return {"result": f"processed_{data}"}

 def process_data(self, data):
 try:
 return self.breaker.call(self._unreliable_call, data)
 except CircuitBreakerError:
 logger.error(f"Schutzschalter ist offen! Kein API-Aufruf für die Daten: {data}")
 # Fallback-Logik hier: Rückgabe von zwischengespeicherten Daten, Standardwert oder ein spezifischerer Fehler
 return {"result": "fallback_data", "source": "circuit_breaker"}
 except Exception as e:
 logger.error(f"Fehler während des API-Aufrufs (nicht im Zusammenhang mit dem Schutzschalter): {e}")
 raise

if __name__ == "__main__":
 client = ExternalAPIClient()
 for i in range(15):
 print(f"\n--- Versuch {i+1} ---")
 try:
 result = client.process_data(f"item_{i}")
 print(f"Ergebnis: {result}")
 except Exception as e:
 print(f"Fehler behandelt: {e}")
 time.sleep(1) # Simuliere eine Verzögerung zwischen den Aufrufen

Führe dies aus, und du wirst sehen, wie der Schutzschalter sich nach einigen Fehlern öffnet, dann schließlich versucht, halb offen zu sein und vielleicht sogar wieder zu schließen, wenn der simulierte Dienst anfängt, sich zu benehmen.

Strategie 3: Idempotenz für zustandsändernde Operationen

Das ist entscheidend für jeden Agenten, der externe Zustände ändert (z.B. einen Datensatz erstellen, eine E-Mail senden, eine Zahlung einleiten). Wenn dein Agent versucht, eine Aktion auszuführen, und das Netzwerk aussetzt oder der externe Dienst eine Zeitüberschreitung hat, wie weißt du dann, ob die Aktion tatsächlich stattgefunden hat?

Wenn du einfach ohne Berücksichtigung der Idempotenz wiederholst, könntest du versehentlich die Aktion zweimal ausführen. Stell dir vor, du sendest dieselbe E-Mail zweimal oder, schlimmer noch, belastest einen Kunden doppelt. Das ist nicht gut.

Eine Operation ist idempotent, wenn das mehrmalige Ausführen denselben Effekt hat wie das einmalige Ausführen. Zum Beispiel ist das Setzen eines Wertes (SET x = 5) idempotent. Ein Wert zu inkrementieren (x = x + 1) ist es nicht.

Wie man Idempotenz erreicht:

• Verwenden Sie einzigartige Anfrage-IDs: Bei einer API-Anfrage, die den Zustand ändert, fügen Sie eine einzigartige, vom Client generierte ID in den Anfrageheader ein (z. B. X-Idempotency-Key). Der externe Dienst kann diesen Schlüssel dann verwenden, um doppelte Anfragen zu erkennen und die ursprüngliche Antwort ohne erneute Verarbeitung zurückzugeben.
• Gestalten Sie idempotente APIs: Wenn Sie die API kontrollieren, entwerfen Sie Endpunkte, die von Natur aus idempotent sind. Statt eines „Bestellung erstellen“-Endpunkts könnte es einen „Bestellung upsert“-Endpunkt geben, der basierend auf einer einzigartigen Bestell-ID erstellen oder aktualisieren kann.
• Überprüfen Sie den Status vor dem Wiederholen: Nach einem fehlgeschlagenen, zustandsändernden Vorgang, wenn die API dies unterstützt, fragen Sie den Status der Ressource mithilfe der einzigartigen ID ab, bevor Sie einen erneuten Versuch unternehmen.

Auch wenn ich keinen direkten Codeausschnitt dafür habe (es geht mehr um API-Design und clientseitige Logik), hier ist, wie der Denkprozess Ihres Agents aussehen könnte:

# Pseudo-Code des Agents für eine idempotente Operation
transaction_id = generate_unique_id()
payload = {"data": "some_value", "idempotency_key": transaction_id}

try:
 response = external_payment_api.process_charge(payload)
 # Erfolg! Speichern Sie transaction_id und Antwort.
except (ConnectionError, TimeoutError, APIError) as e:
 # Oh nein, es ist fehlgeschlagen. Ist die Zahlung trotzdem durchgegangen?
 logger.warning(f"Zahlung fehlgeschlagen, überprüfe den Status mit ID: {transaction_id}")
 try:
 status_response = external_payment_api.get_transaction_status(transaction_id)
 if status_response.get("status") == "completed":
 logger.info(f"Zahlung {transaction_id} war tatsächlich erfolgreich bei der Überprüfung des Wiederversuchs.")
 # Behandeln Sie es als Erfolg, speichern Sie die Info.
 else:
 logger.info(f"Zahlung {transaction_id} ist tatsächlich fehlgeschlagen, Wiederholungsversuch (mit demselben idempotency key) versuchen.")
 # Hier würden Sie mit der *gleichen* transaction_id erneut versuchen
 # Die Zahlungs-API sollte dies erkennen und keine doppelte Abbuchung vornehmen.
 response = external_payment_api.process_charge(payload)
 # ... behandeln Sie den Erfolg/Misserfolg des Wiederholungsversuchs
 except Exception as check_e:
 logger.error(f"Konnte den Transaktionsstatus für {transaction_id} nicht überprüfen: {check_e}")
 # Muss für eine manuelle Prüfung protokolliert werden oder in die Dead-Letter-Warteschlange verschoben werden

Dies erfordert die Zusammenarbeit mit dem externen Dienst, aber es ist ein entscheidendes Muster, um wirklich zuverlässige Agents zu entwickeln, die finanzielle oder andere sensible Operationen behandeln.

Strategie 4: Fallbacks und sanfte Degradierung

Manchmal ist ein externer Dienst einfach völlig nicht verfügbar, und es gibt keine Hoffnung auf einen erneuten Versuch oder Warten. In diesen Fällen stürzt ein guter Agent nicht einfach ab; er findet einen Weg, eine degradierte, aber dennoch nützliche Erfahrung zu bieten.

Das könnte bedeuten:

• Verwendung von zwischengespeicherten Daten: Wenn Ihr Agent spezifische Daten von einem Dienst benötigt, dieser aber nicht verfügbar ist, können Sie eine veraltete Version aus einem Cache verwenden?
• Bereitstellung von Standardwerten: Wenn ein KI-Modell zur Sentimentanalyse nicht verfügbar ist, können Sie alle Eingaben für eine gewisse Zeit einfach als „neutral“ oder „unbekannt“ klassifizieren, anstatt den gesamten Agentenfluss zu unterbrechen?
• Wechsel zu einem Backup-Dienst: Wenn Ihre primäre Übersetzungs-API nicht verfügbar ist, können Sie Anfragen an einen sekundären, möglicherweise weniger leistungsfähigen oder teureren Dienst weiterleiten?
• Überspringen optionaler Schritte: Wenn ein nicht kritischer Anreicherungsprozess fehlschlägt, kann der Agent einfach ohne diese Anreicherung fortfahren und möglicherweise eine Warnung protokollieren?
• Benachrichtigung von Benutzern/Betriebsleitern: Mindestens sollte der Agent sanft fehlschlagen und das Problem klar an den Benutzer oder Systembetreiber kommunizieren.

Meine Anekdote über den Ausfall des Benachrichtigungsdienstes? Mein Fallback war einfach: Wenn mein benutzerdefinierter Benachrichtigungsdienst ausfiel, protokollierte der Agent einfach das Ereignis lokal und schickte mir eine E-Mail mit den Worten „Hey, Ihr Benachrichtigungsdienst ist wahrscheinlich wieder down, überprüfen Sie die Protokolle.“ Nicht ideal für Endbenutzer, aber es verhinderte, dass der gesamte Agent blockierte, und stellte sicher, dass ich wusste, dass etwas nicht stimmte.

Umsetzbare Erkenntnisse für Ihr nächstes Agentenprojekt

1. Gehen Sie von einem Ausfall aus: Entwerfen Sie Ihren Agenten von Grund auf mit der Erwartung, dass externe Abhängigkeiten ausfallen.
2. Implementieren Sie Wiederholungen mit exponentiellem Backoff: Verwenden Sie Bibliotheken wie Tenacity (Python) oder ähnliche Muster in anderen Sprachen für temporäre Fehler.
3. Setzen Sie Schaltkreisschutzmechanismen ein: Verhindern Sie Kaskadeneffekte und sparen Sie Ressourcen, indem Sie den Schaltkreis „auslösen“, wenn ein Dienst konstant ausfällt. Pybreaker ist ein guter Anfang.
4. Priorisieren Sie Idempotenz bei Zustandsänderungen: Stellen Sie sicher, dass Operationen wie Zahlungen oder die Erstellung von Aufzeichnungen nicht dupliziert werden, wenn ein Wiederholungsversuch erfolgt. Verwenden Sie einzigartige IDs.
5. Planen Sie für sanfte Degradierung: Identifizieren Sie kritische vs. nicht kritische Abhängigkeiten und bauen Sie Fallbacks. Was ist das „wenigstens schlechte“, was Ihr Agent tun kann, wenn eine Abhängigkeit ausfällt?
6. Überwachen Sie intensiv: All diese Strategien erzeugen Protokolle. Stellen Sie sicher, dass Sie diese Protokolle sammeln und analysieren, um zu verstehen, *warum* Dinge fehlschlagen und wie oft.

Denksichere Agents zu bauen, ist nicht nur eine Frage cleverer Algorithmen oder leistungsstarker Modelle. Es geht grundlegend darum, Solidität in jede Schicht zu integrieren, insbesondere wenn man mit der unordentlichen Realität externer Abhängigkeiten umgeht. Wenn Sie diese Strategien anwenden, verbringen Sie weniger Zeit mit dem Debuggen mysteriöser Agentenausfälle und mehr Zeit damit, wirklich nützliche, zuverlässige autonome Systeme zu entwickeln.

Was sind Ihre bevorzugten Strategien, um mit fehleranfälligen externen Diensten umzugehen? Hinterlassen Sie einen Kommentar unten, ich würde gerne Ihre Erfahrungen und Lösungen hören!

🕒 Published: March 27, 2026