ユーザー投稿コンテンツ（テキスト・画像）をAIで自動判定するロジック

ご挨拶

初めまして。
アジアクエスト デジタルエンジニアリング部 AI/DATAエンジニアリング課の辻口です。 現在私が参画しているプロジェクトにて、ユーザーが投稿したテキストや画像が適切かどうかをAIで自動判定する仕組みを構築しました。
今回は、その仕組みと実装についてご紹介します。

概要

Webサービスにおいて、ユーザーが投稿するテキストや画像には、不適切な内容が含まれる場合があります。
これらを人手で一つずつ確認するのは運用コストが高いため、Amazon Bedrockを活用した自動判定の仕組みを構築しました。

アーキテクチャ図

詳細

処理概要

1. LambdaがS3から対象の画像を取得する
2. 事前に用意したプロンプトと画像データを合わせてBedrockへ送信する
3. Bedrockから判定結果（JSON）を受け取る

プロンプト（一部抜粋）

ユーザーが投稿した画像を、以下のカテゴリに基づき評価してください。

■ 分類カテゴリ（優先度順）
・SS（重大な違反）：盗作・著作権侵害・なりすまし、違法行為の描写、性的な内容
・S（深刻な問題）：個人情報（住所・氏名・電話番号等）、暴言・嫌がらせ、ヘイトスピーチ
・A（ポリシー違反）：外部リンク、広告・販促、報酬についての言及、勧誘行為
・B（品質的な問題）：非対応言語、スパム・繰り返し、絵文字のみの画像
・C（参考情報）：サービス運営・価格に関するコメント

■ 返答形式（JSON）
{
  "status": "ok" | "false" | "and",
  "matches": ["SS", "S", "A", "B", "C"],
  "reason": "判定理由を日本語で簡潔に記載"
}

■ statusの判定基準
・"false"：上記カテゴリに1つでも該当する場合
・"and"：AIでは判断しきれず、人間の確認が必要な場合
・"ok"：問題・懸念点が見当たらない場合

いずれの場合も"reason"に判定理由を日本語で返してください。

プロンプト作成のポイント

• できる限り大枠から詳細へ記述する
  • 優先度の高いものから判定するよう促す
  • 大きな枠組みから詳細に詰めていく（推奨）
• 出力形式の大枠を先に記述する
  • 詳細な条件はその後に書く

サンプルコード

import ast
import base64
import json
import mimetypes
import re
from typing import Any, Dict, Optional, Union

import boto3

# AWSクライアントの初期化
s3_client = boto3.client("s3")
bedrock_client = boto3.client("bedrock-runtime")

# 使用するBedrockモデルID
BEDROCK_MODEL_ID = "anthropic.claude-3-sonnet-20240229-v1:0"


# ============================================================
# 1. S3から画像を取得
# ============================================================
def get_image_from_s3(bucket: str, key: str) -> bytes:
    """S3バケットから画像オブジェクトをバイト列として取得する。"""
    try:
        resp = s3_client.get_object(Bucket=bucket, Key=key)
        return resp["Body"].read()
    except Exception as e:
        raise Exception(f"S3取得エラー: {repr(e)}")


# ============================================================
# 2. ファイル名からMIMEタイプを推測
#    - 画像系のMIMEタイプのみ返却、それ以外はNone
# ============================================================
def guess_image_mime(filename: str) -> Optional[str]:
    """ファイル名から画像のMIMEタイプを推測する。"""
    mt, _ = mimetypes.guess_type(filename)
    return mt if (mt and mt.startswith("image/")) else None


