Gemini 2.5 Pro の空間理解（バウンディングボックス・セグメンテーション）を本番アプリで使いこなす

import base64
from PIL import Image
from io import BytesIO
 
class Segment(BaseModel):
    label: str
    box_2d: List[int]
    mask: str = Field(description="data:image/png;base64,... 形式の base64 マスク")
 
class SegmentationResult(BaseModel):
    segments: List[Segment]
 
def segment_objects(image_path: str, target: str) -> SegmentationResult:
    client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    image = Image.open(image_path)
 
    prompt = (
        f"画像から『{target}』に該当する物体ごとに、バウンディングボックスとセグメンテーションマスクを返してください。"
        "mask は data:image/png;base64,... 形式で、box_2d の領域にフィットさせた PNG にしてください。"
        "存在しない場合は segments を空配列で返してください。"
    )
    resp = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=[prompt, image],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=SegmentationResult,
            temperature=0.1,
        ),
    )
    return SegmentationResult.model_validate_json(resp.text)
 
def decode_mask_to_full_image(seg: Segment, img_w: int, img_h: int) -> Image.Image:
    """base64マスクを元画像サイズの2値マスクに展開する。"""
    # data:image/png;base64,XXX のヘッダを除去
    b64 = seg.mask.split(",", 1)[1] if "," in seg.mask else seg.mask
    box_png = Image.open(BytesIO(base64.b64decode(b64))).convert("L")
 
    # ピクセル座標のボックス
    x1, y1, x2, y2 = to_pixel_box(seg.box_2d, img_w, img_h)
    box_w, box_h = max(1, x2 - x1), max(1, y2 - y1)
 
    # ボックスにフィットしている小マスクを、ボックスのサイズへリサイズ
    box_png = box_png.resize((box_w, box_h), Image.NEAREST)
 
    # 元画像サイズの黒キャンバスに合成
    full = Image.new("L", (img_w, img_h), 0)
    full.paste(box_png, (x1, y1))
    return full

def iou(a: List[int], b: List[int]) -> float:
    """[ymin, xmin, ymax, xmax] 同士の IoU。"""
    ay1, ax1, ay2, ax2 = a
    by1, bx1, by2, bx2 = b
    iy1, ix1 = max(ay1, by1), max(ax1, bx1)
    iy2, ix2 = min(ay2, by2), min(ax2, bx2)
    iw, ih = max(0, ix2 - ix1), max(0, iy2 - iy1)
    inter = iw * ih
    if inter == 0:
        return 0.0
    area_a = (ay2 - ay1) * (ax2 - ax1)
    area_b = (by2 - by1) * (bx2 - bx1)
    return inter / (area_a + area_b - inter)
 
def filter_hallucinations(
    detections: List[Detection],
    img_w: int, img_h: int,
    min_conf: float = 0.5,
) -> List[Detection]:
    out = []
    for d in detections:
        if d.confidence < min_conf:
            continue
        x1, y1, x2, y2 = to_pixel_box(d.box_2d, img_w, img_h)
        bw, bh = x2 - x1, y2 - y1
        if bw <= 0 or bh <= 0:
            continue
        ratio = (bw * bh) / float(img_w * img_h)
        if ratio < 0.001 or ratio > 0.95:
            continue
        out.append(d)
    return out
 
def cross_check(detections_run1, detections_run2, iou_threshold: float = 0.5):
    """2回の検出結果のうち、IoU >= 閾値で重なるものだけ残す。"""
    consistent = []
    for d1 in detections_run1:
        for d2 in detections_run2:
            if d1.label == d2.label and iou(d1.box_2d, d2.box_2d) >= iou_threshold:
                consistent.append(d1)
                break
    return consistent

def smart_resize(img: Image.Image, max_side: int = 1280) -> Image.Image:
    w, h = img.size
    longest = max(w, h)
    if longest <= max_side:
        return img
    scale = max_side / longest
    return img.resize((round(w * scale), round(h * scale)), Image.LANCZOS)

// Cloudflare Workers 想定
import { GoogleGenAI } from "@google/genai";
 
interface DetectInput {
  imageUrl: string;
  target: string;
}
 
