STUDY NOTES

第31回: LangChain 1.0 Middleware を 3 シナリオで活用 — PII / レジリエンス / ツール選択を Streamlit で体験¶

第29回で create_agent と Middleware の仕組みを学んだ。第31回はそれを 3 つの実用シナリオで Streamlit デモアプリ化する応用回。「ミドルウェアを 1 つずつ試す」ではなく、複数を組み合わせた production 級の構成を体験できる。

シナリオ	役割	使用 Middleware
1. メール処理エージェント	PII フィルタ + 履歴要約 + 人間承認	`PIIMiddleware` × 2 + `SummarizationMiddleware` + `HumanInTheLoopMiddleware`
2. レジリエンスエージェント	無限ループ防止 + コスト制御 + リトライ + フォールバック	`ModelCallLimitMiddleware` + `ToolCallLimitMiddleware` × 2 + `ModelFallbackMiddleware` + `ToolRetryMiddleware` + `ModelRetryMiddleware`
3. ツール選択エージェント	12ツールから動的に最適な 3 つを選択	`LLMToolSelectorMiddleware`（subclass で選択結果ログ採取）

位置付け: 第29回の Notebook が「教科書」なら、第31回は「実戦演習」。各シナリオがproduction の典型ニーズ（コンプライアンス、SLA、コスト最適化）に対応していて、そのままテンプレとして流用できる。

全体像¶

31/
├── run.py                                ← `streamlit run` を subprocess で起動
├── src/sd_31/
│   ├── app.py                            ← Streamlit ナビゲーション（3 シナリオ切替）
│   ├── agents/
│   │   ├── email_agent.py                ← シナリオ1: PII + 要約 + HITL
│   │   ├── resilient_agent.py            ← シナリオ2: リトライ + フォールバック + 制限
│   │   └── tool_selector_agent.py        ← シナリオ3: LLMToolSelector（log 採取 subclass）
│   ├── controllers/
│   │   └── base.py                       ← invoke_agent / resume_agent（HITL 中断 + 再開）
│   ├── pages/
│   │   ├── scenario1.py / scenario2.py / scenario3.py
│   │   └── common.py
│   └── mock/
│       ├── email_agent.json              ← メールモック
│       ├── resilient_agent.json          ← 検索/DB モック
│       └── tool_selector_agent.json      ← weather / news / stocks 等の応答
└── tests/                                 ← pytest（agent 単位 + controller 単位）

シナリオ1（メール処理）の典型フロー:

sequenceDiagram
    participant User
    participant ST as Streamlit
    participant Ctrl as invoke_agent
    participant Agent as email_agent
    participant PII as PIIMiddleware ×2
    participant Sum as SummarizationMiddleware
    participant HITL as HumanInTheLoopMiddleware
    participant LLM as Claude Sonnet 4.5
    participant Tool as send_email tool

    User->>ST: 「田中さんに返信して」
    ST->>Ctrl: invoke_agent(message, thread_id)
    Ctrl->>Agent: agent.invoke({messages: [...]})

    Agent->>PII: PIIMiddleware (input)
    Note over PII: メールアドレスを [REDACTED] に置換<br/>電話番号があれば PIIDetectionError raise
    PII-->>Agent: マスキング済み input

    Agent->>Sum: SummarizationMiddleware
    Note over Sum: 1000 tokens 超なら<br/>Haiku で古い履歴を要約
    Sum-->>Agent: 圧縮済み messages

    Agent->>LLM: 推論
    LLM-->>Agent: AIMessage(tool_calls=[send_email(...)])

    Agent->>HITL: tool_call 検査
    Note over HITL: send_email は<br/>承認必須リストにある
    HITL-->>Ctrl: __interrupt__ 発火

    Ctrl-->>ST: AgentResponse(status="pending_approval", approval_info=...)
    ST->>User: 承認ダイアログ表示

    User->>ST: 「承認」クリック
    ST->>Ctrl: resume_agent(decision="approve", thread_id)
    Ctrl->>Agent: agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}))
    Agent->>Tool: send_email(...) 実行
    Tool-->>Agent: 送信完了
    Agent-->>Ctrl: AIMessage（最終応答）
    Ctrl-->>ST: AgentResponse(status="success", message=...)
    ST->>User: 「メール送信しました」

