LangChain Zuverlässigkeit: Streaming, Batching, Wiederholungsversuche, Fallback-Mechanismen, Middleware

Bis jetzt hatte der Helpdesk-Mitarbeiter nur eine bewusst eingeschränkte Form von Freiheit. Er darf die Zuständigkeiten für Dienstleistungen sowie aktuelle Vorfälle einsehen, muss aber vor dem Verfassen einer Ticketnotiz innehalten, die Argumente aufzeigen, die er verwenden will, und auf einen Menschen warten. Diese Regelung funktioniert solange, wie jede Modellaufruf- und Toolaufrufoperation schnell antwortet. In der Praxis ist das jedoch nicht der Fall.

Ein Modellaufruf kann fünfzehn Sekunden in Anspruch nehmen. Ein Anbieter könnte aufgrund von Rate-Limits den Datenverkehr für eine ganze Minute blockieren. Ein einzelnes Ticket in einer Batch-Verarbeitung könnte fehlerhafte Eingaben enthalten. Ein Backup-Modell könnte anders antworten als das Hauptmodell, und ein Benutzer könnte den Browser schließen, während ein genehmigtes Tool noch läuft. Jeder dieser Fälle erfordert eine benannte Verhaltensregel, die durch einen Test untermauert wird.

Dieser Schritt fügt fünf weitere Einschränkungen für den Mitarbeiter hinzu. Bei einer einzelnen Anfrage:

one request
  -> stream honest progress
  -> retry a small transient failure
  -> use a labeled fallback if attempts are exhausted
  -> stop at time and call limits

Für eine Menge unzusammenhängender Anfragen:

many independent requests
  -> run a bounded batch
  -> keep one result or error beside each input

Die Regel gilt stets: Verbergen Sie einen Fehler nicht hinter einer fröhlichen Antwort.

Das Ticket, mit dem wir weiterhin testen

Der Testfall bleibt unverändert: Die EU-Zahlungsvorgänge funktionieren nach dem Deployment um 14:05 nicht mehr, wobei in den Logs von checkout-api der Fehler „Database Connection Refused“ auftritt. Der Benutzer bittet den Agenten, den Verantwortlichen sowie aktuelle Vorfälle herauszufinden, einen nächsten Schritt vorzuschlagen und nach der Freigabe eine Notiz zu INC-2048 hinzuzufügen.

Auf dem einfachen Weg liest der Agent den Verantwortlichen und die aktuellen Vorfälle, schreibt eine Empfehlung, legt einen Vorschlag für eine Notiz vor, wartet auf die Freigabe, schreibt einmal und gibt das Ergebnis zurück. Um es schwieriger zu machen: Reichen Sie gleich drei weitere Tickets ein, lassen Sie einen Anruf beim Provider zweimal fehlschlagen, machen Sie ein Ticket ungültig und unterbrechen Sie die Verbindung zum Kunden mitten im ersten Antrag. Alles, was folgt, ist darauf ausgelegt, genau damit umzugehen.

Geben Sie Fehlern zunächst Namen

"The call failed" doesn't tell you what to do. Sort failures into categories that lead to different actions.

• Temporary provider failure. A reset connection, a request timeout, or a provider reply that clearly asks you to come back later. Retry a few times with a pause in between.
• Rate limit. The provider is guarding a quota or its capacity. Follow any retry information it sends, slow the caller down and lower concurrency. Repeating immediately only makes it worse.
• Invalid input. The ticket lacks an ID, the schema doesn't match, or the text exceeds an application limit. Answer with a clear client error, and don't resend the same malformed request.
• Authorization failure. The caller isn't allowed to read a runbook or write to the ticket. Halt there; no switching credentials and no asking another model to work around the refusal.
• Fehler im Tool-Betrieb. Das Ticket ist bereits geschlossen, der Übergang ist nicht erlaubt oder es existiert bereits eine ähnliche Notiz. Wandeln Sie die Antwort des Tools in einen sinnvollen Anwendungsfehler um, da ein erneuter Versuch möglicherweise fehlerhaft sein könnte.
• Fehler bei der Validierung der Modellausgabe. Der Anbieter hat Daten zurückgegeben, die den Ausgabevertrag verletzen. Bei einem nur zum Lesen bestimmten Ergebnis kann ein Reparaturversuch akzeptabel sein, es muss jedoch weiterhin ein endgültiger Fehlerweg vorhanden sein.
• Unbekannter Fehler. Die Anwendung konnte den Fehler nicht klassifizieren. Schließen Sie den Fehler ab, dokumentieren Sie die Fehlerart und untersuchen Sie ihn – behandeln Sie niemals jeden Ausnahmefall als vorübergehend.

