Design PatternsIntegrationLLM APISoftware ArchitecturePython

LLM API統合パターン：本番環境のデザインパターン

約1分

プルリクエストのタイトルは「Add GPT-5.5 fallback」。差分は本来20行で済むところが340行——コピペされたリトライデコレータ、ハードコードされたモデル文字列、ほぼ同一のcall_llm_with_retryバリアントが5つ。

あなたのレビュー：「ここは抽象化が必要です」。著者：「具体的にはどの抽象化ですか？」

この記事はその問いに答えます。

Gang of Fourの6つのパターンをLLM API領域に翻訳しました。モデル選択のFactory、プロンプトのStrategy、ストリーミングのObserver、リトライとロギングのDecorator、フォールバックのChain of Responsibility、エージェントループのTemplate Method——それぞれに本番用Pythonコードと、置き換えるアンチパターンを添えています。

パターン1：Factory——中央集権的なモデルインスタンス化

問題

chat.pyに"gpt-5.5"。summarizer.pyに"claude-sonnet-4-20250514"。classifier.pyに"deepseek-v4-flash"。モデル移行とは、コードベース全体での検索と置換——そして、本番環境でしか読み込まれない設定ファイルに漏れがないことを祈ることです。

パターン

中央集権的なレジストリを持つModelFactory。ハードコードされた文字列ではなく、タスクタイプがどのモデルを使うかを決定します。環境変数により、カナリアデプロイとロールバックが可能になります。モデルのメタデータ（capabilities、cost tier、context window）はモデルIDと一緒に管理されます。実装にはOpenAI Python SDK——OpenAI互換APIの標準クライアントライブラリ——を使用します。そのAsyncOpenAIクライアントが以下のFactoryを支えます。

from dataclasses import dataclass
from openai import AsyncOpenAI

@dataclass
class ModelSpec:
    model_id: str
    provider: str
    capabilities: list[str]       # ["chat", "vision", "tools", "json_mode"]
    cost_tier: str                # "cheap", "mid", "frontier"
    context_window: int

class ModelFactory:
    def __init__(self, base_url: str, api_key: str):
        self.client = AsyncOpenAI(base_url=base_url, api_key=api_key)
        self.registry: dict[str, ModelSpec] = {}
        self._load_registry()

    def create(self, task_type: str, requirements: list[str] = None) -> tuple[AsyncOpenAI, ModelSpec]:
        model_id = os.getenv(f"MODEL_OVERRIDE_{task_type.upper()}", None)
        if model_id:
            spec = self.registry[model_id]
        else:
            spec = self._select_by_capability(task_type, requirements or [])
        return self.client, spec

    def _select_by_capability(self, task_type: str, requirements: list[str]) -> ModelSpec:
        candidates = [
            m for m in self.registry.values()
            if all(req in m.capabilities for req in requirements)
        ]
        tier_map = {"classification": "cheap", "generation": "mid", "review": "frontier"}
        tier = tier_map.get(task_type, "mid")
        return next((m for m in candidates if m.cost_tier == tier), candidates[0])

統一APIエンドポイント——全プロバイダーに1つのbase_url——は、Factoryの設定をO(Nプロバイダー)からO(1エンドポイント＋Nモデル文字列)へ縮小します。APIキーは1つ。クライアントインスタンスは1つ。レジストリ内のすべてのモデルがそれを通じてアクセス可能です。この単一エントリポイントアーキテクチャの構築は、APIキー認証から始まります——レジストリ内のすべてのモデルへのアクセスを制御する1つの認証情報が、Nキー×Mプロバイダーの乱雑さを排除します。

これが置き換えるアンチパターン

各呼び出し箇所にハードコードされたモデル文字列。モデルの非推奨化はコードベース全体の検索と置換を引き起こし——漏れに気づく最初の兆候は、本番環境での404エラーです。

パターン2：Strategy——プラグイン可能なプロンプトテンプレート

問題

