Gemini の構造化出力で入力種別ごとに型を変える — anyOf 判別ユニオンの実運用メモ

import os
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def variant(kind: str, props: dict, required: list[str]) -> types.Schema:
    # 判別子 kind は単一値 enum + 先頭固定。これがバリアントを一意に決める
    properties = {"kind": types.Schema(type=types.Type.STRING, enum=[kind])}
    properties.update(props)
    return types.Schema(
        type=types.Type.OBJECT,
        properties=properties,
        required=["kind"] + required,
        property_ordering=["kind"] + list(props.keys()),
    )
 
app_review = variant(
    "app_review",
    {
        "rating": types.Schema(type=types.Type.INTEGER, minimum=1, maximum=5),
        "summary": types.Schema(type=types.Type.STRING),
        "feature_area": types.Schema(
            type=types.Type.STRING,
            enum=["onboarding", "billing", "performance", "other"],
        ),
    },
    required=["rating", "summary"],
)
 
support_inquiry = variant(
    "support_inquiry",
    {
        "summary": types.Schema(type=types.Type.STRING),
        "urgency": types.Schema(type=types.Type.STRING, enum=["low", "normal", "high"]),
    },
    required=["summary", "urgency"],
)
 
refund_request = variant(
    "refund_request",
    {
        "order_id": types.Schema(type=types.Type.STRING),
        "reason": types.Schema(type=types.Type.STRING),
    },
    required=["order_id", "reason"],
)
 
batch_schema = types.Schema(
    type=types.Type.ARRAY,
    items=types.Schema(any_of=[app_review, support_inquiry, refund_request]),
)
 
resp = client.models.generate_content(
    model="gemini-3.5-flash",
    contents=INPUT_ITEMS_AS_TEXT,
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=batch_schema,
        temperature=0,
    ),
)

from typing import Annotated, Literal, Union
from pydantic import BaseModel, Field, TypeAdapter, ValidationError
 
class AppReview(BaseModel):
    kind: Literal["app_review"]
    rating: int = Field(ge=1, le=5)
    summary: str
    feature_area: str | None = None
 
class SupportInquiry(BaseModel):
    kind: Literal["support_inquiry"]
    summary: str
    urgency: Literal["low", "normal", "high"]
 
class RefundRequest(BaseModel):
    kind: Literal["refund_request"]
    order_id: str
    reason: str
 
Item = Annotated[
    Union[AppReview, SupportInquiry, RefundRequest],
    Field(discriminator="kind"),
]
batch_adapter = TypeAdapter(list[Item])
 
def parse_batch(raw_json: str) -> tuple[list, list]:
    """検証を通った要素と、落ちた要素（生データ）を分けて返します。"""
    import json
    data = json.loads(raw_json)
    ok, rejected = [], []
    for entry in data:
        try:
            ok.append(batch_adapter.validate_python([entry])[0])
        except ValidationError as e:
            rejected.append({"raw": entry, "error": e.errors()})
    return ok, rejected

import { z } from "zod";
 
const Item = z.discriminatedUnion("kind", [
  z.object({
    kind: z.literal("app_review"),
    rating: z.number().int().min(1).max(5),
    summary: z.string(),
    feature_area: z.enum(["onboarding", "billing", "performance", "other"]).optional(),
  }),
  z.object({
    kind: z.literal("support_inquiry"),
    summary: z.string(),
    urgency: z.enum(["low", "normal", "high"]),
  }),
  z.object({
    kind: z.literal("refund_request"),
    order_id: z.string().min(1),
    reason: z.string(),
  }),
]);
const Batch = z.array(Item);
 
async function classifyWithRepair(input: string, callModel: (prompt: string) => Promise<string>) {
  const first = await callModel(input);
  const parsed = Batch.safeParse(JSON.parse(first));
  if (parsed.success) return { items: parsed.data, repaired: false };
 
  // 修復は1回だけ。失敗箇所だけを示して直させる
  const repairPrompt =
    `次の JSON はスキーマ検証に失敗しました。kind の値と必須フィールドを満たすよう、` +
    `構造だけを修正して同じ配列を返してください。\n` +
    `エラー: ${JSON.stringify(parsed.error.issues.slice(0, 5))}\n` +
    `JSON: ${first}`;
  const second = await callModel(repairPrompt);
  const retry = Batch.safeParse(JSON.parse(second));
  if (retry.success) return { items: retry.data, repaired: true };
 
  // 2回目も駄目なら握りつぶさず DLQ へ
  throw new DeadLetter("schema_validation_failed", { first, second });
}