LangChain @before_agent und @after_agent

before_agent und after_agent sind Hooks auf Agent-Ebene, die jeweils einmal vor und nach der Ausführung des Agenten ausgeführt werden. Sie eignen sich für Initialisierung, Vorverarbeitung, Nachverarbeitung und statistische Analyse.

before_agent -- Vorbereitung, bevor der Agent startet

before_agent wird ausgeführt, bevor der Agent offiziell mit der Ausführung beginnt, nur einmal. Hier können Sie Eingabe-Vorverarbeitung, Validierung von Benutzerinformationen, Ressourceninitialisierung usw. durchführen.

Szenario 1: Eingabe-Vorverarbeitung -- Benutzereingabe automatisch korrigieren

Beispiel

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import before_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool

@before_agent
def preprocess_input(state, runtime):
"""Benutzereingabe verarbeiten, bevor der Agent startet"""
messages = state.get("messages", [])
if not messages:
return None

# Letzte Nachricht des Benutzers abrufen
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

# Automatisch höfliche Phrasen hinzufügen (falls der Benutzer direkt fragt)
greetings = ["Hi", "Hello", "hi", "hello", "Hey"]
if content and not any(content.lower().startswith(g) for g in greetings):
# Nicht ändern, direkt zurückgeben
pass

return None

@tool
def search_course(keyword: str) -> str:
"""Kurse auf Example suchen"""
courses = {
"python": "Python3 Grundlagen-Tutorial (Kostenlos, 30 Kapitel)",
"html": "HTML Grundlagen-Tutorial (Kostenlos, 25 Kapitel)",
}
return courses.get(keyword.lower(), f"Keine Kurse zu {keyword} gefunden")

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="Du bist der Kursberater für Example.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Python-Kurse")]
})
print(f"Antwort: {result['messages'][-1].content}")

Szenario 2: Zugriffskontrolle -- Berechtigungsprüfung

Beispiel

from langchain.agents.middleware import before_agent

@before_agent
def access_control(state, runtime):
"""Prüfen, ob der Benutzer die Berechtigung hat, den Agenten zu verwenden"""
# Benutzerinformationen aus runtime.context abrufen
context = runtime.context
if context is None:
return None

user_role = context.get("user_role", "guest")

# Gastbenutzer können nur eingeschränkte Funktionen verwenden
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# Prüfen, ob eingeschränkte Funktionen betroffen sind
restricted_keywords = ["delete", "manage", "configure", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content="Sie haben nicht die ausreichende Berechtigung, diesen Vorgang auszuführen. Bitte melden Sie sich an und versuchen Sie es erneut."
)]
}

return None

after_agent -- Verarbeitung nach Abschluss des Agenten

after_agent wird ausgeführt, nachdem der Agent die gesamte Verarbeitung abgeschlossen hat (nur einmal). Hier können Sie die endgültige Ausgabe formatieren, statistische Informationen aufzeichnen, Ressourcen bereinigen usw.

Szenario 3: Statistische Analyse -- Aufzeichnung von Konversationsdaten

Beispiel

from langchain.agents.middleware import after_agent

@after_agent
def conversation_stats(state, runtime):
"""Konversationsinformationen zählen und sie an das Ergebnis anhängen"""
messages = state.get("messages", [])

# Statistische Daten
model_calls = 0
tool_calls = 0
total_chars = 0

for msg in messages:
if msg.type == "ai":
model_calls += 1
if hasattr(msg, 'tool_calls') and msg.tool_calls:
tool_calls += len(msg.tool_calls)
if hasattr(msg, 'content') and msg.content:
total_chars += len(str(msg.content))

# Statistische Informationen über einen benutzerdefinierten Stream senden
runtime.stream_writer({
"type": "stats",
"model_calls": model_calls,
"tool_calls": tool_calls,
"total_messages": len(messages),
"total_chars": total_chars,
})

return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[conversation_stats],
system_prompt="Du bist der Kursberater für Example.",
)

# Verwende stream_mode=["updates", "custom"], um benutzerdefinierte Ereignisse zu empfangen
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Python-Kurse finden")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"Statistik: {chunk}")

Ausgabe:

统计信息: {'type': 'stats', 'model_calls': 2, 'tool_calls': 1,
            'total_messages': 4, 'total_chars': 127}

Szenario 4: Formatierung der Ausgabe -- Vereinheitlichung des Antwortstils

Beispiel

from langchain.agents.middleware import after_agent
from langchain.messages import AIMessage

