Antigravity × Python Structured Output 実践ガイド — Pydantic と Google Gen AI SDK で作る型安全な AI データ抽出パイプライン

「モデルに JSON で返してと頼んだのに、なぜかマークダウンのコードブロックで返ってきた」——こういう経験、一度はあるのではないでしょうか。

Antigravity で Python のエージェントシステムを開発していると、LLM の出力を後続の処理に渡す場面が必ず出てきます。そのたびに正規表現で JSON を切り出したり、json.loads() のエラーを握りつぶしたりしていると、コードは動いているように見えても、本番で静かに壊れていきます。

Google Gen AI SDK v1.0 には Structured Output という機能があります。Pydantic モデルをそのままレスポンスのスキーマとして渡すと、モデルがそのスキーマに準拠した JSON だけを生成するよう制約をかけられる機能です。ただ、ドキュメントのサンプルコードを動かせるようになるのと、これを本番パイプラインで安定稼働させるのとでは、学ぶべきことが大きく違います。

ここでは私が複数のデータ抽出プロジェクトで試行錯誤した経験をもとに、Antigravity 環境での Structured Output 実装を体系的に解説します。

Structured Output の仕組みと、単純な JSON モードとの決定的な違い

Google Gen AI SDK v1.0 で Structured Output を使う場合、大きく2つのアプローチがあります。response_mime_type を "application/json" に設定する「JSON モード」と、response_schema に Pydantic モデルを渡す「Structured Output モード」です。

見た目は似ていますが、内部動作がまったく異なります。

JSON モードはモデルに「JSON を出力するよう努力してください」という指示を追加するだけです。モデルが解釈した「JSON っぽいもの」が返ってくるため、フィールドの過不足・型の違い・追加フィールドなどが頻繁に発生します。

一方の Structured Output は、モデルの出力トークンをスキーマに準拠したものに制限する「制約デコーディング」と呼ばれる技術が使われています。モデルが生成できるトークンの選択肢を、スキーマに合致するものだけに絞り込むため、出力の構造が保証されます。

# config.py — Structured Output の基本セットアップ
from google import genai
from google.genai import types
from pydantic import BaseModel, Field
from typing import Literal, Optional
import os
 
client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
 
# Pydantic モデルの定義
class InvoiceItem(BaseModel):
    name: str = Field(description="商品またはサービスの名称")
    quantity: int = Field(ge=1, description="数量（1以上）")
    unit_price: float = Field(ge=0, description="単価（税抜き）")
 
class Invoice(BaseModel):
    invoice_number: str = Field(description="請求書番号（例: INV-2026-001）")
    vendor_name: str = Field(description="請求元の会社名")
    total_amount: float = Field(ge=0, description="合計金額（税込み）")
    items: list[InvoiceItem] = Field(description="明細リスト")
    currency: Literal["JPY", "USD", "EUR"] = Field(default="JPY")
 
# Structured Output を使った呼び出し
def extract_invoice(document_text: str) -> Invoice:
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=f"以下の請求書テキストから情報を抽出してください:\n\n{document_text}",
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=Invoice,  # Pydantic モデルを直接渡す
        ),
    )
    # response.parsed は自動的に Pydantic モデルのインスタンスになる
    return response.parsed

ここで重要なのは response.parsed です。response.text ではなく response.parsed を使うと、SDK が自動的に JSON をパースして Pydantic モデルのインスタンスを返してくれます。型補完も効くため、Antigravity 上での開発体験が大幅に向上します。

Pydantic スキーマ設計の実践パターン

Field の description が品質を左右する理由

Structured Output を使い始めたとき、最初に直面するのが「スキーマは正しいのに、値の内容がおかしい」という問題です。スキーマは守られているのに、vendor_name に住所が入っていたり、total_amount に小計が入っていたりするケースです。

原因のほとんどは Field(description=...) の不足です。モデルはフィールド名と description を手がかりに「何を入れるべきか」を判断します。フィールド名だけでは曖昧な場合、モデルが独自に解釈してしまいます。

from datetime import date
 
# 悪い例：description が不足していて品質が安定しない
class ContractBad(BaseModel):
    party_a: str
    party_b: str
    amount: float
    date: str
 
