Claude Vision API 実装パターン — 画像解析・PDF 処理・OCR を本番品質で動かすまで

from PIL import Image
import io, base64
 
def resize_for_claude(path: str, max_edge: int = 1568, quality: int = 85) -> tuple[str, str]:
    """長辺を max_edge に収め、JPEG 圧縮して Base64 を返します"""
    img = Image.open(path)
    if img.mode in ("RGBA", "P"):
        img = img.convert("RGB")
    ratio = max_edge / max(img.size)
    if ratio < 1:
        img = img.resize(
            (int(img.width * ratio), int(img.height * ratio)),
            Image.LANCZOS,
        )
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=quality)
    data = base64.standard_b64encode(buf.getvalue()).decode("utf-8")
    return data, "image/jpeg"
 
def estimate_image_tokens(width: int, height: int) -> int:
    return (width * height) // 750

import base64
from pathlib import Path
 
pdf_data = base64.standard_b64encode(Path("report.pdf").read_bytes()).decode("utf-8")
 
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    messages=[{
        "role": "user",
        "content": [
            {"type": "document",
             "source": {"type": "base64", "media_type": "application/pdf", "data": pdf_data}},
            {"type": "text", "text": "この決算資料から売上高・営業利益・前年比をテーブルで抽出してください。"},
        ],
    }],
)
print(message.content[0].text)

from pypdf import PdfReader, PdfWriter
import io
 
def split_pdf(path: str, pages_per_chunk: int = 90) -> list[bytes]:
    """100ページ制限に収まるよう PDF を分割します"""
    reader = PdfReader(path)
    chunks = []
    for start in range(0, len(reader.pages), pages_per_chunk):
        writer = PdfWriter()
        for page in reader.pages[start:start + pages_per_chunk]:
            writer.add_page(page)
        buf = io.BytesIO()
        writer.write(buf)
        chunks.append(buf.getvalue())
    return chunks

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{
        "role": "user",
        "content": [
            {"type": "document",
             "source": {"type": "base64", "media_type": "application/pdf", "data": pdf_data},
             "citations": {"enabled": True}},
            {"type": "text", "text": "解約条項の要点をまとめてください。"},
        ],
    }],
)
 
for block in message.content:
    if block.type == "text":
        print(block.text)
        for c in getattr(block, "citations", None) or []:
            print(f"  └ 出典: {c.start_page_number}〜{c.end_page_number - 1}ページ")

TABLE_TOOL = {
    "name": "record_tables",
    "description": "画像から抽出した表データを記録します",
    "input_schema": {
        "type": "object",
        "properties": {
            "tables": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "title": {"type": "string", "description": "表のタイトルまたは見出し"},
                        "headers": {"type": "array", "items": {"type": "string"}},
                        "rows": {
                            "type": "array",
                            "items": {"type": "array", "items": {"type": "string"}},
                        },
                        "notes": {"type": "string", "description": "欠損・判読困難箇所の注記"},
                    },
                    "required": ["headers", "rows"],
                },
            }
        },
        "required": ["tables"],
    },
}
 
def extract_tables(image_path: str) -> dict:
    data, media_type = resize_for_claude(image_path)  # OCR系は1568pxを維持
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        tools=[TABLE_TOOL],
        tool_choice={"type": "tool", "name": "record_tables"},
        messages=[{
            "role": "user",
            "content": [
                {"type": "image",
                 "source": {"type": "base64", "media_type": media_type, "data": data}},
                {"type": "text",
                 "text": "画像内のすべての表を抽出してください。"
                         "セルが判読できない場合は空文字にし、notes に位置を記録してください。"},
            ],
        }],
    )
    for block in message.content:
        if block.type == "tool_use":
            return block.input  # ここは常にスキーマ準拠の dict
    return {"tables": []}

import pandas as pd
 
result = extract_tables("financial_report.png")
for i, table in enumerate(result["tables"]):
    df = pd.DataFrame(table["rows"], columns=table["headers"])
    df.to_csv(f"table_{i}.csv", index=False)
    if table.get("notes"):
        print(f"table_{i}: 要確認 — {table['notes']}")

UI_REVIEW_TOOL = {
    "name": "record_ui_issues",
    "description": "UI スクリーンショットの問題点を記録します",
    "input_schema": {
        "type": "object",
        "properties": {
            "issues": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "severity": {"type": "string", "enum": ["high", "medium", "low"]},
                        "category": {"type": "string",
                                     "enum": ["text-truncation", "contrast", "tap-target", "layout", "other"]},
                        "location": {"type": "string", "description": "画面内の位置の説明"},
                        "description": {"type": "string"},
                        "suggestion": {"type": "string"},
                    },
                    "required": ["severity", "category", "description"],
                },
            }
        },
        "required": ["issues"],
    },
}

from concurrent.futures import ThreadPoolExecutor, as_completed
 
def analyze_one(path: str, prompt: str) -> tuple[str, str]:
    data, media_type = resize_for_claude(path, max_edge=1092)  # 分類タスクは小さめで十分
    message = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 分類は Haiku で足りる
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image",
                 "source": {"type": "base64", "media_type": media_type, "data": data}},
                {"type": "text", "text": prompt},
            ],
        }],
    )
    return path, message.content[0].text
 
