Claude API の構造化レスポンスを本番で安定させる — tool_use と JSON Schema、多層検証の実装メモ

# Before — 「厳密にしすぎて壊れる」スキーマの例
"input_schema": {
    "type": "object",
    "properties": {
        "primary_category": {
            "type": "string",
            "enum": ["nature", "abstract", "city", "minimal", "people"],  # other がない
        },
        "color_palette": {
            "type": "array",
            "items": {"type": "string", "pattern": "^#[0-9A-Fa-f]{6}$"},  # 強すぎる正規表現
            "minItems": 5,
            "maxItems": 5,
        },
    },
    "required": ["primary_category", "color_palette"],
}

# After — 「ギリギリ厳しい」スキーマの例
"input_schema": {
    "type": "object",
    "properties": {
        "primary_category": {
            "type": "string",
            "enum": ["nature", "abstract", "city", "minimal", "people", "other"],
        },
        "color_palette": {
            "type": "array",
            "items": {"type": "string", "minLength": 4, "maxLength": 7},  # #RGB / #RRGGBB
            "minItems": 1,
            "maxItems": 5,
        },
    },
    "required": ["primary_category", "color_palette"],
    "additionalProperties": False,
}

from dataclasses import dataclass
 
@dataclass
class ClassificationResult:
    primary_category: str
    tags: list[str]
    confidence: float
    needs_human_review: bool
    domain_violations: list[str]
 
 
def validate_domain(raw: dict) -> ClassificationResult:
    violations: list[str] = []
 
    # ドメインルール1: 信頼度が高いのにレビュー必要は矛盾
    if raw.get("confidence", 0) > 0.92 and raw.get("needs_human_review"):
        violations.append("high_confidence_but_review_requested")
 
    # ドメインルール2: "other" が選ばれた時点で必ずレビュー対象
    if raw.get("primary_category") == "other" and not raw.get("needs_human_review"):
        raw["needs_human_review"] = True
        violations.append("auto_promoted_other_to_review")
 
    # ドメインルール3: タグの先頭/末尾の空白を除去し空文字を弾く
    cleaned_tags = [t.strip() for t in raw.get("tags", []) if t and t.strip()]
    if not cleaned_tags:
        violations.append("empty_tags_after_cleaning")
        cleaned_tags = ["uncategorized"]
 
    return ClassificationResult(
        primary_category=raw.get("primary_category", "other"),
        tags=cleaned_tags,
        confidence=float(raw.get("confidence", 0.0)),
        needs_human_review=bool(raw.get("needs_human_review", False)),
        domain_violations=violations,
    )

// TypeScript SDK の場合の partial JSON ハンドリング例
import Anthropic from "@anthropic-ai/sdk";
import { parse as parsePartial } from "partial-json";
 
const client = new Anthropic();
 
async function classifyWithProgressHint(imageInput: any) {
  let accumulatedJson = "";
  let preview: { primary_category?: string } = {};
  let finalInput: Record<string, unknown> | null = null;
 
  const stream = client.messages.stream({
    model: "claude-haiku-4-5",
    max_tokens: 512,
    tools: [TOOL_SCHEMA],
    tool_choice: { type: "tool", name: "classify_wallpaper" },
    messages: [{ role: "user", content: [imageInput, { type: "text", text: "分類してください" }] }],
  });
 
  for await (const event of stream) {
    if (event.type === "content_block_delta" && event.delta.type === "input_json_delta") {
      accumulatedJson += event.delta.partial_json;
      try {
        // 寛容パースで進捗だけ得る
        preview = parsePartial(accumulatedJson);
        if (preview.primary_category) {
          notifyUiHint(preview.primary_category);
        }
      } catch {
        // partial の段階での失敗は無視
      }
    }
    if (event.type === "content_block_stop" && event.content_block?.type === "tool_use") {
      // ← ここで初めて確定値を使う
      finalInput = event.content_block.input as Record<string, unknown>;
    }
  }
 
  if (!finalInput) {
    throw new Error("tool_use did not complete");
  }
  return finalInput;
}

