Ich baue zuverlässige Agenten aus unzuverlässigen Komponenten: meine Entwicklungsstrategie

Hallo zusammen, Leo hier von agntdev.com! Heute möchte ich über etwas sprechen, das mich in letzter Zeit sehr beschäftigt, besonders während ich an einigen neuen Agentenprojekten arbeite. Es geht um den Teil „Dev“ der Agentenentwicklung, genauer gesagt, wie wir zuverlässige Agenten aus unzuverlässigen Komponenten bauen. Ja, Sie haben richtig gehört. Denn ganz ehrlich, das ist die Realität für die meisten von uns, oder?

Wir arbeiten normalerweise nicht mit perfekt gestalteten und unternehmensgerechten APIs und Diensten. Viel häufiger setzen wir Open-Source-Modelle, fragwürdige Drittanbieter-APIs mit zweifelhaften Ratenlimits und vielleicht sogar einige selbstgemachte Mikroservices zusammen, die, sagen wir mal, eine eigene Persönlichkeit haben. Und dennoch besteht die Erwartung, dass unsere Agenten einfach… funktionieren. Konsistent. Zuverlässig. Selbst wenn die zugrunde liegenden Komponenten ihre Macken haben.

Ich bin diesen Weg schon so oft gegangen. Ich erinnere mich an ein Projekt aus dem letzten Jahr, bei dem ich einen Agenten baute, um meine Open-Source-Beiträge zu verwalten. Er sollte mit der GitHub-API, einem auf einem kostenlosen Niveau gehosteten Sentiment-Analysemodell und einem benutzerdefinierten Benachrichtigungsdienst interagieren, den ich an einem Wochenende zusammengebastelt hatte. Jeder dieser Dienste hatte seine Eigenheiten. GitHub schränkte mich manchmal unerwartet ein, das Sentiment-Modell konnte manchmal ablaufen, und mein Benachrichtigungsdienst… naja, sagen wir einfach, er hatte die Angewohnheit, nach einer Stunde Betrieb seine Manieren zu vergessen. Wenn ich nicht ernsthafte Schutzmaßnahmen implementiert hätte, wäre alles wie ein Kartenhaus zusammengebrochen.

Heute möchte ich einige Strategien und praktische Modelle teilen, die ich angenommen habe, um meine Agenten widerstandsfähiger zu machen, auch wenn die Elemente, aus denen sie bestehen, alles andere als zuverlässig sind.

Die unvermeidliche Wahrheit: Dinge werden brechen

Zunächst einmal akzeptieren wir dies als Eingeständnis. Keine API ist zu 100 % verfügbar. Kein Modell ist zu 100 % genau. Kein Netzwerk ist zu 100 % stabil. Sobald Sie das akzeptieren, können Sie anfangen, für den Fehler zu entwerfen, was gegenintuitiv Ihren Agenten effektiver macht.

Das Problem, das ich oft sehe, insbesondere bei neueren Entwicklern, die die Arbeit mit Agenten erkunden, ist, dass sie den Erfolg bei jedem externen Aufruf voraussetzen. Sie schreiben Code wie diesen:

response = external_api.call_method(data)
# Annehmen, dass die Antwort immer perfekt ist und fortfahren
processed_data = process_response(response)

Und dann, wenn external_api.call_method einen Verbindungsfehler auslöst, oder einen 500 zurückgibt, oder einfach ein fehlerhaftes JSON sendet, steht der gesamte Agent still. Wir können das besser machen.

Strategie 1: Solide Versuche mit exponentiellem Backoff

Das ist wahrscheinlich die grundlegendste Technik, und doch wird sie oft schlecht implementiert oder gar nicht. Sofort nach einem Fehler zu versuchen, ist normalerweise eine schlechte Idee. Wenn der externe Dienst ausgefallen ist, schlagen Sie nur weiter darauf ein, was die Situation verschlimmern oder zu einem Ratenlimit führen kann.

Der Schlüssel ist der exponentielle Backoff. Das bedeutet, dass man zwischen den Versuchen immer längere Zeiträume wartet. Das gibt dem externen Dienst die Gelegenheit, sich zu erholen, und reduziert die Belastung, die Sie ihm auferlegen.

Beispiel: Python mit Tenacity

Für Python ist meine Bibliothek der Wahl dafür Tenacity. Sie erleichtert es, Wiederholungslogik auf unglaublich saubere Weise 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