Zusammen bestimmen diese Kategorien, welche Fehler erneut versucht werden, welche auf einen Ersatzverfahren zurückgreifen und welche direkt an den Aufrufer weitergeleitet werden.

Projektstruktur und Einrichtung

Der Snapshot für diesen Schritt enthält:

langchain-helpdesk/
├── app.py
├── reliability.py
├── batch_runner.py
├── stream_view.py
├── middleware_config.py
└── tests/
    ├── test_reliability.py
    └── test_stream_view.py

Installieren Sie die Pakete:

python -m pip install -U \
  langchain \
  langchain-core \
  langgraph \
  pytest

Die Tests rufen niemals einen Anbieter auf.

Streaming liefert Inhalt; es überprüft nicht

Ein leeres Bildschirmbild während die Tools laufen und das Modell arbeitet, ist eine schlechte Erfahrung. Streaming behebt dies, indem es im Laufe der Arbeit Ereignisse auslöst. LangChain’s aktuelles Agent-Streaming bietet Ihnen drei Ansichten:

• updates: abgeschlossene Agentenschritte
• messages: Modelltokens zusammen mit Metadaten
• custom: Fortschrittsdaten, die Ihre Anwendung definiert

Sie enthalten unterschiedliche Arten von Informationen, daher sollten sie nicht alle als eine einzige Antwort dargestellt werden.

Eine einfache Stream-Schleife

for chunk in agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": request_text,
            }
        ]
    },
    config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    send_to_client(chunk)

Eine detaillierte Erklärung: agent.stream(...) startet denselben Ablauf wie invoke(...), liefert jedoch während der Ausführung Teile des Ergebnisses. Das erste Argument ist die Zustandsaktualisierung, bei der es sich hier um eine Benachrichtigung des Benutzers handelt. config enthält Einstellungen, die auf die Anfrage beschränkt sind – wie beispielsweise die Thread-ID –, und sensible Daten gehören nicht dazu. stream_mode=[„updates“, „custom“] fordert Daten nach Abschluss jedes Schritts sowie eigene Fortschrittsmeldungen an; daher werden in diesem Beispiel keine Texte Token für Token zurückgegeben. version=„v2“ wählt das aktuelle Format der Blöcke, bei dem jeder Block einen Typ, einen Namespace sowie eine Datenträgerinhalte enthält. send_to_client(chunk) bezieht sich auf Ihren eigenen Code, der beispielsweise Server-Sent Events, WebSocket-Nachrichten oder Zeilen in einer Konsole ausgeben kann.

Rohdaten der Blöcke für die Benutzeroberfläche übersetzen

Der Snapshot wandelt die Blöcke um, bevor sie eine Benutzeroberfläche erreichen:

@dataclass(frozen=True)
class ProgressEvent:
    kind: str
    label: str
    payload: Any

def readable_events(chunks):
    for chunk in chunks:
        chunk_type = chunk.get("type")
        data = chunk.get("data")        if chunk_type == "updates" and isinstance(data, dict):
            for step_name, step_data in data.items():
                yield ProgressEvent(
                    kind="step",
                    label=step_name,
                    payload=step_data,
                )

Dieser Adapter bietet der Benutzeroberfläche einen kleinen, stabilen Vertrag. Er kann Zeilen wie diese anzeigen:

Checking service ownership...
Reading recent incidents...
Waiting for ticket-note approval...
Ticket note approved and written.

Was es vor Abschluss des Laufs auf keinen Fall anzeigen darf, ist Folgendes:

Done.

