LLM Structured Output 実践ガイド — JSON Mode、Tool Use、Pydanticスキーマ検証

はじめに
プロバイダー別Structured Output比較
Pydanticによるスキーマ検証の自動化
- 基本パターン
- リトライパターン（Self-Healing）
LiteLLMによるプロバイダー統合
Instructorライブラリの活用
プロダクションパイプラインの構築
- FastAPI + Structured Output
- バッチ処理パイプライン
まとめ

はじめに

LLMの出力をプログラム的に処理するには、**構造化された形式（JSON、XMLなど）**が必須です。プロンプトに「JSONで応答して」と入れるだけでは不十分です — スキーマの不一致、フィールドの欠落、型の誤りなど、さまざまな問題が発生します。

この記事では、主要LLMプロバイダーのStructured Output機能を比較し、プロダクションで安定的に使用する方法を解説します。

プロバイダー別Structured Output比較

OpenAI: response_format + Structured Outputs

from openai import OpenAI
from pydantic import BaseModel
from typing import List, Optional

client = OpenAI()

# 方法1: JSON Mode（基本）
response = client.chat.completions.create(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "応答をJSONで返してください。"},
        {"role": "user", "content": "ソウルの有名なレストランを3つ推薦して"}
    ],
    response_format={"type": "json_object"}
)
# JSONは保証されるが、スキーマは保証されない

# 方法2: Structured Outputs（スキーマ保証）
class Restaurant(BaseModel):
    name: str
    cuisine: str
    price_range: str
    rating: float
    address: str

class RestaurantList(BaseModel):
    restaurants: List[Restaurant]
    total_count: int

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "ソウルの有名なレストランを推薦してください。"},
        {"role": "user", "content": "韓国料理のレストラン3つ"}
    ],
    response_format=RestaurantList
)

result = response.choices[0].message.parsed
print(result.restaurants[0].name)  # 型安全！

Anthropic: Tool Useを使ったStructured Output

import anthropic
from typing import List

client = anthropic.Anthropic()

# AnthropicはTool Useを活用したStructured Output
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=[
        {
            "name": "extract_restaurants",
            "description": "レストラン情報を構造化された形式で抽出",
            "input_schema": {
                "type": "object",
                "properties": {
                    "restaurants": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "cuisine": {"type": "string"},
                                "price_range": {
                                    "type": "string",
                                    "enum": ["$", "$$", "$$$", "$$$$"]
                                },
                                "rating": {"type": "number"},
                                "address": {"type": "string"}
                            },
                            "required": ["name", "cuisine", "price_range"]
                        }
                    },
                    "total_count": {"type": "integer"}
                },
                "required": ["restaurants", "total_count"]
            }
        }
    ],
    tool_choice={"type": "tool", "name": "extract_restaurants"},
    messages=[
        {"role": "user", "content": "ソウルの韓国料理レストラン3つ推薦して"}
    ]
)

# Tool Useの結果から構造化データを抽出
tool_use = next(
    block for block in response.content
    if block.type == "tool_use"
)
restaurants = tool_use.input["restaurants"]

Google Gemini: responseSchema

import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")

model = genai.GenerativeModel(
    "gemini-2.0-flash",
    generation_config=genai.GenerationConfig(
        response_mime_type="application/json",
        response_schema={
            "type": "object",
            "properties": {
                "restaurants": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "cuisine": {"type": "string"},
                            "rating": {"type": "number"}
                        }
                    }
                }
            }
        }
    )
)

response = model.generate_content("ソウルの韓国料理レストラン3つ")
import json
data = json.loads(response.text)

Pydanticによるスキーマ検証の自動化

基本パターン

from pydantic import BaseModel, Field, validator
from typing import List, Optional, Literal
from enum import Enum
import json

class PriceRange(str, Enum):
    CHEAP = "$"
    MODERATE = "$$"
    EXPENSIVE = "$$$"
    VERY_EXPENSIVE = "$$$$"

class Restaurant(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    cuisine: str = Field(..., description="料理の種類")
    price_range: PriceRange
    rating: float = Field(..., ge=0.0, le=5.0)
    address: Optional[str] = None
    tags: List[str] = Field(default_factory=list, max_length=10)

    @validator('rating')
    def round_rating(cls, v):
        return round(v, 1)

class RestaurantResponse(BaseModel):
    restaurants: List[Restaurant] = Field(..., min_length=1, max_length=20)
    query: str
    total_count: int

# LLMレスポンスのパース + 検証
def parse_llm_response(raw_json: str) -> RestaurantResponse:
    """LLMレスポンスをパースしてPydanticで検証"""
    try:
        data = json.loads(raw_json)
        return RestaurantResponse(**data)
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON: {e}")
    except Exception as e:
        raise ValueError(f"Schema validation failed: {e}")

リトライパターン（Self-Healing）

from tenacity import retry, stop_after_attempt, retry_if_exception_type

class StructuredOutputParser:
    def __init__(self, client, model: str, schema: type[BaseModel]):
        self.client = client
        self.model = model
        self.schema = schema

    @retry(
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type(ValueError)
    )
    def parse(self, prompt: str) -> BaseModel:
        """スキーマ検証失敗時にエラーメッセージを含めてリトライ"""
        schema_json = self.schema.model_json_schema()

        messages = [
            {
                "role": "system",
                "content": f"以下のJSONスキーマに従って応答してください:\n{json.dumps(schema_json, indent=2)}"
            },
            {"role": "user", "content": prompt}
        ]

        # 前回の試行でエラーがあれば含める
        if hasattr(self, '_last_error'):
            messages.append({
                "role": "user",
                "content": f"前回の応答でエラーが発生しました: {self._last_error}\n正しいJSONで再度応答してください。"
            })

        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            response_format={"type": "json_object"}
        )

        raw = response.choices[0].message.content
        try:
            data = json.loads(raw)
            result = self.schema(**data)
            if hasattr(self, '_last_error'):
                del self._last_error
            return result
        except Exception as e:
            self._last_error = str(e)
            raise ValueError(str(e))