シナリオ2（レジリエンス）の Middleware スタック:

flowchart TD
    Input[ユーザ入力] --> MCLM[ModelCallLimitMiddleware<br/>run_limit=5]
    MCLM --> TCLM1[ToolCallLimitMiddleware<br/>thread_limit=10, run_limit=10]
    TCLM1 --> TCLM2[ToolCallLimitMiddleware<br/>tool_name=search_web<br/>run_limit=3]
    TCLM2 --> MRM[ModelRetryMiddleware<br/>max_retries=2, backoff=2.0]
    MRM --> MFM[ModelFallbackMiddleware<br/>fallback to Haiku]
    MFM --> TRM[ToolRetryMiddleware<br/>max_retries=2, backoff=2.0]
    TRM --> LLM[Claude Sonnet 4.5]
    LLM -->|tool_call| Tool[search_web 30%失敗<br/>query_database 20%失敗<br/>analyze_data]
    Tool -->|失敗| TRM
    Tool -->|成功| Response[最終応答]
    LLM -->|API失敗| MRM
    MRM -->|リトライ枯渇| MFM

    style MCLM fill:#FCE4EC,stroke:#E7157B
    style TCLM1 fill:#FCE4EC,stroke:#E7157B
    style TCLM2 fill:#FCE4EC,stroke:#E7157B
    style MRM fill:#E1F5FE,stroke:#0288D1
    style MFM fill:#E1F5FE,stroke:#0288D1
    style TRM fill:#E1F5FE,stroke:#0288D1

シナリオ3（ツール選択）— 12 ツールから 3 つに絞る:

sequenceDiagram
    participant User
    participant Agent as create_agent (12 tools)
    participant TSel as LoggingToolSelectorMiddleware<br/>(wraps LLMToolSelectorMiddleware)
    participant HaikuSel as Haiku 4.5<br/>(tool selection)
    participant Sonnet as Claude Sonnet 4.5<br/>(main inference)
    participant Tool as weather tool

    User->>Agent: 「東京の天気を教えて」
    Agent->>TSel: wrap_model_call(request)
    TSel->>HaikuSel: 12 tools の中から最適な 3 つを選んで
    HaikuSel-->>TSel: [weather, search, calendar] を選択
    Note over TSel: always_include=["search"] で<br/>search は必ず含まれる
    Note over TSel: _selected_tools にログ記録
    TSel->>Sonnet: 推論（tools=3 個だけ context に乗せる）
    Sonnet->>Tool: weather("東京") 呼び出し
    Tool-->>Sonnet: 「晴れ、25℃...」
    Sonnet-->>Agent: 「東京の天気は晴れ、気温25℃です」

使用ライブラリ・原理¶

`PIIMiddleware` — 個人情報の検出とマスキング/ブロック¶

PIIMiddleware(
    "email",
    strategy="redact",                # マスキング ([REDACTED] に置換)
    apply_to_input=True,
    apply_to_tool_results=True,       # ツール戻り値もマスク
),
PIIMiddleware(
    "phone_number",
    detector=phone_pattern,           # カスタム正規表現
    strategy="block",                 # 検出したら PIIDetectionError raise
    apply_to_input=True,
),

設計:

strategy="redact" → 検出して置換（処理続行）
strategy="block" → 検出して PIIDetectionError raise（処理中断）
apply_to_input / apply_to_tool_results で適用範囲を制御
第1引数の "email" / "phone_number" はビルトイン detector のプリセット名。detector=正規表現 でカスタム化可能

`SummarizationMiddleware` — トークン上限に応じた自動要約¶

