LangChain ミドルウェアフック -- @before_model と @after_model

before_model と after_model は最もよく使われる2つの ミドルウェア（Middleware） フックです。これらは毎回のモデル呼び出しの前後に実行され、コンテンツフィルタリング、メッセージ前処理、レスポンス審査などに適しています。

@before_model——モデル呼び出し前のインターセプト

@before_model は毎回モデルを呼び出す前に実行されます。ここでメッセージを変更したり、コンテキスト条件を注入したり、モデル呼び出しを直接スキップしたりできます。

シナリオ 1：メッセージ前処理——会話の長さを制限

例

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):
    """メッセージ履歴の長さを制限し、コンテキストが長すぎるのを防ぐ"""
    messages = state.get("messages", [])

    # system message + 最新の6件のメッセージを保持
    MAX_MESSAGES = 6
    if len(messages) > MAX_MESSAGES:
        # 最新のメッセージに切り詰める
        trimmed = messages[-MAX_MESSAGES:]
        # 最初のメッセージがユーザーメッセージであることを確認
        if trimmed and trimmed[0].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=[limit_context],
    system_prompt="あなたはサンプル のアシスタントです。",
)

# 複数ターンの会話をシミュレーション
result = agent.invoke({
    "messages": [
        HumanMessage(content="1ターン目"),
        HumanMessage(content="2ターン目"),
        HumanMessage(content="3ターン目"),
        HumanMessage(content="4ターン目"),
        HumanMessage(content="5ターン目"),
        HumanMessage(content="6ターン目"),
        HumanMessage(content="7ターン目"),
    ]
})
print(f"メッセージ数: {len(result['messages'])}")
print(f"返信: {result['messages'][-1].content}")

実行結果：

メッセージ数: 8
返答: こんにちは！複数ターンの会話テストを行っているようですね。何かお手伝いできることはありますか？

シナリオ 2：コンテンツフィルタリング——NGワードをブロック

例

from langchain.agents.middleware import before_model


SENSITIVE_WORDS = ["パスワード", "銀行カード番号", "身分証番号"]


@before_model
def content_filter(state, runtime):
    """ユーザーメッセージにNGワードが含まれているか確認し、含まれていればインターセプトする"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]
    content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

    for word in SENSITIVE_WORDS:
        if word in content:
            print(f"[インターセプト] NGワードを検出: {word}")
            # jump_to="end" で直接終了し、モデルに返信させない
            return {
                "jump_to": "end",
                "messages": [
                    HumanMessage(content=f"申し訳ありませんが、安全のため、「{word}」を含むリクエストは処理できません。")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[content_filter],
    system_prompt="あなたはサンプル のアシスタントです。",
)

# 正常な質問
result = agent.invoke({
    "messages": [HumanMessage(content="Python はどうやって始めればいいですか？")]
})
print(f"正常な質問: {result['messages'][-1].content[:80]}")

# NGワードを含む質問
result = agent.invoke({
    "messages": [HumanMessage(content="私の銀行カード番号は何ですか？調べてもらえますか？")]
})
print(f"\n"NGワードを含む質問: {result['messages'][-1].content}")

実行結果：

[拦截] 检测到敏感词: 银行卡号
通常の質問: Python 入门可以从安装 Python 环境开始...

機密の質問: 抱歉，为了安全，不能处理包含「银行卡号」的请求。

@after_model——モデル呼び出し後の処理

@after_model はモデルの返信後に実行され、モデル出力の審査、重要な情報の抽出、後続の指示の追加などに適しています。

シナリオ 3：レスポンス内容の審査

例

from langchain.agents.middleware import after_model


FORBIDDEN_TOPICS = ["政治", "暴力", "ポルノ"]


@after_model
def response_audit(state, runtime):
    """モデルの返信を審査し、禁止トピックに関連する場合は置き換える"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]
    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"返信に「{topic}」関連の内容が含まれていることを検出し、置き換えました"
            })
            # 上書きするメッセージを返す（add_messages reducer 経由）
            from langchain.messages import AIMessage
            return {
                "messages": [
                    AIMessage(content="申し訳ありませんが、この質問にはお答えできません。"
                                      "プログラミング学習に関する内容を質問してください。")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[response_audit],
    system_prompt="あなたはサンプル のアシスタントです。",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Python にはどのような学習リソースがありますか？")]
})
print(f"正常な返信: {result['messages'][-1].content}")

シナリオ 4：プロンプト情報を自動追加

例

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


@after_model
def append_disclaimer(state, runtime):
    """毎回のモデル返信後に免責事項を自動的に追加する"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]

    # AI の最終返信（tool_calls がないもの）のときのみ追加
    if (last_msg.type == "ai"
        and last_msg.content
        and not (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls)):
        # 最後の AI メッセージを置き換えて、免責事項を追加する
        return {
            "messages": [
                AIMessage(
                    content=(
                        last_msg.content
                        + "\n\n---\n*上記の内容はサンプル AI アシスタントによって生成されたものであり、参考までに提供されています。*
                    )
                )
            ]
        }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[append_disclaimer],
    system_prompt="あなたはサンプルのアシスタントです。",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Python は学びやすいですか？")]
})
print(result["messages"][-1].content)

実行結果：

Python 是一门非常适合初学者的编程语言，它的语法简洁、贴近自然语言...
---
*上記の内容はサンプルのAIアシスタントが生成したもので、参考までにご覧ください。*

can_jump_to——フロー遷移制御

before_model と after_model では、can_jump_to パラメータと jump_to 状態を使用して Agent のフローを制御できます。

例

from langchain.agents.middleware import before_model


@before_model(can_jump_to=["end"])  # ジャンプ可能なターゲットを宣言
def conditional_exit(state, runtime):
    """特定の条件下で Agent を直接終了する"""
    messages = state.get("messages", [])
    if not messages:
        return None

    # ユーザーが「さようなら」と言ったら、直接終了
    last_content = str(messages[-1].content)
    if last_content.strip() in ["さようなら", "バイバイ", "bye"]:
        return {
            "jump_to": "end",  # Agent を直接終了
            "messages": [{"role": "assistant", "content": "さようなら！またのご利用をお待ちしております。"}]
        }

    return None
can_jump_to の値意味適用シナリオ
["end"]終了にジャンプ可能条件付き終了、安全インターセプト
["model"]モデルにジャンプバック可能モデルに再処理させる必要がある
["tools"]ツールノードにジャンプ可能モデルをスキップしてツールを直接実行
["model", "end"]モデルまたは終了にジャンプ可能複数の条件分岐

can_jump_to でターゲットを宣言しない場合、jump_to は無視されます。これは、ミドルウェアが不正なノードに誤ってジャンプするのを防ぐ安全メカニズムです。

その他の拡張