カテゴリを30から34に増やした日 — 8,000枚を全部かけ直さない差分再分類の設計

CREATE TABLE asset_labels (
  asset_id         TEXT PRIMARY KEY,
  label            TEXT NOT NULL,      -- 採用ラベル
  runner_up        TEXT,               -- 2位ラベル
  confidence       REAL NOT NULL,      -- 1位の確信度 0.0-1.0
  runner_up_conf   REAL,               -- 2位の確信度
  margin           REAL NOT NULL,      -- confidence - runner_up_conf
  taxonomy_version INTEGER NOT NULL,   -- 分類時のカテゴリ体系のバージョン
  classified_with  TEXT NOT NULL,      -- 使用モデル名（固定バージョン）
  classified_at    TEXT NOT NULL
);
CREATE INDEX idx_margin ON asset_labels(margin);
CREATE INDEX idx_label_version ON asset_labels(label, taxonomy_version);

from pydantic import BaseModel, Field
from google import genai
 
class LabelGuess(BaseModel):
    label: str
    confidence: float = Field(ge=0.0, le=1.0)
 
class Classification(BaseModel):
    top: LabelGuess
    runner_up: LabelGuess
 
client = genai.Client(api_key="YOUR_API_KEY")
MODEL = "gemini-3.5-flash"  # エイリアスではなく固定バージョンを台帳に残す
 
def classify(image_bytes: bytes, categories: list[str]) -> Classification:
    resp = client.models.generate_content(
        model=MODEL,
        contents=[
            {"inline_data": {"mime_type": "image/jpeg", "data": image_bytes}},
            "以下のカテゴリから最も適切なものと、次点を1つずつ選び、"
            "それぞれの確信度を0.0-1.0で返してください。\n"
            + "\n".join(f"- {c}" for c in categories),
        ],
        config={
            "response_mime_type": "application/json",
            "response_schema": Classification,
            "temperature": 0.0,
        },
    )
    return Classification.model_validate_json(resp.text)

import numpy as np
from google import genai
 
client = genai.Client(api_key="YOUR_API_KEY")
EMBED_MODEL = "gemini-embedding-2"
 
def embed(texts: list[str]) -> np.ndarray:
    resp = client.models.embed_content(
        model=EMBED_MODEL,
        contents=texts,
        config={"task_type": "SEMANTIC_SIMILARITY", "output_dimensionality": 768},
    )
    vecs = np.array([e.values for e in resp.embeddings], dtype=np.float32)
    # 768次元に縮めた場合は再正規化が必要（縮約後はノルムが1でない）
    return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)
 
def impacted_labels(old_labels: dict[str, str],
                    new_labels: dict[str, str],
                    threshold: float = 0.62) -> set[str]:
    """old_labels / new_labels は {ラベル名: 一文の定義} """
    old_names = list(old_labels)
    new_names = list(new_labels)
    old_vec = embed([f"{k}: {v}" for k, v in old_labels.items()])
    new_vec = embed([f"{k}: {v}" for k, v in new_labels.items()])
 
    sim = new_vec @ old_vec.T          # (new, old) のコサイン類似度
    impacted: set[str] = set()
    for i, _ in enumerate(new_names):
        for j, old in enumerate(old_names):
            if sim[i, j] >= threshold:
                impacted.add(old)
    return impacted

# Before: 全件を無条件に再分類する
def backfill_all(conn, categories, taxonomy_version):
    rows = conn.execute("SELECT asset_id FROM asset_labels").fetchall()
    for (asset_id,) in rows:                     # 8,142 回
        result = classify(load_image(asset_id), categories)
        upsert(conn, asset_id, result, taxonomy_version)

# After: 影響ラベル ∪ 低マージン資産だけを再分類する
MARGIN_THRESHOLD = 0.15   # 確信度の揺れ幅（実測 ±0.03）の5倍を取った
 
def select_targets(conn, impacted: set[str], current_version: int) -> list[str]:
    placeholders = ",".join("?" for _ in impacted)
    by_label = conn.execute(
        f"SELECT asset_id FROM asset_labels "
        f"WHERE label IN ({placeholders}) AND taxonomy_version < ?",
        (*impacted, current_version),
    ).fetchall()
 
    by_margin = conn.execute(
        "SELECT asset_id FROM asset_labels "
        "WHERE margin < ? AND taxonomy_version < ?",
        (MARGIN_THRESHOLD, current_version),
    ).fetchall()
 
    return sorted({r[0] for r in by_label} | {r[0] for r in by_margin})
 
def backfill_selective(conn, categories, taxonomy_version, impacted):
    targets = select_targets(conn, impacted, taxonomy_version)
    print(f"targets: {len(targets)} / {count_all(conn)}")
    for asset_id in targets:                     # 1,180 回
        result = classify(load_image(asset_id), categories)
        upsert(conn, asset_id, result, taxonomy_version)
 
    # 再分類しなかった資産は、体系バージョンだけ進めて「判定済み」にする
    conn.execute(
        "UPDATE asset_labels SET taxonomy_version = ? "
        "WHERE taxonomy_version < ? AND asset_id NOT IN "
        "(SELECT value FROM json_each(?))",
        (taxonomy_version, taxonomy_version, json.dumps(targets)),
    )
    conn.commit()