interface Detection {
  label: string;
  box_2d: [number, number, number, number];
  confidence: number;
  source: "gemini" | "yolo";
}
 
const SOFT_FAIL_PATTERNS = [
  /quota/i,
  /rate.?limit/i,
  /server.?error/i,
];
 
async function detectWithGemini(input: DetectInput, env: Env): Promise<Detection[]> {
  const ai = new GoogleGenAI({ apiKey: env.GEMINI_API_KEY });
  const imageBytes = await (await fetch(input.imageUrl)).arrayBuffer();
  const result = await ai.models.generateContent({
    model: "gemini-2.5-pro",
    contents: [
      {
        role: "user",
        parts: [
          {
            text: `画像から『${input.target}』を検出。box_2dは[ymin,xmin,ymax,xmax](0-1000)。`
          },
          {
            inlineData: {
              mimeType: "image/jpeg",
              data: btoa(String.fromCharCode(...new Uint8Array(imageBytes))),
            },
          },
        ],
      },
    ],
    config: {
      responseMimeType: "application/json",
      responseSchema: {
        type: "object",
        properties: {
          detections: {
            type: "array",
            items: {
              type: "object",
              properties: {
                label: { type: "string" },
                box_2d: { type: "array", items: { type: "integer" } },
                confidence: { type: "number" },
              },
              required: ["label", "box_2d", "confidence"],
            },
          },
        },
        required: ["detections"],
      },
      temperature: 0.1,
    },
  });
  const parsed = JSON.parse(result.text ?? "{\"detections\":[]}");
  return parsed.detections.map((d: Detection) => ({ ...d, source: "gemini" as const }));
}
 
async function detectWithYolo(input: DetectInput, env: Env): Promise<Detection[]> {
  // 自前 YOLO 推論サーバー（Replicate, Modal, 自社GPU）に飛ばすイメージ
  const r = await fetch(env.YOLO_ENDPOINT, {
    method: "POST",
    headers: { "Authorization": `Bearer ${env.YOLO_TOKEN}` },
    body: JSON.stringify({ image_url: input.imageUrl, target: input.target }),
  });
  if (!r.ok) return [];
  const json = await r.json<{ detections: Detection[] }>();
  return json.detections.map((d) => ({ ...d, source: "yolo" }));
}
 
export async function detect(input: DetectInput, env: Env): Promise<Detection[]> {
  try {
    const dets = await detectWithGemini(input, env);
    if (dets.length > 0) return dets;
    // ゼロ件は信用せず、確認のために YOLO も走らせる
    const fallback = await detectWithYolo(input, env);
    return fallback;
  } catch (e: any) {
    const msg = String(e?.message ?? e);
    const recoverable = SOFT_FAIL_PATTERNS.some((p) => p.test(msg));
    if (recoverable) {
      console.warn("Gemini soft-fail, falling back to YOLO:", msg);
      return detectWithYolo(input, env);
    }
    throw e;
  }
}

import json
from pathlib import Path
 
def evaluate(predicted, expected, iou_thr=0.4):
    # IoU >= iou_thr かつ同じラベルを「マッチ」とみなして precision/recall を返す
    matched_pred = set()
    matched_exp = set()
    for i, e in enumerate(expected):
        for j, p in enumerate(predicted):
            if j in matched_pred:
                continue
            if e["label"] != p["label"]:
                continue
            if iou(e["box_2d"], p["box_2d"]) >= iou_thr:
                matched_pred.add(j)
                matched_exp.add(i)
                break
    precision = len(matched_pred) / max(1, len(predicted))
    recall = len(matched_exp) / max(1, len(expected))
    return precision, recall
 
def run_regression_suite(fixtures_dir: str, threshold_p: float = 0.7, threshold_r: float = 0.7):
    failures = []
    for case in Path(fixtures_dir).glob("*.json"):
        spec = json.loads(case.read_text())
        result = detect_objects(spec["image_path"], spec["target"])
        precision, recall = evaluate(
            [d.model_dump() for d in result.detections],
            spec["expected"],
        )
        if precision < threshold_p or recall < threshold_r:
            failures.append({"case": case.name, "precision": precision, "recall": recall})
    return failures