ビジネスロジックにインライン化されたプロンプト文字列。チェックアウトのトーンを変えるには、チェックアウト、サポート、オンボーディングのコードに散らばった"You are a helpful shopping assistant..."をすべて見つけなければなりません。2つのプロンプトバリアントをA/Bテストするには、すべての呼び出し箇所でif/elseスパゲッティが必要になります。

パターン

PromptStrategyインターフェース。ユースケースまたは実験バリアントごとの具象実装。フィーチャーフラグまたはA/Bテストバケットによる実行時選択。各ストラテジーはバージョン管理された成果物です——プロンプトレジストリはバージョンをストラテジークラスにマップします。

from abc import ABC, abstractmethod

class PromptStrategy(ABC):
    version: str

    @abstractmethod
    def build_messages(self, context: dict) -> list[dict]:
        """Build the messages array for this prompt strategy."""

class CheckoutV3(PromptStrategy):
    version = "checkout_v3.2"

    def build_messages(self, context: dict) -> list[dict]:
        return [
            {"role": "system", "content": CHECKOUT_SYSTEM_V3},
            {"role": "user", "content": f"<cart>{context['cart']}</cart>"}
        ]

class PromptRouter:
    def __init__(self, strategies: dict[str, PromptStrategy]):
        self.strategies = strategies

    def select(self, feature_flags: dict, task: str) -> PromptStrategy:
        variant = feature_flags.get(f"prompt_{task}", "default")
        return self.strategies[variant]

チェックアウトプロンプトV3とV4をA/Bテストするとき、フィーチャーフラグを切り替えるだけです。コード変更はゼロ。評価スイート（テストガイドを参照）がどちらのバリアントが勝つかを測定します。

これが置き換えるアンチパターン

ビジネスロジック全体に散らばったプロンプト文字列。トーンを変えるには、コピペされたすべてのバリアントを見つける必要があります。デプロイログを追跡しない限り、ユーザーがどのバージョンのプロンプトを受け取ったのかを知る方法はありません。

パターン3：Observer——疎結合なストリーミングコンシューマー

問題

ストリーミングループには、TTS合成、UIチャンクレンダリング、コスト追跡、ロギングがすべて絡み合っています。新しいコンシューマー（アナリティクス、翻訳オーバーレイ、監査記録）を追加するには、コアの生成ループを変更する必要があります。3回追加した後、ループは200行になり、誰も触りたくなくなります。

パターン

StreamObserverインターフェース。各コンシューマーに対応する具象オブザーバー。ジェネレーターはオブザーバーに通知しますが、オブザーバーが何をするかは知りません。疎結合です。オブザーバーは独立して追加、削除、置換できます。

class StreamObserver(ABC):
    @abstractmethod
    async def on_token(self, token: str, sequence: int): ...
    @abstractmethod
    async def on_complete(self, full_response: str, usage: dict): ...
    @abstractmethod
    async def on_error(self, error: Exception): ...

class StreamObservable:
    def __init__(self, client: AsyncOpenAI):
        self.client = client
        self.observers: list[StreamObserver] = []

    def attach(self, observer: StreamObserver): self.observers.append(observer)

    async def stream(self, **kwargs):
        stream = await self.client.chat.completions.create(stream=True, **kwargs)
        full_response = ""
        try:
            async for chunk in stream:
                token = chunk.choices[0].delta.content or ""
                full_response += token
                await asyncio.gather(*[
                    o.on_token(token, len(full_response)) for o in self.observers
                ])
            await asyncio.gather(*[
                o.on_complete(full_response, usage) for o in self.observers
            ])
        except Exception as e:
            await asyncio.gather(*[o.on_error(e) for o in self.observers])

1つのオブザーバーがクラッシュしてもストリームは止まりません——エラーはオブザーバーごとに隔離されます。CostTrackerオブザーバーを追加します。TTSOutputオブザーバーを追加します。どちらも相手の存在を知りません。

これが置き換えるアンチパターン

