LangChain Agent middlewareで、柔軟なAIエージェントを作ろう

概要

2025/9/8に、langchainのblogで、middlewareに関して発表されました。

Try it in LangChain 1.0 alpha
blogの最後に記載された以下のインストールコマンドからわかりますが、2025/09/12時点ではまだpreです。

pip install --pre -U langchain

LangChain v1 is under active development and is not recommended for production use.

本番環境利用は推奨されていないので、ご注意ください。

読み進めるにあたり

LangChainの初学者向けの内容ではありません。

公式ドキュメントのCore conceptsパートの内容がおおまかに把握されている方向けです。
（以下画像の緑で囲った箇所）

middleware

今回の内容に入っていきます。

LangChainに限った話ではないので、ある程度の開発経験があれば出会う話かと思います。
リクエストとレスポンスの処理パイプラインの中間で実行される関数やコンポーネントですが、代表的な用途としては以下が挙げられます。

• 認証・認可
• ロギング
• エラーハンドリング

今回は、LangChainでも構築できるようになった話をしていきたいと思います！

Built-in middleware

ここは記事内容の紹介と感想程度ですが、
以下3つのパターンでmiddlewareを構築する方法が記載されていました。それぞれKey keatures:とUse Cases:が記載されています。

• Summarization
• Human-in-the-loop
• Anthropic prompt caching

Summarizationに関して、
説明文とコード例から、どういった場合に利用するのかイメージがしやすかったです。

Long-running conversations that exceed token limits

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="openai:gpt-4o",
    tools=[weather_tool, calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",
            max_tokens_before_summary=4000,  # Trigger summarization at 4000 tokens
            messages_to_keep=20,  # Keep last 20 messages after summary
            summary_prompt="Custom prompt for summarization...",  # Optional
        ),
    ],
)

Custom Middleware

今回の本題はこちらになります。

以下は冒頭に載せたdocsから引用した画像です。
緑で囲った３か所が、記載通りですがmiddlewareです。
model：openai:gpt-4oやclaude-sonnet-4-20250514のことです。
（実装としてはBaseChatModelオブジェクトを通じて設定するので、厳密には文字列単体ではないです。）

それぞれドキュメントの内容を参考にして、処理を構築してみました。

• before_model：ここはログ出力のみ
• modify_model_request：会話が長くなってきた場合のモデル切り替え
• after_model：会話が一定の長さを超えたら、強制終了

agent.invokeを通じて行う処理

今回は、middlewareの挙動確認が趣旨のため、アプリケーション内容に関しては、明確な意味はありません。他愛のない会話を繰り返すくらいですね。

この部分のコードは生成AIにお任せしました。

今回動作確認を行ったアプリケーションは、
以下リポジトリにて公開していますので、必要であればご利用ください。

ANTHROPIC_API_KEYの準備

最近Cluadeを利用することが多いこともあり、Anthropicのmodelを使って動作確認を行っていますので、リポジトリのコードのまま動作確認されたい場合は、Anthropicのキーをご準備ください。

before_model

model実行前に動いていることを確認するだけの処理しかないので、特に記載することはありません。

modify_model_request

会話が長くなってきた場合に、モデルを高性能なものに切り替えることができます。
docs通りの実装をすると、以下のエラーがでます。

Current number of messages: 1
Middleware: Using claude-3-7-sonnet-20250219 for short conversations.
Error during testing: 'str' object has no attribute 'bind_tools'

以下のように、
__init__メソッドでChatAnthropicクラス（BaseChatModel継承クラス）のオブジェクトを定義する形に修正しています。