Teilweise Text ist kein endgültiger Text

Streaming-Token lassen eine Antwort schneller erscheinen, erhöhen aber weder die Sicherheit noch die Genauigkeit. Das Modell kann einen Satz beginnen und anschließend eine andere Richtung einschlagen, ein Tool kann nach dem Versand der Tokens versagen, und eine Richtlinienprüfung kann die vorgeschlagene Aktion ablehnen.

Verwenden Sie explizite Ereigniszustände: running, waiting_for_approval, completed, failed und cancelled. Nur completed weist ein endgültiges Ergebnis auf.

Bestimmen Sie, was ein Abbruch bedeutet

Ein geschlossener Browser beweist nicht, dass der Server gestoppt wurde. Legen Sie die Richtlinie im Voraus fest:

• Stellen Sie arbeitsspeicherbasierte Aufgaben ab, wenn ein Abbruch möglich ist und niemand auf das Ergebnis wartet.
• Lassen Sie genehmigte, dauerhafte Aufgaben zu Ende laufen, falls der Benutzer vorhat, später wieder darauf zuzugreifen.
• Gehen Sie niemals davon aus, dass eine Schreiboperation nicht stattgefunden hat, nur weil die Antwort verloren ging.

Suchen Sie die Operation anhand der Idempotenzschlüssel vor dem erneuten Schreibversuch nach.

Streaming und Stornierung hängen zusammen, sind aber keine einzige Funktion.

Unabhängige Eingaben in Batches verarbeiten

LangChain-Runnables bieten batch und abatch an. Die Standardfunktion batch führt unabhängige Aufrufe in einem Threadpool aus, was sich für viele netzwerkgebundene Operationen eignet; ein vom Anbieter bereitgestelltes Runnable kann dies durch eine native Batch-API ersetzen. Der Helpdesk-Mitarbeiter nimmt mehrere unverwandte Tickets entgegen:

outputs = runnable.batch(
    tickets,
    config={"max_concurrency": 4},
    return_exceptions=True,
)

tickets ist die geordnete Liste der Eingaben. max_concurrency begrenzt die Anzahl der gleichzeitig ausgeführten Aufrufe, wobei vier nur als Beispiel dient; bestimmen Sie Ihre Zahl anhand der Quoten des Anbieters, der Kapazität der Tools, der Kosten sowie der gemessenen Latenzzeit. Mit return_exceptions=True führt das Versagen eines Elements nicht mehr dazu, dass die Ergebnisse der anderen verloren gehen. Anschließend ordnet die Anwendung jedes Ausgabeergebnis seinem jeweiligen Eingabewert zu:

for ticket, output in zip(tickets, outputs, strict=True):
    if isinstance(output, BaseException):
        save_failure(ticket.ticket_id, type(output).__name__)
    else:
        save_recommendation(ticket.ticket_id, output)

Dadurch bleibt ein Ergebnis pro Ticket erhalten.

Überspringen Sie die parallele Verarbeitung in diesen Fällen: Der zweite Eintrag hängt vom ersten ab; jeder Eintrag aktualisiert dasselbe Datensatz ohne Konkurrenzkontrolle; die Ausführungsreihenfolge ist selbst eine Geschäftsregel; der Anbieter bietet Transaktionen in Bulk mit unterschiedlichen Garantien an; oder ein Benutzer kann so viele Einträge einsenden, dass die gemeinsame Kapazität ausgeschöpft wird. Die Gruppierung der Aufgaben verteilt die Last nur, sie lässt sie nicht verschwinden.

Wiederholen Sie den kleinstmöglichen sicheren Bereich

Durch Wiederholen wird die Arbeit erneut ausgeführt, daher sollten Sie sorgfältig darauf achten, was Sie wiederholen. Das Erneuten eines gesamten Workflows kann das Abrufen, die Aufrufe des Modells sowie den Schreibvorgang im Ticket erneut auslösen. Ein sichererer Ansatz besteht darin, nur den Schritt des Lesens zu wiederholen, von dem bekannt ist, dass er vorübergehend fehlschlägt. Dafür wird im Snapshot ein spezieller Fehler deklariert:

class TemporaryProviderError(RuntimeError):
    """A short-lived error that may succeed when tried again."""

und umschließt einen einzigen ausführbaren Codeabschnitt:

primary_call = RunnableLambda(primary).with_retry(
    retry_if_exception_type=(TemporaryProviderError,),
    wait_exponential_jitter=False,
    stop_after_attempt=3,
)

RunnableLambda(primary) wandelt eine einfache Python-Funktion in ein ausführbares Objekt um. retry_if_exception_type fungiert als Allowlist – ein ungültiges Ticket passt daher nicht und wird direkt an den Aufrufer zurückgesendet. Durch die Einstellung wait_exponential_jitter=False bleibt der Offline-Test schnell und deterministisch. stop_after_attempt=3 zählt jede Versuchsanstrengung, was insgesamt drei Versuche bedeutet und nicht drei Wiederholungsversuche zusätzlich zum ersten. Gegenüber einem echten Netzwerkdienst ist ein exponentielles Backoff mit Jitter in der Regel die bessere Wahl, da dadurch verhindert wird, dass alle Worker gleichzeitig auf den Dienst zugreifen.

Schreibvorgänge sollten nicht blind wiederholt werden

Stellen Sie sich vor, der Ticket-Service fügt die Notiz hinzu, dann wird die Verbindung unterbrochen, bevor die Erfolgsantwort ankommt. Der Anrufer sieht einen Zeitüberschreitungsfehler, doch die Schreiboperation könnte bereits erfolgt sein, und ein naiver Neversuch würde die Notiz zweimal hinzufügen. Verwenden Sie einen Idempotenzschlüssel:

operation_id = "ticket-note:INC-2048:approval-7331"

Die Ticket-Service-Speicherung erfolgt zusammen mit dem ersten erfolgreichen Schreibvorgang. Wenn derselbe Schlüssel erneut auftaucht, wird das gespeicherte Ergebnis zurückgegeben und die Änderung nicht ein zweites Mal angewendet. Wenn das externe System keine Idempotenz gewährleisten kann, sollte man seinen aktuellen Zustand vor einem erneuten Versuch überprüfen und akzeptieren, dass einige Fehlerzustände nicht vollständig beseitigt werden können.

Eine Alternative ist ein anderer Weg

Sobald die zulässigen Versuche aufgebraucht sind, kann ein als Alternative verwendbares Programm übernehmen:

reliable = primary_call.with_fallbacks(
    [RunnableLambda(conservative_fallback)],
    exceptions_to_handle=(TemporaryProviderError,),
)

Es läuft nur für den genannten vorübergehenden Fehler, und was es zurückgibt, ist Folgendes:

Recommendation(
    ticket_id=ticket.ticket_id,
    text=(
        "The recommendation service is temporarily unavailable. "
        "Keep the ticket open and compare the failed deployment "
        "with the last known-good release."
    ),
    source="fallback",
)

In der Antwort wird zugegeben, was passiert ist. Es wird nicht behauptet, dass nach den Runbooks gesucht wurde oder dass das Ticket aktualisiert wurde.

Auf ein anderes Modell umzusteigen ist auch nicht dasselbe wie das Erneut-Starten des ersten Modells. Das zweite Modell kann sich bei Toolaufrufen anders verhalten, eine kleinere Kontextgrenze haben, strukturierte Ausgaben auf andere Weise unterstützen, andere Sicherheitsmechanismen anwenden, in Bezug auf Latenz und Kosten abweichen oder unter einer anderen Datenrichtlinie des Anbieters fallen. Behandeln Sie den Fallback-Weg als eigene Konfiguration und testen Sie ihn getrennt. Der beste Fallback ist manchmal überhaupt kein Modell, sondern nur eine feste, konservative Nachricht oder eine Warteschlange für menschliche Überprüfung.

Timeouts existieren in mehreren Ebenen

Ein einziger Timeout-Wert reicht nicht aus. Weisen Sie separate Budgets für die Verbindungserrichtung, eine einzelne Modellanfrage, eine einzelne Tool-Anfrage, den gesamten Agentenaufruf, das Warten auf Freigabe sowie die Client-Verbindung zu. Das Budget für den gesamten Ablauf muss jedes individuelle Anfragenbudget übertreffen, benötigt aber dennoch eine Obergrenze.

