Verschärfung eines Python LangChain Agents mit sieben integrierten Middleware-Komponenten
Erfahren Sie, wie das Middleware von LangChain 1.0 Summarisierung, Aufrufbeschränkungen, Wiederholungsversuche, Modell-Fallback-Möglichkeiten, Redaktion sensibler Daten sowie menschliche Freigabe zu einem Gemini-Agenten hinzufügt, ohne dessen Kernlogik anzutasten.
Das Erstellen eines LangChain-Agenten, der in einem Notebook Fragen beantwortet, dauert nur wenige Minuten. Einen Agenten zu erstellen, dem man in der Produktion vertrauen kann, ist schwieriger: Er darf weder im Endlosschleifen feststecken und das API-Budget aufbrauchen, noch die Kartennummer eines Kunden an den Modellanbieter weiterleiten oder E-Mails versenden, die von niemandem genehmigt wurden. LangChain 1.0 begegnet diesen betrieblichen Problemen mit Middleware – einer Schicht aus Hooks um den Agenten-Loop, die man als einfache Liste konfigurieren kann. In diesem Leitfaden wird zunächst das zugrundeliegende Konzept erläutert, anschließend werden sieben Middleware auf einen von Gemini unterstützten Python-Agenten angewendet, und zum Schluss wird eine eigene Middleware vorgestellt, damit Sie genau erkennen können, was jeder Hook ändert und wann er zum Einsatz kommt.
Checkpoints um den Agenten-Loop herum
Stellen Sie sich einen Flughafen vor. Das Ziel ist es, von einer Stadt in eine andere zu fliegen, doch um diese einzige Handlung herum gibt es eine Reihe von Kontrollpunkten: Die Anmeldung bestätigt die Identität, Sicherheitsscans überprüfen das Gepäck, der Abflugbereich prüft die Boarding-Pässe und nach der Landung übernimmt die Gepäckausgabe. Keiner von ihnen fliegt das Flugzeug, und der Pilot überprüft nicht das Gepäck. Jede Ebene erledigt eine Aufgabe vor oder nach der Hauptaktion.
Middleware wendet dieselbe Idee auf Agenten an. Der Kern eines Agenten ist eine Schleife: Das Modell wird aufgerufen, es darf Werkzeuge auswählen, diese werden ausgeführt und der Vorgang wird wiederholt, bis das Modell eine endgültige Antwort liefert. Middleware fügt ohne Änderungen an dieser Schleife Kontrollpunkte ein:
- Das Entfernen von Kartennummern, bevor der Text das LLM erreicht, ist der Sicherheitsscanner.
- Die Anforderung, dass eine Person eine ausgehende E-Mail freigibt, ist der Abflugbereich.
- Das Anhalten nach zehn Modellaufrufen, um Ausgaben zu begrenzen, ist ein Schutzmechanismus.
Falls Sie mit TypeScript arbeiten, behandelt der begleitende Artikel zu LangChain Guardrails und Middleware die gleichen Konzepte aus der JavaScript-Sicht; dieser Leitfaden konzentriert sich hingegen auf Python und auf die konkreten eingebauten Klassen.
Die Hooks, die das Middleware verwenden kann
Der Loop stellt in jeder Phase Hooks zur Verfügung, an die sich das Middleware anschließen kann:
before_agentundafter_agentwerden einmal ausgeführt, am Anfang und am Ende einer Aufrufung.before_modelundafter_modelwerden jedes Mal ausgelöst, wenn der Loop den Modellaufruf vornehmen oder gerade abgeschlossen hat.wrap_model_callundwrap_tool_callumhüllen den eigentlichen Aufruf, sodass er neu versucht, ersetzt oder aus dem Cache bereitgestellt werden kann.
Das ist das gesamte mentale Modell. Man fügt Middleware hinzu, indem man eine Liste an create_agent übergeben wird, wie in diesem Beispiel, das E-Mail-Redaktion, eine Anrufbegrenzung sowie Wiederholungsversuche von Tools kombiniert:
agent = create_agent(
model=model,
tools=[my_tool],
middleware=[
PIIMiddleware("email", strategy="redact"),
ModelCallLimitMiddleware(run_limit=5),
ToolRetryMiddleware(max_retries=3),
],
)
Die Reihenfolge der Liste ist wichtig. Die Middleware werden nacheinander angewendet, ähnlich wie die Schichten einer Zwiebel um den Agenten herum, sodass ein zuerst aufgeführter Redaktionsschritt die rohe Eingabe vor allen anderen sieht.
Einrichtung und ein Basise-Agent
Die Middleware benötigt LangChain 1.0 oder neuer, daher sollte man sie mit dem -U-Flag installieren, um ältere Versionen zu aktualisieren. Die Beispiele verwenden Gemini über dessen kostenloses Tier; man kann einen Schlüssel in Google AI Studio erstellen.
!pip install -qU langchain langchain-google-genai
Importieren Sie os sowie die Klassen für das Gemini-Chat-Modell:
import os
from langchain_google_genai import ChatGoogleGenerativeAI
Anschließend wird das Modell konfiguriert. Der Auszug gibt die Schlüsselwerte nur zu Demonstrationszwecken direkt an; in echtem Code sollte GOOGLE_API_KEY aus der Umgebung exportiert oder stattdessen aus einem Secrets-Manager geladen werden, anstatt ihn im Quellcode zu platzieren. Eine Temperatur von null sorgt dafür, dass die Ausgaben reproduzierbar bleiben:
os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)
Das Thema jedes Experiments ist ein minimaler Agent ohne Middleware. Dafür werden create_agent sowie der tool-Decorator benötigt:
from langchain.agents import create_agent
from langchain_core.tools import tool
Es gibt ein künstliches Wettertool, das stets von Sonnenschein berichtet, und dieses wird mit einer einzigen Benachrichtigung des Benutzers aufgerufen:
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)
Jeder Abschnitt darunter umschließt einen weiteren Checkpoint für diesen Agenten.
1. SummarizationMiddleware für begrenzten Speicher
In langen Gesprächen wächst die Nachrichtengeschichte stetig, bis sie das Kontextfenster überschreitet, und jedes zusätzliche Token wird berechnet. SummarizationMiddleware überprüft die Größe der Geschichte in before_model; wenn sie einen Schwellenwert überschreitet, fasst es ältere Nachrichten zu einem Zusammenfassung zusammen und behält nur die neuesten in ihrem ursprünglichen Wortlaut bei.
from langchain.agents.middleware import SummarizationMiddleware
Die untenstehende Konfiguration verwendet dasselbe Modell zur Erstellung von Zusammenfassungen, triggert bei 10 Nachrichten und lässt die letzten 4 unverändert. Der Test erstellt eine gefälschte Geschichte mit sechs Städten (zwölf Nachrichten) und fragt anschließend, welche Stadt zuerst kam:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
SummarizationMiddleware(
model=model, # which LLM writes the summary
trigger=("messages", 10), # summarize when history hits 10 messages
keep=("messages", 4), # keep the 4 most recent messages intact
),
],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history)) # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"])) # 6
Dreizehn Nachrichten kommen herein, und sechs bleiben übrig: die Zusammenfassung, die vier erhalten gebliebenen Nachrichten sowie die neue Antwort. Das Modell antwortet weiterhin mit „Delhi“, weil dieser Fakt in die Zusammenfassung übernommen wurde. Das Zählen der Nachrichten macht das Verhalten in einer Demo leicht erkennbar, doch in der Produktion überwachen tokenbasierte Auslöser wie („tokens“, 3000) oder ein Bruchteil des Kontextfensters wie („fraction“, 0.8) die tatsächlichen Kosten und setzen weitaus bessere Grenzen. Beachten Sie, dass Zusammenfassungen verlustbehaftet sind: Genauzahlen oder Identifikatoren, die früh in einem Gespräch erwähnt werden, könnten nicht überleben.
2. Aufrufbegrenzungen als Kosten-Notabschalter
Der teuerste Fehler eines Agenten ist ein unkontrollierter Schleifenprozess, bei dem das Modell und die Tools sich ständig gegenseitig aufrufen und Minutenlang Kosten verursachen, bevor jemand etwas bemerkt. Zwei Middleware-Elemente setzen hier eine harte Obergrenze:
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
Hier ist das Modell auf drei Aufrufe pro Ausführung beschränkt, wobei exit_behavior="end" verwendet wird, damit der Agent sauber stoppt anstatt eine Ausnahme auszulösen, und die Tools auf zwei Aufrufe. Die Anfrage fordert absichtlich nacheinander sechs Städte an:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
# "end" = stop gracefully instead of raising an error
ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
ToolCallLimitMiddleware(run_limit=2),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)
Der Agent benötigt sechs Abfragen, erreicht seine Grenzen und beendet die Ausführung ordnungsgemäß mit den vorliegenden Teilergebnissen. Ein thread_limit-Parameter steht außerdem zur Verfügung, um die Anzahl der Aufrufe über einen gesamten Konversationsthread statt nur eine einzelne Ausführung zu begrenzen. Diese beiden Maßnahmen dienen als einfache Absicherung; wählen Sie Grenzwerte, die deutlich über dem Bedarf legitimer Anfragen liegen, damit sie nur bei echten Fehlverläufen aktiviert werden.
3. ToolRetryMiddleware für unzuverlässige Abhängigkeiten
Echte Tools versagen: HTTP-Aufrufe laufen ab und Verbindungen werden unterbrochen. ToolRetryMiddleware verwendet wrap_tool_call, um Fehler zu erkennen und mit exponentiellem Backoff erneut zu versuchen.
from langchain.agents.middleware import ToolRetryMiddleware
Zur Veranschaulichung zählt ein Tool zur Überwachung von Aktienkursen seine Versuche und löst bei den ersten beiden fehlgeschlagenen Versuchen einen ConnectionError aus, bevor es erfolgreich ist. Das Middleware-Modul erlaubt bis zu drei Wiederholungsversuche, wobei zunächst eine einsekündige Verzögerung eintritt und diese bei jedem Versuch verdoppelt wird:
attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
"""Get the current stock price for a ticker symbol."""
attempt_counter["count"] += 1
print(f" [tool called — attempt #{attempt_counter['count']}]")
if attempt_counter["count"] < 3:
raise ConnectionError("API timeout — please retry")
return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
model=model,
tools=[flaky_stock_price],
middleware=[
ToolRetryMiddleware(
max_retries=3, # retry a failed tool up to 3 times
initial_delay=1.0, # wait 1s before first retry
backoff_factor=2.0, # double the wait each time: 1s, 2s, 4s
),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)
Das Tool versagt zweimal, das Middleware-Modul wartet und versucht es erneut – beim dritten Versuch gelingt es. Aus Sicht des Agents ist nichts schiefgelaufen. Wiederholungsversuche sind nur für idempotente Operationen wie Lesevorgänge sicher; bei Tools, die beispielsweise Karten abbuchen oder Nachrichten senden, könnte eine Wiederholung zu denselben Nebeneffekten führen.
4. ModelFallbackMiddleware für Ausfälle des Anbieters
Dieselbe Idee der Widerstandsfähigkeit kann den Aufruf des Modells schützen. Wenn das primäre Modell aufgrund von Rate Limiting oder einem Ausfall beispielsweise versagt, wiederholt dieses Middleware den Anfragenversuch an die Backup-Modelle in der von Ihnen angegebenen Reihenfolge:
from langchain.agents.middleware import ModelFallbackMiddleware
Im Beispiel wird das leichtere Gemini-Modell als primär verwendet und ein zweites Gemini-Modell als Backup hinzugefügt:
backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
model=model, # primary: gemini-3.5-flash-lite
tools=[get_weather],
middleware=[ModelFallbackMiddleware(backup_model)],
)
Solange das primäre Modell funktionsfähig ist, wird man nichts bemerken – genau das ist auch der Zweck. Der Nutzen zeigt sich erst dann, wenn Ihr Anbieter Probleme hat. Für eine stärkere Absicherung sollten Sie ein Backup von einem anderen Anbieter in Betracht ziehen, da Ausfälle oft alle Modelle hinter derselben API betreffen.
5. PIIMiddleware für sensible Daten
Oftmals möchten Sie nicht, dass E-Mails, Kartennummern oder IP-Adressen überhaupt an den Modellanbieter gesendet werden. PIIMiddleware scannt den Text in before_model, bevor das Modell ihn sieht, und wendet eine von vier Strategien an: redact, mask, hash oder block.
from langchain.agents.middleware import PIIMiddleware
Dieser Agent verfügt über keine Tools. Er ersetzt E-Mail-Adressen vollständig durch einen Platzhalter und maskiert Kartennummern so, dass nur die letzten vier Ziffern übrig bleiben; beide Regeln werden auf die Benutzereingaben angewendet:
agent = create_agent(
model=model,
tools=[],
middleware=[
# Replace emails entirely with [REDACTED_EMAIL]
PIIMiddleware("email", strategy="redact", apply_to_input=True),
# Mask credit cards — keeps last 4 digits
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Draft a support reply to priya.sharma@example.com confirming her card "
"4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)
Das Modell erstellt seine Antwort, ohne jemals die echte Adresse oder die vollständige Kartennummer zu erhalten. Sie können auch einen benutzerdefinierten PII-Typ mit Ihrer eigenen regulären Ausdrucksregel registrieren, um beispielsweise jeden Text zu blockieren, der wie eine interne API-Schlüssel aussieht:
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")
Die auf Mustern basierende Erkennung fängt gut strukturierte Werte ein, aber nicht jede kreative Schreibweise – betrachten Sie sie daher als eine Schutzschicht und nicht als Garantie für Konformität.
6. HumanInTheLoopMiddleware für Freigabestufen
Einige Aktionen sind aufgrund ihrer Folgen zu wichtig, um vollständig autonom ausgeführt zu werden: Versenden von E-Mails, Löschung von Datensätzen oder Durchführung von Zahlungen. HumanInTheLoopMiddleware stoppt den Agenten unmittelbar vor dem Ausführen eines sensiblen Tools, wartet auf eine Entscheidung eines Menschen und setzt die Ausführung anschließend fort.
Dies funktioniert anders als die vorherigen Middlewares. Ein pausierter Agent wird nicht beendet. Ein Checkpointer speichert seinen vollständigen Zustand, und die Thread-ID dient als Schlüssel, um diesen später wiederzufinden und fortzusetzen. Der Prüfer kann den Aufruf freigeben, seine Argumente ändern oder ihn ablehnen.
Die Importe bringen das Middleware, einen in-Memory-Checkpointer von LangGraph sowie den Command-Typ zum Wiederaufnehmen mit sich:
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
Der untenstehende Agent verfügt über ein send_email-Tool, markiert es zur Unterbrechung und speichert den pausierten Zustand in InMemorySaver. Die erste Aufrufung auf dem Thread demo-1 bittet den Agenten, einen Manager per E-Mail zu kontaktieren:
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to the given recipient."""
return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
model=model,
tools=[send_email],
middleware=[
# Pause and ask a human whenever the agent wants to call send_email
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
],
checkpointer=InMemorySaver(), # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
{"messages": [{"role": "user", "content":
"Send an email to boss@company.com saying the report is ready."}]},
config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])
Die Ausführung stoppt, bevor das Tool gestartet wird. Der Eintrag __interrupt__ im Ergebnis zeigt den ausstehenden Aufruf mit Empfänger, Betreff und Inhalt an – es wurde jedoch nichts gesendet. Die Genehmigung wird als Command auf demselben Thread gesendet, wodurch die pausierte Ausführung wieder aufgenommen wird:
# Step 2: approve and resume
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config, # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)
Anstelle von approve können Sie reject mit einem Grund oder edit mit geänderten Argumenten senden. In einer echten Anwendung wäre hier der Punkt, an dem ein Genehmigungsbildschirm angezeigt wird. Zwei praktische Hinweise: InMemorySaver verliert seinen Zustand, wenn der Prozess neu gestartet wird, weshalb Produktivsysteme einen persistenten Checkpointer benötigen; außerdem hat sich das Format des Wiederaufnahmepakets zwischen den LangChain-Versionen geändert, daher sollten Sie dies in der Middleware-Dokumentation für die von Ihnen genutzte Version überprüfen.
7. Erstellen Sie Ihr eigenes Middleware mit einem Dekorator
Falls keine integrierten Lösungen ausreichen, ist ein benutzerdefiniertes Middleware-Modul einfach zu erstellen, da jeder Hook einen entsprechenden Dekorator besitzt. Die notwendigen Importe sind der before_model-Dekorator sowie der AgentState-Typ:
from langchain.agents.middleware import before_model, AgentState
Dieses Beispiel protokolliert, wie viele Nachrichten bei jedem Aufruf des Modells gesendet werden sollen, und wird wie jedes vorgefertigte Middleware-Element der Liste hinzugefügt:
@before_model
def log_before_model(state: AgentState, runtime) -> None:
print(f" [middleware] Calling model with {len(state['messages'])} messages")
# Returning None = observe only.
# Returning a dict would UPDATE the agent's state (e.g., trim messages)
return Noneagent = create_agent(
model=model,
tools=[get_weather],
middleware=[log_before_model], # plugs in like any prebuilt middleware
)
Der Rückgabewert ist die wichtige Designentscheidung. Der Rückgabe von None bedeutet, dass das Middleware-Element lediglich beobachtet. Die Rückgabe eines Wörterbuchs aktualisiert den Zustand des Agents, wodurch Nachrichten entfernt, Kontext eingefügt oder benutzerdefinierte Einschränkungen durchgesetzt werden können. Für die anderen Hooks existieren ebenfalls Dekoratoren: @before_agent, @after_model, @wrap_model_call, @wrap_tool_call sowie @dynamic_prompt zur Erstellung von Systemanfragen zur Laufzeit.
Kernpunkte
- Middleware trennt operative Aspekte von der Agentenlogik: Der Loop bleibt unverändert, während Kontrollpunkte als einfache Liste um ihn herum angeordnet werden und an
create_agentübergeben werden.
TodoListMiddleware, LLMToolSelectorMiddleware und ContextEditingMiddleware ist in der offiziellen Referenz dokumentiert.Weitere Literatur
- Von einem Einzelknoten-Chatbot zu einem MCP-gestützten Agent in LangGraph — Erstellen Sie eine LangGraph-Anwendung Schicht für Schicht: Zustand und Reduzierer, Kanten, Tool-Loops, checkpointierte Threads, drei Streaming-Modi sowie Tools über MCP bereitgestellt.
- Hybride Agenten-Memory: Kombination von BM25 und Vektorsuche mit RRF in Python — Erfahren Sie, warum reine Vektorsuche als Agenten-Memory versagt, wie Reciprocal Rank Fusion BM25 und dichte Ergebnisse in Python kombiniert und wann GraphRAG-Zusammenfassungen hilfreich sind.