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;
}

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}")