LangChain @before_agent と @after_agent
before_agent と after_agent は Agent レベルのフックで、Agent の実行前と完了後にそれぞれ 1 回ずつ実行されます。初期化、前処理、後処理、統計分析に適しています。
before_agent -- Agent の開始前の準備作業
before_agent は Agent が正式に実行を開始する前に実行され、1回だけ実行されます。ここで、入力の前処理、ユーザー情報の検証、リソースの初期化などを行うことができます。
シナリオ 1：入力前処理 -- ユーザー入力を自動修正
例
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):
"""Agent の開始前にユーザー入力を処理"""
messages = state.get("messages", [])
if not messages:
return None
# ユーザーの最後のメッセージを取得
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
# ユーザーが直接質問した場合、丁寧な言葉を自動追加
greetings = ["こんにちは", "ごきげんよう", "hi", "hello", "やあ"]
if content and not any(content.lower().startswith(g) for g in greetings):
# 変更せずにそのまま返す
pass
return None
@tool
def search_course(keyword: str) -> str:
"""サンプル でコースを検索"""
courses = {
"python": "Python3 基本チュートリアル（無料、30章）",
"html": "HTML 基本チュートリアル（無料、25章）",
}
return courses.get(keyword.lower(), f"{keyword} に関するコースが見つかりません")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="あなたはサンプル のコースアドバイザーです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python コース")]
})
print(f"返信: {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):
"""Agent の開始前にユーザー入力を処理"""
messages = state.get("messages", [])
if not messages:
return None
# ユーザーの最後のメッセージを取得
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""
# ユーザーが直接質問した場合、丁寧な言葉を自動追加
greetings = ["こんにちは", "ごきげんよう", "hi", "hello", "やあ"]
if content and not any(content.lower().startswith(g) for g in greetings):
# 変更せずにそのまま返す
pass
return None
@tool
def search_course(keyword: str) -> str:
"""サンプル でコースを検索"""
courses = {
"python": "Python3 基本チュートリアル（無料、30章）",
"html": "HTML 基本チュートリアル（無料、25章）",
}
return courses.get(keyword.lower(), f"{keyword} に関するコースが見つかりません")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="あなたはサンプル のコースアドバイザーです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python コース")]
})
print(f"返信: {result['messages'][-1].content}")
シナリオ 2：アクセス制御 -- 権限チェック
例
from langchain.agents.middleware import before_agent
@before_agent
def access_control(state, runtime):
"""ユーザーに Agent を使用する権限があるかチェック"""
# runtime.context からユーザー情報を取得
context = runtime.context
if context is None:
return None
user_role = context.get("user_role", "guest")
# ゲストユーザーは限られた機能のみ使用可能
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# 制限機能に関係するかチェック
restricted_keywords = ["削除", "管理", "設定", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content=現在の権限ではこの操作を実行できません。ログインしてもう一度お試しください。
)]
}
return None
@before_agent
def access_control(state, runtime):
"""ユーザーに Agent を使用する権限があるかチェック"""
# runtime.context からユーザー情報を取得
context = runtime.context
if context is None:
return None
user_role = context.get("user_role", "guest")
# ゲストユーザーは限られた機能のみ使用可能
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# 制限機能に関係するかチェック
restricted_keywords = ["削除", "管理", "設定", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content=現在の権限ではこの操作を実行できません。ログインしてもう一度お試しください。
)]
}
return None
after_agent -- Agent 完了後の処理
after_agent は Agent がすべての処理を完了した後に実行されます（1回だけ）。ここで、最終出力のフォーマット、統計情報の記録、リソースのクリーンアップなどを行うことができます。
シナリオ 3：統計分析 -- 会話データを記録
例
from langchain.agents.middleware import after_agent
@after_agent
def conversation_stats(state, runtime):
"""会話情報を集計して結果に追加"""
messages = state.get("messages", [])
# データを集計
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))
# 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="あなたはサンプル のコースアドバイザーです。",
)
# stream_mode=["updates", "custom"] を使用してカスタムイベントを受信
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Python コースを調べて")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"統計情報: {chunk}")
@after_agent
def conversation_stats(state, runtime):
"""会話情報を集計して結果に追加"""
messages = state.get("messages", [])
# データを集計
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))
# 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="あなたはサンプル のコースアドバイザーです。",
)
# stream_mode=["updates", "custom"] を使用してカスタムイベントを受信
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Python コースを調べて")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"統計情報: {chunk}")
実行結果：
统计信息: {'type': 'stats', 'model_calls': 2, 'tool_calls': 1,
'total_messages': 4, 'total_chars': 127}
シナリオ 4：出力フォーマット -- 返信スタイルの統一
例
from langchain.agents.middleware import after_agent
from langchain.messages import AIMessage
@after_agent
def format_output(state, runtime):
"""結果にフォーマットされた要約情報を追加"""
messages = state.get("messages", [])
if not messages:
return None
# 最後の AI メッセージ（最終返信）を探す
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break
if last_ai:
# 統計情報
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)
footer = (
f"\n\n---\n"
f"> この会話は合計 {len(messages)} 件のメッセージで、"
f"ツールが {tool_count} 回呼び出されました。"\n"
f"> サンプル AI アシスタントによってサポートされています。"
)
# 最終返信に追加
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="あなたはサンプル のコースアドバイザーです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python にはどんなコースがありますか？")]
})
print(result["messages"][-1].content)
from langchain.messages import AIMessage
@after_agent
def format_output(state, runtime):
"""結果にフォーマットされた要約情報を追加"""
messages = state.get("messages", [])
if not messages:
return None
# 最後の AI メッセージ（最終返信）を探す
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break
if last_ai:
# 統計情報
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)
footer = (
f"\n\n---\n"
f"> この会話は合計 {len(messages)} 件のメッセージで、"
f"ツールが {tool_count} 回呼び出されました。"\n"
f"> サンプル AI アシスタントによってサポートされています。"
)
# 最終返信に追加
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="あなたはサンプル のコースアドバイザーです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="Python にはどんなコースがありますか？")]
})
print(result["messages"][-1].content)
実行結果：
サンプルにはPython3基礎チュートリアルが全30章あり、完全無料で、Python初心者の入門に最適です。 --- > 本次对话共进行 4 条消息，调用了 1 次工具。 > 由サンプル AI 助手提供支持。
4つのフックの完全な連携例
例
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
# ----- すべてのフックを定義 -----
@before_agent
def init_session(state, runtime):
"""開始: セッションの初期化"""
print(">>> セッション開始")
return None
@before_model
def pre_model_check(state, runtime):
# 各モデル呼び出しの前
msg_count = len(state.get("messages", []))
print(f" [モデル前] メッセージ数: {msg_count}")
return None
@after_model
def post_model_check(state, runtime):
# 各モデル呼び出しの後
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [モデル後] ツール呼び出しが必要")
return None
@after_agent
def finish_session(state, runtime):
# 終了: リソースのクリーンアップ
total = len(state.get("messages", []))
print(f<<< セッション終了、合計 {total} 件のメッセージ)
return None
# ----- Agent を作成 -----
@tool
def get_weather(city: str) -> str:
"""天気を調べる"""
return f"{city}: 晴れ"
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="あなたはアシスタントです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="杭州の天気は？")]
})
print(f"\n"最終返信: {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
# ----- すべてのフックを定義 -----
@before_agent
def init_session(state, runtime):
"""開始: セッションの初期化"""
print(">>> セッション開始")
return None
@before_model
def pre_model_check(state, runtime):
# 各モデル呼び出しの前
msg_count = len(state.get("messages", []))
print(f" [モデル前] メッセージ数: {msg_count}")
return None
@after_model
def post_model_check(state, runtime):
# 各モデル呼び出しの後
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [モデル後] ツール呼び出しが必要")
return None
@after_agent
def finish_session(state, runtime):
# 終了: リソースのクリーンアップ
total = len(state.get("messages", []))
print(f<<< セッション終了、合計 {total} 件のメッセージ)
return None
# ----- Agent を作成 -----
@tool
def get_weather(city: str) -> str:
"""天気を調べる"""
return f"{city}: 晴れ"
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="あなたはアシスタントです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="杭州の天気は？")]
})
print(f"\n"最終返信: {result['messages'][-1].content}")
実行結果：
>>> 会话开始 [model前] 消息数: 2 [model后] 需要工具调用 [model前] 消息数: 3 <<< 会话结束，共 4 条消息 最終返答: 杭州は今日晴れ、お出かけに最適です。
Middleware フックのまとめ
| フック | 実行回数 | いつ使うか | 主な機能 |
|---|---|---|---|
| before_agent | 1 回 | 権限チェック、入力前処理、リソース初期化 | jump_to="end" で早期終了できる |
| before_model | ループごと | メッセージの切り詰め、コンテンツフィルタリング、コンテキスト注入 | jump_to でフローを制御できる |
| wrap_model_call | ループごと | リトライ、デグラデーション、キャッシュ、プロンプト変更 | モデル実行の完全制御 |
| after_model | ループごと | レスポンスレビュー、コンテンツ追加、ログ | モデル出力を置き換え可能 |
| wrap_tool_call | ツール呼び出しごと | ツールのリトライ、キャッシュ、パラメータ書き換え | ツール実行の完全制御 |
| after_agent | 1 回 | 出力フォーマット、統計分析、クリーンアップ | 最終状態の変更 |