【実務の泥臭い話】LLM構造化出力（Structured Outputs）の本番スキーママイグレーションと、Pydantic v2によるエラーハンドリング戦略
はじめに
OpenAIやAnthropicが提供する Structured Outputs（構造化出力） は、LLMの返すべきJSONの型を100%保証してくれる強力な機能です。
しかし、これを本番環境で運用し始めると、以下のような**「スキーマ変更の壁」**にぶつかります。
-
ビジネスロジックの変更: 既存のPydanticスキーマに新しい必須フィールドを追加したいが、古いプロンプトや古いモデル（Cache）が古い形式のJSONを返してしまい、APIが
ValidationErrorでクラッシュする。 - モデルの気まぐれ: スキーマを複雑にしすぎると、LLMが型は守っても「中身のバリデーション（例：文字列の長さやフォーマット）」に失敗する。
本記事では、本番環境を止めずにLLMの出力スキーマを安全にアップデート（マイグレーション）するための**「バージョン管理戦略」**と、Pydantic v2の機能を駆使したエラーハンドリングを解説します。
1. アンチパターンと理想のマイグレーションフロー
❌ やりがちなアンチパターン
アプリケーション側のPydanticモデル（スキーマ）を直接書き換えてデプロイする。
👉 結果: デプロイの瞬間に、まだ古いプロンプトで動いているLLMコンテキストや、レポ―ト生成などの非同期タスク（Celery/Cloud Tasks）のキューに残っていたタスクが型エラーで全滅する。
⭕ 理想の戦略：スキーマのバージョン管理（V1/V2並行運用）
LLMに渡すスキーマは、DBのマイグレーションと同様に**「互換性を保ちながら段階的に移行する」**のが鉄則です。
フロー図 (Mermaid)
2. 【実装】Pydantic v2 を使った「型崩れ救済」の実装コード
今回は「ユーザーのプロフィール抽出」を例にします。
-
V1:
nameとageのみ -
V2:
first_nameとlast_nameに分割、さらにemail（必須）を追加
LLMが古い形式、あるいは不完全な形式で出力してきた場合でも、システムをクラッシュさせずにデフォルト値や移行ロジックで救済するコードです。
from typing import Optional, Union
from pydantic import BaseModel, Field, EmailStr, model_validator
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
load_dotenv()
# --- 1. スキーマ定義 ---
class UserProfileV1(BaseModel):
"""古いスキーマ"""
name: str
age: int
class UserProfileV2(BaseModel):
"""新しいスキーマ (本番ロジックが期待する型)"""
first_name: str
last_name: str
age: int
email: EmailStr = Field(default="unknown@example.com") # 必須化されたが救済のためにデフォルト値を設定
# --- 2. 移行期用の「堅牢なラッパースキーマ」 ---
class MigrationUserSchema(BaseModel):
"""
LLMからの出力を受け止めるためのコンテナ。
LLMには最新(V2)の形式で出すように指示するが、V1が来てもパースできるようにする。
"""
data: Union[UserProfileV2, UserProfileV1]
@model_validator(mode="after")
def convert_v1_to_v2(self) -> "MigrationUserSchema":
# もしLLMが古いV1形式でデータを返してきた場合、内部でV2に自動変換する
if isinstance(self.data, UserProfileV1):
names = self.data.name.split(" ", 1)
first_name = names[0]
last_name = names[1] if len(names) > 1 else ""
# V2にマイグレーション
self.data = UserProfileV2(
first_name=first_name,
last_name=last_name,
age=self.data.age,
email="migrated_user@example.com" # 補完ロジック
)
return self
# --- 3. 実行ロジック ---
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 構造化出力をシリアライズ
# ※ 新しいMigrationUserSchemaの形式で出力させる
structured_llm = model.with_structured_output(MigrationUserSchema)
def process_user_data(input_text: str):
try:
# LLMの呼び出し
result = structured_llm.invoke(
f"以下のテキストからユーザー情報を抽出してください。最新のV2形式(first_name, last_name)を意識してください。\nテキスト: {input_text}"
)
# 最終的にビジネスロジックに渡るデータは、100% V2の型であることが保証される
final_profile: UserProfileV2 = result.data
return final_profile
except Exception as e:
# Structured Outputs自体が弾かれた場合の最後の砦（リトライやフォールバック）
print(f"厳密なバリデーションエラー: {e}")
return None
if __name__ == "__main__":
# テスト1: LLMが正しくV2に近い形で思考した場合
text_ok = "私の名前は山田太郎、25歳です。連絡先は taro@example.com です。"
profile = process_user_data(text_ok)
print("成功パターン:", profile)
# テスト2: LLMが「name」として一塊で返してしまった（V1相当）場合でも、convert_v1_to_v2 が救済する
text_old_style = "田中次郎、30歳。"
profile_fallback = process_user_data(text_old_style)
print("救済パターン:", profile_fallback)
3. 実務で導入すべき3つのTips
① Pydanticの Field(default=...) と default_factory の賢い使い分け
LLMに「必須（Required）」としてスキーマを認識させつつ、アプリケーション側でのパースエラーを防ぎたい場合は、スキーマ定義を2つに分けます。
-
LLM用のスキーマ:
email: EmailStr（デフォルト値なし＝JSON Schema上でrequiredになる） -
受取・救済用のスキーマ:
email: Optional[EmailStr] = None（パースエラーを防ぎ、後段の関数でデフォルト値を埋める）
② model_validator(mode="before") で生JSONの段階でクレンジングする
LLMがどうしても {"age": "25歳"} のように文字列で返してしまい、Pydanticの型キャスト（int型への変換）に失敗することがあります。
その場合は、mode="before" を使って、Pydanticのチェックが走る前に正規表現などで数字だけを抽出する「前処理」をモデル内に組み込むと、コントローラー層が汚れません。
③ プロンプトのバージョンとスキーマの固定
with_structured_output に渡すPydanticモデルを変更する際は、必ずプロンプトのバージョン（Gitのコミットハッシュやプロンプト管理ツールのタグ）も同期させてデプロイしてください。片方だけが変わると、LLMのハルシネーション（幻覚）率が跳ね上がります。
まとめ
Structured Outputsは強力ですが、本番環境で「仕様変更」が発生した瞬間、諸刃の剣へと変わります。
- スキーマは一気に変えず、Union型で新旧を受け止める
- Pydanticのバリデーター層で型変換（アダプター）を吸収する
この2点を徹底するだけで、深夜のValidationErrorによるアラート通知に怯える日々から解放されます。AIアプリケーションを「おもちゃ」から「堅牢なシステム」へ進化させましょう。
少しニッチな運用寄りの話でしたが、参考になりましたら LGTM と ストック をお願いします！
「こういうパターンのパースエラーで困っている」などあれば、ぜひコメント欄で教えてください。