SummarizationMiddleware(
    model="anthropic:claude-haiku-4-5",
    trigger=("tokens", 1_000),       # 1000 tokens 超でトリガー
),

trigger は ("tokens", N) か ("messages", N) か
要約モデルは別指定。メインを Sonnet、要約を Haiku に分けてコスト削減
第29回 Notebook では keep=("messages", 10) も指定していたが、この回はデフォルト挙動に任せる

`HumanInTheLoopMiddleware` — ツール実行前の承認要求¶

HumanInTheLoopMiddleware(
    interrupt_on={
        "send_email": {"allowed_decisions": ["approve", "reject"]},
        "read_email": False,                # 承認不要
        "list_emails": False,
    }
)

__interrupt__ フックで処理を中断 → 呼び出し側が Command(resume={"decisions": [{"type": "approve"}]}) で再開。第22回の手書き interrupt() + Command(resume=...) パターンを Middleware にカプセル化。

`ModelCallLimitMiddleware` / `ToolCallLimitMiddleware`¶

ModelCallLimitMiddleware(run_limit=5, exit_behavior="end"),
ToolCallLimitMiddleware(thread_limit=10, run_limit=10),       # 全ツール合計
ToolCallLimitMiddleware(tool_name="search_web", run_limit=3),  # search_web 個別

run_limit = 1 リクエスト内の上限（= 1 invoke 内）
thread_limit = 同じ thread_id 内の累計上限（= セッション内）
exit_behavior="end" で上限到達時に終了。"raise" で例外送出も選べる
tool_name=... 引数で特定ツールだけに limit を設定できる → 高コストツール（API 呼び出しが重い）を個別制限

`ModelFallbackMiddleware` / `ToolRetryMiddleware` / `ModelRetryMiddleware`¶

ModelFallbackMiddleware("anthropic:claude-haiku-4-5"),       # 失敗時の差し替え先
ToolRetryMiddleware(max_retries=2, backoff_factor=2.0, initial_delay=1.0),
ModelRetryMiddleware(max_retries=2, backoff_factor=2.0),

ModelFallbackMiddleware: メイン LLM が RateLimitError や APIError で失敗 → 代替モデルに切替
*RetryMiddleware: 例外時の指数バックオフリトライ。initial_delay × backoff_factor^N で wait
3 つ組み合わせると 「リトライ 2 回 → だめなら Haiku に fallback → だめなら exit」 の堅牢チェーン

`LLMToolSelectorMiddleware` + subclass で選択ログ採取¶

class LoggingToolSelectorMiddleware(LLMToolSelectorMiddleware):
    """選択結果を記録する LLMToolSelectorMiddleware"""

    def wrap_model_call(self, request, call_next):
        def logging_call_next(modified_request):
            selected_tool_names = [t.name for t in modified_request.tools]
            _selected_tools.clear()
            _selected_tools.extend(selected_tool_names)
            return call_next(modified_request)

        return super().wrap_model_call(request, logging_call_next)


LoggingToolSelectorMiddleware(
    model="anthropic:claude-haiku-4-5",      # 選択器の LM
    max_tools=3,                              # 最大3ツール選ぶ
    always_include=["search"],                # 常に含めるツール
)

仕組み:

main LLM に渡す前に Haiku に「12 ツール一覧 + ユーザ質問」を渡して 3 つに絞らせる
絞った 3 ツールだけを main LLM の context に乗せる
20 個超えるツールがあるエージェントで必須（main LLM のコンテキスト圧迫を防ぐ）
wrap_model_call を上書きして call_next の前後にロギングを挟むパターン

ファイル別の役割¶

