LangChain ミドルウェアパターン (Python)

概要

テナント A がプロンプトを送信します：「alice@acme.com からのこのサポートチケットについて、彼女の滞納請求書について概要を作成してください。」 チェーンのキャッシングミドルウェアが PII 削除ミドルウェアの前に実行されたため、メールアドレスを含む生のプロンプトがキャッシュキーの一部になりました。30 秒後、テナント B がセマンティック的に同一のプロンプトを送信します（異なるテナント、異なる顧客、同じ形式）。キャッシュヒット。テナント B のユーザーは alice@acme.com とその滞納請求書について名前が記載された概要を取得します。これは本番環境での問題カタログエントリ P24 であり、実在の事故クラスです — ポストモーテムには「コスト削減のためキャッシングを追加したが、1 時間以内に顧客の PII を別のテナントに漏らした」と書かれています。

兄弟的な障害モード：

• P25 — リトライミドルウェアが 429 でモデル呼び出しを 2 回実行します。両方の試行が on_llm_end を発火させます。トークン使用量アグリゲータが両方を合計します。単一の論理的な呼び出しが 2 つとして請求され、テナントのセッションごとの予算が真の使用量の 50% で超過します。
• P10 — エージェントは曖昧なプロンプトで 15 回以上ループします。デフォルトのコスト上限はありません。セッションごとのトークン予算ミドルウェアはこれを解決します。なければ、単一の「アカウントについて教えてください」プロンプトが数千のトークンを消費できます。
• P34 — Runnable.invoke はプロンプトインジェクションをサニタイズしません。"Ignore previous instructions and..." を含む RAG ドキュメントが逐語的に続きます。ガードレールミドルウェアはあなたのインジェクション防御です。なければ、間接的なプロンプトインジェクションはワンラインエクスプロイトです。
• P61 — set_llm_cache(InMemoryCache()) はプロンプト文字列のみをハッシュします。異なるツールバインディングを持つ 2 つのチェーンは同じキャッシュ応答を返します。ツールはキャッシュキーで黙って無視されます。

このスキルは、LangChain 1.0 チェーンと LangGraph 1.0 エージェントの標準的なミドルウェア順序を定義します。順序不変行列（隣接する各ペアは交換した場合に名前付きの障害モードを持っています）、6 つの参照実装、プロンプト プラス バインド済みツール プラス tenant_id を含むキャッシュキーハッシュ、request_id で重複排除するリトライテレメトリ、および毎回のビルドで順序不変性をアサートする統合テストパターンを含みます。

ピン留め：langchain-core 1.0.x、langchain 1.0.x、langgraph 1.0.x。問題カタログアンカー：P10、P24、P25、P34、P61、P27、P29、P30、P33 への補足的な参照付き。

前提条件

• Python 3.10 以上
• langchain-core >= 1.0, < 2.0
• langgraph >= 1.0, < 2.0（エージェントミドルウェア用）
• 少なくとも 1 つのプロバイダーパッケージ：pip install langchain-anthropic（または openai）
• オプション：正規表現を超えた PII NER には presidio-analyzer + presidio-anonymizer
• オプション：マルチワーカーキャッシュとレート制限には redis + langchain-redis

手順

ステップ 1 — 標準的なミドルウェア順序を採用する

本番環境に進む全ての LangChain 1.0 チェーンと LangGraph 1.0 エージェントはこの順序でミドルウェアを適用します：

user → redact → guardrail → budget → cache → retry → model

• redact → cache (P24)： キャッシュキーは PII フリーである必要があります。そうしないとテナント A の PII がヒット時にテナント B に漏洩します
• guardrail → cache： インジェクション満載のプロンプトは決してキャッシュエントリになってはいけません
• budget → cache： キャッシュヒットは RPS に対してカウントされます。ループがヒット単独では セッションを DoS できないようにまず予算をチェックします
• cache → retry： キャッシュヒットはリトライを迂回します。リトライはモデル呼び出しのみをラップします

本番チェーンは通常 4～6 のミドルウェアレイヤー を実行し、レイヤーあたり <1ms のオーバーヘッド（ベンチ：p50 0.3ms/レイヤー、p99 0.9ms を 100 リクエストサンプルで）があります。完全なペアワイズマトリックスとベンチマークスクリプトについては ordering-invariants.md を参照してください。

ステップ 2 — PII 削除ミドルウェア

可逆的なプレースホルダーで エンティティをマスクしますので、呼び出し元は出力に再挿入できます — ただしキャッシュキーとモデルプロンプトは削除されたテキストのみを見ます。

import re
from typing import Any

