Claude APIで画像の不適切コンテンツを検出・除外する実装メモ

AI

ユーザーが画像をアップロードできる機能を作ろうとして、最初に頭を悩ませたのが「不適切な画像をどう弾くか」という問題でした。テキストのモデレーションは前にも触ったことがあったんですが、画像は初めてで、想像以上に考えることが多かったです。

この記事では、Claude APIのVision機能を使って画像の不適切コンテンツを検出し、処理フローから除外する実装について書いていきます。正直まだ完成形ではないんですが、動くところまでは来たのでまとめておきます。

  • Claude APIに画像を渡す基本的な方法
  • モデレーション用プロンプトの設計方針
  • 不適切コンテンツの検出・除外を行う実装コード
  • バッチ処理での並列化とレートリミットの扱い
  • 実際にやってみてわかったハマりポイント

そもそもなぜClaudeで画像モデレーションなのか

画像の不適切コンテンツ検出といえば、Google Cloud VisionのSafeSearch APIとか、AWS RekognitionのContent Moderationとか、専用のサービスがすでに存在しています。そっちを使えば話が早いんですが、今回は以下の理由でClaudeを選びました。

  • すでにテキストモデレーションでClaudeを使っていたので、APIキーを統一したかった
  • 「なぜ不適切なのか」という理由をある程度自然言語で返してほしかった
  • カテゴリの定義を柔軟に変えたかった(日本語コンテキストで判断させたい場面があった)

ClaudeのVision(画像の理解・分析)に対応しているのは、少なくとも「Claude 3 と Claude 4 系の画像対応モデル」です(モデルによってはテキスト専用もあるので、そこだけ注意です)。

余談ですが、専用サービスと比べるとコストと精度のトレードオフはあります。大量の画像を高速に処理したいなら素直に専用サービスを使うのが正直おすすめです。

画像をAPIに渡す基本の形

まず前提として、ClaudeのMessages APIに画像を渡す方法は主に2つあります。

  • Base64エンコード:画像データをBase64に変換してリクエストに含める
  • URLで指定:公開されている画像のURLを直接渡す

ローカルファイルを扱う場合はBase64が基本になります。こんな形です。

import anthropic
import base64

def load_image_as_base64(image_path: str) -> tuple[str, str]:
    with open(image_path, "rb") as f:
        data = base64.standard_b64encode(f.read()).decode("utf-8")
    
    # 拡張子からメディアタイプを判定(雑な実装なので本番は要改善)
    ext = image_path.rsplit(".", 1)[-1].lower()
    media_type_map = {
        "jpg": "image/jpeg",
        "jpeg": "image/jpeg",
        "png": "image/png",
        "gif": "image/gif",
        "webp": "image/webp",
    }
    media_type = media_type_map.get(ext, "image/jpeg")
    return data, media_type

あと、Files APIでファイルを事前にアップロードして、Messagesのリクエスト内ではfile_id参照にするやり方もあります。同じ画像(や同じ資料)を何度も使うケースだと、毎回Base64を詰め込むより扱いやすくなる場面があります。

不適切コンテンツ検出のためのプロンプト設計

ここが一番悩んだところです。テキストのモデレーションと違って、画像の場合は「何を問題とみなすか」の定義がより大事になってきます。

Claudeの公式ドキュメントでも、unsafe categories(不適切カテゴリ定義)を用意して分類させるアプローチが紹介されています。これを画像向けに応用するとこんな感じになります。

UNSAFE_IMAGE_CATEGORIES = {
    "Sexual": "性的な行為や露出を含む画像。成人向けコンテンツ全般。",
    "Violence": "暴力的・残酷な描写を含む画像。リアルな怪我・血液の描写など。",
    "HateSpeech": "特定の人種・宗教・性別などを侮辱・差別する視覚的コンテンツ。",
    "ChildSafety": "未成年者が関わる性的・有害なコンテンツ。絶対に許可しない。",
    "IllegalContent": "違法行為を直接描写・助長する画像。",
}

カテゴリの粒度はユースケース次第で変えてください。自分の場合はユーザー投稿型のサービスを想定していたので、このくらいの分類にしました。

