LangChain @before_agent and @after_agent
before_agent and after_agent are Agent-level hooks that execute once before and after the Agent runs, respectively. They are suitable for initialization, preprocessing, post-processing, and statistical analysis.
before_agent -- Preparation before the Agent starts
before_agent runs before the Agent officially starts executing, only once. You can do input preprocessing, user information validation, resource initialization, etc. here.
Scenario 1: Input Preprocessing -- Auto-correct User Input
Example
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):
"""Process user input before the Agent starts"""
messages = state.get("messages", [])
if not messages:
return None
# Get the user's last message
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
# Automatically add polite phrases (if the user asks directly)
greetings = ["Hi", "Hello", "hi", "hello", "Hey"]
if content and not any(content.lower().startswith(g) for g in greetings):
# Do not modify, return directly
pass
return None
@tool
def search_course(keyword: str) -> str:
"""Search courses on Example"""
courses = {
"python": "Python3 Basic Tutorial (Free, 30 chapters)",
"html": "HTML Basic Tutorial (Free, 25 chapters)",
}
return courses.get(keyword.lower(), f"No courses found related to {keyword}")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="You are the course consultant for Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python courses")]
})
print(f"Reply: {result['messages'][-1].content}")
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):
"""Process user input before the Agent starts"""
messages = state.get("messages", [])
if not messages:
return None
# Get the user's last message
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
# Automatically add polite phrases (if the user asks directly)
greetings = ["Hi", "Hello", "hi", "hello", "Hey"]
if content and not any(content.lower().startswith(g) for g in greetings):
# Do not modify, return directly
pass
return None
@tool
def search_course(keyword: str) -> str:
"""Search courses on Example"""
courses = {
"python": "Python3 Basic Tutorial (Free, 30 chapters)",
"html": "HTML Basic Tutorial (Free, 25 chapters)",
}
return courses.get(keyword.lower(), f"No courses found related to {keyword}")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="You are the course consultant for Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python courses")]
})
print(f"Reply: {result['messages'][-1].content}")
Scenario 2: Access Control -- Permission Check
Example
from langchain.agents.middleware import before_agent
@before_agent
def access_control(state, runtime):
"""Check if the user has permission to use the Agent"""
# Get user information from runtime.context
context = runtime.context
if context is None:
return None
user_role = context.get("user_role", "guest")
# Guest users can only use limited features
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# Check if it involves restricted features
restricted_keywords = ["delete", "manage", "configure", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content="You do not have sufficient permission to perform this operation. Please log in and try again."
)]
}
return None
@before_agent
def access_control(state, runtime):
"""Check if the user has permission to use the Agent"""
# Get user information from runtime.context
context = runtime.context
if context is None:
return None
user_role = context.get("user_role", "guest")
# Guest users can only use limited features
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# Check if it involves restricted features
restricted_keywords = ["delete", "manage", "configure", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content="You do not have sufficient permission to perform this operation. Please log in and try again."
)]
}
return None
after_agent -- Processing after the Agent completes
after_agent executes after the Agent completes all processing (only once). You can format the final output, record statistical information, clean up resources, etc. here.
Scenario 3: Statistical Analysis -- Recording Conversation Data
Example
from langchain.agents.middleware import after_agent
@after_agent
def conversation_stats(state, runtime):
"""Count conversation information and append it to the result"""
messages = state.get("messages", [])
# Statistical data
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))
# Send statistical information via custom stream
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="You are the course consultant for Example.",
)
# Use stream_mode=["updates", "custom"] to receive custom events
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Find Python courses")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"Statistics: {chunk}")
@after_agent
def conversation_stats(state, runtime):
"""Count conversation information and append it to the result"""
messages = state.get("messages", [])
# Statistical data
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))
# Send statistical information via custom stream
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="You are the course consultant for Example.",
)
# Use stream_mode=["updates", "custom"] to receive custom events
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Find Python courses")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"Statistics: {chunk}")
Output:
统计信息: {'type': 'stats', 'model_calls': 2, 'tool_calls': 1,
'total_messages': 4, 'total_chars': 127}
Scenario 4: Formatting Output -- Unifying Reply Style
Example
from langchain.agents.middleware import after_agent
from langchain.messages import AIMessage
@after_agent
def format_output(state, runtime):
"""Append formatted summary information to the result"""
messages = state.get("messages", [])
if not messages:
return None
# Find the last AI message (final reply)
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break
if last_ai:
# Statistical information
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)
footer = (
f"\n\n---\n"
f"> This conversation had {len(messages)} messages,"
f"Called the tool {tool_count} times.\n"
f"> Powered by Example AI Assistant."
)
# Append to the final reply
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="You are the course consultant for Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="What courses does Python have?")]
})
print(result["messages"][-1].content)
from langchain.messages import AIMessage
@after_agent
def format_output(state, runtime):
"""Append formatted summary information to the result"""
messages = state.get("messages", [])
if not messages:
return None
# Find the last AI message (final reply)
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break
if last_ai:
# Statistical information
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)
footer = (
f"\n\n---\n"
f"> This conversation had {len(messages)} messages,"
f"Called the tool {tool_count} times.\n"
f"> Powered by Example AI Assistant."
)
# Append to the final reply
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="You are the course consultant for Example.",
)
result = agent.invoke({
"messages": [HumanMessage(content="What courses does Python have?")]
})
print(result["messages"][-1].content)
Output:
Example offers the Python3 basics tutorial in 30 chapters, completely free and perfect for Python beginners. --- > 本次对话共进行 4 条消息，调用了 1 次工具。 > 由Example AI 助手提供支持。
Complete Collaboration Example of Four Hooks
Example
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
# ----- Define all hooks -----
@before_agent
def init_session(state, runtime):
"""Start: Initialize session"""
print(">>> Session started")
return None
@before_model
def pre_model_check(state, runtime):
"""Before each model call"""
msg_count = len(state.get("messages", []))
print(f" [Before model] Message count: {msg_count}")
return None
@after_model
def post_model_check(state, runtime):
"""After each model call"""
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [After model] Tool call required")
return None
@after_agent
def finish_session(state, runtime):
"""End: Clean up resources"""
total = len(state.get("messages", []))
print(f"<<< Session ended, {total} messages in total")
return None
# ----- Create Agent -----
@tool
def get_weather(city: str) -> str:
"""Query weather"""
return f"{city}: Sunny"
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="You are an assistant.",
)
result = agent.invoke({
"messages": [HumanMessage(content="What is the weather in Hangzhou?")]
})
print(f"\n"Final reply: {result['messages'][-1].content}")
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
# ----- Define all hooks -----
@before_agent
def init_session(state, runtime):
"""Start: Initialize session"""
print(">>> Session started")
return None
@before_model
def pre_model_check(state, runtime):
"""Before each model call"""
msg_count = len(state.get("messages", []))
print(f" [Before model] Message count: {msg_count}")
return None
@after_model
def post_model_check(state, runtime):
"""After each model call"""
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [After model] Tool call required")
return None
@after_agent
def finish_session(state, runtime):
"""End: Clean up resources"""
total = len(state.get("messages", []))
print(f"<<< Session ended, {total} messages in total")
return None
# ----- Create Agent -----
@tool
def get_weather(city: str) -> str:
"""Query weather"""
return f"{city}: Sunny"
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="You are an assistant.",
)
result = agent.invoke({
"messages": [HumanMessage(content="What is the weather in Hangzhou?")]
})
print(f"\n"Final reply: {result['messages'][-1].content}")
Output:
>>> 会话开始 [model前] 消息数: 2 [model后] 需要工具调用 [model前] 消息数: 3 <<< 会话结束，共 4 条消息 Final reply: Hangzhou is sunny today, great for going out.
Middleware Hook Summary
| Hook | Execution Count | When to Use | Key Capabilities |
|---|---|---|---|
| before_agent | Once | Permission check, input preprocessing, resource initialization | Can use jump_to="end" to terminate early |
| before_model | Each loop | Message trimming, content filtering, context injection | Can control flow with jump_to |
| wrap_model_call | Each loop | Retry, fallback, caching, prompt modification | Full control over model execution |
| after_model | Each loop | Response review, content appending, logging | Can replace model output |
| wrap_tool_call | Each tool call | Tool retry, caching, parameter rewriting | Full control over tool execution |
| after_agent | Once | Output formatting, statistical analysis, cleanup | Final state modification |