ファイル	役割
`app.py`	Streamlit ナビゲーション。サイドバーで 3 シナリオ切替
`agents/email_agent.py`	シナリオ1: 3 ツール + 4 ミドルウェア + InMemorySaver + mock email JSON
`agents/resilient_agent.py`	シナリオ2: 3 ツール（30% / 20% 失敗）+ 6 ミドルウェア + 実行ログ採取
`agents/tool_selector_agent.py`	シナリオ3: 12 ツール + LoggingToolSelectorMiddleware
`controllers/base.py`	中核。`invoke_agent` (PIIDetectionError ハンドル + interrupt 検出) + `resume_agent` (Command(resume=...))
`pages/scenario*.py`	各シナリオの Streamlit ページ
`pages/common.py`	共通ユーティリティ（thread_id 管理、メッセージ表示）
`mock/*.json`	デモ用モックデータ（メール、Web 検索結果、DB、weather など）

行レベルの工夫（中核ロジックの抜粋）¶

① 4 ミドルウェアの組合せ (`email_agent.py:115-156`)¶

def create_email_agent():
    phone_pattern = r"(?:\+?\d{1,3}[\s.-]?)*\(?\d{2,4}\)?[\s.-]?\d{3,4}[\s.-]?\d{4}"

    agent = create_agent(
        model="anthropic:claude-sonnet-4-5",
        tools=[read_email, send_email, list_emails],
        system_prompt=EMAIL_AGENT_SYSTEM_PROMPT,
        checkpointer=InMemorySaver(),
        middleware=[
            # メールアドレスをマスク
            PIIMiddleware(
                "email",
                strategy="redact",
                apply_to_input=True,
                apply_to_tool_results=True,                                   # ①
            ),
            # 電話番号をブロック
            PIIMiddleware(
                "phone_number",
                detector=phone_pattern,                                       # ②
                strategy="block",
                apply_to_input=True,
            ),
            # 長い履歴を自動要約
            SummarizationMiddleware(
                model="anthropic:claude-haiku-4-5",
                trigger=("tokens", 1_000),                                    # ③
            ),
            # send_email実行前に承認を要求
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "send_email": {"allowed_decisions": ["approve", "reject"]},  # ④
                    "read_email": False,
                    "list_emails": False,
                }
            ),
        ],
    )
    return agent

行	やってること	なぜそうする
①	`apply_to_tool_results=True` でツール戻り値もマスク	`read_email` で取得したメール本文に含まれるアドレスを LLM に渡す前にマスク。外部 API から個人情報が漏れ込むのを防ぐ
②	`detector=phone_pattern` カスタム正規表現	日本の電話番号フォーマット（市外局番含む）に対応。ビルトイン `"phone_number"` は米国フォーマット前提
③	`trigger=("tokens", 1_000)`	デモなので 1k tokens でトリガー。本番は 50k-100k 程度が現実的
④	`send_email` だけ承認必須、他はスルー	副作用の重いツール（メール送信）だけ HITL、読み取り系は自動。プロダクション HITL の典型パターン

② 6 ミドルウェアでレジリエンス強化 (`resilient_agent.py:137-160`)¶

def create_resilient_agent():
    agent = create_agent(
        model="anthropic:claude-sonnet-4-5",
        tools=[search_web, query_database, analyze_data],
        system_prompt=RESILIENT_AGENT_SYSTEM_PROMPT,
        checkpointer=InMemorySaver(),
        middleware=[
            ModelCallLimitMiddleware(run_limit=5, exit_behavior="end"),       # ①
            ToolCallLimitMiddleware(thread_limit=10, run_limit=10),
            ToolCallLimitMiddleware(tool_name="search_web", run_limit=3),     # ②
            ModelFallbackMiddleware("anthropic:claude-haiku-4-5"),            # ③
            ToolRetryMiddleware(max_retries=2, backoff_factor=2.0, initial_delay=1.0),
            ModelRetryMiddleware(max_retries=2, backoff_factor=2.0),
        ],
    )
    return agent