"""input_json_delta から進捗ヒントを取り出す3方式のコストを手元で測る（API 不要）"""
import json, time
 
PAYLOAD = {
    "primary_category": "nature",
    "secondary_categories": ["landscape", "mountain", "sunrise", "forest",
                             "lake", "autumn", "minimal", "gradient"],
    "color_tone": "warm",
    "confidence": 0.82,
    "needs_human_review": False,
    "palette": [{"hex": "#f2a65a", "ratio": 0.34},
                {"hex": "#2b3a67", "ratio": 0.28},
                {"hex": "#e8e1d9", "ratio": 0.21}],
    "notes": "朝焼けの稜線。彩度が高めのため sunset と混同されやすい。前景の樹影が強いので forest との重複タグを許容する。",
    "locale_hints": {"ja": "朝焼けの山", "en": "sunrise over the ridge"},
}
RAW = json.dumps(PAYLOAD, ensure_ascii=False)
DELTAS = [RAW[i:i + 12] for i in range(0, len(RAW), 12)]  # 12文字ずつに刻む
 
 
def scan(buf):
    """buf を1度だけ走査して (深さスタック, 文字列の途中か, 最後のカンマ位置) を返す"""
    stack, in_str, esc, last_comma = [], False, False, -1
    for i, ch in enumerate(buf):
        if esc:
            esc = False
        elif in_str and ch == "\\":
            esc = True
        elif ch == '"':
            in_str = not in_str
        elif not in_str:
            if ch in "{[":
                stack.append("}" if ch == "{" else "]")
            elif ch in "}]" and stack:
                stack.pop()
            elif ch == "," and len(stack) == 1:
                last_comma = i
    return stack, in_str, last_comma
 
 
def repair(buf):
    """開いたままの構造を閉じ、その時点で読める JSON 文字列にする"""
    stack, in_str, last_comma = scan(buf)
    if in_str:                       # 値の途中で切れている → 直前の完了地点まで巻き戻す
        buf = buf[:last_comma] if last_comma >= 0 else buf[:buf.index("{") + 1]
        stack, _, _ = scan(buf)
    buf = buf.rstrip().rstrip(",:").rstrip()
    return buf + "".join(reversed(stack))
 
 
def naive(deltas):
    """毎 delta、素の json.loads を試す"""
    buf, tries, ok, first = "", 0, 0, None
    for i, d in enumerate(deltas):
        buf += d
        tries += 1
        try:
            obj = json.loads(buf)
        except ValueError:
            continue
        ok += 1
        if first is None and obj.get("primary_category"):
            first = i + 1
    return tries, ok, first
 
 
def repair_always(deltas):
    """毎 delta、修復してからパースする（寛容パーサ相当）"""
    buf, tries, ok, first = "", 0, 0, None
    for i, d in enumerate(deltas):
        buf += d
        tries += 1
        try:
            obj = json.loads(repair(buf))
        except ValueError:
            continue
        ok += 1
        if first is None and isinstance(obj, dict) and obj.get("primary_category"):
            first = i + 1
    return tries, ok, first
 
 
def gated(deltas):
    """深さ1の値が閉じた delta でだけ修復＋パースする"""
    buf, tries, ok, first = "", 0, 0, None
    depth, in_str, esc, seen = 0, False, False, 0
    for i, d in enumerate(deltas):
        buf += d
        trigger = False
        while seen < len(buf):
            ch = buf[seen]
            seen += 1
            if esc:
                esc = False
            elif in_str and ch == "\\":
                esc = True
            elif ch == '"':
                in_str = not in_str
                trigger |= (not in_str and depth == 1)
            elif not in_str:
                if ch in "{[":
                    depth += 1
                elif ch in "}]":
                    depth -= 1
                    trigger |= (depth == 1)
                elif ch == "," and depth == 1:
                    trigger = True
        if not trigger:
            continue
        tries += 1
        try:
            obj = json.loads(repair(buf))
        except ValueError:
            continue
        ok += 1
        if first is None and isinstance(obj, dict) and obj.get("primary_category"):
            first = i + 1
    return tries, ok, first
 
 
