YouTipLangchain Before After Model
Title: LangChain Middleware Hooks — @before_model and @after_model
before_model and after_model are the two most commonly used middleware hooks. They execute before and after each model call, suitable for content filtering, message preprocessing, response moderation, etc.
* * *
## @before_model — Interception Before Model Call
@before_model executes before each model call. You can modify messages here, inject context conditions, or skip the model call directly.
### Scenario 1: Message Preprocessing — Limiting Conversation Length
## Example
```python
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):
"""Limit message history length to prevent excessive context"""
messages = state.get("messages",[])
# Keep system message + last 6 messages
MAX_MESSAGES =6
if len(messages)> MAX_MESSAGES:
# Trim to recent messages
trimmed = messages[-MAX_MESSAGES:]
# Ensure the first message is a user message
if trimmed and trimmed.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=,
system_prompt="You are the TUTORIAL assistant.",
)
# Simulate multi-turn conversation
result = agent.invoke({
"messages": [
HumanMessage(content="First round"),
HumanMessage(content="Second round"),
HumanMessage(content="Third round"),
HumanMessage(content="Fourth round"),
HumanMessage(content="Fifth round"),
HumanMessage(content="Sixth round"),
HumanMessage(content="Seventh round"),
]
})
print(f"Message count: {len(result['messages'])}")
print(f"Reply: {result['messages'].content}")
```
Output:
Message count: 8Reply: Hello! It looks like you're conducting a multi-turn conversation test. How can I help you?
### Scenario 2: Content Filtering — Blocking Sensitive Words
## Example
```python
from langchain.agents.middleware import before_model
SENSITIVE_WORDS =["password","bank card number","ID number"]
@before_model
def content_filter(state, runtime):
"""Check if user messages contain sensitive words, intercept if found"""
messages = state.get("messages",[])
if not messages:
return None
last_msg = messages
content =str(last_msg.content)if hasattr(last_msg,'content')else""
for word in SENSITIVE_WORDS:
if word in content:
print(f" Sensitive word detected: {word}")
# jump_to="end" ends directly, preventing model response
return{
"jump_to": "end",
"messages": [
HumanMessage(content=f"Sorry, for security reasons, requests containing '{word}' cannot be processed.")
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=,
system_prompt="You are the TUTORIAL assistant.",
)
# Normal question
result = agent.invoke({
"messages": [HumanMessage(content="How do I get started with Python?")]
})
print(f"Normal question: {result['messages'].content[:80]}")
# Sensitive question
result = agent.invoke({
"messages": [HumanMessage(content="What is my bank card number? Can you help me check?")]
})
print(f"\n Sensitive question: {result['messages'].content}")
```
Output:
Sensitive word detected: bank card numberNormal question: Getting started with Python begins with installing the Python environment...Sensitive question: Sorry, for security reasons, requests containing 'bank card number' cannot be processed.
* * *
## @after_model — Post-Model Call Processing
@after_model executes after the model responds, suitable for moderating model output, extracting key information, appending follow-up instructions, etc.
### Scenario 3: Response Content Moderation
## Example
```python
from langchain.agents.middleware import after_model
FORBIDDEN_TOPICS =["politics","violence","pornography"]
@after_model
def response_audit(state, runtime):
"""Audit model response, replace if involving forbidden topics"""
messages = state.get("messages",[])
if not messages:
return None
last_msg = messages
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"Response contains '{topic}' related content, has been replaced"
})
# Return an overriding message (via add_messages reducer)
from langchain.messages import AIMessage
return{
"messages": [
AIMessage(content="Sorry, I cannot answer this question."
"Please ask about programming learning related content.")
]
}
return None
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=,
system_prompt="You are the TUTORIAL assistant.",
)
result = agent.invoke({
"messages