行	やってること	なぜそうする
①	LLM 呼び出し 1 リクエスト 5 回まで	ReAct ループが暴走しない保険。5 回で `exit_behavior="end"` なので強制終了
②	`search_web` だけ 3 回 / リクエスト制限	高コスト or 外部 API rate limit に弱いツールだけ厳しく制限
③	Sonnet が落ちたら Haiku にフォールバック	過負荷時 Sonnet が 529（overload）を返すことがあるので、Haiku で部分的に処理継続

Middleware の実行順序: リストの先頭から外側に巻かれる。[A, B, C] だと A が外、C が内。ModelCallLimitMiddleware を先頭に置くことで、リトライが何回起きても run_limit を超えれば止まる。

③ subclass で `wrap_model_call` を入れ子に拡張 (`tool_selector_agent.py:38-48`)¶

class LoggingToolSelectorMiddleware(LLMToolSelectorMiddleware):
    """選択結果を記録するLLMToolSelectorMiddleware"""

    def wrap_model_call(self, request, call_next):
        def logging_call_next(modified_request):                              # ①
            selected_tool_names = [t.name for t in modified_request.tools]
            _selected_tools.clear()
            _selected_tools.extend(selected_tool_names)                       # ②
            return call_next(modified_request)

        return super().wrap_model_call(request, logging_call_next)            # ③

行	やってること	なぜそうする
①	`call_next` をラップする関数 `logging_call_next` を作る	`super().wrap_model_call(request, logging_call_next)` に渡すことで、親クラスが選択処理を実行 → その後で logging_call_next が呼ばれる順序になる
②	`modified_request.tools` には選択後の 3 ツールが入っている	親 `LLMToolSelectorMiddleware` が tools を絞ってから call_next に渡す仕様を利用
③	`super().wrap_model_call` で親の選択ロジックは丸ごと再利用	DRY。subclass で観測だけ追加する綺麗なパターン

④ Controller での割り込み検出 + 再開 (`controllers/base.py:160-240`)¶

def invoke_agent(agent, user_message, thread_id):
    try:
        config = {"configurable": {"thread_id": thread_id}}
        previous_messages = _get_state_messages(agent, thread_id)

        result = agent.invoke(
            {"messages": [{"role": "user", "content": user_message}]},
            config=config,
        )

        # HITL 中断検出                                                       # ①
        if isinstance(result, dict) and "__interrupt__" in result:
            interrupt_info = result["__interrupt__"][0].value if result["__interrupt__"] else {}
            response_text = _extract_new_response(result, previous_messages)
            return AgentResponse(
                status="pending_approval",
                message=response_text,
                approval_info=interrupt_info,
            )

        # 通常応答
        response_text = _extract_new_response(result, previous_messages)
        if response_text is None:
            response_text = _extract_response(result)
        return AgentResponse(status="success", message=response_text)

    except PIIDetectionError as e:                                            # ②
        pii_type = "電話番号" if "phone" in str(e).lower() else "個人情報"
        return AgentResponse(
            status="pii_blocked",
            message=f"入力に{pii_type}が含まれているため、処理をブロックしました。",
        )
    except Exception as e:
        return AgentResponse(status="error", message=f"エラー: {str(e)}")


def resume_agent(agent, decision, thread_id):
    try:
        result = agent.invoke(
            Command(resume={"decisions": [{"type": decision}]}),              # ③
            config={"configurable": {"thread_id": thread_id}},
        )
        response_text = _extract_response(result)
        return AgentResponse(status="success", message=response_text)
    except Exception as e:
        return AgentResponse(status="error", message=f"エラー: {str(e)}")

行	やってること	なぜそうする
①	`"__interrupt__" in result` で HITL 中断を検出	LangChain Middleware は中断時に state に `__interrupt__` を入れる仕様。これを見て Streamlit が承認ダイアログを出す
②	`PIIDetectionError` を専用ハンドル	`PIIMiddleware(strategy="block")` がこの例外を投げる。ユーザに「PII でブロックされた」と日本語で説明
③	`Command(resume={"decisions": [{"type": decision}]})` で再開	第22回の `Command(resume=...)` と同じパターンだが、`decisions` リスト形式は HITL Middleware 固有

