LangChain ミドルウェア（Middleware）

LangChain Middleware（ミドルウェア）は LangChain の最も強力な機能です。Agent 実行の各段階にカスタムロジックを挿入でき、リトライ、フォールバック、キャッシュ、コンテンツフィルタリング、ログ記録などの機能を実現できます。Agent 自体のコードを変更する必要はありません。

Middleware とは

Middleware は Agent の実行フローにおけるフック（Hook）です。各フックを使用すると、特定の時点でカスタムコードを実行できます：

例

# Middleware の直感的な理解：
# Agent の実行フローが次のようになっていると仮定します：

# 1. ユーザー入力 → 2. モデル思考 → 3. 場合によりツール呼び出し → 4. モデル再思考 → 5. 結果出力

# Middleware を使用すると、この5つの段階の間にカスタムロジックを挿入できます：
# 1. ユーザー入力
# ↓ [before_agent フック：ログ記録、権限チェック]
# 2. モデル思考
# ↓ [before_model フック：メッセージ前処理]
# ↓ [wrap_model_call フック：リトライ、フォールバック、キャッシュ]
# ↓ [after_model フック：コンテンツ監査]
# 3. ツール実行
# ↓ [wrap_tool_call フック：ツール呼び出しリトライ]
# 4. モデル思考に戻る（完了までループ）
# ↓ [after_agent フック：結果のフォーマット、統計分析]
# 5. 結果出力

6つのフックポイント

LangChain の Middleware は6つのフックを提供し、実行タイミングによって2つのカテゴリに分類されます：

フック実行頻度実行位置主な用途
before_agent1回Agent 開始前初期化、権限チェック、入力前処理
before_model各ループモデル呼び出し前メッセージ前処理、動的コンテキスト注入
wrap_model_call各ループモデル呼び出しのラップリトライ、フォールバック、キャッシュ、リクエスト書き換え
after_model各ループモデル呼び出し後コンテンツ監査、レスポンスフィルタリング、ログ
wrap_tool_callツール呼び出しごとツール実行のラップツールリトライ、結果キャッシュ、パラメータ書き換え
after_agent1回Agent 終了後出力のフォーマット、統計、リソースのクリーンアップ

2つの使用方法

Middleware はクラス継承またはデコレータの2つの方法で使用できます：

方法1：デコレータ（推奨）

例

from langchain.agents.middleware import before_model, after_model

# デコレータ方式：シンプルで直感的
@before_model
def log_before(state, runtime):
    """モデル呼び出しのたびにログを記録"""
    msg_count = len(state.get("messages", []))
    print(f"[before_model] 現在のメッセージ数: {msg_count}")
    return None


@after_model
def log_after(state, runtime):
    """モデル呼び出しのたびにログを記録"""
    last_msg = state["messages"][-1] if state.get("messages") else None
    if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
        print(f"[after_model] モデルがツール呼び出しを要求しました")
    return None

方法2：クラス継承（複雑なロジックに適している）

例

from langchain.agents.middleware import AgentMiddleware


class LoggingMiddleware(AgentMiddleware):
    """カスタムログミドルウェア"""

    @property
    def name(self) -> str:
        # カスタムミドルウェア名（デフォルトはクラス名）
        return "logging"

    def before_agent(self, state, runtime):
        """Agent 開始前のロジック"""
        print("[Logging] Agent の実行を開始")
        return None

    def before_model(self, state, runtime):
        """モデル呼び出し前のロジック"""
        msg_count = len(state.get("messages", []))
        print(f“[Logging] モデルを呼び出す準備中、現在 {msg_count} 件のメッセージ”)
        return None

    def after_model(self, state, runtime):
        """モデル呼び出し後のロジック"""
        print("[Logging] モデル呼び出し完了")
        return None

    def after_agent(self, state, runtime):
        """Agent 終了後のロジック"""
        print([Logging] Agent の実行が終了しました)
        return None

完全なライフサイクル例

例

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import (
    before_agent, after_agent,
    before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool


@before_agent
def start_log(state, runtime):
    """Agent 開始前"""
    print(">>> [before_agent] Agent 開始 <<<")
    runtime.stream_writer({"type": "lifecycle", "phase": "start"})
    return None


@before_model
def pre_model(state, runtime):
    """モデル呼び出しのたびに"""
    msg_count = len(state.get("messages", []))
    print(f" -> [before_model] {msg_count} 件目のメッセージ")
    return None


@after_model
def post_model(state, runtime):
    """モデル呼び出しのたびに"""
    last = state["messages"][-1] if state.get("messages") else None
    if hasattr(last, 'tool_calls') and last.tool_calls:
        tools = [tc['name'] for tc in last.tool_calls]
        print(f" <- [after_model] ツール要求: {tools}")
    else:
        content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
        print(f" <- [after_model] 直接返信: {content}...")
    return None


@after_agent
def end_log(state, runtime):
    """Agent 終了後"""
    total_msgs = len(state.get("messages", []))
    print(f<<< [after_agent] Agent 終了、合計 {total_msgs} 件のメッセージ <<<)
    return None


@tool
def get_weather(city: str) -> str:
    """天気を検索"""
    return f"{city}: 晴れ、25°C"


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[start_log, pre_model, post_model, end_log],
    system_prompt="あなたはアシスタントです。",
)

print("\n========== 1つ目の質問（ツールが必要） ==========")
result = agent.invoke({
    "messages": [HumanMessage(content="杭州の天気は？")]
})

print(f"\n最終返信: {result['messages'][-1].content}")

print("\n========== 2つ目の質問（ツール不要） ==========")
result = agent.invoke({
    "messages": [HumanMessage(content="こんにちは")]
})
print(f"\n最終返信: {result['messages'][-1].content}")

実行結果：

========== 第一个问题（需要工具） ==========
>>> [before_agent] Agent 开始 <<<
  -> [before_model] 第 2 条消息
  <- [after_model] 请求工具: ['get_weather']
  -> [before_model] 第 3 条消息
  <- [after_model] 直接回复: 杭州今天晴，气温25°C。...
<<< [after_agent] Agent 结束，共 4 条消息 <<<

最終返答: 杭州今天晴，气温25°C。

========== 第二个问题（无需工具） ==========
>>> [before_agent] Agent 开始 <<<
  -> [before_model] 第 1 条消息
  <- [after_model] 直接回复: 你好！有什么可以帮你的？...
<<< [after_agent] Agent 结束，共 2 条消息 <<<

最終返答: こんにちは！何かお手伝いできることはありますか？

出力からわかること：

  • before_agent と after_agent：各質問で1回だけ実行される
  • before_model と after_model：モデル呼び出しのたびに実行される（1つ目の質問ではモデルが2回呼び出されたため、それぞれ2回実行される）

Middleware の戻り値

Middleware の戻り値は、Agent の状態を変更するかどうか、またはフローを制御するかを決定します：

戻り値効果例
None状態を変更せず、通常のフローを続行純粋なログ記録
dictAgent の状態を更新（現在の状態にマージ）{"custom_field": "value"} を返す
jump_to を含む dict指定されたノードにジャンプ{"jump_to": "end"} を返す

返された dict は Agent 状態の reducer を介してマージされます。messages フィールドには add_messages reducer が使用されるため、返された messages は上書きではなく追加されます。

その他の拡張