if __name__ == "__main__":
    print(f"payload={len(RAW)} chars / deltas={len(DELTAS)} (12 chars each)")
    for name, fn in (("naive json.loads", naive),
                     ("repair every delta", repair_always),
                     ("gated repair", gated)):
        start = time.perf_counter()
        for _ in range(2000):
            tries, ok, first = fn(DELTAS)
        ms = (time.perf_counter() - start) / 2000 * 1000
        print(f"{name:20} parses={tries:3} ok={ok:3} "
              f"first_hint=delta {first}/{len(DELTAS)}  {ms:.3f} ms/stream")

import time
from anthropic import APIStatusError, APITimeoutError
 
def classify_with_observability(image_block: dict, max_retries: int = 3) -> ClassificationResult:
    last_error: Exception | None = None
    for attempt in range(1, max_retries + 1):
        try:
            raw = classify_wallpaper_tool(image_block)
            result = validate_domain(raw)
            log_metric("classification.success", {"attempt": attempt})
            return result
        except APITimeoutError as e:
            last_error = e
            log_metric("classification.timeout", {"attempt": attempt})
            time.sleep(2 ** attempt)
        except APIStatusError as e:
            last_error = e
            log_metric("classification.api_status", {"status": e.status_code, "attempt": attempt})
            if e.status_code == 400:
                # スキーマ違反系は2回で打ち切り
                if attempt >= 2:
                    raise
            time.sleep(2 ** attempt)
 
    raise RuntimeError(f"classification failed after {max_retries} attempts: {last_error}")

"""structured_call.py — Claude API で構造化レスポンスを安定して取り出す共通レイヤ。"""
from __future__ import annotations
import time
import logging
from dataclasses import dataclass
from typing import Any, Callable, TypeVar
 
from anthropic import Anthropic, APIStatusError, APITimeoutError
 
T = TypeVar("T")
logger = logging.getLogger(__name__)
client = Anthropic()
 
 
@dataclass
class StructuredResult[T]:
    value: T
    domain_violations: list[str]
    attempt_count: int
 
 
def call_with_tool(
    *,
    model: str,
    tool: dict,
    messages: list[dict],
    domain_validator: Callable[[dict], tuple[T, list[str]]],
    max_retries: int = 3,
    extended_thinking: bool = False,
) -> StructuredResult[T]:
    """tool_use で構造化レスポンスを取り、ドメイン検証を通して返す。"""
    create_kwargs: dict[str, Any] = {
        "model": model,
        "max_tokens": 1024,
        "tools": [tool],
        "tool_choice": {"type": "tool", "name": tool["name"]},
        "messages": messages,
    }
    if extended_thinking:
        create_kwargs["thinking"] = {"type": "enabled", "budget_tokens": 4096}
 
    last_error: Exception | None = None
    for attempt in range(1, max_retries + 1):
        try:
            resp = client.messages.create(**create_kwargs)
            for block in resp.content:
                if block.type == "tool_use" and block.name == tool["name"]:
                    value, violations = domain_validator(dict(block.input))
                    return StructuredResult(value=value, domain_violations=violations, attempt_count=attempt)
            raise RuntimeError("tool_use block not found")
 
        except APITimeoutError as e:
            last_error = e
            logger.warning("timeout attempt=%s err=%s", attempt, e)
            time.sleep(min(2 ** attempt, 8))
 
        except APIStatusError as e:
            last_error = e
            logger.warning("api_status attempt=%s status=%s", attempt, e.status_code)
            if e.status_code == 400 and attempt >= 2:
                # スキーマ違反系は2回で打ち切り
                break
            time.sleep(min(2 ** attempt, 8))
 
    raise RuntimeError(f"structured call failed: {last_error}")