LangChain Modellaufruf-Abfangen -- @wrap_model_call
@wrap_model_call ist der leistungsfähigste Hook in der Middleware.
Anders als before/after beobachtet @wrap_model_call nicht nur; es kannden Modellausführungsprozess vollständig steuern— wiederholen, degradieren, cachen oder sogar das Modell überspringen und direkt eine voreingestellte Antwort verwenden.
Den Handler-Callback verstehen
Der Kern von @wrap_model_call ist eineHandler-Callback-FunktionNur der Aufruf von handler(request) führt das Modell tatsächlich aus; wenn nicht aufgerufen, wird das Modell übersprungen.
Beispiel
# Grundlegende Struktur von wrap_model_call
# request: enthält alle Informationen wie Modell, Nachrichten, Werkzeuge usw.
# handler: ein aufrufbares Objekt; nur die Ausführung ruft das Modell tatsächlich auf
@wrap_model_call
def my_middleware(request, handler):
# Du kannst alles vor dem Modellaufruf tun
print("Modell wird gleich aufgerufen...")
# Nur der Aufruf von handler(request) führt das Modell tatsächlich aus
response = handler(request)
# Du kannst alles nach dem Modellaufruf tun
print("Modellaufruf abgeschlossen")
return response
# request: enthält alle Informationen wie Modell, Nachrichten, Werkzeuge usw.
# handler: ein aufrufbares Objekt; nur die Ausführung ruft das Modell tatsächlich auf
@wrap_model_call
def my_middleware(request, handler):
# Du kannst alles vor dem Modellaufruf tun
print("Modell wird gleich aufgerufen...")
# Nur der Aufruf von handler(request) führt das Modell tatsächlich aus
response = handler(request)
# Du kannst alles nach dem Modellaufruf tun
print("Modellaufruf abgeschlossen")
return response
Szenario 1: Wiederholungsmechanismus
Dies ist das häufigste Szenario — Modellaufrufe können aufgrund von Netzwerkproblemen fehlschlagen, und automatische Wiederholungen können die Zuverlässigkeit verbessern:
Beispiel
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Automatisch wiederholen, wenn der Modellaufruf fehlschlägt, bis zu 3 Mal"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Wiederholung erfolgreich] Versuch {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Inkrementelles Warten: 2s, 4s
print(f" [Wiederholung] Versuch {attempt + 1} fehlgeschlagen, erneuter Versuch nach {wait_time} Sekunden...")
import time
time.sleep(wait_time)
# Alle Wiederholungen fehlgeschlagen
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python-Tutorial vorstellen")]
})
print(f"\nAntwort: {result[''][-1].content[:100]}...")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Automatisch wiederholen, wenn der Modellaufruf fehlschlägt, bis zu 3 Mal"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Wiederholung erfolgreich] Versuch {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Inkrementelles Warten: 2s, 4s
print(f" [Wiederholung] Versuch {attempt + 1} fehlgeschlagen, erneuter Versuch nach {wait_time} Sekunden...")
import time
time.sleep(wait_time)
# Alle Wiederholungen fehlgeschlagen
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python-Tutorial vorstellen")]
})
print(f"\nAntwort: {result[''][-1].content[:100]}...")
Szenario 2: Modell-Degradierung / Failover
Wenn das primäre Modell nicht verfügbar ist, automatisch auf ein Backup-Modell umschalten:
Beispiel
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
Primäres Modell und Backup-Modell
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # Token für das degradierte Modell begrenzen
@wrap_model_call
def fallback_on_error(request, handler):
"""Automatisch auf das Backup-Modell umschalten, wenn das primäre Modell fehlschlägt"""
try:
# Versuche, das primäre Modell zu verwenden
return handler(request)
except Exception as e:
print(f"[Degradiert] Primäres Modell fehlgeschlagen: {e}, wechsle zum Backup-Modell...")
# Das Modell in request überschreiben, zum Backup-Modell wechseln
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Eingeschränkt] Backup-Modell ebenfalls fehlgeschlagen: {e2}")
# Beide Modelle sind fehlgeschlagen, eine freundliche Nachricht zurückgeben
return AIMessage(
content="Entschuldigung, der Dienst ist vorübergehend nicht verfügbar. Bitte versuchen Sie es später erneut."
)
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
Primäres Modell und Backup-Modell
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # Token für das degradierte Modell begrenzen
@wrap_model_call
def fallback_on_error(request, handler):
"""Automatisch auf das Backup-Modell umschalten, wenn das primäre Modell fehlschlägt"""
try:
# Versuche, das primäre Modell zu verwenden
return handler(request)
except Exception as e:
print(f"[Degradiert] Primäres Modell fehlgeschlagen: {e}, wechsle zum Backup-Modell...")
# Das Modell in request überschreiben, zum Backup-Modell wechseln
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Eingeschränkt] Backup-Modell ebenfalls fehlgeschlagen: {e2}")
# Beide Modelle sind fehlgeschlagen, eine freundliche Nachricht zurückgeben
return AIMessage(
content="Entschuldigung, der Dienst ist vorübergehend nicht verfügbar. Bitte versuchen Sie es später erneut."
)
request.override() ist eine unveränderliche Methode — sie gibt eine neue Kopie von request zurück, ohne das ursprüngliche Objekt zu verändern. Dadurch ist jeder Aufruf unabhängig und sicher.
Szenario 3: Modellantworten zwischenspeichern
Bei wiederholten Abfragen können Sie Modellantworten zwischenspeichern, um die Kosten für API-Aufrufe zu senken:
Beispiel
from langchain.agents.middleware import wrap_model_call
from langchain.messages import AIMessage
# Ein einfacher In-Memory-Cache
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Modellantworten zwischenspeichern, bei derselben Frage nicht wiederholt aufrufen"""
# Verwende den Inhalt der letzten Benutzernachricht als Cache-Schlüssel
messages = request.messages
if not messages:
return handler(request)
# Cache-Schlüssel generieren
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Übermäßig langen Inhalt kürzen
# Cache prüfen
if cache_key in cache:
print(f"[Cache-Treffer] Zwischengespeichertes Ergebnis direkt zurückgeben")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(aus dem Cache)*")
# Cache-Miss, das Modell aufrufen
result = handler(request)
# Der Inhalt von AIMessage kann in einer Liste gemischt sein, nimm direkt den ersten
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Cache-Miss] Im Cache gespeichert, aktuell {len(cache)} Einträge")
elif hasattr(result, 'model_response'):
# Der Fall von ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Du bist der Assistent von EXAMPLE und musst prägnant antworten.",
)
# Erste Abfrage (Cache-Miss)
result = agent.invoke({
"messages": [{"role": "user", "content": "Was ist Python?"}]
})
print(f"Erstes Mal: {result['messages'][-1].content[:80]}...\n")
# Zweite Abfrage mit derselben Frage (Cache-Hit)
result = agent.invoke({
"messages": [{"role": "user", "content": "Was ist Python?"}]
})
print(f"Zweites Mal: {result['messages'][-1].content[:80]}...")
from langchain.messages import AIMessage
# Ein einfacher In-Memory-Cache
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Modellantworten zwischenspeichern, bei derselben Frage nicht wiederholt aufrufen"""
# Verwende den Inhalt der letzten Benutzernachricht als Cache-Schlüssel
messages = request.messages
if not messages:
return handler(request)
# Cache-Schlüssel generieren
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Übermäßig langen Inhalt kürzen
# Cache prüfen
if cache_key in cache:
print(f"[Cache-Treffer] Zwischengespeichertes Ergebnis direkt zurückgeben")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(aus dem Cache)*")
# Cache-Miss, das Modell aufrufen
result = handler(request)
# Der Inhalt von AIMessage kann in einer Liste gemischt sein, nimm direkt den ersten
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Cache-Miss] Im Cache gespeichert, aktuell {len(cache)} Einträge")
elif hasattr(result, 'model_response'):
# Der Fall von ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Du bist der Assistent von EXAMPLE und musst prägnant antworten.",
)
# Erste Abfrage (Cache-Miss)
result = agent.invoke({
"messages": [{"role": "user", "content": "Was ist Python?"}]
})
print(f"Erstes Mal: {result['messages'][-1].content[:80]}...\n")
# Zweite Abfrage mit derselben Frage (Cache-Hit)
result = agent.invoke({
"messages": [{"role": "user", "content": "Was ist Python?"}]
})
print(f"Zweites Mal: {result['messages'][-1].content[:80]}...")
Run result:
[缓存未命中] 已存入缓存，当前 1 条 第一次: Python 是一种高级编程语言，以简洁易读的语法著称... [缓存命中] 直接返回缓存结果 第二次: Python 是一种高级编程语言，以简洁易读的语法著称... *(来自缓存)*
Szenario 4: Anpassen der Anfrage — Dynamisches Einfügen von Systemnachrichten
Beispiel
from datetime import datetime
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Aktuelle Zeitinformationen vor jedem Modellaufruf einfügen"""
now = datetime.now()
time_context = (
f"Aktuelle Zeit: {now.strftime('%Y Jahr %m Monat %d Tag %H:%M')}."
f"Heute ist {['Montag','Dienstag','Mittwoch','Donnerstag','Freitag','Samstag','Sonntag'][now.weekday()]}."
)
# Zeitinformationen an die vorhandene system_message anhängen
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Verwende override, um eine neue Anfrage zu erstellen
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Aktuelle Zeitinformationen vor jedem Modellaufruf einfügen"""
now = datetime.now()
time_context = (
f"Aktuelle Zeit: {now.strftime('%Y Jahr %m Monat %d Tag %H:%M')}."
f"Heute ist {['Montag','Dienstag','Mittwoch','Donnerstag','Freitag','Samstag','Sonntag'][now.weekday()]}."
)
# Zeitinformationen an die vorhandene system_message anhängen
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Verwende override, um eine neue Anfrage zu erstellen
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
Szenario 5: Kombinieren mehrerer wrap_model_call
Mehrere wrap_model_call-Middlewares werden automatisch in der Reihenfolge kombiniert – die erste befindet sich in der äußersten Schicht:
Beispiel
from langchain.agents.middleware import wrap_model_call
@wrap_model_call
def outer_middleware(request, handler):
"""Äußerste Middleware"""
print("[Äußere] Start")
result = handler(request) # Hier erfolgt der Eintritt in inner_middleware
print("[Äußere] Ende")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Innere Middleware"""
print(" [Innere] Start")
result = handler(request) # Hier wird das Modell tatsächlich aufgerufen
print(" [Innere] Ende")
return result
# Ausführungsreihenfolge:
# [Äußere] Start
# [Innere] Start
# → Tatsächlicher Aufruf des Modells
# [Innere] Ende
# [Äußere] Ende
@wrap_model_call
def outer_middleware(request, handler):
"""Äußerste Middleware"""
print("[Äußere] Start")
result = handler(request) # Hier erfolgt der Eintritt in inner_middleware
print("[Äußere] Ende")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Innere Middleware"""
print(" [Innere] Start")
result = handler(request) # Hier wird das Modell tatsächlich aufgerufen
print(" [Innere] Ende")
return result
# Ausführungsreihenfolge:
# [Äußere] Start
# [Innere] Start
# → Tatsächlicher Aufruf des Modells
# [Innere] Ende
# [Äußere] Ende
Andere ErweiterungenMehrere wrap_model_call-Aufrufe umhüllen Schicht für Schicht wie eine Zwiebel. Die äußerste Schicht wird zuerst ausgeführt und kehrt zuletzt zurück. Dadurch können Sie mehrere unabhängige Funktionen kombinieren — zum Beispiel übernimmt die äußere Schicht die Cache-Prüfung, die innere Schicht die Wiederholungsversuche, ohne sich gegenseitig zu beeinträchtigen.