# 良い例：description で意図を明確にしたスキーマ
class ContractGood(BaseModel):
    party_a: str = Field(
        description="契約における甲の名称（法人名または個人名）"
    )
    party_b: str = Field(
        description="契約における乙の名称（法人名または個人名）"
    )
    amount: float = Field(
        ge=0,
        description="契約金額（税込み・円）。記載がない場合は 0.0 を設定すること"
    )
    contract_date: date = Field(
        description="契約締結日（YYYY-MM-DD 形式で返すこと）"
    )

date 型を使うと、SDK が自動的に文字列を datetime.date オブジェクトに変換してくれます。description に「YYYY-MM-DD 形式で返すこと」と明示することで、モデルの出力が安定します。

Enum と Literal で選択肢を制限する

自由テキストで返ってくる可能性があるフィールドには、Literal または Enum で選択肢を制限するのが非常に効果的です。

from enum import Enum
 
class ContractStatus(str, Enum):
    DRAFT = "draft"
    ACTIVE = "active"
    EXPIRED = "expired"
    TERMINATED = "terminated"
 
class RiskLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
class ContractAnalysis(BaseModel):
    status: ContractStatus = Field(
        description="契約の現在のステータス。有効期限が不明な場合は active とすること"
    )
    risk_level: RiskLevel = Field(
        description="契約内容のリスク評価。免責条項が広範な場合は high または critical"
    )
    auto_renewal: bool = Field(
        description="自動更新条項が含まれているかどうか"
    )
    penalty_exists: bool = Field(
        description="違約金・損害賠償条項が含まれているかどうか"
    )
    summary: str = Field(
        max_length=500,
        description="契約内容の要点を200文字以内で要約したもの"
    )

str, Enum の継承は重要で、これにより response.parsed.status が文字列として扱えると同時に、ContractStatus.ACTIVE との比較も安全に行えます。

Optional フィールドと Union 型の扱い

すべてのフィールドが必ず抽出できるとは限りません。特に非構造化テキストからの情報抽出では、Optional なフィールドの設計が重要です。

class PersonalInfo(BaseModel):
    full_name: str = Field(description="氏名（フルネーム）")
    email: Optional[str] = Field(
        default=None,
        description="メールアドレス。記載がない場合は null を設定すること"
    )
    phone: Optional[str] = Field(
        default=None,
        description="電話番号（ハイフン区切り）。記載がない場合は null を設定すること"
    )
    department: Optional[str] = Field(
        default=None,
        description="所属部署。個人の場合は null を設定すること"
    )

Optional フィールドの description に「記載がない場合は null を設定すること」と明示するのは必須です。これを書かないと、モデルが推測で値を生成してしまいます。

本番で直面するよくある実装の落とし穴

実際にプロダクションで Structured Output を運用していて、繰り返し直面したポイントを3つお伝えします。

落とし穴1: response.parsed が None を返すケース

response.parsed が None になるケースがいくつかあります。最も多いのは、入力テキストがスキーマのフィールドに対応する情報をまったく含まない場合です。もう一つは、モデルが制約デコーディングを適用できないほど複雑なスキーマを渡したときです。

from pydantic import BaseModel, ValidationError
import logging
 
logger = logging.getLogger(__name__)
 
def safe_extract(document_text: str, schema_class: type[BaseModel]) -> BaseModel | None:
    """エラーハンドリング付きの Structured Output 呼び出し"""
    try:
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=f"以下のテキストから情報を抽出してください:\n\n{document_text}",
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=schema_class,
            ),
        )
 
        if response.parsed is not None:
            return response.parsed
 
        # フォールバック: response.text から手動パース
        if response.text:
            try:
                import json
                raw_data = json.loads(response.text)
                return schema_class.model_validate(raw_data)
            except (json.JSONDecodeError, ValidationError) as e:
                logger.warning(f"フォールバックパース失敗: {e}")
 
        logger.error("モデルが空のレスポンスを返しました")
        return None
 
    except Exception as e:
        logger.error(f"構造化抽出に失敗: {type(e).__name__}: {e}")
        raise RuntimeError(f"構造化抽出に失敗しました: {e}") from e

落とし穴2: ネストが深いスキーマでの精度低下

スキーマのネストが3階層以上になると、モデルの精度が落ちることがあります。私の経験では、深い階層は意図的に分割するか、「2フェーズ抽出」に切り替えるアプローチが安定します。