学んだこと（要点）¶

PIIMiddleware は apply_to_tool_results=True が肝。ツール戻り値の PII が LLM の context に漏れる事故を防げる
HumanInTheLoopMiddleware の interrupt_on={...} はツール単位で柔軟。副作用の重いツールだけ HITL、他は素通しが production の定石
Middleware の組合せ順は重要。リトライ系は実行系の外、Limit 系は最外（暴走防止優先）
ToolCallLimitMiddleware を tool_name 指定で 2 重に重ねることで、全体 limit と個別 limit を両立できる
LLMToolSelectorMiddleware を subclass して wrap_model_call を拡張するパターンは、観測やカスタマイズに非常に便利
Command(resume={"decisions": [...]}) の dict 形式は HITL Middleware 固有。第22回の単純な Command(resume=Feedback(...)) とは違う
第29回で「Middleware の使い方」を学び、第31回で「production を意識した複数組合せ」を学ぶ流れが連載構成として優秀
モックデータを JSON 外出しにすることで、デモ実行に API キーが要らない部分が増えて学習しやすい設計（Streamlit ナビゲーションは API 必要だがツール内部はモック）
_selected_tools / _execution_log のモジュールレベルリストでメトリクス採取するパターンは、第24回の memory シングルトンと同じ思想。production では state-ful Middleware（クラスベース）に移すべき

拡張アイデア¶

シナリオ追加: コンテキスト編集 — ContextEditingMiddleware で古い tool 結果を削除するデモを追加。長期会話での context 肥大化を体感
シナリオ追加: TodoList Agent — TodoListMiddleware を使って Claude Code 風 TODO 駆動エージェントを最小コードで実装。第24回の自前実装と比較
PIIMiddleware のカスタム detector を増やす — マイナンバー / クレジットカード番号 / 住所を別 PIIMiddleware で追加。日本の法令対応
HITL に reject 後のフォローアップ — ユーザが reject した時、「なぜ reject したか」を聞いて agent に渡し、修正案を再生成させる
メトリクスを Prometheus にエクスポート — _execution_log を prometheus_client の Counter に置き換えて、Grafana で可視化
シナリオ間の Middleware 共有 — PIIMiddleware をシナリオ2にも追加してみる。ModelFallbackMiddleware をシナリオ1にも追加してみる。Middleware の組合せ自由度を体感
apply_to_output=True を試す — PIIMiddleware で LLM 出力もマスクする実験。LLM が思いついたメアドが意図せず出力される事故を防げる

現代版に移植するなら¶

1. API キー設定は `.env.op` + `op run` に切り替える（CLAUDE.md ルール 8）¶

# 31/.env.op
ANTHROPIC_API_KEY=op://Personal/anthropic-api-key/credential
LANGCHAIN_API_KEY=op://Personal/langsmith-api-key/credential
LANGCHAIN_PROJECT=sd-31
LANGCHAIN_TRACING_V2=true

起動:

op run --env-file=.env.op -- uv run python run.py
op run --env-file=.env.op -- uv run pytest tests/ -v

2. モデル ID を最新へ¶

claude-sonnet-4-5 / claude-haiku-4-5 を claude-sonnet-4-6 / claude-haiku-4-5-20251001 等に変更。email_agent.py:121, resilient_agent.py:140, tool_selector_agent.py:316 の 3 箇所をハードコードしているのでまとめて差し替え。

3. モジュールレベルシングルトン (`_selected_tools`, `_execution_log`, `_sent_emails`) を state-ful Middleware に¶

production では複数セッション安全性が必要。第29回の LoggingMiddleware(AgentMiddleware) のように __init__ で self.execution_log = [] を持たせるほうが堅い。第24回でも同じ問題を指摘した。

4. mock JSON の代わりに本物の API 接続¶