@after_agent
def format_output(state, runtime):
"""Formatierte Zusammenfassungsinformationen an das Ergebnis anhängen"""
messages = state.get("messages", [])
if not messages:
return None

# Letzte KI-Nachricht finden (finale Antwort)
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break

if last_ai:
# Statistische Informationen
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)

footer = (
f"\n\n---\n"
f"> Diese Unterhaltung hatte {len(messages)} Nachrichten,"
f"Das Tool wurde {tool_count} Mal aufgerufen.\n"
f"> Bereitgestellt von Example AI Assistant."
)

# An die finale Antwort anhängen
return {
"messages": [
AIMessage(content=last_ai.content + footer)
]
}

return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[format_output],
system_prompt="Du bist der Kursberater für Example.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Welche Kurse hat Python?")]
})
print(result["messages"][-1].content)

Ausgabe:

Example offers the Python3 basics tutorial in 30 chapters, completely free and perfect for Python beginners.

---
> 本次对话共进行 4 条消息，调用了 1 次工具。
> 由Example AI 助手提供支持。

Vollständiges Kollaborationsbeispiel mit vier Hooks

Beispiel

from langchain.agents.middleware import (
before_agent, after_agent, before_model, after_model
)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool

# ----- Alle Hooks definieren -----

@before_agent
def init_session(state, runtime):
"""Start: Sitzung initialisieren"""
print(">>> Sitzung gestartet")
return None

@before_model
def pre_model_check(state, runtime):
"""Vor jedem Modellaufruf"""
msg_count = len(state.get("messages", []))
print(f" [Vor dem Modell] Nachrichtenanzahl: {msg_count}")
return None

@after_model
def post_model_check(state, runtime):
"""Nach jedem Modellaufruf"""
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [Nach Modell] Tool-Aufruf erforderlich")
return None

@after_agent
def finish_session(state, runtime):
"""Ende: Ressourcen aufräumen"""
total = len(state.get("messages", []))
print(f"<<< Sitzung beendet, insgesamt {total} Nachrichten")
return None

# ----- Agent erstellen -----

@tool
def get_weather(city: str) -> str:
"""Wetter abfragen"""
return f"{city}: Sonnig"

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[init_session, pre_model_check, post_model_check, finish_session],
system_prompt="Du bist ein Assistent.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Wie ist das Wetter in Hangzhou?")]
})
print(f"\n"Endgültige Antwort: {result['messages'][-1].content}")

Ausgabe:

>>> 会话开始
  [model前] 消息数: 2
  [model后] 需要工具调用
  [model前] 消息数: 3
<<< 会话结束，共 4 条消息

Final reply: Hangzhou is sunny today, great for going out.

Middleware-Hook-Zusammenfassung

Hook	Ausführungsanzahl	Wann verwenden	Schlüsselfähigkeiten
before_agent	Einmal	Berechtigungsprüfung, Eingabevorverarbeitung, Ressourceninitialisierung	Kann jump_to="end" verwenden, um frühzeitig abzubrechen
before_model	Jede Schleife	Nachrichten-Kürzung, Inhaltsfilterung, Kontextinjektion	Kann den Kontrollfluss mit jump_to steuern
wrap_model_call	Jede Schleife	Wiederholung, Fallback, Caching, Prompt-Modifikation	Vollständige Kontrolle über die Modellausführung
after_model	Jede Schleife	Überprüfung der Antwort, Anhängen von Inhalten, Protokollierung	Kann die Modellausgabe ersetzen
wrap_tool_call	Jeder Tool-Aufruf	Tool-Wiederholung, Caching, Parameterumschreibung	Vollständige Kontrolle über die Tool-Ausführung
after_agent	Einmal	Ausgabeformatierung, statistische Analyse, Bereinigung	Änderung des endgültigen Zustands

Weitere Erweiterungen

Agent Entwicklungskonfiguration

Referenzhandbuch

LangChain @before_agent und @after_agent

before_agent -- Vorbereitung, bevor der Agent startet

Szenario 1: Eingabe-Vorverarbeitung -- Benutzereingabe automatisch korrigieren

Beispiel

Szenario 2: Zugriffskontrolle -- Berechtigungsprüfung

Beispiel

after_agent -- Verarbeitung nach Abschluss des Agenten

Szenario 3: Statistische Analyse -- Aufzeichnung von Konversationsdaten

Beispiel

Szenario 4: Formatierung der Ausgabe -- Vereinheitlichung des Antwortstils

Beispiel

Vollständiges Kollaborationsbeispiel mit vier Hooks

Beispiel

Middleware-Hook-Zusammenfassung