生成ループにインライン化されたすべてのストリーミングコンシューマーロジック。アナリティクス計測の追加には、TTSを処理するのと同じ関数の編集が必要で——変数名を打ち間違えて音声出力にリグレッションを起こすリスクがあります。

パターン4：Decorator——散らかりのない運用レイヤー

問題

10行のLLM呼び出しが、60行のリトライロジック、コスト追跡、構造化ロギング、エラーハンドリングに囲まれています。8つの呼び出し箇所で、少しずつ異なるパラメータでコピペされています。

パターン

コアのLLM呼び出しを包むレイヤー化されたデコレータ。各デコレータは単一の責務を持ちます。呼び出し箇所ごとに異なる組み合わせに合成されます。

@with_retry(max_retries=3, backoff="exponential", retry_on=[429, 503])
@with_cost_tracking(budget_per_call=5.00)
@with_structured_logging(log_level="DEBUG")
async def core_llm_call(client, model_spec, messages):
    return await client.chat.completions.create(
        model=model_spec.model_id, messages=messages
    )

リトライデコレータは、指数バックオフとジッターで一時的なエラーを処理します——リトライすべきでないエラータイプ（400、401、403）は即座に通過させます。レート制限の仕組みと429処理アーキテクチャ全体は、レート制限処理ガイドで解説しています——このパターンは、そのロジックをカプセル化し、呼び出し箇所全体で一貫して適用します。コスト追跡デコレータはgen_ai.usageをログし、1回の呼び出しコストが予算を超えるとアラートします。どちらのデコレータも相手の存在を知りません。スタックの順序が重要です：リトライを最外層に（失敗したリトライもコスト追跡されるように）、ロギングを最内層に（最終レスポンスを見られるように）。

このパターンは、レート制限処理ロジックをカプセル化して、すべての呼び出し箇所で一貫して適用する方法を示します。同じカプセル化の原則はプロンプトキャッシュにも適用されます——@with_cacheデコレータは、API呼び出しが発生する前に繰り返しまたは類似したリクエストをインターセプトします。TokSpanのプロンプトキャッシングドキュメントは、デコレータが包むAPIレベルのキャッシングの仕組みを解説しています。

これが置き換えるアンチパターン

すべてのLLM呼び出しの周りにコピペされた運用ボイラープレート。一貫性のないリトライパラメータ。8つの呼び出し箇所のうち3つでコスト追跡が欠落。各呼び出し箇所が少しずつ違うやり方をしているため、どのロギング形式が「正しい」のか誰も知りません。

パターン5：Chain of Responsibility——フォールバックパイプライン

問題

ネストされたtry/exceptブロックにハードコードされたモデルフェイルオーバー。try gpt-5.5 —except: try claude-sonnet —except: try deepseek —except: return error。フォールバックモデルを追加するか、チェーンの順序を変えるには、ブロック全体を書き直す必要があります。呼び出し箇所ごとに少しずつ異なるチェーンがあります。

パターン

ModelHandlerオブジェクトのチェーン。各ハンドラーは自分のモデルとリクエストの処理方法を知っています。失敗した場合（非一時的なエラー、タイムアウト、しきい値以下の品質）は、次のハンドラーに渡します。チェーンの構成はコードではなく設定に置かれます。

class ModelHandler(ABC):
    def __init__(self, model_spec: ModelSpec):
        self.model_spec = model_spec
        self._next: ModelHandler | None = None

    def set_next(self, handler: "ModelHandler") -> "ModelHandler":
        self._next = handler
        return handler

    async def handle(self, request: dict) -> dict | None:
        try:
            result = await self._call_model(request)
            if self._quality_check(result):
                return result
        except NonRetryableError:
            pass
        if self._next:
            return await self._next.handle(request)
        return None

class FallbackChain:
    def __init__(self):
        self.head: ModelHandler | None = None
        self.circuit_breaker: dict[str, int] = {}  # model_id —consecutive failures

    async def execute(self, request: dict) -> dict:
        if not self.head:
            raise RuntimeError("Empty fallback chain")
        return await self.head.handle(request)