Legen Sie das Timeout-Verhalten neben die Operation, die es begrenzt: Provider-Timouts beim Provider-Client, Tool-Timouts beim Tool-Client sowie die Frist für Webanfragen an der Servicegrenze. Wenn ein Timeout auslöst, notieren Sie, welche Schicht die Arbeit beendet hat. Ein einfaches „Timed out“ ohne angegebene Grenze ist schwer zu beheben.

Middleware umschließt Agentenaufrufe

In den früheren Schritten wurden hauptsächlich feste Ketten verwendet, und der Freigabeschritt führte Agent-Middleware ein. Middleware eignet sich auch gut für Regeln, die viele Aufrufe von Modellen oder Tools abdecken sollten: Wiederholung fehlgeschlagener Modellaufvorgänge, Begrenzung der Anzahl der Modellaufrufe, Begrenzung der Anzahl der Toolaufrufe, Filterung von Tools für den aktuellen Benutzer, Hinzufügen von Anfrage-Metadaten für Protokolle und Spuren, Ausblenden ausgewählter personenbezogener Daten sowie Abbruch, wenn das Budget einer Anwendung aufgebraucht ist. Der Snapshot verwendet die eingebaute Middleware:

def reliability_middleware():
    return [
        ModelRetryMiddleware(
            max_retries=2,
            retry_on=(
                TemporaryProviderError,
                TimeoutError,
                ConnectionError,
            ),
            on_failure="error",
            initial_delay=0.5,
            backoff_factor=2.0,
            max_delay=4.0,
            jitter=True,
        ),
        ModelCallLimitMiddleware(
            run_limit=8,
            exit_behavior="error",
        ),
        ToolCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ]

ModelRetryMiddleware umschließt die Aufrufe der Modelle, wobei ein erster Aufruf zusammen mit zwei Wiederholungsversuchen insgesamt höchstens drei Versuche ermöglicht. ModelCallLimitMiddleware verhindert, dass ein Modellloop endlos läuft. ToolCallLimitMiddleware stoppt einen übermäßigen Einsatz von Tools innerhalb einer Ausführung. Diese Kontrollmechanismen erfüllen getrennte Aufgaben: Eine Aufrufbegrenzung ersetzt keine Zeitüberschreitung, eine Zeitüberschreitung ersetzt keine Freigabe und eine Freigabe ersetzt keine Autorisierung.

Die Reihenfolge ist wichtig

Hooks, die vor einem Aufruf ausgeführt werden, folgen der Reihenfolge der Liste, diejenigen, die danach ausgeführt werden, in umgekehrter Reihenfolge, und verschachtelte Hooks sind so angeordnet wie Funktionsaufrufe. Wenn Middleware A Middleware B umschließt, kann eine Wiederholung in A die Arbeit von B erneut ausführen; daher sollte man immer die Reihenfolge festhalten, wenn zwei Komponenten beide wiederholen, Modelle ändern oder den Zustand verändern könnten. Die Freigabe für eine Schreiboperation sollte außerhalb jeder Wiederholung liegen, die den bereits genehmigten Nebeneffekt erneut auslösen könnte.

Kundenspezifische Hooks sind einfacher Code

LangChain bietet node-basierte Hooks (before_model, after_model) sowie wrap-basierte Hooks (wrap_model_call, wrap_tool_call). Node-basierte Hooks eignen sich für Überprüfungen und Zustandsaktualisierungen an festgelegten Stellen. Wrap-basierte Hooks werden für Code verwendet, der eine Aufrufsequenz umschließen muss oder entscheidet, ob der Handler nullmal, einmal oder mehrmals ausgeführt wird. Allerdings sollten nicht alle Anwendungsregeln zu Middleware werden. Zustandswechsel von Tickets, Autorisierungsentscheidungen sowie Regeln zur Datenbesitzverteilung lassen sich besser als benannte Anwendungsdienste verfolgen.

