LLM I/Oとデータ連携基盤の相性のいい作り方

要約

LLM を業務に組み込むとき、最も詰まるのは「LLMの入出力（I/O）」と「データ連携基盤」の境界設計です。LLMは確率的にテキストを返すコンポーネントで、データ基盤は決定的にレコードを扱います。この性質の違いを設計で吸収しなければ、運用が壊れます。

本記事では、LLM I/O の特性を整理したうえで、データ連携基盤との相性が良い設計パターン（構造化出力・冪等性・スキーマ駆動・観測可能性）を、実装パターンとともに解説します。

---

LLM I/O の特性

LLMは従来のAPIや関数とは性質が違います。データ基盤と組み合わせる前に、まずこの特性を理解しておく必要があります。

確率的・非決定論的

同じ入力でも、出力が毎回少し違います。temperature=0 でも完全な再現性は保証されません。レコードの id フィールドのように決定的に同じ値が必要な場面では、LLM出力をそのまま使ってはいけません。

コンテキスト長の制約

入力できるトークン数に上限があります（4K〜200K程度）。データソースから取ってきた1万件のレコードを丸投げすることはできません。チャンキング・要約・段階的処理を前提とする必要があります。

コストとレイテンシ

クラウドAPIなら従量課金、ローカルLLMなら推論時間が支配的なコストです。1レコードずつ呼ぶか、N件まとめて呼ぶかで、コストもレイテンシも桁が変わります。

出力の自由度

人間相手なら自由テキストでよくても、データ基盤に書き戻すには 構造化された型 が必要です。JSON Schema を強制しないと、後段のパース処理が壊れやすくなります。

---

相性のいい設計パターン

データ連携基盤とLLMを組み合わせる際の、4つの基本原則を順に見ていきます。

1. 構造化出力を強制する

LLMの出力は必ず JSON Schema で型を縛ります。Ollama / vLLM / OpenAI / Anthropic / Gemini いずれも format や response_format で構造化出力を指定できます。

schema = {
    "type": "object",
    "properties": {
        "category": {"type": "string",

ポイントは以下です。

• enum で値を限定する（カテゴリ分類などはほぼ全ケースで使える）
• maxLength でフィールド長を縛る（DBのカラム長と揃える）
• required で必須フィールドを宣言（None回避）

これだけで後段の処理が劇的に安定します。

2. 冪等性を確保する

LLMが確率的である以上、同じレコードに対して複数回呼ばれることを前提にします。重複呼び出しでDBが汚れないよう、冪等キー を設計します。

def make_idempotency_key(record_id: str, version: str, prompt_hash: str) -> str:
    """同じレコード × 同じプロンプト → 同じキー。"""

プロンプトのバージョンが変わったら自動的にキャッシュが切り替わるよう、prompt_hash をキーに含めるのが定石です。

3. スキーマ駆動でデータ変換を定義する

「どういう入力を、どういうプロンプトで、どういう出力スキーマに変換するか」をコードではなく データ（設定） として持つと、変更が容易になります。

class TransformLLMConfig(BaseModel):
    name: str
    model: str
    system_prompt: str
    user_prompt_template: str  # Jinja2形式で {{ field }} を埋め込む
    output_schema:

この設計なら、UIから設定を編集するだけでパイプラインを更新でき、コード変更が要りません。

4. 観測可能性を組み込む

LLM処理は「失敗したのに気付かない」ことが起こりやすい領域です。以下を必ずログに残します。

---

データ連携基盤側の設計

LLM側の特性を受け止めるため、基盤側にも対応する仕掛けが必要です。

入力境界：チャンキングと前処理

LLMに渡す前に、データを適切な粒度に分割します。

• テキスト：トークナイザを使って、モデルのコンテキスト長の60〜70%を上限に分割
• テーブル：行単位 / 列単位で分割。スキーマ情報をプロンプトに含める
• PDF/画像：ページ単位 or セクション単位

出力境界：バリデーション → 永続化

LLMの出力は pydantic で受け、検証通過したものだけをDBに書きます。スキーマ違反のレコードは Dead Letter Queue に逃がし、人間がレビューする運用にします。

スループット制御

asyncio.Semaphore で同時実行数を絞るのが基本です。

sem = asyncio.Semaphore(8)  # 同時8並列

async def bounded_call(record: dict) -> dict:
    async 

---

アーキテクチャ全体像

[ データソース ]
   ↓ (CSV/SQL/REST/SaaS)
[ ソースオペレーター（pydantic検証）]
   ↓ (dict ストリーム)
[ 前処理：チャンキング・正規化 ]   ↓
   ↓

---

アンチパターン集

1. プロンプトをコードに直書き → 設定として外に出す
2. JSON出力をスキーマなしで扱う → 必ず JSON Schema で縛る
3. 失敗時の生出力を捨てる → パース失敗時こそ生の出力をログに残す
4. 大量レコードを単発で叩く → バッチ処理 or 並列度制御を必ず入れる
5. 冪等性を考えない → 冪等キーは設計初期から仕込む
6. プロンプトのバージョン管理がない → prompt_hash をデータに残す

---

まとめ

LLM I/O とデータ連携基盤を綺麗に繋ぐコツは、LLMの確率的・非決定的な性質を、基盤側の決定的な処理で包む ことです。

1. 構造化出力：JSON Schema で型を縛る
2. 冪等性：プロンプトハッシュ込みのキーで重複処理を防ぐ
3. スキーマ駆動：プロンプトと変換ルールを設定として外出し
4. 観測可能性：プロンプト・出力・レイテンシ・失敗を全て記録