💨

【LLM基礎#4】壊れないJSON出力で「運用コスト」を下げる(Structured Outputs / Function Callin

に公開

はじめに

#1で「トークン=料金の単位」を押さえて、#3で「出力の揺れ(temperature/top-k/top-p)」も触れました。
次に現場で起こりがちなことをあげていきます。

  • LLMの出力がJSONとして壊れる
  • JSONはパースできるけど、必要なキーが無い/型が違う/値が変
  • 結果、リトライ地獄 → トークン増 → コスト増 → 運用が死ぬ

なので#4のゴールはシンプルに:

「壊れない入出力」を作って、再実行・手直し・人手確認を減らす(=運用コストを下げる)を目標に進めたいと思います


1. “壊れる”って具体的に何が起きる?

例えば、アプリ側がこういうJSONを期待してるとします。

{"id":"INC-1024","priority":"high","summary":"..."}

でもLLMは平気で以下のように出力することがあります。

  • priority が urgent になってる(想定外)
  • id が空
  • 余計なキーが混ざる
  • そもそもJSONの括弧が閉じてない

つまり、
「人間にはそれっぽく見えるけど、システムには使えない」出力が普通に出る。


2. JSON mode と Structured Outputs の違い(ここ重要)

OpenAIには「JSONっぽく出す」方法が複数ありますが、強さが違います。

JSON mode

  • JSONとしては出ることを狙う
  • ただし “スキーマ(型や必須キー)まで守る保証” は弱い
    (= パースはできても中身が壊れる) (OpenAI)

Structured Outputs(JSON Schemaで縛る)

  • JSON Schemaに一致する出力を強制できる仕組み (OpenAI Platform)
  • strict: true を使う例が公式に載ってる (OpenAI Platform)
  • さらに “Function Calling でも Structured Outputs を使える” と明記されてる (OpenAI)

つまり、
JSON mode = 「JSONの形」まで
Structured Outputs = 「JSONの中身(型・必須・enum)」まで
となる


3. 運用者目線での“勝ちパターン”構成

運用で強いのはこの3点セット。

  1. スキーマ(JSON Schema)を用意
  2. LLM出力をバリデーション(通らなければ即失敗扱い)
  3. 失敗時のリトライ戦略(エラー内容を返して再生成)

リトライ戦略は「何でも再試行」じゃなくて、429(Rate Limit)や5xx(サーバ側)だけ指数バックオフみたいに絞るのが基本です。 (OpenAI Platform)


4. コード実験①:Pydanticでスキーマ化 → バリデーションで壊れた出力を弾く

前提(使うライブラリ)

  • Python 3.x
  • pydantic(JSON Schema生成と検証)
  • (API叩くなら)openai

PydanticはモデルからJSON Schemaを生成できます。 (Pydantic)

実験コード(ローカルで動く)

from typing import Literal, Optional
from pydantic import BaseModel, Field, ValidationError
import json

# 期待する“正しい形”をコードで固定する
class Ticket(BaseModel):
    id: str = Field(min_length=1)
    priority: Literal["low", "medium", "high"]
    title: str = Field(min_length=1, max_length=80)
    summary: str = Field(min_length=1, max_length=200)
    next_action: Optional[str] = Field(default=None, max_length=120)

# ✅ 正常ケース
valid = {
    "id": "INC-1024",
    "priority": "high",
    "title": "VPN接続が不安定",
    "summary": "複数ユーザーで断続的に切断。",
    "next_action": "端末ログ収集"
}

# ❌ 壊れたケース(LLMがやりがち)
invalid = {
    "id": "",
    "priority": "urgent",   # enum外
    "title": "",            # 空
    "summary": "x" * 250    # 長すぎ
}

print("=== JSON Schema(抜粋) ===")
schema = Ticket.model_json_schema()
excerpt = {
    "type": "object",
    "properties": {
        "id": schema["properties"]["id"],
        "priority": schema["properties"]["priority"],
        "summary": schema["properties"]["summary"],
    },
    "required": schema["required"],
}
print(json.dumps(excerpt, ensure_ascii=False, indent=2))

