LLMを業務システムに組み込む実践的な設計パターン

LLMのAPI呼び出しは簡単だ。難しいのは、業務システムとして運用することだ

OpenAIやAnthropicのAPIを叩いてテキストを生成するだけなら、数十行のコードで済む。しかし、それを業務システムとして本番運用するとなると、考えるべきことが桁違いに増える。

私たちはAI・DX・システム開発の受託を通じて、複数のプロジェクトでLLMを業務システムに組み込んできた。その中で繰り返し直面した設計課題がある。レイテンシー、コスト、出力の安定性、エラーハンドリング——APIのドキュメントには書かれていない、本番固有の問題だ。

本記事では、3つの設計パターンと、それぞれのトレードオフを共有する。

パターン1: 同期呼び出し + キャッシュ（低レイテンシー要件）

ユーザーの操作に対してリアルタイムにLLMの応答を返す場合の設計。

from hashlib import sha256
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは{role}です。簡潔に回答してください。"),
    ("human", "{query}"),
])
chain = prompt | llm

_cache: dict[str, str] = {}

async def get_response(role: str, query: str) -> str:
    cache_key = sha256(f"{role}:{query}".encode()).hexdigest()
    if cache_key in _cache:
        return _cache[cache_key]
    result = await chain.ainvoke({"role": role, "query": query})
    _cache[cache_key] = result.content
    return result.content

適するケース: FAQ応答、フォーム入力補助、定型文書生成。同じ入力に同じ出力が返る前提が成り立つ場合。

設計上の注意: temperature=0 でも完全に決定論的ではない。キャッシュヒット率は80-90%程度を目安に設計する。本番ではRedis等のTTL付きキャッシュストアを使い、ドメインに応じて有効期間を設定する。

選ばなかったアプローチ: ストリーミング（SSE）。ユーザー体感は改善するが、キャッシュとの相性が悪い。キャッシュが効く用途では同期呼び出しの方がシンプルで運用コストが低い。

パターン2: 非同期バッチ処理（コスト最適化）

大量のドキュメントを一括処理する場合の設計。請求書の要約生成、レポートの自動分類など。

import asyncio
from dataclasses import dataclass

@dataclass
class Job:
    doc_id: str
    content: str
    result: str | None = None

async def process_batch(jobs: list[Job], concurrency: int = 5) -> list[Job]:
    semaphore = asyncio.Semaphore(concurrency)
    async def process_one(job: Job) -> Job:
        async with semaphore:
            try:
                result = await chain.ainvoke({
                    "role": "文書分類アシスタント",
                    "query": f"以下の文書を分類してください:\n{job.content[:2000]}"
                })
                job.result = result.content
            except Exception as e:
                job.result = f"ERROR: {str(e)}"
        return job
    return await asyncio.gather(*[process_one(j) for j in jobs])

適するケース: 月次レポート生成、データクレンジング、大量メール分類。レイテンシー制約が緩く、スループットが重要な場合。

コスト設計: GPT-4o-miniのバッチAPIで50%削減可能（結果返却に最大24時間）。並行数はレートリミットとの兼ね合い。私たちのプロジェクトでは1,000件/日の処理で月額$30-50程度。

選ばなかったアプローチ: リアルタイム逐次処理。1,000件を逐次処理すると40分以上かかる。Semaphoreで並行制御する方がスループットもAPI効率も良い。

パターン3: Structured Output + バリデーション（出力安定性）

LLMの出力をシステムの入力として使う場合、最も重要なのは出力フォーマットの安定性だ。

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class ClassificationResult(BaseModel):
    category: str = Field(description="分類カテゴリ", enum=["問い合わせ", "クレーム", "要望", "その他"])
    confidence: float = Field(description="確信度 0.0-1.0", ge=0.0, le=1.0)
    summary: str = Field(description="要約（50文字以内）", max_length=50)
    requires_escalation: bool = Field(description="エスカレーションが必要か")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(ClassificationResult)

async def classify_ticket(content: str) -> ClassificationResult | None:
    try:
        return await structured_llm.ainvoke(
            f"以下のサポートチケットを分類してください:\n{content}"
        )
    except Exception:
        return None

適するケース: チケット分類、データ抽出、フォーム自動入力。LLMの出力を後続処理のインプットとして使う場合。

設計上の注意: with_structured_output はOpenAI・Anthropic双方で使える。GPT-4o-miniでの成功率は95-98%程度。残り2-5%はフォールバック（人間分類に回す）が発動する。

選ばなかったアプローチ: プロンプトでJSON指示する方法。with_structured_output 以前はこの方法を使っていたが、パース失敗率が10-15%と高かった。型安全な方法に移行した方が運用コストが低い。

3パターンの使い分け

多くのプロジェクトでは、これらを組み合わせて使う。例えば、チケット分類（パターン3）をバッチ処理（パターン2）で回すケースは多い。

まだ解決できていないこと

モデル更新時の出力変化。 OpenAIがモデルを更新すると、同じプロンプトでも出力が変わることがある。pytest + 回帰テストスイートで変化を自動検知しているが、対応は手動だ。

コストの予測精度。 トークン数は入力テキストの長さに依存するため、月額コストの事前予測が難しい。実績データが3ヶ月分溜まるまではバッファ込みの概算で運用している。

マルチモデル切り替え。 GPT-4o-miniでは精度が出ない一部のタスクでGPT-4oにフォールバックする仕組みを検討中だが、コストが10倍になる。タスク別の閾値設計がまだ確立していない。

APIを叩くのは開発の始まりに過ぎない。本番で生き残るのは、トレードオフを設計に織り込んだシステムだけだ。

関連記事