def extract_in_phases(document: str) -> tuple[BaseModel, BaseModel]:
    """大きなドキュメントを段階的に抽出する 2 フェーズアプローチ"""
 
    # Phase 1: 概要情報を抽出
    top_level = safe_extract(document, TopLevelSchema)
 
    if top_level is None:
        raise ValueError("Phase 1 の抽出に失敗しました")
 
    # Phase 2: 詳細情報を抽出（前の結果をコンテキストに含める）
    detail_prompt = (
        f"以下のドキュメントから詳細情報を抽出してください。\n"
        f"概要情報として以下が取得済みです: {top_level.model_dump_json()}\n\n"
        f"ドキュメント:\n{document}"
    )
    detail_level = safe_extract(detail_prompt, DetailLevelSchema)
 
    return top_level, detail_level

落とし穴3: 長文ドキュメントでのトークン上限と精度低下

Gemini 2.5 Flash のコンテキストウィンドウは大きいですが、長文では注意が分散して精度が下がります。チャンク分割にはオーバーラップを持たせることで、文脈の断絶を防げます。

def extract_from_long_document(
    document: str,
    schema_class: type[BaseModel],
    chunk_size: int = 4000,
    overlap: int = 500,
) -> list[BaseModel]:
    """長文をチャンク分割して構造化抽出し、結果をリストで返す"""
    results = []
    start = 0
 
    while start < len(document):
        end = min(start + chunk_size, len(document))
        chunk = document[start:end]
 
        try:
            result = safe_extract(chunk, schema_class)
            if result is not None:
                results.append(result)
        except RuntimeError as e:
            # チャンク失敗は記録してスキップ（全体を止めない設計）
            logger.warning(f"チャンク抽出失敗（start={start}）: {e}")
 
        # オーバーラップ付きで次のチャンクへ
        start = end - overlap if end < len(document) else len(document)
 
    return results

3つのプロダクション対応ユースケース

ユースケース1: 請求書・領収書の自動解析

バリデーションロジックを Pydantic の @field_validator に組み込むことで、モデルの計算ミスをデータ受け取り時点で検出できます。

from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional
from datetime import date
 
class TaxInfo(BaseModel):
    tax_rate: float = Field(ge=0, le=1, description="税率（例: 0.10 が10%）")
    tax_amount: float = Field(ge=0, description="税額")
 
class LineItem(BaseModel):
    description: str = Field(description="品目名または作業内容")
    quantity: float = Field(ge=0, description="数量")
    unit: Optional[str] = Field(default=None, description="単位（個・時間・式 等）")
    unit_price: float = Field(ge=0, description="単価（税抜き）")
    subtotal: float = Field(ge=0, description="小計（quantity × unit_price）")
 
class ReceiptData(BaseModel):
    document_type: Literal["invoice", "receipt", "estimate"] = Field(
        description="書類の種類。請求書=invoice、領収書=receipt、見積書=estimate"
    )
    document_number: Optional[str] = Field(
        default=None,
        description="書類番号。記載がない場合は null"
    )
    issue_date: date = Field(description="発行日（YYYY-MM-DD 形式）")
    vendor_name: str = Field(description="発行元（請求元・販売元）の名称")
    client_name: Optional[str] = Field(
        default=None,
        description="請求先・宛先の名称。記載がない場合は null"
    )
    items: list[LineItem] = Field(description="明細リスト")
    subtotal: float = Field(ge=0, description="税抜き合計金額")
    tax: TaxInfo = Field(description="税金情報")
    total: float = Field(ge=0, description="税込み合計金額")
 
    @field_validator("total")
    @classmethod
    def validate_total_consistency(cls, v: float, info) -> float:
        """合計 = 税抜き合計 + 税額 の整合性チェック（1円以内の丸め誤差を許容）"""
        data = info.data
        if "subtotal" in data and "tax" in data and data["tax"] is not None:
            expected = data["subtotal"] + data["tax"].tax_amount
            if abs(v - expected) > 1.0:
                logger.warning(
                    f"合計金額の不整合検出: total={v}, subtotal+tax={expected:.2f}"
                )
        return v

@field_validator は Pydantic v2 のデコレーターです。モデルが計算を誤るケースは実際にあるため、このようなドメイン検証は本番で非常に重要です。