print("\n=== valid を検証 ===")
t = Ticket.model_validate(valid)
print(json.dumps(t.model_dump(), ensure_ascii=False, indent=2))

print("\n=== invalid を検証 ===")
try:
    Ticket.model_validate(invalid)
except ValidationError as e:
    # エラーは “どこがどう壊れたか” を機械的に返してくれる
    print(e.errors())

実行結果(例)

JSON Schema(抜粋)

{
  "type": "object",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string",
      "minLength": 1
    },
    "priority": {
      "title": "Priority",
      "enum": [
        "low",
        "medium",
        "high"
      ],
      "type": "string"
    },
    "summary": {
      "title": "Summary",
      "type": "string",
      "minLength": 1,
      "maxLength": 200
    }
  },
  "required": [
    "id",
    "priority",
    "title",
    "summary"
  ]
}

valid は通る

{
  "id": "INC-1024",
  "priority": "high",
  "title": "VPN接続が不安定",
  "summary": "複数ユーザーで断続的に切断。",
  "next_action": "端末ログ収集"
}

invalid は落ちる(=壊れた出力を自動で排除できる)

[
  {'type': 'string_too_short', 'loc': ('id',), ...},
  {'type': 'literal_error', 'loc': ('priority',), ...},
  {'type': 'string_too_short', 'loc': ('title',), ...},
  {'type': 'string_too_long', 'loc': ('summary',), ...}
]

結局は
LLMの出力を「信用しない前提」で、型で殴って落とすのが運用の出発点になると思います。


5. コード実験②:Structured Outputsで“最初から壊れないJSON”を出させる

Structured Outputsは Responses API の text.format で JSON Schema を渡して使えます(公式例あり)。 (OpenAI Platform)

APIサンプル(Python)

import os, json
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Literal, Optional

class Ticket(BaseModel):
    id: str = Field(min_length=1)
    priority: Literal["low", "medium", "high"]
    title: str = Field(min_length=1, max_length=80)
    summary: str = Field(min_length=1, max_length=200)
    next_action: Optional[str] = Field(default=None, max_length=120)

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

schema = Ticket.model_json_schema()

response = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=[
        {"role": "system", "content": "社内ITヘルプデスクの一次受付です。ユーザー文からチケットを作成してください。"},
        {"role": "user", "content": "VPNが昨日から切れたり繋がったりします。Windows更新を入れてからかも。急ぎです。"}
    ],
    text={
        "format": {
            "type": "json_schema",
            "name": "ticket",
            "schema": schema,
            "strict": True
        }
    }
)

print(response.output_text)          # 公式例でも output_text を使っている :contentReference[oaicite:7]{index=7}
ticket = json.loads(response.output_text)
print(ticket)

6. “コスト改善”にどう効くの?(運用者視点のまとめ)

Structured Outputs + バリデーションは、地味に金を救います。

  • 再試行が減る:壊れたJSONで無限リトライしない
  • 人手修正が減る:ログ確認・手直しの時間が減る
  • 出力を短くできる:説明文を出させず、必要フィールドだけ返させる
  • トークンを計測して改善できる:Responses APIは usage を返す (OpenAI Platform)
  • 出力上限を強制できる:max_output_tokens がある (OpenAI Platform)
  • 長すぎる入力は挙動を選べる:truncation=auto/disabled (OpenAI Platform)
  • キャッシュ運用の入口もある:prompt_cache_key / prompt_cache_retention (OpenAI Platform)
  • 処理ティアも選べる:service_tier(default/flex等) (OpenAI Platform)

次回(#5の予告)

#5は「RAGの前にやるべきデータ設計」の学習をしたいと思います。
RAGは検索が外れたら全部外れるので、チャンク設計・メタデータ・再ランキングの土台が先に必要になるので。

次はいよいよ知識を食わせて正確にするフェーズに取り掛かりたいと思います。

Discussion