スキーマの進化と後方互換性
スキーマをバージョン管理し、過去の抽出結果を移行して、移行期間中に並列検証を実行することで、長期稼働する抽出パイプラインの破壊的なスキーマ変更を管理します。
「スキーマの進化と後方互換性」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Engineering Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Engineering Academyコースには全4レッスンが含まれています。
スキーマが時間とともに変化する理由
抽出スキーマは不変ではありません。ビジネス要件は変化し、新しいドキュメントタイプが登場し、最初から取得しておくべきだったフィールドに気づくこともあります。稼働中のパイプラインでスキーマを変更すると、後方互換性の問題が生じます。既存の抽出レコードは古いスキーマを使い、新しいレコードは新しいスキーマを使うためです。この移行を安全に管理するのが、スキーマ進化です。
スキーマのバージョン管理
各スキーマにバージョン番号を割り当て、すべての抽出レコードと一緒に保存してください。スキーマを変更したら、バージョンを上げます。これにより、スキーマのバージョンでレコードを検索したり、古いレコードにマイグレーションを実行したり、バージョンごとに異なる検証ロジックを維持したりできます。すべての出力モデルにschema_versionという文字列フィールドを追加するだけでも十分です。
from pydantic import BaseModel
from typing import Literal
class InvoiceV1(BaseModel):
schema_version: Literal['1.0'] = '1.0'
vendor: str
total_amount: float
class InvoiceV2(BaseModel):
schema_version: Literal['2.0'] = '2.0'
vendor: str
vendor_tax_id: str | None = None # new field
total_amount: float
currency: str = 'USD' # new field with default追加的変更と破壊的変更
追加的変更は安全です。Optionalフィールドやデフォルト値を持つフィールドの追加であれば、古い抽出コードや古いレコードが壊れることはありません。一方、破壊的変更は危険です。フィールド名の変更、型のstringからintへの変更、フィールドの削除は、下流の利用側を壊します。常に追加的変更を優先してください。破壊的変更が避けられない場合は、新しいメジャースキーマバージョンを作成し、管理された手順で移行します。
# Safe: additive change - add optional field
class ProductV2(BaseModel):
name: str
price: float
sku: str | None = None # NEW optional field - backward safe
category: str = 'general' # NEW with default - backward safe
# Risky: breaking change - rename or retype
# class ProductV2(BaseModel):
# product_name: str # RENAMED from name - breaks consumers
# price_cents: int # RETYPED from float - breaks dataデータベースへのスキーマバージョンの保存
各レコードをどのバージョンが生成したか常に把握できるよう、抽出結果テーブルにスキーマバージョンを含めてください。抽出データ全体を保存するjsonbカラムと、スキーマバージョンを保存するschema_versionテキストカラムを用意するのが一般的なパターンです。これにより、バージョンを考慮したクエリを記述し、トラフィックの少ない時間帯に古いレコードを選択的に移行できます。
-- PostgreSQL table design
CREATE TABLE extractions (
doc_id TEXT PRIMARY KEY,
schema_version TEXT NOT NULL,
extracted_data JSONB NOT NULL,
extracted_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_schema_version ON extractions(schema_version);
-- Query old records needing migration
SELECT doc_id, extracted_data
FROM extractions
WHERE schema_version = '1.0'
LIMIT 1000;マイグレーションスクリプトの作成
スキーマのバージョン移行ごとにマイグレーションスクリプトを作成し、古いレコードを読み込み、新しい形式に変換して、新しいバージョンとして書き戻します。失敗してもデータベースが中途半端な移行状態にならないよう、トランザクションを使い、小さなバッチに分けて移行を実行してください。移行の完了を検証するまでは、必ず古いスキーマを利用可能な状態に保ちます。
import asyncpg
import json
async def migrate_v1_to_v2(pool, batch_size=100):
async with pool.acquire() as conn:
rows = await conn.fetch(
'SELECT doc_id, extracted_data FROM extractions WHERE schema_version=$1 LIMIT $2',
'1.0', batch_size
)
for row in rows:
old = row['extracted_data']
new_data = {
'schema_version': '2.0',
'vendor': old['vendor'],
'vendor_tax_id': None, # unknown for old records
'total_amount': old['total_amount'],
'currency': 'USD' # assume USD for old records
}
await conn.execute(
'UPDATE extractions SET extracted_data=$1, schema_version=$2 WHERE doc_id=$3',
json.dumps(new_data), '2.0', row['doc_id']
)移行中の並列検証
スキーマの移行中は、並列検証を実行します。受信したドキュメントの一部を対象に、古いスキーマと新しいスキーマの両方で同時に抽出してください。結果を比較し、新しいスキーマが古いスキーマで取得できていたすべての内容に加えて、新しいフィールドも取得できることを確認します。統計的に有意なサンプルで安定した同等性が確認できるまで、古いスキーマを廃止しないでください。
async def parallel_validate(text: str) -> dict:
v1_result, v2_result = await asyncio.gather(
extract_with_schema(text, InvoiceV1),
extract_with_schema(text, InvoiceV2)
)
discrepancy = (
v1_result.vendor != v2_result.vendor or
abs(v1_result.total_amount - v2_result.total_amount) > 0.01
)
if discrepancy:
log_discrepancy(text, v1_result, v2_result)
return {'v1': v1_result, 'v2': v2_result, 'discrepancy': discrepancy}スキーマ展開のためのフィーチャーフラグ
フィーチャーフラグを使って、パイプラインを古いスキーマから新しいスキーマへ切り替えるタイミングを制御してください。これにより、新しいスキーマをトラフィックの一部に段階的に展開し、エラー率を監視できます。問題が発生した場合も、コードを再デプロイせずに即座にロールバックできます。LaunchDarklyのようなフィーチャーフラグサービスでも、単純なデータベースの1行でも利用できます。
import os
def get_active_schema():
version = os.environ.get('EXTRACTION_SCHEMA_VERSION', '1.0')
schemas = {
'1.0': InvoiceV1,
'2.0': InvoiceV2,
}
return schemas.get(version, InvoiceV1)
async def extract_document(text: str):
SchemaClass = get_active_schema()
return await extract_with_schema(text, SchemaClass)Union型による利用側の互換性
抽出データを読み取る下流の利用側は、複数のスキーマバージョンを適切に処理する必要があります。利用側のコードで判別共用体を使い、schema_versionフィールドに基づいて正しい解析ロジックを選択してください。これは条件分岐のif-elseチェーンを書くより堅牢で、バージョン3が登場した場合にも拡張しやすくなります。
from pydantic import BaseModel
from typing import Union, Annotated
from typing import Literal
def parse_extraction(raw: dict) -> Union[InvoiceV1, InvoiceV2]:
version = raw.get('schema_version', '1.0')
if version == '1.0':
return InvoiceV1(**raw)
elif version == '2.0':
return InvoiceV2(**raw)
else:
raise ValueError(f'Unknown schema version: {version}')デプロイ前のスキーマ変更のテスト
新しいスキーマをデプロイする前に、完全な回帰テストセットに対して実行してください。これは、期待される出力が既知である代表的なドキュメントを集めたものです。古いスキーマと新しいスキーマについて、フィールドごとのF1スコアを比較します。いずれかのフィールドでF1が低下した場合、新しいスキーマの説明がモデルを混乱させたことを意味します。リリースする前に、そのフィールドの説明を修正してください。
def eval_schema_on_test_set(test_cases: list, SchemaClass) -> dict:
field_f1 = {}
for case in test_cases:
result = extract_with_schema(case['text'], SchemaClass)
for field in case['expected']:
expected = case['expected'][field]
actual = getattr(result, field, None)
# Update precision/recall counters
update_metrics(field_f1, field, expected, actual)
return {k: compute_f1(v) for k, v in field_f1.items()}スキーマの非推奨化への対応
新しい抽出でスキーマバージョンが使われなくなったら、そのバージョンを非推奨にできます。非推奨化とは、そのバージョンでの新規レコードの受け入れを停止し、古いレコードは読み取り可能な状態に保ち、古いレコードを移行またはアーカイブする廃止日を設定することです。すべての利用側が解析コードをアップグレードすべき時期を把握できるよう、変更履歴に非推奨化を記録してください。
DEPRECATED_VERSIONS = {'1.0'}
SUNSET_DATE = '2026-09-01'
def warn_if_deprecated(version: str):
if version in DEPRECATED_VERSIONS:
import warnings
warnings.warn(
f'Schema version {version} is deprecated. '
f'It will be removed after {SUNSET_DATE}. '
'Migrate consumers to version 2.0.',
DeprecationWarning,
stacklevel=2
)変更履歴とコミュニケーション
すべてのスキーマ変更には、変更内容、理由、移行手順、想定される影響を説明する変更履歴のエントリを添える必要があります。変更をデプロイする前に、抽出データを利用するすべてのチームへ変更履歴のエントリを共有してください。多くのスキーマ移行の失敗は技術的な問題ではなく、変更が予定されていることを利用側に知らせていなかったことが原因で起こります。
# CHANGELOG.md entry format:
# ## Schema v2.0 (2026-07-01)
# ### Changes
# - ADDED: vendor_tax_id (Optional[str]) - VAT/EIN extracted from header
# - ADDED: currency (str, default='USD') - detected from symbol/code
# ### Migration
# Run: python scripts/migrate_v1_to_v2.py --batch-size=500
# ### Consumers
# - billing-service: update parse_extraction() to handle v2
# - audit-service: query now supports currency filterクイックチェック
抽出パイプラインにおけるスキーマ進化と後方互換性の理解度を確認しましょう。
レッスンのまとめ
このレッスンでは、スキーマのバージョン管理によってすべての抽出レコードと一緒にバージョン識別子を保存し、選択的に移行できること、追加的変更は安全である一方、フィールド名の変更や型の変更には慎重な移行が必要なこと、そして並列検証によって古いスキーマを廃止する前に新しいスキーマを検証できることを学びました。次は、TTFTとTPOTのメトリクスを使ってLLMのレイテンシーを測定します。
よくある質問
「スキーマの進化と後方互換性」レッスンは無料ですか？
はい。「スキーマの進化と後方互換性」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。
「スキーマの進化と後方互換性」で何を学びますか？
スキーマをバージョン管理し、過去の抽出結果を移行して、移行期間中に並列検証を実行することで、長期稼働する抽出パイプラインの破壊的なスキーマ変更を管理します。 ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Engineering Academyを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「スキーマの進化と後方互換性」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Engineering Academyレッスンでコードを書いて実行できますか？
はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。