OpenAI 互換レイヤーで JSON の形が揃わないとき、既存コードを残したまま response_json_schema へ寄せる3段階

import logging
import os
from typing import Literal, Optional
 
from openai import OpenAI
from pydantic import BaseModel, ValidationError
 
# モデル名は使う前に ai.google.dev の Models ページで提供状況を確かめてください
MODEL = "gemini-4-hydrogen"
SYSTEM = "壁紙の説明文から、最も近いタグを1つだけ選んでください。"
 
 
class WallpaperTag(BaseModel):
    label: Literal["landscape", "portrait", "abstract", "other"]
    reason: str
 
 
compat = OpenAI(
    api_key=os.environ["GEMINI_API_KEY"],
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
)
 
 
def classify_compat(caption: str) -> Optional[WallpaperTag]:
    completion = compat.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": caption},
        ],
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "wallpaper_tag",
                "schema": WallpaperTag.model_json_schema(),
            },
        },
    )
    text = completion.choices[0].message.content or ""
    try:
        return WallpaperTag.model_validate_json(text)
    except ValidationError as e:
        # 形が崩れた応答は捨てずに残す。どの経路で崩れたかを後で数えるため
        logging.warning("compat schema mismatch: %s | raw=%r", e.errors()[0]["type"], text[:200])
        return None

from typing import Optional
 
from google import genai
from google.genai import types
from pydantic import ValidationError
 
# 先ほどの MODEL・SYSTEM・WallpaperTag をそのまま使う
native = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
 
def classify_native(caption: str) -> Optional[WallpaperTag]:
    res = native.models.generate_content(
        model=MODEL,
        contents=caption,
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM,
            response_mime_type="application/json",
            # Pydantic の定義から作った JSON Schema を、互換レイヤーと同じものとして渡す
            response_json_schema=WallpaperTag.model_json_schema(),
        ),
    )
    try:
        return WallpaperTag.model_validate_json(res.text or "")
    except ValidationError as e:
        logging.warning("native schema mismatch: %s | raw=%r", e.errors()[0]["type"], (res.text or "")[:200])
        return None

import os
from typing import Optional
 
 
def classify(caption: str) -> Optional[WallpaperTag]:
    backend = os.environ.get("TAG_BACKEND", "compat")
    if backend == "native":
        return classify_native(caption)
    return classify_compat(caption)
 
 
def shadow_compare(captions: list[str]) -> None:
    """同じ入力を両経路に送り、ラベルの食い違いと検証失敗を数える"""
    diff = fail_compat = fail_native = 0
    for c in captions:
        a, b = classify_compat(c), classify_native(c)
        fail_compat += a is None
        fail_native += b is None
        if a and b and a.label != b.label:
            diff += 1
            print(f"DIFF {a.label} / {b.label} : {c[:60]}")
    print(f"件数={len(captions)} 食い違い={diff} 互換の失敗={fail_compat} SDKの失敗={fail_native}")