LangChain Middleware-Hooks -- @before_model und @after_model
before_model und after_model sind die beiden am häufigsten verwendeten Middleware-Hooks. Sie werden vor und nach jedem Modellaufruf ausgeführt und eignen sich für Inhaltsfilterung, Nachrichten-Vorverarbeitung, Antwortüberprüfung usw.
@before_model — Abfangen vor dem Modellaufruf
@before_model wird vor jedem Modellaufruf ausgeführt. Hier können Sie Nachrichten ändern, Kontextbedingungen einfügen oder den Modellaufruf vollständig überspringen.
Szenario 1: Nachrichten-Vorverarbeitung — Begrenzung der Konversationslänge
Beispiel
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@before_model
def limit_context(state, runtime):
"""Begrenze die Länge des Nachrichtenverlaufs, um übermäßig langen Kontext zu vermeiden"""
messages = state.get("messages", [])
# Systemnachricht + die 6 neuesten Nachrichten behalten
MAX_MESSAGES = 6
if len(messages) > MAX_MESSAGES:
# Auf die neuesten Nachrichten kürzen
trimmed = messages[-MAX_MESSAGES:]
# Sicherstellen, dass die erste Nachricht eine Benutzernachricht ist
if trimmed and trimmed[0].type != "human":
trimmed = trimmed[1:]
return {"messages": trimmed}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[limit_context],
system_prompt="Du bist der Assistent des EXAMPLE-Tutorials.",
)
# Simuliere eine Multi-Turn-Konversation
result = agent.invoke({
"messages": [
HumanMessage(content="Runde 1"),
HumanMessage(content="Runde 2"),
HumanMessage(content="Runde 3"),
HumanMessage(content="Runde 4"),
HumanMessage(content="Runde 5"),
HumanMessage(content="Runde 6"),
HumanMessage(content="Runde 7"),
]
})
print(f"Nachrichtenanzahl: {len(result['messages'])}")
print(f"Antwort: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@before_model
def limit_context(state, runtime):
"""Begrenze die Länge des Nachrichtenverlaufs, um übermäßig langen Kontext zu vermeiden"""
messages = state.get("messages", [])
# Systemnachricht + die 6 neuesten Nachrichten behalten
MAX_MESSAGES = 6
if len(messages) > MAX_MESSAGES:
# Auf die neuesten Nachrichten kürzen
trimmed = messages[-MAX_MESSAGES:]
# Sicherstellen, dass die erste Nachricht eine Benutzernachricht ist
if trimmed and trimmed[0].type != "human":
trimmed = trimmed[1:]
return {"messages": trimmed}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[limit_context],
system_prompt="Du bist der Assistent des EXAMPLE-Tutorials.",
)
# Simuliere eine Multi-Turn-Konversation
result = agent.invoke({
"messages": [
HumanMessage(content="Runde 1"),
HumanMessage(content="Runde 2"),
HumanMessage(content="Runde 3"),
HumanMessage(content="Runde 4"),
HumanMessage(content="Runde 5"),
HumanMessage(content="Runde 6"),
HumanMessage(content="Runde 7"),
]
})
print(f"Nachrichtenanzahl: {len(result['messages'])}")
print(f"Antwort: {result['messages'][-1].content}")
Ausgabe:
Message count: 8 Reply: Hello! It looks like you are testing multi-turn conversation. How can I help you?
Szenario 2: Inhaltsfilterung — Blockieren sensibler Wörter
Beispiel
from langchain.agents.middleware import before_model
SENSITIVE_WORDS = ["Passwort", "Bankkartennummer", "Personalausweisnummer"]
@before_model
def content_filter(state, runtime):
"""Prüfen, ob Benutzernachrichten sensible Wörter enthalten, und diese gegebenenfalls abfangen"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
for word in SENSITIVE_WORDS:
if word in content:
print(f"[Blockiert] Sensibles Wort erkannt: {word}")
# jump_to="end" beendet direkt und verhindert, dass das Modell antwortet
return {
"jump_to": "end",
"messages": [
HumanMessage(content=f"Entschuldigung, aus Sicherheitsgründen können Anfragen, die '{word}' enthalten, nicht verarbeitet werden.")
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[content_filter],
system_prompt="Du bist der Assistent des EXAMPLE-Tutorials.",
)
# Normale Frage
result = agent.invoke({
"messages": [HumanMessage(content="Wie fange ich mit Python an?")]
})
print(f"Normale Frage: {result['messages'][-1].content[:80]}")
# Sensible Frage
result = agent.invoke({
"messages": [HumanMessage(content="Wie lautet meine Bankkartennummer? Können Sie mir helfen, sie zu überprüfen?")]
})
print(f"\nSensible Frage: {result['messages'][-1].content}")
SENSITIVE_WORDS = ["Passwort", "Bankkartennummer", "Personalausweisnummer"]
@before_model
def content_filter(state, runtime):
"""Prüfen, ob Benutzernachrichten sensible Wörter enthalten, und diese gegebenenfalls abfangen"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
for word in SENSITIVE_WORDS:
if word in content:
print(f"[Blockiert] Sensibles Wort erkannt: {word}")
# jump_to="end" beendet direkt und verhindert, dass das Modell antwortet
return {
"jump_to": "end",
"messages": [
HumanMessage(content=f"Entschuldigung, aus Sicherheitsgründen können Anfragen, die '{word}' enthalten, nicht verarbeitet werden.")
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[content_filter],
system_prompt="Du bist der Assistent des EXAMPLE-Tutorials.",
)
# Normale Frage
result = agent.invoke({
"messages": [HumanMessage(content="Wie fange ich mit Python an?")]
})
print(f"Normale Frage: {result['messages'][-1].content[:80]}")
# Sensible Frage
result = agent.invoke({
"messages": [HumanMessage(content="Wie lautet meine Bankkartennummer? Können Sie mir helfen, sie zu überprüfen?")]
})
print(f"\nSensible Frage: {result['messages'][-1].content}")
Ausgabe:
[拦截] 检测到敏感词: 银行卡号 Normal question: Python 入门可以从安装 Python 环境开始... Sensitive question: 抱歉，为了安全，不能处理包含「银行卡号」的请求。
@after_model — Verarbeitung nach dem Modellaufruf
@after_model wird ausgeführt, nachdem das Modell geantwortet hat, geeignet zum Überprüfen der Modellausgabe, zum Extrahieren wichtiger Informationen, zum Anhängen von Folgeanweisungen usw.
Szenario 3: Überprüfung des Antwortinhalts
Beispiel
from langchain.agents.middleware import after_model
FORBIDDEN_TOPICS = ["politics", "violence", "pornography"]
@after_model
def response_audit(state, runtime):
"""Überprüfe Modellantworten und ersetze sie, wenn sie verbotene Themen beinhalten"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
for topic in FORBIDDEN_TOPICS:
if topic in content:
runtime.stream_writer({
"type": "warning",
"message": f"Antwort wurde als Inhalt mit Bezug zu '{topic}' erkannt und wurde ersetzt"
})
# Eine überschreibende Nachricht zurückgeben (über add_messages-Reducer)
from langchain.messages import AIMessage
return {
"messages": [
AIMessage(content="Entschuldigung, ich kann diese Frage nicht beantworten."
"Bitte fragen Sie nach Inhalten zum Programmierlernen.")
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[response_audit],
system_prompt="Du bist der Assistent des EXAMPLE-Tutorials.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Welche Lernressourcen sind für Python verfügbar?")]
})
print(f"Normale Antwort: {result['messages'][-1].content}")
FORBIDDEN_TOPICS = ["politics", "violence", "pornography"]
@after_model
def response_audit(state, runtime):
"""Überprüfe Modellantworten und ersetze sie, wenn sie verbotene Themen beinhalten"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
for topic in FORBIDDEN_TOPICS:
if topic in content:
runtime.stream_writer({
"type": "warning",
"message": f"Antwort wurde als Inhalt mit Bezug zu '{topic}' erkannt und wurde ersetzt"
})
# Eine überschreibende Nachricht zurückgeben (über add_messages-Reducer)
from langchain.messages import AIMessage
return {
"messages": [
AIMessage(content="Entschuldigung, ich kann diese Frage nicht beantworten."
"Bitte fragen Sie nach Inhalten zum Programmierlernen.")
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[response_audit],
system_prompt="Du bist der Assistent des EXAMPLE-Tutorials.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Welche Lernressourcen sind für Python verfügbar?")]
})
print(f"Normale Antwort: {result['messages'][-1].content}")
Szenario 4: Automatisches Anhängen von Prompt-Informationen
Beispiel
from langchain.agents.middleware import after_model
from langchain.messages import AIMessage
@after_model
def append_disclaimer(state, runtime):
"""Automatisch einen Haftungsausschluss nach jeder Modellantwort anhängen"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Nur anhängen, wenn es die endgültige Antwort der KI ist (keine tool_calls)
if (last_msg.type == "ai"
and last_msg.content
and not (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls)):
# Die letzte KI-Nachricht ersetzen und den Haftungsausschluss hinzufügen
return {
"messages": [
AIMessage(
content=(
last_msg.content
+ "\n\n---\n*Der obige Inhalt wurde vom EXAMPLE-KI-Assistenten generiert und dient nur als Referenz.*
)
)
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[append_disclaimer],
system_prompt="Du bist der Assistent von EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Ist Python leicht zu lernen?")]
})
print(result["messages"][-1].content)
from langchain.messages import AIMessage
@after_model
def append_disclaimer(state, runtime):
"""Automatisch einen Haftungsausschluss nach jeder Modellantwort anhängen"""
messages = state.get("messages", [])
if not messages:
return None
last_msg = messages[-1]
# Nur anhängen, wenn es die endgültige Antwort der KI ist (keine tool_calls)
if (last_msg.type == "ai"
and last_msg.content
and not (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls)):
# Die letzte KI-Nachricht ersetzen und den Haftungsausschluss hinzufügen
return {
"messages": [
AIMessage(
content=(
last_msg.content
+ "\n\n---\n*Der obige Inhalt wurde vom EXAMPLE-KI-Assistenten generiert und dient nur als Referenz.*
)
)
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[append_disclaimer],
system_prompt="Du bist der Assistent von EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Ist Python leicht zu lernen?")]
})
print(result["messages"][-1].content)
Ausgabe:
Python 是一门非常适合初学者的编程语言，它的语法简洁、贴近自然语言... --- *The above content was generated by the Example AI assistant and is for reference only.*
can_jump_to — Fluss-Sprung-Steuerung
In before_model und after_model kannst du den Ablauf des Agenten über den Parameter can_jump_to und den Zustand jump_to steuern:
Beispiel
from langchain.agents.middleware import before_model
@before_model(can_jump_to=["end"]) # Deklariere die Ziele, zu denen gesprungen werden kann
def conditional_exit(state, runtime):
"""Beende den Agenten direkt unter bestimmten Bedingungen"""
messages = state.get("messages", [])
if not messages:
return None
# Wenn der Benutzer "goodbye" sagt, direkt beenden
last_content = str(messages[-1].content)
if last_content.strip() in ["Goodbye", "Bye", "bye"]:
return {
"jump_to": "end", # Beende den Agenten direkt
"messages": [{"role": "assistant", "content": "Auf Wiedersehen! Wir freuen uns darauf, Sie beim nächsten Mal zu bedienen."}]
}
return None
@before_model(can_jump_to=["end"]) # Deklariere die Ziele, zu denen gesprungen werden kann
def conditional_exit(state, runtime):
"""Beende den Agenten direkt unter bestimmten Bedingungen"""
messages = state.get("messages", [])
if not messages:
return None
# Wenn der Benutzer "goodbye" sagt, direkt beenden
last_content = str(messages[-1].content)
if last_content.strip() in ["Goodbye", "Bye", "bye"]:
return {
"jump_to": "end", # Beende den Agenten direkt
"messages": [{"role": "assistant", "content": "Auf Wiedersehen! Wir freuen uns darauf, Sie beim nächsten Mal zu bedienen."}]
}
return None
| can_jump_to Wert | Bedeutung | Geeignete Szenarien |
|---|---|---|
| ["end"] | Kann zum Ende springen | Bedingter Ausstieg, Sicherheitsabfang |
| ["model"] | Kann zum Modell zurückspringen | Wenn das Modell erneut verarbeiten muss |
| ["tools"] | Kann zum Tool-Knoten springen | Das Modell überspringen und das Tool direkt ausführen |
| ["model", "end"] | Kann zum Modell oder zum Ende springen | Mehrere bedingte Verzweigungen |
Andere ErweiterungenWenn das Ziel nicht in can_jump_to deklariert ist, wird jump_to ignoriert. Dies ist ein Sicherheitsmechanismus, um zu verhindern, dass Middleware versehentlich zu illegalen Knoten springt.