# ============================================================
# 3. Bedrock API用のペイロード構築
#    - テキスト入力 → そのままtextブロックとして追加
#    - 画像入力 → Base64エンコードしてimageブロックとして追加
# ============================================================
def build_bedrock_payload(
    input_obj: Union[str, bytes],
    prompt: str,
    media_type: str = "image/png",
    max_tokens: int = 1024,
) -> Dict[str, Any]:
    """Bedrock (Claude) へ送信するリクエストペイロードを構築する。"""
    # まずプロンプトをセット
    content = [{"type": "text", "text": prompt}]

    if isinstance(input_obj, str):
        # テキスト投稿の場合
        content.append({"type": "text", "text": input_obj})
    elif isinstance(input_obj, (bytes, bytearray)):
        # 画像投稿の場合 → Base64エンコードして送信
        content.append({
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": media_type,
                "data": base64.b64encode(input_obj).decode("utf-8"),
            },
        })
    else:
        return {}

    return {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": content}],
    }


# ============================================================
# 4. Bedrock API呼び出し
# ============================================================
def call_bedrock(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Bedrockモデルを呼び出し、レスポンスをdictで返す。"""
    try:
        resp = bedrock_client.invoke_model(
            modelId=BEDROCK_MODEL_ID,
            body=json.dumps(payload),
            contentType="application/json",
            accept="application/json",
        )
        result = json.loads(resp["body"].read())
        # レスポンスのtextフィールドからJSON部分を抽出
        return parse_bedrock_response(result["content"][0]["text"])
    except Exception as e:
        raise Exception(f"Bedrock呼び出しエラー: {repr(e)}")


# ============================================================
# 5. Bedrockレスポンスのパース
#    LLMの出力は必ずしも綺麗なJSONとは限らないため、
#    複数の方法でパースを試みる
# ============================================================
def parse_bedrock_response(output: Any) -> Dict[str, Any]:
    """Bedrockの応答テキストからJSONオブジェクトを抽出する。"""
    if isinstance(output, (bytes, bytearray)):
        output = output.decode("utf-8", errors="replace")
    text = str(output).strip()

    # 試行1: そのままJSONパース（二重エンコード対策込み）
    try:
        obj = json.loads(text)
        if isinstance(obj, str):
            obj = json.loads(obj)
        if isinstance(obj, dict):
            return obj
    except json.JSONDecodeError:
        pass

    # 試行2: ```json ... ``` などのコードフェンスを除去してパース
    cleaned = re.sub(
        r"^\s*```(?:json)?\s*|\s*```\s*$", "",
        text, flags=re.IGNORECASE | re.MULTILINE,
    )
    # 説明文が混在している場合、最初の {...} ブロックだけ抽出
    m = re.search(r"\{.*\}", cleaned, flags=re.DOTALL)
    if m:
        try:
            obj = json.loads(m.group(0).strip())
            if isinstance(obj, dict):
                return obj
        except json.JSONDecodeError:
            pass

    # 試行3: Pythonリテラル形式（シングルクォート等）への対応
    try:
        obj = ast.literal_eval(cleaned)
        if isinstance(obj, dict):
            return obj
    except Exception:
        pass

    # すべて失敗した場合は空dictを返却
    return {}

発展：実運用に向けた構成例

今回はS3 → Lambda → Bedrockというシンプルな構成で紹介しましたが、実際のサービスに組み込む場合は以下のようなAWSサービスと組み合わせることで、より実用的な構成にできます。

• API Gateway + Lambda：外部システムからHTTPリクエストで判定を呼び出せるようにする。投稿データをAPIで受け取り、Lambdaで判定処理を実行する構成です。
• Step Functions：判定処理をワークフローとして管理する。例えば「リクエスト受付 → 非同期でAI判定 → 結果をコールバックAPIへPOST」といった複数ステップの処理をStep Functionsで制御することで、可視性・エラーハンドリングが向上します。
• コールバック方式：判定結果をWeb Server側のAPIにPOSTで返す構成にすれば、非同期処理との相性が良くなります。呼び出し元はAPIを叩いた時点で200を返し、判定完了後にコールバックで結果を受け取る形です。

これらを組み合わせることで、今回紹介した判定ロジックをそのまま活かしつつ、実際のサービスに組み込むことが可能です。
ぜひ用途に合わせて構成を検討してみてください。