ハンドラーで3回連続して失敗すると——サーキットブレーカーが一時的にチェーンから外します。クールダウン期間の後、テストリクエストとともに再追加されます。マルチモデルアーキテクチャガイドがルーティング戦略を詳しく解説しています——このパターンは形式化されたチェーン実装を提供します。

これが置き換えるアンチパターン

呼び出し箇所全体にコピペされたネスト型try/exceptフォールバックロジック。一貫性のないチェーンの順序。サーキットブレーカーなし——2番目の位置にある劣化したモデルは、成功することなくすべてのフォールバックにレイテンシを加えます。

パターン6：Template Method——標準化されたエージェントループ

問題

エージェントごとに、ツール呼び出しループが少しずつ異なります。while Trueを使うものもあれば、for i in range(max_iterations)を使うものもあります。ループ制限自体を忘れているものもあります。エージェント間で動作が一貫しません。無期限にループできるエージェントにはコスト暴走のリスクがあります。

パターン

固定されたスケルトンを持つAgentLoopテンプレートメソッド：plan——ツール実行——observe——次の判断。サブクラスはフックメソッドをオーバーライドして独自の動作を実装します。スケルトンは、すべてのエージェントが同じ安全特性（ループ制限、タイムアウト、コスト上限、構造化エラーハンドリング）を継承することを保証します。

class AgentLoop(ABC):
    def __init__(self, max_iterations: int = 15, timeout: float = 120.0, cost_cap: float = 5.00):
        self.max_iterations = max_iterations
        self.timeout = timeout
        self.cost_cap = cost_cap

    async def run(self, task: str) -> dict:
        start = time.time()
        total_cost = 0.0
        for i in range(self.max_iterations):
            if time.time() - start > self.timeout:
                return {"status": "timeout", "partial_result": self._build_partial()}
            if total_cost > self.cost_cap:
                return {"status": "cost_cap_exceeded"}

            plan = await self.plan(task)            # Hook: override
            tool = await self.select_tool(plan)      # Hook: override
            result = await self.execute(tool)        # Hook: override
            total_cost += result.get("cost", 0)

            if await self.should_stop(i, result):    # Hook: override
                return await self.synthesize()

    @abstractmethod
    async def plan(self, task: str) -> dict: ...
    @abstractmethod
    async def select_tool(self, plan: dict) -> str: ...
    @abstractmethod
    async def execute(self, tool: str) -> dict: ...

シングルエージェントガイドがツール呼び出しループの基本を解説しています。このパターンはデザインパターンの視点を提供します：安全性の保証を願望ではなく構造にする形式化されたテンプレート。

これが置き換えるアンチパターン

各エージェントが独自のループを実装します。一貫性のない安全ガード。誰かがmax_iterationsチェックなしで”while True”バージョンをコピーしたために、無限にループできるエージェント。

クイックリファレンス：どのパターンをいつ使うか？

あなたの状況使うパターン
コードベースに3つ以上のモデル文字列がある場合Factory —モデル選択を一元化します
if/else で実装されたプロンプトのA/Bテストがある場合Strategy —プロンプトのバリエーションをカプセル化します
生成コードに結合されたストリーミングコンシューマーがある場合Observer —イベント駆動設計で結合を解消します
すべてのLLM呼び出しの周りに60行のボイラープレートがある場合Decorator —運用上の関心事をレイヤー化します
モデルのフェイルオーバー用にネストされたtry/exceptがある場合Chain of Responsibility —設定可能なフォールバックを提供します
一貫性のないループを持つ複数のエージェントがある場合Template Method —安全ガードで標準化します

コードベースサイズ別の導入順序： 小規模（<5K行、1〜2ユースケース）——DecoratorとFactoryから始めます。中規模（5〜50K行）——StrategyとChain of Responsibilityを追加します。大規模（50K行超、複数エージェント）——ObserverとTemplate Methodを追加します。