プロンプトはこんな形で組みます。

def build_moderation_prompt(categories: dict) -> str:
    category_str = "\n".join(
        f"- {name}: {desc}" for name, desc in categories.items()
    )
    return f"""
以下の画像を確認し、不適切なコンテンツが含まれているかを判定してください。

判定するカテゴリ:
{category_str}

以下のJSON形式のみで回答してください。余分なテキストは不要です。
{{
  "is_unsafe": true または false,
  "violated_categories": ["カテゴリ名", ...],
  "risk_level": 0〜10の整数(0=安全、10=最高リスク),
  "reason": "判定理由を1〜2文で(安全な場合は空文字列)"
}}
"""

JSONで返させるのは後続の処理でパースしやすいからです。ただClaudeがたまに余計なテキストを前後に付けてくることがあって、そこは注意が必要です(後述)。

実際のモデレーション関数

全体をまとめた関数がこちらです。

import anthropic
import json
import re

client = anthropic.Anthropic()

def moderate_image(image_path: str, threshold: int = 3) -> dict:
    """
    画像の不適切コンテンツを検出する。
    threshold: このリスクレベル以上で「不適切」と判定(0〜10)
    """
    image_data, media_type = load_image_as_base64(image_path)
    prompt = build_moderation_prompt(UNSAFE_IMAGE_CATEGORIES)

    response = client.messages.create(
        model="claude-haiku-3-5-20241022",  # 画像対応モデルを指定(例)
        max_tokens=512,
        temperature=0,  # 判定ブレを減らすため0固定
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": prompt,
                    },
                ],
            }
        ],
    )

    # contentは複数ブロックになることがあるので、textだけ連結して扱う
    raw_text = "".join([b.text for b in response.content if getattr(b, "type", None) == "text"])
    result = parse_json_response(raw_text)
    result["should_exclude"] = (
        result.get("is_unsafe", False)
        or result.get("risk_level", 0) >= threshold
    )
    return result


def parse_json_response(text: str) -> dict:
    # Claudeが```json ... ```で囲む場合があるので取り除く
    cleaned = re.sub(r"```json\s*|\s*```", "", text).strip()
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        # ここ注意:パース失敗時は安全側に倒す(除外扱い)
        return {
            "is_unsafe": True,
            "violated_categories": [],
            "risk_level": -1,
            "reason": "レスポンスのパースに失敗しました",
        }

parse_json_response の中でパース失敗時に is_unsafe: True を返しているのは意図的です。判定できなかった場合は安全側(除外する方向)に倒したほうがいいと思っているので。

モデレーション処理は軽量モデルから始めて、精度が足りなければ上位モデルに切り替えるというのはやりやすい方針だと思います。ただし、モデル名は時期で増えたり変わったりするので、使う直前に公式の「Models」一覧から画像対応かどうかを確認するのが確実です。

複数画像をまとめて処理する(バッチ除外フロー)

ユーザー投稿の処理とか、画像のバッチ処理を想定すると、複数ファイルを一気に処理したくなります。

from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

def moderate_images_batch(image_paths: list[str], max_workers: int = 5) -> list[dict]:
    results = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_path = {
            executor.submit(moderate_image, path): path
            for path in image_paths
        }
        for future in as_completed(future_to_path):
            path = future_to_path[future]
            try:
                result = future.result()
                result["image_path"] = path
                results.append(result)
            except Exception as e:
                results.append({
                    "image_path": path,
                    "is_unsafe": True,
                    "should_exclude": True,
                    "reason": f"処理エラー: {e}",
                })

    return results


def filter_safe_images(image_paths: list[str]) -> tuple[list[str], list[dict]]:
    """安全な画像のパスリストと、除外された画像の情報を返す"""
    all_results = moderate_images_batch(image_paths)
    safe = [r["image_path"] for r in all_results if not r.get("should_exclude")]
    excluded = [r for r in all_results if r.get("should_exclude")]
    return safe, excluded

ThreadPoolExecutorで並列処理を入れています。max_workers=5 はひとつの目安で、実際はアカウントのレートリミット(requests / tokens)次第で調整が必要です。429が返ってきたときは、レスポンスヘッダのretry-afterを見て待つのが基本になります。