class MyMiddlewareModifyModel(AgentMiddleware):
    """
    モデルリクエストを動的に変更するミドルウェア

    このミドルウェアは、LLMモデルへのリクエストを実行時に変更
    会話の長さに応じて使用するモデルを動的に切り替え、
    短い会話では軽量なモデル、長い会話では高性能なモデルを使用
    """

    def __init__(self):
        # モデルインスタンスを事前に作成
        self.short_model = ChatAnthropic(
            model="claude-3-7-sonnet-20250219",
            betas=["extended-cache-ttl-2025-04-11"],
        )
        self.long_model = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            betas=["extended-cache-ttl-2025-04-11"],
        )

    def modify_model_request(
        self, request: ModelRequest, state: AgentState
    ) -> ModelRequest:
        if len(state["messages"]) > 3:
            request.model = self.long_model
            print(
                "Middleware: Switched to claude-sonnet-4-20250514 for long conversations."
            )
        else:
            request.model = self.short_model
            print(
                "Middleware: Using claude-3-7-sonnet-20250219 for short conversations."
            )
        return request

model以外の選択肢

今回はコードをつかった関係で、reques.modelとしていましたが、他の選択肢にsystem_promptやtoolsもありました。かなり柔軟なフローが組めそうです。

after-model

ここでは処理を終了する条件を実装しています。

class MyMiddlewareAfterModel(AgentMiddleware):
    """
    モデル実行後に動作するミドルウェア

    このミドルウェアは、LLMモデルの実行完了後に動作
    会話の長さをチェックして一定の制限を超えた場合に会話を終了するよう指示
    """

    def after_model(self, state: AgentState) -> dict[str, Any] | None:
        # モデル実行後の処理（ログ記録、メトリクス収集など）
        print("Middleware: After model processing...")
        if len(state["messages"]) > 5:
            return {
                "jump_to": "__end__",
                "messages": [
                    AIMessage("I'm sorry, the conversation has been terminated.")
                ],
            }
        return state

jump_to: __end__後の挙動でエラー

「終了条件を満たした場合に、後続の処理は実行せず、そこで強制終了する」と勝手に思っていましたが、実際は違いました。

私の実装修正で機能するかもしれませんが、そこは今後の確認点とさせてください。
終了条件を満たした後、agent.invokeを実行して以下のエラーが発生しました。

Error during testing: Error code: 400 - {'type': 'error', 'error': {'type': 'invalid_request_error', 'message': 'messages.6: tool_use ids were found without tool_result blocks immediately after: toolu_01CSoL3rJH6EZsM5DpnRztvW. Each tool_use block must have a corresponding tool_result block in the next message.'}, 'request_id': 'req_011CT3FZJsYV8f2P9heunWFX'}

今回はこのエラーが発生しない形で早期終了を実現しましたが、
エラー内容で調べてみると、claude-codeのリポジトリでかなりの数（150件）以上のBug報告が上がっていました。この記事公開の２週間前にOpenしたばかりのIssueです。

暫定対応

終了条件を満たした場合に、AIMessageとして、I'm sorry, the conversation has been terminated.を渡しています。agent.invokeのレスポンス内容を確認して、その内容にこの文言があれば、処理を終了する形にしています。

    if(final_message_content == "I'm sorry, the conversation has been terminated."):
        print("Conversation terminated by middleware due to length.\n")
        return

Agent Jump

終了条件を関係するのでここで記載します。docsに以下のコードサンプルはありましたが、
インポート "langchain.agents.types" を解決できませんでしたと表示されていたので、最新の状態とは乖離がありそうです。ここも、もう少し確認しておきたい点です。

from langchain.agents.types import AgentState, AgentUpdate, AgentJump
from langchain.agents.middleware import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    def after_model(self, state: AgentState) -> AgentUpdate | AgentJump | None:
        return {
        "messages": ...,
        "jump_to": "model"
    }

まとめ

middlewareを導入することで、柔軟なフロー作成ができそうということを体感できました。

今回は試してみただけなので、実践的な構成にはなっていません。実際はどういった構成にするか、そういったところで頭を悩ませるべきかと思ったので、今回の学習内容を活かして、もう少し実績的な内容にも取り組んでいければと思います。

記事内でも記載していますが、確認しておきたい点はいくつかあります。ドキュメントやリポジトリの内容確認もしつつ、適宜キャッチアップを継続できればと思います。

ありがとうございました。

他、参考記事