_REDACTORS = [
    ("EMAIL", re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")),
    ("PHONE", re.compile(r"\+?\d[\d\s\-\(\)]{7,}\d")),
    ("SSN",   re.compile(r"\b\d{3}-\d{2}-\d{4}\b")),
    ("CC",    re.compile(r"\b(?:\d[ -]*?){13,16}\b")),
]

def redact(text: str) -> tuple[str, dict[str, str]]:
    pmap: dict[str, str] = {}
    for label, pattern in _REDACTORS:
        for i, match in enumerate(pattern.findall(text)):
            token = f"<{label}_{i}>"
            pmap[token] = match
            text = text.replace(match, token)
    return text, pmap

def redaction_middleware(inputs: dict[str, Any]) -> dict[str, Any]:
    redacted, pmap = redact(inputs["input"])
    return {**inputs, "input": redacted, "_pii_map": pmap}

名前、住所、カスタムエンティティについては、Presidio の AnalyzerEngine が 20 以上のエンティティタイプをカバーしています。正規表現対 spaCy 対 Presidio トレードオフマトリックス、GDPR/HIPAA/PCI-DSS エンティティリスト、再挿入パターン（削除されていない出力を 元のテナントのみ に返す — クロステナント填補しない）については pii-redaction.md を参照してください。

ステップ 3 — ガードレールミドルウェア

インジェクションパターンを事前に検出し、ユーザーコンテンツをラップしてモデルはそれをデータとして扱います。2 つのレイヤー：パターンマッチ（90% のケースを安価にキャッチ）プラスプロンプトラップ（滑り抜けるものを中和）。

INJECTION_PATTERNS = [
    re.compile(r"ignore (all |the )?(previous|prior|above) (instructions|rules)", re.I),
    re.compile(r"system prompt (is|was|now)", re.I),
    re.compile(r"you are now (a |an )?", re.I),
    re.compile(r"</?(system|instruction|prompt)>", re.I),
]

class GuardrailViolation(Exception):
    pass

def guardrail_middleware(inputs: dict[str, Any],
                        allowed_tools: set[str] | None = None) -> dict[str, Any]:
    for pattern in INJECTION_PATTERNS:
        if pattern.search(inputs["input"]):
            raise GuardrailViolation(f"Injection pattern matched: {pattern.pattern!r}")
    wrapped = f"<user_input>\n{inputs['input']}\n</user_input>"
    out = {**inputs, "input": wrapped}
    if allowed_tools is not None:
        out["_tool_allowlist"] = allowed_tools
    return out

ラップせずにモデルが「命令が何であるかを知る」ことに依存しないでください。

ステップ 4 — トークン予算ミドルウェア（セッションごと / テナントごと）

P10 に直接対処します — エージェントは曖昧なプロンプトで 15 回以上ループし、数千のトークンを消費します。予算ミドルウェアはセッションが上限を超えた場合、モデル呼び出しの前に例外を発生させます。

from dataclasses import dataclass, field
from collections import defaultdict
from threading import Lock

class BudgetExceeded(Exception): pass

@dataclass
class TokenBudget:
    ceiling: int = 50_000           # tokens per session
    _usage: dict[str, int] = field(default_factory=lambda: defaultdict(int))
    _lock: Lock = field(default_factory=Lock)

    def record(self, session_id: str, tokens: int) -> None:
        with self._lock:
            self._usage[session_id] += tokens

    def check(self, session_id: str) -> None:
        with self._lock:
            used = self._usage[session_id]
        if used >= self.ceiling:
            raise BudgetExceeded(f"Session {session_id}: {used}/{self.ceiling}")

budget = TokenBudget(ceiling=50_000)

def budget_middleware(inputs: dict[str, Any]) -> dict[str, Any]:
    budget.check(inputs.get("session_id") or "anonymous")
    return inputs

BaseCallbackHandler.on_llm_end とペアリングして usage_metadata.input_tokens + output_tokens で budget.record(...) を呼び出します。マルチワーカーデプロイの場合、Redis で TokenBudget をサポートします — プロセスごとの辞書はプロセスごとです（P29）。

ステップ 5 — ツール対応キー付きキャッシングミドルウェア

P61 はブービートラップです：InMemoryCache() はプロンプト文字列のみをハッシュするため、異なるツールリストを持つ 2 つのチェーンは同じキャッシュ応答を返します。プロンプト + バインド済みツール + テナント id 上のカスタムキーを使用します。

import hashlib, json
from typing import Callable

def cache_key(prompt: str, bound_tools: list[dict] | None, tenant_id: str) -> str:
    """Blake2b-16 ハッシュ。ツール対応、テナント対応、\\x1f セパレータ経由の衝突安全。"""
    h = hashlib.blake2b(digest_size=16)
    h.update(prompt.encode("utf-8")); h.update(b"\x1f")
    if bound_tools:
        h.update(json.dumps(bound_tools, sort_keys=True).encode("utf-8"))
    h.update(b"\x1f"); h.update(tenant_id.encode("utf-8"))
    return h.hexdigest()

def cache_middleware(get: Callable[[str], Any | None], put: Callable[[str, Any], None]):
    def _run(inputs: dict[str, Any]) -> dict[str, Any]:
        key = cache_key(inputs["input"], inputs.get("_bound_tools"),
                        inputs.get("tenant_id", "default"))
        hit = get(key)
        if hit is not None:
            return {**inputs, "_cache_hit": True, "output": hit}
        inputs["_cache_key"] = key
        return inputs
    return _run

キャッシュキーは削除されたプロンプト（ステップ 2 が最初に実行）で計算される 必要 があり、ツールスキーマを 含む必要 があります。バックエンド比較（InMemoryCache / SQLiteCache / RedisCache / RedisSemanticCache）、無効化戦略（TTL、スキーマバージョンバンプ、テナント全体パージ）、および Unicode 正規化と P62 を含む完全な落とし穴リストについては cache-key-design.md を参照してください。

ステップ 6 — テレメトリタグ付きリトライミドルウェア

P25：リトライは 429 でモデル呼び出しを 2 回実行し、両方の試行が on_llm_end を発火させ、アグリゲータが両方を合計し、テナント予算が真の使用量の 50% で超過します。修正：最初の試行に安定した request_id をアタッチし、アグリゲータに request_id ごとに 追加ではなく置換 するようにして、成功した最後の試行のみがカウントされます。

import time, uuid

RETRYABLE = (TimeoutError, ConnectionError,
             # プロバイダー固有 — あなたのプロバイダー SDK からインポート：
             # anthropic.RateLimitError, anthropic.APITimeoutError,
             # openai.RateLimitError, openai.APITimeoutError,
             )

def retry_middleware(max_retries: int = 2, base_delay: float = 1.0):
    def _run(inputs: dict[str, Any]) -> dict[str, Any]:
        request_id = inputs.get("request_id") or str(uuid.uuid4())
        return {**inputs, "request_id": request_id}
    return _run

完全なリトライループ、request_id ごとの重複排除アグリゲータ、プロバイダー固有の再試行可能例外リスト（Anthropic / OpenAI / Gemini）、ジッターを持つ指数バックオフ、デッドアップストリーム上のリトライストームを停止するサーキットブレーカーについては retry-telemetry.md を参照してください。

ステップ 7 — ミドルウェアをチェーンに構成する

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

# 順序が重要です。なぜについてはステップ 1 を参照してください。
chain = (
    RunnableLambda(redaction_middleware)
    | RunnableLambda(guardrail_middleware)
    | RunnableLambda(budget_middleware)
    | RunnableLambda(cache_middleware(cache_get, cache_put))
    | RunnableLambda(retry_middleware(max_retries=2))
    | model                             # ChatAnthropic / ChatOpenAI
)

LangGraph エージェントの場合、同じレイヤーが適用されますが、条件付きエッジを持つノード として配線されます — 違反時に END にルーティングする budget ノード、インジェクションマッチ時にエラーハンドラーにルーティングする guardrail ノード、など。LangGraph 適応については references/ordering-invariants.md を参照してください。

ステップ 8 — 統合テスト：順序不変性をアサートする

順序はコードレビューでは見えなくなります。誰かがキャッシュを削除の上に移動するまで。すべてのコミット時に実行されるテストで不変性をアサートします。

def test_cache_key_does_not_leak_pii():
    """P24 — キャッシュキーは生ではなく REDACTED プロンプトから構築されます。"""
    a = redaction_middleware({"input": "Ticket from alice@acme.com", "tenant_id": "T1"})
    b = redaction_middleware({"input": "Ticket from bob@other.com",  "tenant_id": "T1"})
    assert cache_key(a["input"], None, "T1") == cache_key(b["input"], None, "T1")

def test_cache_key_tenant_isolation():
    """P24/P33 — 同じプロンプト、異なるテナント、異なるキャッシュキー。"""
    assert cache_key("notes", None, "T1") != cache_key("notes", None, "T2")

def test_cache_key_tool_aware():
    """P61 — 同じプロンプト、異なるツールバインディング、異なるキャッシュキー。"""
    assert cache_key("p", [{"name":"search"}], "T") != cache_key("p", [{"name":"code_exec"}], "T")

CI で実行します。失敗は誰かが順序不変性を破ったことを意味します — チェーンは修正されるまでマージしません。

出力

• 標準順序で構成された 6 つのミドルウェアレイヤー：redact → guardrail → budget → cache → retry → model
• プレースホルダーマップ付き可逆 PII 削除（メール、電話、SSN、クレジットカード；名前/住所については Presidio オプション）
• インジェクションパターン検出とユーザーコンテンツラップ付きガードレールミドルウェア
• スレッドセーフカウンター付きセッションごと / テナントごとのトークン予算
• プロンプト + バインド済みツールスキーマ + テナント id を含むキャッシュキーハッシュ（P61 と P24 を修正）
• トークンアグリゲータが重複排除するように request_id タグ付きリトライミドルウェア（P25 を修正）
• 順序不変性をアサートする統合テスト

エラー処理

エラー / 障害モード	原因	修正
テナント B がキャッシュヒット時にテナント A の PII を受け取る	キャッシュの前の削除（P24） — 生 PII がキャッシュキーに入った	並べ替え：削除が最初に実行されます；キャッシュキーは削除されたプロンプト + tenant_id で構築
トークン使用量アグリゲータがリトライ後に実際の使用量の 2 倍を報告	リトライ二重カウント（P25） — 両方の試行が on_llm_end を発火させ、アグリゲータが合計	最初の試行に request_id をアタッチ；アグリゲータが request_id で重複排除
異なるバインド済みツールを持つ 2 つのチェーンが同じキャッシュ応答を返す	P61 — InMemoryCache() はプロンプト文字列のみをハッシュし、ツールスキーマではない	3 つすべてで blake2b で cache_key(prompt, bound_tools, tenant_id) を使用
エージェントが曖昧なプロンプトで 15 回以上ループ；請求スパイク	トークン予算なし（P10） — recursion_limit=25 デフォルトはコスト上限なし	budget_middleware をキャッシュの前に挿入；セッションが上限を超えた場合 BudgetExceeded を発生させ
モデルが RAG ドキュメント内の "Ignore previous instructions and..." に従う	ガードレールなし（P34） — Runnable.invoke はプロンプトインジェクションをサニタイズしない	guardrail_middleware を削除の後、キャッシュの前に挿入；ユーザー入力を <user_input> タグでラップ
正当なプロンプトで GuardrailViolation が発生	過度に熱心なインジェクションパターンマッチ	references/ordering-invariants.md のパターンを調整；反復のために偽陽性をログ
ツールスキーマを変更したデプロイ後のキャッシュポイズニング	古いキャッシュエントリが古いツールリストを参照	schema_version 定数をバンプしてキャッシュキーに含める
マルチワーカーデプロイで予算追跡ドリフト	P29 アナログ — プロセス内辞書はワーカーごとのみ	TokenBudget を Redis または別の共有ストアでサポート
ローカル開発中に KeyboardInterrupt 中にリトライが発火	P07 — デフォルト exceptions_to_handle は Python < 3.12 で KeyboardInterrupt を含む	リトライ可能な例外を明示的にリスト；BaseException をキャッチしない

例

正しい順序でエンドツーエンドチェーンを構築する

ステップ 7 の構成は 6 つのレイヤーを順序で示します。本番コードでは通常、これは build_chain(tenant_id: str, allowed_tools: set[str]) のようなファクトリに存在し、テナントスコープのキャッシュバックエンドと予算インスタンスを閉じます。ファクトリは順序を明示的かつテスト可能にします。

LangGraph エージェントバージョン

LangGraph エージェント内の同じ 6 つのレイヤーは 6 つのノードプラス条件付きエッジになります。budget は違反時に END にルーティング；guardrail はインジェクションマッチ時に error_handler にルーティング；cache はヒット時に END にルーティング。適応されたグラフトポロジーについては references/ordering-invariants.md を参照してください。

キャッシュポイズニングインシデントのデバッグ

ポストモーテムテンプレート：（1）キャッシュエントリを列挙、（2）キーが削除前後で構築されたかチェック、（3）ログ内の最初のクロステナントヒットを特定、（4）テナントプレフィックスまたはフル フラッシュで削除、（5）ステップ 8 からの順序統合テストを追加してこれが再発しないようにする。

リソース

• LangChain 1.0 / LangGraph 1.0 リリース発表
• LangChain How-to: caching
• LangChain callbacks
• LangGraph middleware / pre_model_hook
• Microsoft Presidio (PII detection)
• OWASP LLM01: Prompt Injection
• パック問題カタログ：docs/pain-catalog.md（エントリ P10、P24、P25、P34、P61、プラス P27、P29、P30、P33）
• 関連リファレンス：ordering-invariants.md、pii-redaction.md、cache-key-design.md、retry-telemetry.md