LangChain Middleware
Die LangChain-Middleware ist die leistungsstärkste Funktion von LangChain. Sie ermöglicht es Ihnen, benutzerdefinierte Logik an verschiedenen Stellen der Agent-Ausführung einzufügen, um Wiederholungsversuche, Degradierung, Caching, Inhaltsfilterung, Protokollierung und andere Funktionen umzusetzen — ohne den Code des Agenten selbst zu ändern.
Was ist Middleware
Middleware besteht ausHooks (Hook)im Agent-Ausführungsablauf. Jeder Hook ermöglicht es Ihnen, benutzerdefinierten Code zu bestimmten Zeitpunkten auszuführen:
Beispiel
# Intuitives Verständnis von Middleware:
# Nehmen wir an, der Agent-Ausführungsablauf sieht so aus:
# 1. Benutzereingabe → 2. Modell denkt nach → 3. Möglicher Aufruf von Tools → 4. Modell denkt erneut nach → 5. Ergebnis ausgeben
# Middleware ermöglicht es Ihnen, benutzerdefinierte Logik zwischen diesen 5 Phasen einzufügen:
# 1. Benutzereingabe
# ↓ [before_agent-Hook: Protokollierung, Berechtigungsprüfungen]
# 2. Modell-Denkprozess
# ↓ [before_model-Hook: Nachrichten-Vorverarbeitung]
# ↓ [wrap_model_call-Hook: Wiederholung, Degradation, Caching]
# ↓ [after_model-Hook: Inhaltsmoderation]
# 3. Tool-Ausführung
# ↓ [wrap_tool_call-Hook: Tool-Aufruf-Wiederholung]
# 4. Zurück zum Modell-Denkprozess (Schleife bis Abschluss)
# ↓ [after_agent-Hook: Ergebnisformatierung, statistische Analyse]
# 5. Ergebnis ausgeben
# Nehmen wir an, der Agent-Ausführungsablauf sieht so aus:
# 1. Benutzereingabe → 2. Modell denkt nach → 3. Möglicher Aufruf von Tools → 4. Modell denkt erneut nach → 5. Ergebnis ausgeben
# Middleware ermöglicht es Ihnen, benutzerdefinierte Logik zwischen diesen 5 Phasen einzufügen:
# 1. Benutzereingabe
# ↓ [before_agent-Hook: Protokollierung, Berechtigungsprüfungen]
# 2. Modell-Denkprozess
# ↓ [before_model-Hook: Nachrichten-Vorverarbeitung]
# ↓ [wrap_model_call-Hook: Wiederholung, Degradation, Caching]
# ↓ [after_model-Hook: Inhaltsmoderation]
# 3. Tool-Ausführung
# ↓ [wrap_tool_call-Hook: Tool-Aufruf-Wiederholung]
# 4. Zurück zum Modell-Denkprozess (Schleife bis Abschluss)
# ↓ [after_agent-Hook: Ergebnisformatierung, statistische Analyse]
# 5. Ergebnis ausgeben
Sechs Hook-Punkte
Die Middleware von LangChain bietet 6 Hooks, unterteilt nach Ausführungszeitpunkt in zwei Kategorien:
| Hook | Ausführungshäufigkeit | Ausführungsposition | Hauptzweck |
|---|---|---|---|
| before_agent | Einmal | Bevor der Agent startet | Initialisierung, Berechtigungsprüfungen, Eingabevorverarbeitung |
| before_model | Jede Schleifeniteration | Vor dem Modellaufruf | Nachrichten-Vorverarbeitung, dynamische Kontextinjektion |
| wrap_model_call | Jede Schleifeniteration | Umschließt den Modellaufruf | Wiederholung, Degradierung, Caching, Umschreiben von Anfragen |
| after_model | Jede Schleifeniteration | Nach dem Modellaufruf | Inhaltsmoderation, Antwortfilterung, Protokollierung |
| wrap_tool_call | Jeder Tool-Aufruf | Umschließt die Tool-Ausführung | Tool-Wiederholung, Ergebnis-Caching, Parameter-Neuschreibung |
| after_agent | Einmal | Nach Beendigung des Agenten | Ausgabeformatierung, Statistiken, Ressourcenbereinigung |
Zwei Arten der Verwendung
Middleware kann auf zwei Arten verwendet werden: Klassenvererbung oder Dekoratoren.
Methode 1: Dekorator (Empfohlen)
Beispiel
from langchain.agents.middleware import before_model, after_model
# Dekorator-Ansatz: einfach und intuitiv
@before_model
def log_before(state, runtime):
"""Vor jedem Modellaufruf protokollieren"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Aktuelle Nachrichtenanzahl: {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Nach jedem Modellaufruf protokollieren"""
last_msg = state["messages"][-1] if state.get("messages") else None
if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
print(f"[after_model] Modell hat einen Tool-Aufruf angefordert")
return None
# Dekorator-Ansatz: einfach und intuitiv
@before_model
def log_before(state, runtime):
"""Vor jedem Modellaufruf protokollieren"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Aktuelle Nachrichtenanzahl: {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Nach jedem Modellaufruf protokollieren"""
last_msg = state["messages"][-1] if state.get("messages") else None
if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
print(f"[after_model] Modell hat einen Tool-Aufruf angefordert")
return None
Methode 2: Klassenvererbung (geeignet für komplexe Logik)
Beispiel
from langchain.agents.middleware import AgentMiddleware
class LoggingMiddleware(AgentMiddleware):
"""Benutzerdefinierte Logging-Middleware"""
@property
def name(self) -> str:
# Benutzerdefinierter Middleware-Name (standardmäßig Klassenname)
return "logging"
def before_agent(self, state, runtime):
"""Logik, bevor der Agent startet"""
print("[Protokollierung] Agent hat mit der Ausführung begonnen")
return None
def before_model(self, state, runtime):
"""Logik vor dem Modellaufruf"""
msg_count = len(state.get("messages", []))
print(f"[Protokollierung] Modellaufruf wird vorbereitet, aktuell {msg_count} Nachrichten")
return None
def after_model(self, state, runtime):
"""Logik nach dem Modellaufruf"""
print("[Protokollierung] Modellaufruf abgeschlossen")
return None
def after_agent(self, state, runtime):
"""Logik nach Ende des Agenten"""
print("[Protokollierung] Agentenausführung beendet")
return None
class LoggingMiddleware(AgentMiddleware):
"""Benutzerdefinierte Logging-Middleware"""
@property
def name(self) -> str:
# Benutzerdefinierter Middleware-Name (standardmäßig Klassenname)
return "logging"
def before_agent(self, state, runtime):
"""Logik, bevor der Agent startet"""
print("[Protokollierung] Agent hat mit der Ausführung begonnen")
return None
def before_model(self, state, runtime):
"""Logik vor dem Modellaufruf"""
msg_count = len(state.get("messages", []))
print(f"[Protokollierung] Modellaufruf wird vorbereitet, aktuell {msg_count} Nachrichten")
return None
def after_model(self, state, runtime):
"""Logik nach dem Modellaufruf"""
print("[Protokollierung] Modellaufruf abgeschlossen")
return None
def after_agent(self, state, runtime):
"""Logik nach Ende des Agenten"""
print("[Protokollierung] Agentenausführung beendet")
return None
Vollständiges Lebenszyklusbeispiel
Beispiel
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_agent, after_agent,
before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@before_agent
def start_log(state, runtime):
"""Bevor der Agent startet"""
print(">>> [before_agent] Agent gestartet <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Vor jedem Modellaufruf"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Nachricht {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""Nach jedem Modellaufruf"""
last = state["messages"][-1] if state.get("messages") else None
if hasattr(last, 'tool_calls') and last.tool_calls:
tools = [tc['name'] for tc in last.tool_calls]
print(f" <- [after_model] Angefordertes Tool: {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Direkte Antwort: {content}...")
return None
@after_agent
def end_log(state, runtime):
"""Nachdem der Agent endet"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] Agent beendet, {total_msgs} Nachrichten insgesamt <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Wetter abfragen"""
return f"{city}: Sonnig, 25°C"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[start_log, pre_model, post_model, end_log],
system_prompt="Du bist ein Assistent.",
)
print("\n========== Erste Frage (erfordert Tool) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Wie ist das Wetter in Hangzhou?")]
})
print(f"\nEndgültige Antwort: {result['messages'][-1].content}")
print("\n========== Zweite Frage (kein Tool erforderlich) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hello")]
})
print(f"\nEndgültige Antwort: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_agent, after_agent,
before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@before_agent
def start_log(state, runtime):
"""Bevor der Agent startet"""
print(">>> [before_agent] Agent gestartet <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Vor jedem Modellaufruf"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Nachricht {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""Nach jedem Modellaufruf"""
last = state["messages"][-1] if state.get("messages") else None
if hasattr(last, 'tool_calls') and last.tool_calls:
tools = [tc['name'] for tc in last.tool_calls]
print(f" <- [after_model] Angefordertes Tool: {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Direkte Antwort: {content}...")
return None
@after_agent
def end_log(state, runtime):
"""Nachdem der Agent endet"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] Agent beendet, {total_msgs} Nachrichten insgesamt <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Wetter abfragen"""
return f"{city}: Sonnig, 25°C"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[start_log, pre_model, post_model, end_log],
system_prompt="Du bist ein Assistent.",
)
print("\n========== Erste Frage (erfordert Tool) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Wie ist das Wetter in Hangzhou?")]
})
print(f"\nEndgültige Antwort: {result['messages'][-1].content}")
print("\n========== Zweite Frage (kein Tool erforderlich) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hello")]
})
print(f"\nEndgültige Antwort: {result['messages'][-1].content}")
Ausgabe:
========== 第一个问题（需要工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 2 条消息 <- [after_model] 请求工具: ['get_weather'] -> [before_model] 第 3 条消息 <- [after_model] 直接回复: 杭州今天晴，气温25°C。... <<< [after_agent] Agent 结束，共 4 条消息 <<< Final reply: 杭州今天晴，气温25°C. ========== 第二个问题（无需工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 1 条消息 <- [after_model] 直接回复: 你好！有什么可以帮你的？... <<< [after_agent] Agent 结束，共 2 条消息 <<< Final reply: Hello! How can I help you?
An der Ausgabe können Sie sehen:
- before_agent und after_agent: wurden nur einmal pro Frage ausgeführt
- before_model und after_model: wurden bei jedem Modellaufruf ausgeführt (die erste Frage rief das Modell zweimal auf, also liefen beide jeweils zweimal)
Rückgabewert der Middleware
Der Rückgabewert der Middleware bestimmt, ob der Agent-Status geändert oder der Ablauf gesteuert wird:
| Rückgabewert | Wirkung | Beispiel |
|---|---|---|
| Keine | Ändert keinen Status; setzt den normalen Ablauf fort | Nur Protokollierung |
| dict | Aktualisiert den Agent-Status (wird mit dem aktuellen Status zusammengeführt) | Gib {"custom_field": "value"} zurück |
| Dict, das jump_to enthält | Zum angegebenen Knoten springen | Gib {"jump_to": "end"} zurück |
Andere ErweiterungenDas zurückgegebene Dict wird durch den Reducer des Agent-Status zusammengeführt. Für das Feld messages wird der add_messages-Reducer verwendet, sodass zurückgegebene Nachrichten angehängt und nicht überschrieben werden.