Testen der Fehlerwege offline

Der Snapshot enthält fünf Tests, von denen keiner einen Provider benötigt.

Ein vorübergehender Fehler wird beim zweiten Versuch erfolgreich abgehandelt:

result = runnable.invoke(ticket())

assert calls == 2
assert result.source == "primary"

Ausgebrannte Versuche verwenden den beschrifteten Ersatzweg:

assert calls == 3
assert result.source == "fallback"
assert "temporarily unavailable" in result.text

Ungültige Eingaben werden weder erneut versucht noch versteckt:

with pytest.raises(InvalidTicketError):
    runnable.invoke(ticket())

assert calls == 1

Ein Misserfolg einer Charge löscht keinen anderen Erfolg aus:

assert results[0].status == "completed"
assert results[1].status == "failed"
assert results[1].error_type == "RuntimeError"

Der Fortschritt bleibt getrennt von den Modell-Token:

assert [(event.kind, event.label) for event in events] == [
    ("step", "model"),
    ("progress", "custom"),
    ("token", "model"),
]

Führen Sie sie aus:

pytest -q

Erwartetes Ergebnis:

5 passed

Diese Tests belegen Ihre lokalen Steuerungsregeln. Sie beweisen jedoch nicht, dass die Fehler eines Anbieters korrekt abgebildet werden – dafür ist ein kleiner Integrationstest mit dem echten Anbieter-SDK sowie ein kontrolliertes Testkonto erforderlich.

Tragen Sie die Fehlermatrix auf

Fügen Sie vor der Veröffentlichung eine solche Tabelle in den Code oder die Dokumentation ein:

failure                         retry?   fallback?   client result
connection reset               yes      maybe       503 if exhausted
provider rate limit            yes      maybe       503 or queued
invalid ticket                 no       no          422
authorization denied           no       no          403
structured output invalid      limited  maybe       502 if exhausted
ticket already closed          no       no          409
unknown exception              no       no          500 + incident ID

Die genauen Statuscodes hängen von Ihrer API ab. Entscheidend ist, dass jede Fehlerklasse eine klare Vorgehensweise hat.

Noch fehlend

Das Laufzeitverhalten ist jetzt klarer, doch es handelt sich weiterhin um einen Übersichtsbeitrag für die Kommandozeile. Es gibt keine authentifizierte HTTP-API, keine Start- und Bereitschaftsprüfungen, kein Laden von Geheimnissen, keine strukturierten Anfragenprotokolle, keinen dauerhaften Speicher für Aufgaben, keine Bereitstellungskonfiguration, keine echten Integrationstests mit Anbietern, kein Bewertungssatz vor der Veröffentlichung und keine Überprüfung der Produktionsprotokolle. Eine Servicestruktur, die diese Funktionen hinzufügt, folgt später.

Stand des Helpdesk-Workflows

Der Workflow verfügt nun über sichtbare Zustände und begrenzte Wiederherstellungsmöglichkeiten:

request
  -> progress events
  -> small retry boundary
  -> labeled fallback when allowed
  -> model and tool call limits
  -> approval before write
  -> final result or named failure

batch
  -> bounded concurrency
  -> one outcome per ticket

Keine Kontrollmaßnahme beseitigt die Unsicherheit vollständig. Ziel ist es, diese sichtbar zu halten, einzuschränken und testbar zu machen. Danach folgt der gesamte Helpdesk-Ablauf hinter einer API – mit Konfigurationen, Geheimnissen, Anfragenummern, Zeitlimits, Protokollen sowie Verfolgungshilfen – anschließend deterministische Tests, Bewertungsfälle und Bereitstellungsprüfungen, sowie eine Liste dessen, was ein echtes Unternehmen bereitstellen muss, bevor der Service mit Live-Tickets in Berührung kommt. Alles Genannte wurde am 31. August 2026 anhand der Python-Dokumentation von LangChain zu Themen wie Streaming, ausführbaren Wiederholungsversuchen, Fallback-Mechanismen, Batch-Verarbeitung, Middleware und Testing überprüft; Paket-APIs sowie Fehlerarten der Anbieter können sich ändern.