def analyze_parallel(paths: list[str], prompt: str, workers: int = 4) -> dict:
    results, errors = {}, {}
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(analyze_one, p, prompt): p for p in paths}
        for future in as_completed(futures):
            try:
                path, text = future.result()
                results[path] = text
            except anthropic.RateLimitError:
                errors[futures[future]] = "rate_limit"
            except Exception as e:
                errors[futures[future]] = str(e)
    return {"results": results, "errors": errors}

from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
 
def submit_batch(paths: list[str], prompt: str) -> str:
    requests = []
    for i, path in enumerate(paths):
        data, media_type = resize_for_claude(path, max_edge=1092)
        requests.append(Request(
            custom_id=f"img-{i}-{Path(path).stem}",
            params=MessageCreateParamsNonStreaming(
                model="claude-haiku-4-5-20251001",
                max_tokens=256,
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "image",
                         "source": {"type": "base64", "media_type": media_type, "data": data}},
                        {"type": "text", "text": prompt},
                    ],
                }],
            ),
        ))
    batch = client.messages.batches.create(requests=requests)
    return batch.id
 
def collect_batch(batch_id: str) -> dict:
    batch = client.messages.batches.retrieve(batch_id)
    if batch.processing_status != "ended":
        return {}
    results = {}
    for entry in client.messages.batches.results(batch_id):
        if entry.result.type == "succeeded":
            results[entry.custom_id] = entry.result.message.content[0].text
        else:
            results[entry.custom_id] = f"ERROR: {entry.result.type}"
    return results

def ask_about_pdf(pdf_data: str, question: str) -> str:
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": [
                {"type": "document",
                 "source": {"type": "base64", "media_type": "application/pdf", "data": pdf_data},
                 "cache_control": {"type": "ephemeral"}},  # ここがキャッシュ指定
                {"type": "text", "text": question},
            ],
        }],
    )
    usage = message.usage
    print(f"cache write: {usage.cache_creation_input_tokens}, "
          f"cache read: {usage.cache_read_input_tokens}")
    return message.content[0].text

import hashlib, json
from pathlib import Path
 
class ResultCache:
    def __init__(self, cache_dir: str = ".vision_cache"):
        self.dir = Path(cache_dir)
        self.dir.mkdir(exist_ok=True)
 
    def key(self, image_bytes: bytes, prompt: str) -> str:
        return hashlib.sha256(image_bytes + prompt.encode()).hexdigest()
 
    def get(self, image_bytes: bytes, prompt: str):
        f = self.dir / f"{self.key(image_bytes, prompt)}.json"
        return json.loads(f.read_text()) if f.exists() else None
 
    def put(self, image_bytes: bytes, prompt: str, result: dict):
        f = self.dir / f"{self.key(image_bytes, prompt)}.json"
        f.write_text(json.dumps(result, ensure_ascii=False))

import time
 
def analyze_with_recovery(path: str, prompt: str, max_retries: int = 3):
    """エラー種別ごとに対処を分けた解析ラッパー"""
    try:
        data, media_type = resize_for_claude(path)
    except Exception as e:
        return {"status": "invalid_image", "error": str(e)}  # 隔離キューへ
 
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "image",
                         "source": {"type": "base64", "media_type": media_type, "data": data}},
                        {"type": "text", "text": prompt},
                    ],
                }],
            )
            return {"status": "ok", "text": message.content[0].text,
                    "input_tokens": message.usage.input_tokens}
        except anthropic.BadRequestError as e:
            return {"status": "bad_request", "error": str(e)}  # リトライしない
        except (anthropic.RateLimitError, anthropic.InternalServerError):
            wait = 2 ** attempt * 5
            time.sleep(wait)
    return {"status": "retry_exhausted"}

DIFF_TOOL = {
    "name": "record_differences",
    "description": "2つのドキュメントの相違点を記録します",
    "input_schema": {
        "type": "object",
        "properties": {
            "differences": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "section": {"type": "string", "description": "相違がある箇所"},
                        "document_a": {"type": "string", "description": "1枚目の記載内容"},
                        "document_b": {"type": "string", "description": "2枚目の記載内容"},
                        "significance": {"type": "string", "enum": ["substantive", "formatting"]},
                    },
                    "required": ["section", "document_a", "document_b", "significance"],
                },
            },
            "identical": {"type": "boolean"},
        },
        "required": ["differences", "identical"],
    },
}
 
def compare_documents(path_a: str, path_b: str) -> dict:
    data_a, type_a = resize_for_claude(path_a)
    data_b, type_b = resize_for_claude(path_b)
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        tools=[DIFF_TOOL],
        tool_choice={"type": "tool", "name": "record_differences"},
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "1枚目（旧版）:"},
                {"type": "image",
                 "source": {"type": "base64", "media_type": type_a, "data": data_a}},
                {"type": "text", "text": "2枚目（新版）:"},
                {"type": "image",
                 "source": {"type": "base64", "media_type": type_b, "data": data_b}},
                {"type": "text",
                 "text": "2つのドキュメントを比較し、すべての相違点を記録してください。"
                         "文言の実質的な変更と、レイアウトだけの変更は区別してください。"},
            ],
        }],
    )
    for block in message.content:
        if block.type == "tool_use":
            return block.input
    return {"differences": [], "identical": False}