6つのパターンはすべて、標準のOpenAI互換SDKで動作します。統一APIエンドポイントとは、Factoryの設定が1つのbase_urlとN個のモデル文字列であることを意味します——N個のベースURL×Mプロバイダーではありません。

FAQ

これらのパターンは単純なAPI呼び出しを過剰設計にしませんか？

コードベースにLLM呼び出しが1つしかなく、2つを超えて増えないなら、その通りです——直接のclient.chat.completions.create()で50行が正解です。コードベースが10以上のLLM呼び出し、3以上のモデルバリアント、本番の信頼性要件に達したとき、これらのパターンのROIは最初のインシデントで顕在化します——設定変更で済むはずだった最初のモデル移行、ループ制限の欠落による最初のコスト暴走、ロールバック手段のない最初のプロンプトリグレッション。

最初に実装すべきパターンはどれですか？

Decoratorです。既存のLLM呼び出しを変更せずにレイヤーを重ねられます。1つのデコレータスタック（リトライ、ロギング、コスト追跡）をすべての呼び出し箇所に適用します。本番の信頼性を即座に向上させます。既存コードのリファクタリングはゼロ。次にFactoryです——次にモデルを切り替える必要があるとき、15ファイルではなく1つの設定値を変更するだけで済みます。

これらのパターンはLangChainやLlamaIndexで動作しますか？

共存します。FactoryとStrategyはLangChainの外でよりきれいに動作します——モデル選択とプロンプト管理のフレームワークロックインを防ぎます。ObserverとTemplate MethodはLangChainエージェントの内部に置けます——ループ構造とストリーミングコンシューマーはフレームワーク統合の恩恵を受けます。これらのパターンはLangChainを置き換えるものではありません。選択したどのフレームワークの周囲にもコードを構造化します。

モデルが異なるプロバイダーAPIの背後にある場合、これらのパターンはどう持ちこたえますか？

パターンは複雑になるどころか、実装がよりシンプルになります。Factory：1つのクライアントインスタンスがすべてのモデルをカバーします——設定は1つのbase_urlとN個のモデル文字列であって、N個のベースURL×Mプロバイダーではありません。Chain of Responsibility：1つの統合ポイントを通じてプロバイダーをまたいだフォールバック。Decorator：すべての呼び出しが同じゲートウェイを流れるため、コスト追跡が一貫します。パターン自体はプロバイダー非依存です。統合エンドポイントは、各パターンが管理しなければならない統合面を減らします——これこそ抽象化の要点です。なぜ単一エンドポイントアーキテクチャが業界の標準になりつつあるのかの広い視点については、AI API集約プラットフォームへの移行の分析が、このトレンドの背後にある運用面とコスト面の要因を解説しています。

GoF以外にLLM固有のパターンはありますか？

あります。Semantic Router——ハードコードされたルールではなく、クエリのセマンティクスでルーティング。Guard——すべてのLLM呼び出しの前後で実行される入出力検証パイプライン。Cache-Aside——API呼び出しの前に埋め込み類似度をチェックするセマンティックキャッシュレイヤー。これらは独自の記事を書く価値があるLLMネイティブなパターンです。ここに挙げた6つは意図的です：ほとんどのエンジニアリングチームはすでにGoFパターンを知っています。それらをLLM APIにマッピングすることで、学習曲線をほぼゼロに減らせます。

デザインパターンは高度さのためではありません。構造化ではなくコピペされたコードによって、8箇所で同じバグを持たないようにするためです。

Decoratorから始めます。Factoryを追加します。次のモデル移行は30秒で完了します——検索と置換の午前と、見逃した呼び出し箇所のデバッグの午後ではありません。

このリファレンスをブックマークしてください。次にリトライロジックを4回目のコピペをしている自分に気づいたとき、どの引き出しを開けばいいか分かるでしょう。スケールしてもコードベースを構造化し続けるための、さらなる本番パターンとLLM APIアーキテクチャガイドをお求めなら、ブログを購読してください。