YouTipLangChain @before_agent and @after_agent |
before_agent and after_agentis an Agent-level hook, executed once before and once after the Agent runs. Suitable for initialization, preprocessing, post-processing, and statistical analysis.\\\\n\\\\n\\\\n\\\\n
before_agent——Preparation before Agent starts
\\\\n\\\\nbefore_agentRun before the Agent officially starts execution, only once. You can perform input preprocessing, user information verification, resource initialization, etc. here.
Scenario 1: Input Preprocessing — Automatically Correct User Input
\\\\n\\\\nExample
from dotenv import load_dotenv\\\\n\\\\nload_dotenv()\\\\n\\\\nfrom langchain.agents import create_agent\\\\n\\\\nfrom langchain.agents.middleware import before_agent\\\\n\\\\nfrom langchain.chat_models import init_chat_model\\\\n\\\\nfrom langchain.messages import HumanMessage\\\\n\\\\nfrom langchain.tools import tool\\\\n\\\\n@before_agent\\\\ndef preprocess_input(state, runtime):\\\\n """in Agent Start Before processing user input"""\\\\n messages = state.get("messages", [])\\\\n if not messages:\\\\n return None\\\\n\\\\n # Get the user's last message\\\\n last_msg = messages\\\\n content = str(last_msg.content) if hasattr(last_msg, 'content') else ""\\\\n\\\\n # Automatically add polite phrases (if user asks a question directly)\\\\n greetings = ["Hello", "Hello", "hi", "hello", "Hi"]\\\\n if content and not any(content.lower().startswith(g) for g in greetings):\\\\n # No modification, return directly\\\\n pass\\\\n\\\\n return None\\\\n\\\\n@tool\\\\ndef search_course(keyword: str) -> str:\\\\n """in TUTORIAL Search Course"""\\\\n courses = {\\\\n "python": "Python3 Basic Tutorial (Free, 30 Chapters)",\\\\n "html": "HTML Basic Tutorial (Free, 25 Chapters)",\\\\n }\\\\n return courses.get(keyword.lower(), f"Not found {keyword} Related Courses")\\\\n\\\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\\\nagent = create_agent(\\\\n model=model,\\\\n tools=,\\\\n middleware=,\\\\n system_prompt="You are the Course advisor for TUTORIAL.",\\\\n)\\\\n\\\\nresult = agent.invoke({\\\\n "messages": [HumanMessage(content="Python Course")]\\\\n})\\\\n\\\\nprint(f"Response: {result['messages'].content}")\\\\nScenario 2: Access Control — Permission Check
\\\\n\\\\nExamples
\\\\n\\\\nfrom langchain.agents.middleware import before_agent\\\\n\\\\n@before_agent\\\\ndef access_control(state, runtime):\\\\n """Check if user has permission to use Agent"""\\\\n # From runtime.context Get user information\\\\n context = runtime.context\\\\n if context is None:\\\\n return None\\\\n\\\\n user_role = context.get("user_role", "guest")\\\\n # Guest users can only use limited features\\\\n if user_role == "guest":\\\\n messages = state.get("messages", [])\\\\n if messages:\\\\n last_content = str(messages.content)\\\\n # Check if restricted features are involved\\\\n restricted_keywords = ["Delete", "Management", "Configuration", "admin"]\\\\n if any(kw in last_content for kw in restricted_keywords):\\\\n return {\\\\n "jump_to": "end",\\\\n "messages": [HumanMessage(\\\\n content="Your current permissions are insufficient to perform this operation. Please log in and try again."\\\\n )]\\\\n }\\\\n return None\\\\n\\\\n\\\\n
after_agent——Agent Complete Post processing
\\\\n\\\\nafter_agentExecute after the Agent completes all processing (executed only once). You can format the final output, record statistics, clean up resources, etc. here.
Scenario 3: Statistical Analysis—Recording Conversation Data
\\\\n\\\\nExample
from langchain.agents.middleware import after_agent\\\\n\\\\n@after_agent\\\\ndef conversation_stats(state, runtime):\\\\n """Collect conversation stats and append to result"""\\\\n messages = state.get("messages", [])\\\\n # Statistics data\\\\n model_calls = 0\\\\n tool_calls = 0\\\\n total_chars = 0\\\\n for msg in messages:\\\\n if msg.type == "ai":\\\\n model_calls += 1\\\\n if hasattr(msg, 'tool_calls') and msg.tool_calls:\\\\n tool_calls += len(msg.tool_calls)\\\\n if hasattr(msg, 'content') and msg.content:\\\\n total_chars += len(str(msg.content))\\\\n\\\\n # Send Statistics via custom stream\\\\n runtime.stream_writer({\\\\n "type": "stats",\\\\n "model_calls": model_calls,\\\\n "tool_calls": tool_calls,\\\\n "total_messages": len(messages),\\\\n "total_chars": total_chars,\\\\n })\\\\n return None\\\\n\\\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\\\nagent = create_agent(\\\\n model=model,\\\\n tools=,\\\\n middleware=,\\\\n system_prompt="You are the Course advisor for TUTORIAL.",\\\\n)\\\\n\\\\n# Using stream_mode=["updates", "custom"] Receive custom events\\\\nfor mode, chunk in agent.stream(\\\\n {"messages": [HumanMessage(content="Look up Python Course")]},\\\\n stream_mode=["updates", "custom"],\\\\n):\\\\n if mode == "custom" and chunk.get("type") == "stats":\\\\n print(f"Statistics: {chunk}")\\\\nRunning result:
\\\\n\\\\nStatistics: {'type': 'stats', 'model_calls': 2, 'tool_calls': 1, 'total_messages': 4, 'total_chars': 127}
\\\\n\\\\nScenario 4: Formatted Output - Unified Reply Style
\\\\n\\\\nExample
from langchain.agents.middleware import after_agent\\\\nfrom langchain.messages import AIMessage\\\\n\\\\n@after_agent\\\\ndef format_output(state, runtime):\\\\n """inAppend formatted summary information to the result"""\\\\n messages = state.get("messages", [])\\\\n if not messages:\\\\n return None\\\\n\\\\n # Find the last AI message (final Response)\\\\n last_ai = None\\\\n for msg in reversed(messages):\\\\n if msg.type == "ai" and msg.content:\\\\n last_ai = msg\\\\n break\\\\n\\\\n if last_ai:\\\\n # Statistics\\\\n tool_msgs = [m for m in messages if m.type == "tool"]\\\\n tool_count = len(tool_msgs)\\\\n\\\\n footer = (\\\\n f"nn---n"\\\\n f"> Total messages in this conversation {len(messages)} messages,"\\\\n f"called {tool_count} tools.n"\\\\n f"> by TUTORIAL AI Assistant provides support."\\\\n )\\\\n\\\\n # Append to final Response\\\\n return {\\\\n "messages": [\\\\n AIMessage(content=last_ai.content + footer)\\\\n ]\\\\n }\\\\n return None\\\\n\\\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\\\nagent = create_agent(\\\\n model=model,\\\\n tools=,\\\\n middleware=,\\\\n system_prompt="You are the Course advisor for TUTORIAL.",\\\\n)\\\\n\\\\nresult = agent.invoke({\\\\n "messages": [HumanMessage(content="Python What courses are available?")]\\\\n})\\\\n\\\\nprint(result.content)\\\\nRun result:
\\\\n\\\\nTUTORIAL Includes Python3 Basic Tutorial, Total 30 Chapters, completely free, and highly suitable for Python beginners.---> Total of 4 messages in this conversation,called 1 tool calls.> by TUTORIAL AI Assistant provides support.
\\\\n\\\\n\\\\n\\\\n
Complete Collaboration Example of Four Hooks
\\\\n\\\\nExample
from langchain.agents.middleware import (\\\\n before_agent, after_agent, before_model, after_model\\\\n)\\\\n\\\\nfrom langchain.agents import create_agent\\\\nfrom langchain.chat_models import init_chat_model\\\\nfrom langchain.messages import HumanMessage\\\\nfrom langchain.tools import tool\\\\n\\\\n# ----- Define all hooks -----\\\\n@before_agent\\\\ndef init_session(state, runtime):\\\\n """Start: Initialize session"""\\\\n print(">>> Session start")\\\\n return None\\\\n\\\\n@before_model\\\\ndef pre_model_check(state, runtime):\\\\n """Before each model call"""\\\\n msg_count = len(state.get("messages", []))\\\\n print(f" Message count: {msg_count}")\\\\n return None\\\\n\\\\n@after_model\\\\ndef post_model_check(state, runtime):\\\\n """Post each model call"""\\\\n last = state if state.get("messages") else None\\\\n if last and hasattr(last, 'tool_calls') and last.tool_calls:\\\\n print(f" Tool call required")\\\\n return None\\\\n\\\\n@after_agent\\\\ndef finish_session(state, runtime):\\\\n """End: Clean up resources"""\\\\n total = len(state.get("messages", []))\\\\n print(f"<< str:\\\\n """Query Weather"""\\\\n return f"{city}: Sunny"\\\\n\\\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\\\nagent = create_agent(\\\\n model=model,\\\\n tools=,\\\\n middleware=[init_session, pre_model_check, post_model_check, finish_session],\\\\n system_prompt="You are an assistant.",\\\\n)\\\\n\\\\nresult = agent.invoke({\\\\n "messages": [HumanMessage(content="Hangzhou weather?")]\\\\n})\\\\n\\\\nprint(f"n Final Response: {result['messages'].content}")\\\\nRunning results:
\\\\n\\\\n>> Session start message count: 2 tool call needed message count: 3 <<< Session ended, total 4 messages
\\\\nFinal reply: Hangzhou is sunny today, suitable for travel.
\\\\n\\\\n
Middleware Hook Summary
\\\\n\\\\n| Hook | \\\\nExecution Count | \\\\nWhen to Use | \\\\nKey Capability | \\\\n
|---|---|---|---|
| before_agent | \\\\n1 time | \\\\nPermission check, input preprocessing, resource initialization | \\\\nCan use jump_to="end" to terminate early | \\\\n
| before_model | \\\\nEach loop | \\\\nMessage trimming, content filtering, context injection | \\\\nCan use jump_to to control flow | \\\\n
| wrap_model_call | \\\\nEach loop | \\\\nRetry, fallback, caching, prompt modification | \\\\nFull control over model execution | \\\\n
| after_model | \\\\nEach loop | \\\\nResponse review, content appending, logging | \\\\nCan replace model output | \\\\n
| wrap_tool_call | \\\\nEach tool call | \\\\nTool retry, caching, parameter rewriting | \\\\nFull control over tool execution | \\\\n
| after_agent | \\\\n1 time | \\\\nOutput formatting, statistical analysis, cleanup | \\\\nFinal state modification | \\\\n