ユースケース2: コードレビューの自動スコアリング

class CodeIssue(BaseModel):
    severity: Literal["error", "warning", "info"] = Field(
        description="問題の深刻度。修正必須=error、推奨=warning、参考情報=info"
    )
    category: Literal[
        "security", "performance", "maintainability", "correctness", "style"
    ] = Field(description="問題のカテゴリ")
    line_range: Optional[str] = Field(
        default=None,
        description="問題箇所の行番号（例: '12-15'）。特定できない場合は null"
    )
    description: str = Field(
        max_length=200,
        description="問題の説明（100文字以内で簡潔に）"
    )
    suggestion: str = Field(
        max_length=300,
        description="改善提案（具体的なコード変更を含む場合は省略せず記載）"
    )
 
class CodeReviewResult(BaseModel):
    overall_score: int = Field(
        ge=0, le=100,
        description="コード品質の総合スコア（0-100）。重大な問題があれば30以下"
    )
    issues: list[CodeIssue] = Field(description="検出された問題のリスト（最大10件）")
    strengths: list[str] = Field(description="コードの優れている点（最大3件）")
    summary: str = Field(
        max_length=300,
        description="レビュー全体の要約（100文字以内）"
    )
 
def review_code(code: str, language: str = "python") -> CodeReviewResult:
    """Structured Output を使った自動コードレビュー"""
    prompt = (
        f"以下の {language} コードをレビューしてください。\n"
        f"セキュリティ・パフォーマンス・可読性・正確性の観点で問題を検出し、"
        f"改善提案を含む詳細なレビュー結果を返してください。\n\n"
        f"```{language}\n{code}\n```"
    )
 
    response = client.models.generate_content(
        model="gemini-2.5-pro",  # コードレビューは Pro モデルが精度が高い
        contents=prompt,
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=CodeReviewResult,
        ),
    )
    if response.parsed is None:
        raise ValueError("コードレビューの構造化出力が取得できませんでした")
    return response.parsed

コードレビューには Gemini 2.5 Pro を選んでいます。Flash でも動きますが、セキュリティ問題の検出精度には明確な差があります。API コストと品質のトレードオフとして、コードレビューは Pro が適切だと判断しています。

ユースケース3: 多言語ドキュメントの構造化要約

class DocumentSection(BaseModel):
    heading: str = Field(description="セクションの見出し")
    key_points: list[str] = Field(description="このセクションの重要ポイント（最大3件）")
 
class DocumentSummary(BaseModel):
    title: str = Field(description="ドキュメントのタイトルまたは主題")
    document_language: str = Field(
        description="元のドキュメントの言語（例: ja, en, zh）"
    )
    document_type: Literal[
        "technical", "legal", "business", "academic", "news", "other"
    ] = Field(description="ドキュメントの種類")
    sections: list[DocumentSection] = Field(
        description="主要セクションのリスト（最大5セクション）"
    )
    action_items: list[str] = Field(
        description="このドキュメントから生じるアクションアイテム（最大5件）。なければ空リスト"
    )
    one_line_summary: str = Field(
        max_length=150,
        description="ドキュメント全体を1文（50文字以内）で要約したもの"
    )

多言語ドキュメントの場合、document_language フィールドを設けることで後続の翻訳処理やルーティングを自動化できます。このパターンは、多言語対応が必要な SaaS プロダクトのインジェスチョンパイプラインで非常に役立ちます。

コスト最適化とモデル選択の戦略

Structured Output を本番で使うとき、モデル選択がコストに直接影響します。私の経験から行き着いた使い分け基準は次の通りです。

Gemini 2.5 Flash はシンプルなスキーマ（フィールド数10以下・ネスト2階層まで）の抽出に最適です。請求書解析や基本的なフォーム入力補助には Flash で十分な精度が出て、コストも低く抑えられます。

Gemini 2.5 Pro は複雑なスキーマ・曖昧なテキスト・コードレビューのような高度な判断が必要な処理に使います。コストはかかりますが、複雑なケースでの精度差は無視できません。

コスト最適化の実践的な手法として、「事前フィルタリング」も効果的です。