使い方はこんな感じです。

image_list = [
    "uploads/photo_001.jpg",
    "uploads/photo_002.png",
    "uploads/photo_003.webp",
]

safe_images, excluded_images = filter_safe_images(image_list)

print(f"通過: {len(safe_images)}件")
print(f"除外: {len(excluded_images)}件")

for item in excluded_images:
    print(f"  {item['image_path']} - {item.get('reason', '')}")

やってみてわかったこと・ハマりポイント

temperature は0固定にする

最初 temperature=0.3 くらいで試していたら、同じ画像でも判定結果がたまにブレることがありました。モデレーションは一貫性が重要なので、temperature=0 固定が無難だと思います。

除外閾値の調整が難しい

risk_level の 0〜10 スケール、実際に試してみると 3〜4 あたりに微妙なケースが集中しやすいです。「グロテスクだけどフィクションのイラスト」とか「ちょっとセクシーだけどセーフ寄り」みたいなやつ。閾値をどこに設定するかはユースケース次第で、自分でサンプルを集めてテストするしかないところです。

日本語コンテキストは意外とうまく効く

カテゴリの説明を日本語で書いたら、文化的なコンテキストを含む判断がそこそこうまくいきました。Claudeは安全性のための学習やガードレールが入っている前提なので、プロンプトで「何を不適切として扱うか」を日本語で補足してあげると、判定が安定する印象があります。

コスト感覚

画像の処理コストは、ざっくり「画像も入力トークンとして換算される」タイプで、画像が大きいほどコストも上がります。なので数千枚単位で回すなら、事前に画像サイズを落とす(必要十分な解像度にする)とか、そもそも専用のモデレーションAPIに寄せるとか、どこかで線引きは必要になりそうです。そういえば最近、自分のiPhoneの写真が無駄に高解像度で、アップロード前に縮小する処理を挟むだけでだいぶ財布に優しくなりました。

まだ解決できていないこと

正直、この実装で完璧かというと全然そんなことはなくて、いくつか課題が残っています。

一番気になっているのは偽陰性(見逃し)の問題です。特にイラスト・漫画系のコンテンツはリアル画像より甘めに見える傾向があります。Claudeを含むAIモデルは高性能であっても誤判定する可能性があるため、重要度の高いシステムなら二重チェック(例:Rekognition等との併用)にするのが現実的かもしれません。

あとClaudeの安全設計の都合で、一部の内容は「APIが応答を返せない / エラーになる」ケースがあります。これはこれで「除外すべき画像」と解釈できますが、実装としてはエラーハンドリング(リトライせず隔離、監査ログに残す等)を別途考えておくのがよさそうです。

「大量のコンテンツをまとめて非同期に投げたい」という話だと、Anthropic側にはMessage Batches APIがあります(通常のMessagesとは別枠でレートリミットがある点も含めて要確認)。モデレーションみたいな大量処理には相性が良さそうです。

※この記事にはプロモーションが含まれます

ちなみに、PLAUD NOTE(AIボイスレコーダー。録音から文字起こし・要約まで自動で行える)も気になっています。PLAUD NOTE

まとめ

Claude APIのVision機能を使った画像の不適切コンテンツ検出・除外の実装をざっくりまとめました。

  • 画像はBase64またはURLでAPIに渡せる。ローカルファイルはBase64が基本
  • 不適切カテゴリを定義してJSON形式で返させるとパースしやすい
  • 判定ブレを防ぐために temperature=0 は必須
  • パース失敗時は除外方向(安全側)に倒す設計にしておく
  • バッチ処理はThreadPoolExecutorで並列化できるが、レートリミットに注意
  • 大量処理が必要になったらMessage Batches APIも選択肢に入る

専用のモデレーションAPIと比べると万能ではないですが、「すでにClaudeを使っている」「理由も一緒に返したい」「日本語コンテキストで判断させたい」みたいな場面には割と合っていると思います。閾値の調整は地道にやるしかないですが、そこは運用しながら育てていく感じで。

参考になったらクリックしてもらえると嬉しいです!

Blogmura AIAI Ranking
タイトルとURLをコピーしました