# Simulieren eines unzuverlässigen externen API-Aufrufs
def call_unreliable_api(data):
 if random.random() < 0.6: # 60 % Chance auf Fehler
 logger.warning(f"Der API-Aufruf schlug fehl für die Daten: {data}")
 raise ExternalServiceError("Simuliertes Ausfall oder Timeout der API")
 logger.info(f"Der API-Aufruf war 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 API aufzurufen für: {input_data}")
 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 Versuchen fehlgeschlagen: {e}")
 except Exception as e:
 print(f"Ein unerwarteter Fehler ist aufgetreten: {e}")

In diesem Auszug:

• wait_exponential verlängert die Wartezeiten mit jedem Versuch (4s, dann ~8s, dann ~16s usw., bis maximal 10s).
• stop_after_attempt(5) bedeutet, dass maximal 5 Versuche unternommen werden.
• retry_if_exception_type(ExternalServiceError) stellt sicher, dass nur bei bestimmten Fehlern wiederholt wird, nicht zum Beispiel bei einem KeyboardInterrupt.

Dieses Muster ist ein echter Retter. Ich benutze es für Datenbankverbindungen, HTTP-Anfragen und sogar für die interne Kommunikation zwischen den Modulen von Agenten, wenn ich weiß, dass eines von ihnen vorübergehend überlastet sein könnte.

Strategie 2: Schutzschalter zur Verhinderung von Kaskadenfehlern

Wiederholungen sind großartig für flüchtige Fehler. Aber was passiert, wenn der Dienst komplett ausgefallen ist? Immer wieder zu versuchen, verbraucht Ihre Ressourcen und könnte das Problem für den externen Dienst verschärfen, falls dieser Schwierigkeiten beim Wiederherstellen hat. Hier kommt das Schutzschalter-Modell ins Spiel.

Denken Sie daran wie an einen Sicherungsautomaten in Ihrem Haus. Wenn ein Fehler auftritt (zu viele Fehlversuche), „schaltet er sich ab“, verhindert, dass mehr Strom fließt und schützt das System. Nach einer bestimmten Zeit kann er zurückgesetzt werden, aber er wird nicht weiter versuchen, Strom durch einen kurzgeschlossenen Draht fließen zu lassen.

Für die Agenten überwacht ein Schutzschalter die Aufrufe an einen externen Dienst. Wenn die Fehlerquote einen bestimmten Schwellenwert in einem gegebenen Zeitrahmen überschreitet, „öffnet sich“ der Schaltkreis. Wenn er geöffnet ist, schlagen alle folgenden Aufrufe an diesen Dienst sofort fehl, ohne überhaupt zu versuchen, den Aufruf zu tätigen. Nach einer konfigurierbaren „Wartezeit“ wechselt der Schaltkreis in einen „halb-offenen“ Zustand, der eine begrenzte Anzahl von Testaufrufen zulässt, um zu sehen, ob sich der Dienst erholt hat. Wenn diese erfolgreich sind, schließt er sich; wenn sie fehlschlagen, öffnet er sich wieder.

Warum das für Agenten wichtig ist:

• Ressourcenschonung: Ihr Agent verschwendet keine Zeit und Ressourcen damit, einen fehlerhaften Dienst aufzurufen.
• Schnellere Fehlermeldung: Anstatt auf eine Zeit zu warten, erhält Ihr Agent sofort ein Fehlsignal, was ihm ermöglicht, die Situation zu bewältigen (z.B. eine Fallback-Lösung nutzen, das Problem protokollieren, einen Betreiber benachrichtigen).
• Schützt externe Dienste: Verhindert, dass Ihr Agent einen überlasteten Dienst DDOS-angreift.

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

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):
 # Konfigurieren Sie den Schutzschalter:
 # 3 aufeinanderfolgende Fehler innerhalb von 60 Sekunden öffnen den Kreis.
 # Er bleibt 5 Sekunden lang geöffnet.
 self.breaker = CircuitBreaker(fail_max=3, reset_timeout=5, exclude=[TypeError]) # Keine Unterbrechung bei TypeErrors

 def _unreliable_call(self, data):
 if random.random() < 0.7: # 70 % Fehlerwahrscheinlichkeit
 logger.warning(f"Simulation eines internen API-Fehlers für die Daten: {data}")
 raise ConnectionError("Dienst nicht erreichbar")
 logger.info(f"Der API-Aufruf war 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"Der Kreis ist offen! Kein API-Aufruf für die Daten: {data}")
 # Fallback-Logik hier: Rückgabe von zwischengespeicherten Daten, einem Standardwert oder das Auslösen eines spezifischeren Fehlers
 return {"result": "fallback_data", "source": "circuit_breaker"}
 except Exception as e:
 logger.error(f"Fehler beim API-Aufruf (nicht verbunden 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"Verwalterfehler: {e}")
 time.sleep(1) # Verzögerung zwischen den Aufrufen simulieren

Führen Sie dies aus, und Sie werden sehen, wie der Kreis nach einigen Fehlern öffnet, dann schließlich versucht, sich halb zu öffnen und vielleicht sogar wieder zu schließen, wenn sich der simulierte Dienst besser verhält.

Strategie 3: Idempotenz für statusverändernde Operationen

Es ist entscheidend für jeden Agenten, der den externen Zustand verändert (zum Beispiel, einen Datensatz erstellen, eine E-Mail senden, eine Zahlung initiieren). Wenn Ihr Agent versucht, eine Aktion auszuführen und das Netzwerk instabil ist oder der externe Dienst abläuft, wie wissen Sie, ob die Aktion tatsächlich stattgefunden hat?

Wenn Sie einfach einen weiteren Versuch starten, ohne die Idempotenz zu berücksichtigen, könnten Sie versehentlich die Aktion zweimal ausführen. Stellen Sie sich vor, Sie senden die gleiche E-Mail zweimal oder noch schlimmer, berechnen einen Kunden zweimal. Das ist nicht gut.

Eine Operation ist idempotent, wenn mehrmalige Ausführung den gleichen Effekt hat wie einmalige Ausführung. Zum Beispiel, eine Werte festlegen (SET x = 5) ist idempotent. Eine Werte erhöhen (x = x + 1) ist es nicht.

So erreichen Sie Idempotenz:

• Verwenden Sie eindeutige Anfrage-IDs: Bei einem API-Aufruf, der den Zustand ändert, fügen Sie eine vom Client generierte eindeutige ID im Header der Anfrage hinzu (zum Beispiel X-Idempotency-Key). Der externe Dienst kann dann diesen Schlüssel verwenden, um doppelte Anfragen zu erkennen und die originale Antwort ohne erneute Verarbeitung zurückzugeben.
• Entwickeln Sie idempotente APIs: Wenn Sie die API kontrollieren, gestalten Sie Endpunkte, die von Natur aus idempotent sind. Zum Beispiel, anstatt einen Endpunkt “Bestellung erstellen” zu haben, haben Sie einen Endpunkt “Bestellung upsert”, der basierend auf einer eindeutigen Bestell-ID erstellen oder aktualisieren kann.
• Überprüfen Sie den Status vor dem erneuten Versuch: Nach einem fehlgeschlagenen statusverändernden Vorgang, wenn die API es unterstützt, fragen Sie den Status der Ressource unter Verwendung der eindeutigen ID ab, bevor Sie einen neuen Versuch unternehmen.

Obwohl ich keinen direkten Code dafür habe (es handelt sich mehr um API-Design und Client-seitige Logik), sieht hier möglicherweise die Logik Ihres Agenten aus:

# Pseudocode des Agenten 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 response.
except (ConnectionError, TimeoutError, APIError) as e:
 # Oh nein, das ist fehlgeschlagen. Wurde die Zahlung trotzdem durchgeführt?
 logger.warning(f"Die Zahlung ist fehlgeschlagen, Überprüfung des 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"Die Zahlung {transaction_id} war tatsächlich erfolgreich beim Überprüfen des erneuten Versuchs.")
 # Als erfolgreich betrachten, Informationen speichern.
 else:
 logger.info(f"Die Zahlung {transaction_id} ist tatsächlich fehlgeschlagen, Versuch eines erneuten Versuchs (mit dem gleichen Idempotenzschlüssel).")
 # Hier würden Sie mit derselben transaction_id erneut versuchen
 # Die Zahlungs-API sollte dies erkennen und nicht zweimal abrechnen.
 response = external_payment_api.process_charge(payload)
 # ... Erfolg/Misserfolg des Wiederholungsversuchs verwalten
 except Exception as check_e:
 logger.error(f"Überprüfung des Transaktionsstatus für {transaction_id} fehlgeschlagen: {check_e}")
 # Muss protokolliert werden für eine manuelle Überprüfung oder in die Warteschlange für nichtzustellbare Nachrichten übergehen

Dies erfordert die Zusammenarbeit des externen Dienstes, aber es ist ein kritisches Modell zur Erstellung von wirklich zuverlässigen Agenten, die finanzielle oder andere sensible Operationen verwalten.

Strategie 4: Backup-Lösungen und sanfte Abwicklung

Manchmal ist ein externer Dienst einfach völlig unerreichbar, 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 abgemilderte, 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, aber der Dienst ausgefallen ist, können Sie eine veraltete Version eines Caches verwenden?
• Bereitstellung von Standardwerten: Wenn ein KI-Modell zur Sentiment-Analyse ausgefallen ist, können Sie dann einfach alle Einträge während eines Zeitraums als “neutral” oder “unbekannt” klassifizieren, anstatt den gesamten Agentenfluss zum Scheitern zu bringen?
• Wechsel zu einem Backup-Dienst: Wenn Ihr Hauptübersetzungs-API ausgefallen ist, können Sie Anfragen an einen zweiten Dienst umleiten, der möglicherweise weniger performant oder teurer ist?
• Optionale Schritte überspringen: Wenn ein nicht kritischer Anreicherungs-Schritt ausfällt, kann der Agent dann einfach ohne diese Anreicherung fortfahren und vielleicht eine Warnung protokollieren?
• Benachrichtigung von Benutzern/Betriebsleitern: Mindestens sollten Sie sanft scheitern und das Problem dem Benutzer oder dem Systembetreiber klar kommunizieren.

Meine Anekdote über den Ausfall des Benachrichtigungsdienstes? Meine Backup-Lösung war einfach: Wenn mein benutzerdefinierter Benachrichtigungsdienst ausgefallen war, protokollierte der Agent einfach das Ereignis lokal und sendete eine E-Mail an *mich*, die sagte: “Hey, Ihr Benachrichtigungsdienst ist wahrscheinlich wieder ausgefallen, bitte überprüfen Sie die Protokolle.” Nicht ideal für Endbenutzer, aber es verhinderte, dass der gesamte Agent abstürzte und stellte sicher, dass ich über ein Problem informiert war.

Konkrete Maßnahmen für Ihr nächstes Agentenprojekt

1. Gehen Sie von einem Ausfall aus: Entwerfen Sie Ihren Agenten von Anfang an mit der Erwartung, dass externe Abhängigkeiten ausfallen.
2. Implementieren Sie erneute Versuche mit exponentiellem Backoff: Verwenden Sie Bibliotheken wie Tenacity (Python) oder ähnliche Muster in anderen Sprachen für vorübergehende Fehler.
3. Setzen Sie Schutzschalter ein: Verhindern Sie Kaskadenfehler und erhalten Sie Ressourcen, indem Sie den Kreis “auslösen”, wenn ein Dienst konstant ausfällt. Pybreaker ist ein guter Anfang.
4. Priorisieren Sie Idempotenz für statusverändernde Operationen: Stellen Sie sicher, dass Operationen wie Zahlungen oder die Erstellung von Datensätzen sich nicht duplizieren, wenn ein erneuter Versuch stattfindet. Verwenden Sie eindeutige IDs.
5. Planen Sie für eine sanfte Abwicklung: Identifizieren Sie kritische Abhängigkeiten gegenüber nicht kritischen und erstellen Sie Backup-Lösungen. Was ist die “am wenigsten schlechte” Sache, die Ihr Agent tun kann, wenn eine Abhängigkeit ausfällt?
6. Überwachen Sie aggressiv: All diese Strategien erzeugen Protokolle. Stellen Sie sicher, dass Sie diese Protokolle sammeln und analysieren, um zu verstehen, *warum* Dinge fehlschlagen und wie oft.

Verlässliche Agenten zu bauen, geht nicht nur um kluge Algorithmen oder leistungsstarke Modelle. Es geht grundlegend um die Ingenieureffizienz in jeder Schicht, besonders wenn es um die trüben Realitäten externer Abhängigkeiten geht. Durch die Anwendung dieser Strategien werden Sie weniger Zeit damit verbringen, mysteriöse Agentenabstürze zu debuggen, und mehr Zeit damit, wirklich nützliche und zuverlässige autonome Systeme zu schaffen.

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

🕒 Published: March 29, 2026