# 使用方法
parser = StructuredOutputParser(client, "gpt-4o", RestaurantResponse)
result = parser.parse("ソウルの韓国料理レストラン3つ推薦")

LiteLLMによるプロバイダー統合

import litellm
from pydantic import BaseModel

class ExtractedInfo(BaseModel):
    summary: str
    key_points: list[str]
    sentiment: str
    confidence: float

# OpenAI
response = litellm.completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Kubernetes 1.35リリースを要約して"}],
    response_format=ExtractedInfo
)

# Anthropic（自動的にTool Useに変換）
response = litellm.completion(
    model="claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "Kubernetes 1.35リリースを要約して"}],
    response_format=ExtractedInfo
)

# Gemini
response = litellm.completion(
    model="gemini/gemini-2.0-flash",
    messages=[{"role": "user", "content": "Kubernetes 1.35リリースを要約して"}],
    response_format=ExtractedInfo
)

# 同じコードで3つのプロバイダーを使用可能！

Instructorライブラリの活用

# pip install instructor
import instructor
from openai import OpenAI
from pydantic import BaseModel
from typing import List

client = instructor.from_openai(OpenAI())

class Step(BaseModel):
    explanation: str
    output: str

class MathSolution(BaseModel):
    steps: List[Step]
    final_answer: str
    confidence: float

# Pydanticモデルを直接response_modelとして使用
solution = client.chat.completions.create(
    model="gpt-4o",
    response_model=MathSolution,
    messages=[
        {"role": "user", "content": "2x + 5 = 15を解いて"}
    ],
    max_retries=3  # 自動リトライ
)

print(solution.steps[0].explanation)
print(f"答え: {solution.final_answer}")

# Anthropicも同様にサポート
import anthropic

anthropic_client = instructor.from_anthropic(anthropic.Anthropic())

solution = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    response_model=MathSolution,
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "3x - 7 = 20を解いて"}
    ]
)

プロダクションパイプラインの構築

FastAPI + Structured Output

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import instructor
from openai import OpenAI

app = FastAPI()
client = instructor.from_openai(OpenAI())

class ProductReview(BaseModel):
    sentiment: str  # positive, negative, neutral
    score: float
    key_phrases: List[str]
    summary: str
    language: str

class ReviewRequest(BaseModel):
    text: str
    model: str = "gpt-4o-mini"

@app.post("/analyze", response_model=ProductReview)
async def analyze_review(request: ReviewRequest):
    try:
        result = client.chat.completions.create(
            model=request.model,
            response_model=ProductReview,
            messages=[
                {
                    "role": "system",
                    "content": "製品レビューを分析してください。"
                },
                {"role": "user", "content": request.text}
            ],
            max_retries=2
        )
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

バッチ処理パイプライン

import asyncio
from typing import List
from openai import AsyncOpenAI
import instructor

async_client = instructor.from_openai(AsyncOpenAI())

class ExtractedEntity(BaseModel):
    name: str
    entity_type: str
    confidence: float

class EntityExtractionResult(BaseModel):
    entities: List[ExtractedEntity]
    text_length: int

async def extract_entities(text: str) -> EntityExtractionResult:
    return await async_client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=EntityExtractionResult,
        messages=[
            {"role": "system", "content": "テキストからエンティティを抽出してください。"},
            {"role": "user", "content": text}
        ]
    )

async def batch_extract(texts: List[str], concurrency: int = 5):
    """同時実行数を制限してバッチ処理"""
    semaphore = asyncio.Semaphore(concurrency)

    async def limited_extract(text):
        async with semaphore:
            return await extract_entities(text)

    tasks = [limited_extract(text) for text in texts]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]

    print(f"成功: {len(successes)}, 失敗: {len(failures)}")
    return successes

# 実行
texts = ["ソウルに新しいAIスタートアップが...", "サムスン電子が半導体...", ...]
results = asyncio.run(batch_extract(texts))

まとめ

Structured Outputは、LLMをプロダクションシステムに統合する上で核心となる技術です：

OpenAI: response_format + Structured Outputsでスキーマ100%保証
Anthropic: Tool Useを活用した間接的な方式だが安定的
Instructor/LiteLLM: プロバイダー統合によるコード再利用
Pydantic: スキーマ定義 + 検証の標準
リトライパターン: Self-healingによる安定性確保

クイズ（6問）

Q1. OpenAIのJSON ModeとStructured Outputsの違いは？ JSON Modeは有効なJSONのみを保証、Structured Outputsは指定したスキーマまで保証

Q2. AnthropicでStructured Outputを実装する方式は？ Tool Use（Function Calling）を活用し、input_schemaで構造化された出力を受け取る

Q3. PydanticのField(ge=0.0, le=5.0)は何を意味するか？ 値が0.0以上5.0以下でなければならないという検証条件

Q4. instructorライブラリのmax_retries機能とは？ スキーマ検証失敗時に自動的にリトライして正しい形式を取得する

Q5. バッチ処理におけるasyncio.Semaphoreの役割は？ 同時API呼び出し数を制限してrate limitの超過を防止

Q6. LiteLLMを使用する最大の利点は？ 同じコードでOpenAI、Anthropic、Geminiなど複数のプロバイダーを切り替え可能