シナリオ2 の search_web を Tavily、query_database を SQLite または DuckDB に差し替え。シナリオ1 の email を Gmail API に。本物のリトライ・フォールバックが体感できるようになる。

5. `tests/` の補強¶

test_email_agent.py / test_resilient_agent.py / test_tool_selector_agent.py / test_controller_base.py がすでにあるが、PII 検出の境界ケース、リトライ回数の正確性、ツール選択の決定論性などを増やすと CI で守れる範囲が広がる。

6. `app.py` でモデル切替 UI¶

Streamlit sidebar で「Sonnet / Haiku / Opus」を切替できるようにして、コスト・速度・精度の差を体感。

既知の不具合・注意点¶

ModelFallbackMiddleware("anthropic:claude-haiku-4-5") のモデル ID: バージョン suffix 無しは LiteLLM 側で alias 解決される想定。version pinning したい場合は claude-haiku-4-5-20251001 のように明示推奨
PIIDetectionError の文字列パターンマッチ: if "phone" in str(e).lower() は脆い。例外型自体に pii_type 属性が欲しい
_sent_emails: list[dict] = [] がモジュールレベル: テスト並列化で衝突する。tests/conftest.py で autouse fixture でクリアすべき
ToolRetryMiddleware と random.seed の相性: search_web の 30% 失敗は random.random() < 0.3。seed 固定なしなので毎回挙動が変わる。テストでは random.seed(42) してから呼ぶか、unittest.mock で patch
InMemorySaver の thread_id 自動生成なし: ページごとに thread_id を session_state で持つ必要があるが、scenario*.py の実装に依存
SummarizationMiddleware の trigger=("tokens", 1_000) は短すぎ: 実稼働では会話 3-4 ターンで要約発火しがち。デモ用と割り切るかコメントで明示すべき
HumanInTheLoopMiddleware の interrupt_on={"read_email": False} の False: 「承認不要」を明示する書き方だが、デフォルトと違いがあるか docs を要確認（おそらく明示しなくても同じ）
tool_selector_agent.py:319 の checkpointer=InMemorySaver() が未 import: 実は tool_selector_agent.py の冒頭 import に InMemorySaver がない（agent_module.py ではコメントされてないが）。実行時にエラー出る可能性 — 実機確認すべき
controllers/base.py の previous_messages 差分計算: HITL 中断中にレスポンスを抽出するため append-only 仮定で差分を取るが、SummarizationMiddleware が履歴を書き換えると壊れる。フォールバックでカバーしているが完璧ではない

記事参照¶

Software Design 2026 年 4 月号（推定）連載第31回「LangChain Middleware 実戦 — 3 シナリオで学ぶ」
関連: 第29回 STUDY_NOTES — create_agent + Middleware の入門
関連: 第22回 STUDY_NOTES — Functional API 時代の HITL（interrupt() 手書き）
関連: 第24回 STUDY_NOTES — シングルトン共有メモリの同じパターン
公式 Middleware 一覧: https://docs.langchain.com/oss/python/langchain/middleware
公式 HumanInTheLoopMiddleware: https://docs.langchain.com/oss/python/langchain/middleware/human-in-the-loop
公式 PIIMiddleware: https://docs.langchain.com/oss/python/langchain/middleware/pii

作成: 2026-05-25 ／最終更新: 2026-06-10

STUDY NOTES

第31回: LangChain 1.0 Middleware を 3 シナリオで活用 — PII / レジリエンス / ツール選択を Streamlit で体験¶

全体像¶

使用ライブラリ・原理¶

PIIMiddleware — 個人情報の検出とマスキング/ブロック¶

SummarizationMiddleware — トークン上限に応じた自動要約¶

HumanInTheLoopMiddleware — ツール実行前の承認要求¶

ModelCallLimitMiddleware / ToolCallLimitMiddleware¶

ModelFallbackMiddleware / ToolRetryMiddleware / ModelRetryMiddleware¶

LLMToolSelectorMiddleware + subclass で選択ログ採取¶