Pydanticを極めてAIシステムのデータ契約を設計する

高度なPydanticを使うべき理由

• LLMレスポンスやETL中間データをdictのまま扱うと、型崩れを実行時まで検知できない。
• バリデーションロジックが散在し、変更時に一貫性を保てない。
• ジェネリックなRAGパイプラインやFeature Storeでは、スキーマを再利用する仕組みが必要。

Pydantic v2はBaseModelだけでなく、TypeAdapter, field_validator, dataclass_transform, BaseSettings, GenericModelなど多彩な機能を備えています。ここでは、AIシステム向けにそれらを組み合わせるパターンを解説します。

TypeAdapterでLLMレスポンスを即型付け

from pydantic import BaseModel, TypeAdapter

class LLMAnswer(BaseModel):
    summary: str
    tags: list[str]
    score: float

adapter = TypeAdapter(LLMAnswer)

raw = llm_client.generate(...)
answer = adapter.validate_python(raw)

TypeAdapterはBaseModelをインスタンス化せずに単体でバリデーションを行うユーティリティ。LLM出力をそのままvalidate_pythonに渡せば、フィールド不足や型不一致を即座に検知できます。

field_validator と model_validator

class SearchQuery(BaseModel):
    text: str
    limit: int = 10

    @field_validator('limit')
    @classmethod
    def validate_limit(cls, v: int) -> int:
        if not 1 <= v <= 100:
            raise ValueError('limit must be in [1,100]')
        return v

    @model_validator(mode='after')
    def check_text_and_limit(self):
        if self.text.strip() == '' and self.limit != 1:
            raise ValueError('empty query must have limit 1')
        return self

field_validatorは単一フィールド、model_validatorはモデル全体を検証する。LLMチェーンや検索APIのガードレールに最適。

GenericModelでRAGアーティファクトを統一

from pydantic import BaseModel, Field
from pydantic.generics import GenericModel
from typing import Generic, TypeVar

T = TypeVar('T')  # ペイロード

class Artifact(GenericModel, Generic[T]):
    id: str
    payload: T
    metadata: dict[str, str] = Field(default_factory=dict)

class Passage(BaseModel):
    text: str
    source: str

class Embedding(BaseModel):
    vector: list[float]
    dimension: int

PassageArtifact = Artifact[Passage]
EmbeddingArtifact = Artifact[Embedding]

Genericモデルを使えば、Artifact[Passage]とArtifact[Embedding]を同じ処理パイプラインで扱えます。mypyもpayloadの型を正しく推論。

dataclass_transformでTyped DSLを作る

dataclass_transform = pydantic.dataclasses.dataclass

@dataclass_transform
class Prompt(BaseModel):
    system: str
    user: str

@Prompt.model_validator(mode='after')
def ensure_tokens(prompt: Prompt) -> Prompt:
    if count_tokens(prompt.user) > 4096:
        raise ValueError('too long')
    return prompt

Pydantic v2のdataclassesはdataclass_transform互換なので、mypyにとってもdataclassとして扱われます。

BaseSettingsで階層設定

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix='APP_')

    openai_api_key: str
    redis_url: AnyUrl
    tracing_enabled: bool = False

settings = Settings()  # envから読み込み

env_prefixで環境変数プレフィックスを強制し、設定値の抜け漏れを起動時に検知。Literalで許容値を限定することも可能。

RootModelで非構造データをラップ

class EmbeddingList(RootModel[list[list[float]]]):
    def cosine(self, idx: int, other_idx: int) -> float:
        ...

embeddings = EmbeddingList.model_validate(raw_vectors)

RootModelを使うと、リストや辞書にもメソッドを生やせます。ベクトル集合やスコア配列を型安全に扱える。

field_serializer でAPI出力を整形

class Report(BaseModel):
    created_at: datetime
    score: Decimal

    @field_serializer('created_at', when_used='json')
    def serialize_time(self, dt: datetime) -> str:
        return dt.strftime('%Y-%m-%dT%H:%M:%SZ')

LLMやREST APIの外部公開時にフォーマットを統一できる。

computed_field で派生データを自動計算

class Document(BaseModel):
    tokens: list[str]

    @computed_field(return_type=int)
    def token_count(self) -> int:
        return len(self.tokens)

テンプレートやUIに渡すときに、計算済みフィールドを含められる。

Annotated + BeforeValidator で軽量バリデータ

from typing_extensions import Annotated

PositiveFloat = Annotated[float, Field(gt=0)]

class LLMScore(BaseModel):
    score: PositiveFloat

小さな制約は Annotatedで再利用可能。

TypeAdapter で辞書→TypedDictを検証

class Metric(TypedDict):
    name: str
    value: float

metric_adapter = TypeAdapter(Metric)
metric = metric_adapter.validate_python(raw_metric)

TypedDictにはTypeAdapterを使う。ETL途中の生データを最小コストで検証できる。

カスタムエラーとユーザーフレンドリーなレスポンス

class CustomError(PydanticCustomError):
    def __init__(self, code: str, msg: str) -> None:
        super().__init__('custom_error', msg)
        self.extra = {'code': code}

class Query(BaseModel):
    q: str

    @field_validator('q')
    @classmethod
    def not_empty(cls, v: str) -> str:
        if not v:
            raise CustomError('EMPTY_QUERY', 'query must not be empty')
        return v

API層でエラーコードをそのままHTTPレスポンスに流用できる。

まとめ

• Pydantic v2は単なるバリデータに留まらず、DSLやジェネリックモデル、設定管理まで網羅するツールキット。
• AIシステムではTypeAdapterでLLMレスポンスを即時検証し、GenericModelでRAGやFeature Storeの契約を共通化する。
• dataclass_transformやカスタムエラーを活用すれば、型とバリデーションを統合した堅牢なデータレイヤーが構築できる。

Pydanticを中心に据えたデータ契約を敷き、その上でLLM/ETL/設定管理を組み立てることで、Pythonでも型安全で可観測なAI基盤が実現できます。