LLM I/Oとデータ連携基盤の境界設計｜構造化出力・冪等性・スキーマ駆動・観測可能性

概要

LLMを業務システムに組み込む際には、「LLMの入出力（I/O）」と「データ連携基盤」の境界を設計する必要があります。LLMは確率的にテキストを返すコンポーネントであり、データ基盤は決定的にレコードを扱います。パイプラインを安定して運用するには、この性質の違いを考慮した設計が必要です。

DigitalBaseが社内AI基盤の構築で得た知見をもとに、LLM I/Oの特性を整理したうえで、データ連携基盤との接続に役立つ4つの設計原則（構造化出力・冪等性・スキーマ駆動・観測可能性）を、実装パターンとともに解説します。

LLM I/O の特性

LLMは従来のAPIや関数とは性質が異なります。データ基盤と組み合わせる際には、以下の特性を考慮します。

確率的・非決定論的

同じ入力でも出力が毎回わずかに変動します。temperature=0 を指定しても、完全な再現性は保証されません。レコードの id フィールドのように決定的に同じ値が要求される場面で、LLMの出力をそのまま使うことは避けるべきです。

コンテキスト長の制約

入力できるトークン数には上限があります。近年はモデルによって数万〜100万トークン規模まで拡大していますが、データソースから取得した数万件のレコードを一度に投入できるわけではありません。チャンキング・要約・段階的処理を前提として設計する必要があります。

コストとレイテンシ

クラウド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": ["urgent", "normal", "low"]},
        "summary": {"type": "string", "maxLength": 200},
        "tags": {"type": "array", "items": {"type": "string"}},
        "action_required": {"type": "boolean"},
    },
    "required": ["category", "summary", "action_required"],
}

result = await call_llm(
    model="qwen3:32b",
    prompt=user_text,
    format_schema=schema,
)
parsed = json.loads(result)  # 型が保証された dict

設計上のポイントは以下の3点です。

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

これらの制約は、後段の処理の安定化に役立ちます。

2. 冪等性を確保する

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

def make_idempotency_key(record_id: str, version: str, prompt_hash: str) -> str:
    """同じレコード × 同じプロンプト → 同じキー。"""
    return f"{record_id}:{version}:{prompt_hash[:8]}"

async def process(record: dict):
    key = make_idempotency_key(
        record["id"], "v1", hashlib.sha256(PROMPT.encode()).hexdigest()
    )
    cached = await cache.get(key)
    if cached is not None:
        return cached  # 既に処理済み
    result = await call_llm(...)
    await cache.set(key, result, ttl=86400)
    return result

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

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

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

class TransformLLMConfig(BaseModel):
    name: str
    model: str
    system_prompt: str
    user_prompt_template: str  # Jinja2形式で {{ field }} を埋め込む
    output_schema: dict  # JSON Schema
    mode: str = "map"  # "map" | "filter" | "reduce"
    batch_size: int = 1

この設計であれば、UIから設定を編集するだけでパイプラインを更新でき、コード変更を伴いません。変換ロジックの追加・修正を、エンジニア以外の運用担当者が安全に行える点も利点です。

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

LLM処理では、失敗を検知できるようにする必要があります。最低限、以下の項目をログに残します。

データ連携基盤側の設計

LLMの特性に対応するため、データ連携基盤側にも前処理や検証、実行制御を組み込みます。

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

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

• テキスト：トークナイザを用いて、モデルのコンテキスト長の60〜70%を上限に分割する
• テーブル：行単位／列単位で分割し、スキーマ情報をプロンプトに含める
• PDF・画像：ページ単位またはセクション単位で分割する

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

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

スループット制御

asyncio.Semaphore で同時実行数を制限するのが基本です。クラウドAPIのレート制限やローカルGPUの処理能力に合わせて並列度を調整します。

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

async def bounded_call(record: dict) -> dict:
    async with sem:
        return await call_llm(record)

results = await asyncio.gather(*[bounded_call(r) for r in records])

アーキテクチャ全体像

[ データソース ]
   ↓ (CSV/SQL/REST/SaaS)
[ ソースオペレーター（pydantic検証）]
   ↓ (dict ストリーム)
[ 前処理：チャンキング・正規化 ]
   ↓
[ transform_llm（構造化出力強制 + 冪等キー）]
   ↓ (検証済み dict)
[ ロードオペレーター（DB/API/ファイル）]
   ↓
[ 監視ログ：プロンプト・出力・レイテンシ・エラー ]

アンチパターン集

実装時に避けるべきパターンと、その回避策を整理します。

1. プロンプトをコードに直書きする → 設定として外部に切り出す
2. JSON出力をスキーマなしで扱う → 必ずJSON Schemaで制約する
3. 失敗時の生出力を破棄する → パース失敗時こそ生の出力をログに残す
4. 大量レコードを単発で処理する → バッチ処理または並列度制御を必ず導入する
5. 冪等性を考慮しない → 冪等キーは設計の初期段階から組み込む
6. プロンプトのバージョン管理がない → prompt_hash をデータに残す

まとめ

LLM I/Oとデータ連携基盤を堅牢に接続するための要点は、LLMの確率的・非決定的な性質に、基盤側の決定的な処理で対応することです。

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

これらを設計初期から組み込むことで、LLMを利用した業務データパイプラインの安定運用につなげます。

下記のお悩みなら、ご相談ください

PoC から商用化に進むには、次の 3 点が必要になります。統合型 AI エージェント基盤「DigitalBase」は、 これらを 1 つの製品で提供しています。お気軽にご相談ください。