class DocumentRelevance(BaseModel):
    is_relevant: bool = Field(
        description="このドキュメントに処理対象のデータが含まれているかどうか"
    )
    confidence: float = Field(
        ge=0, le=1,
        description="判定の確信度（0.0-1.0）"
    )
    reason: str = Field(
        max_length=100,
        description="判定の理由を50文字以内で"
    )
 
def filter_and_extract(
    document: str,
    target_schema: type[BaseModel],
    confidence_threshold: float = 0.7,
) -> BaseModel | None:
    """事前フィルタリングでコストを削減する 2 段階処理"""
 
    # Stage 1: 軽量モデルでスクリーニング（低コスト）
    screen_response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=(
            f"このドキュメントに '{target_schema.__name__}' として"
            f"抽出できるデータが含まれているか判定してください:\n\n{document[:1000]}"
        ),
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=DocumentRelevance,
        ),
    )
 
    relevance = screen_response.parsed
    if relevance is None or not relevance.is_relevant:
        return None  # 無関係なドキュメントはここで終了
    if relevance.confidence < confidence_threshold:
        logger.info(f"確信度が低いためスキップ: {relevance.confidence:.2f}")
        return None
 
    # Stage 2: 本処理（対象ドキュメントのみ）
    return safe_extract(document, target_schema)

大量のドキュメントを処理する場合、このフィルタリングだけで API コストを30〜50%削減できることもあります。

Antigravity 上での開発ワークフロー

Structured Output のスキーマを Antigravity で設計するとき、「型補完のフル活用」と「インライン実行による即時検証」の組み合わせが非常に効果的です。

Pydantic モデルは型アノテーションが明示的なため、Antigravity の AI 補完がフィールドの候補を正確に提示してくれます。Field(description=...) の中身も、Antigravity に「このフィールドの description をより明確にして」と頼むと、スキーマ全体のコンテキストを考慮した提案が返ってきます。

テスト時には、response.parsed の代わりに schema_class.model_validate(test_json) でスキーマのバリデーションだけを先に確認する手順が役立ちます。モデルを呼び出さずにスキーマの正しさを検証できるので、設計段階でのフィードバックループが速くなります。

# スキーマ検証だけを行うテストコード（モデル呼び出しなし・テスト高速化）
import pytest
from pydantic import ValidationError
 
class TestInvoiceSchema:
    def test_valid_invoice(self):
        """正常な請求書データでバリデーション成功を確認"""
        data = {
            "invoice_number": "INV-2026-001",
            "vendor_name": "株式会社サンプル",
            "total_amount": 11000.0,
            "items": [{"name": "デザイン費", "quantity": 1, "unit_price": 10000.0}],
            "currency": "JPY"
        }
        invoice = Invoice.model_validate(data)
        assert invoice.currency == "JPY"
        assert invoice.total_amount == 11000.0
 
    def test_invalid_quantity(self):
        """数量が 0 以下の場合にバリデーションエラーになることを確認"""
        with pytest.raises(ValidationError) as exc_info:
            InvoiceItem(name="テスト", quantity=0, unit_price=1000.0)
        assert "quantity" in str(exc_info.value)
 
    def test_optional_fields_default_none(self):
        """Optional フィールドが指定なしで None になることを確認"""
        info = PersonalInfo(full_name="山田太郎")
        assert info.email is None
        assert info.phone is None

Structured Output とユニットテストの組み合わせについては、LangGraph を使ったステートフルエージェント設計ガイドで解説しているテスト戦略も合わせて参照していただくと、エンドツーエンドのパイプラインテストに応用できます。

また、Structured Output を Python SDK から呼び出す基本的なセットアップについては、Google Gen AI SDK Python 本番マスターガイドの環境設定セクションも参考にしてください。

特に型安全なデータパイプラインの設計章は、Structured Output の設計思想を深く理解する上で補完的な知識になります。

今日から始められる一歩

この記事で紹介したパターンのうち、今日すぐに試せる最小単位は safe_extract 関数です。response.text に依存している箇所を1つ選んで safe_extract に置き換えるだけでも、本番での予期しない None エラーと手動 JSON パースのコードが減り始めます。

Structured Output は「モデルに出力フォーマットを強制する機能」ではなく、「型システムとモデルの間に契約を結ぶ設計」だと考えると、Pydantic との組み合わせが自然に見えてきます。そのアプローチで設計を重ねていくことが、長期的に保守しやすい AI パイプラインへの近道になるはずです。