SRA Advent Calendar 2025

Amazon Bedrockでマルチモーダルなエージェントを作ってテストしてみる

Last updated at 2025-12-17Posted at 2025-12-17

TL;DR

Amazon Bedrockは最短でエージェント検証ができる一方、現時点ではマネジメントコンソールのテスト画面やInvokeAgentから画像ファイルを直接入力できないため、マルチモーダル機能の検証には工夫が必要です。
本記事ではそれらの制約をS3＋Lambdaのカスタムアクションで回避してマルチモーダル検証する手順を整理しました。

背景

いまさらBedrock？Amazon Bedrock AgentCoreではなく？

Amazon Bedrock は、FAQチャットボットやナレッジベース統合型エージェントを最速(？)で検証できるAWSのマネージドサービスです。AWSマネジメントコンソールからの設定だけでも、簡単なプロンプトエンジニアリング、メモリ管理、ナレッジベースの作成と連携が自動的に処理されるため、複雑なインフラ構築やコード実装なしで短時間でエージェントを構築・テストできます。
個人的には、RAGを中心としたナレッジベースの検証・評価が一番の使いどころかな、と思っています。

マルチモーダルエージェントをテストする際の制約

Amazon Bedrock ではイメージとテキストをあわせて処理するマルチモーダルなエージェント自体は作成できますが、実際にテストしたい場合、以下の制約があります（2025年12月執筆時点）。

コンソールのテストウィンドウ: 画像ファイルを直接添付して送信することができない
InvokeAgent API: 画像ファイル（JPEG、PNG、JPG）を直接入力として受け付けない

これらの制約により、エージェントのマルチモーダル機能（画像分析）を利用するには、実装レベルで回避策が必要となります。

アーキテクチャ

今回のアプローチ

今回は、S3とカスタムアクション（Lambda関数） を使用して上記の制約を回避し、以下のフローを実装します。

ユーザーがS3に画像をアップロード
その画像パス（URLまたはURI）と質問文をテキストでBedrock エージェントを呼び出し
Bedrock エージェントがアクショングループでカスタムアクション（Lambda）を呼び出し
Lambda関数がS3から画像を取得し、 Bedrock Runtime API（InvokeModel） を使用してモデルに画像を送信して解析
エージェントが分析結果と元の質問文から回答を生成してユーザーに返却

アーキテクチャイメージ

環境準備

利用するAWSサービスとアクセス権限

最低限以下のサービスが利用できるAWSアカウントが必要です。

Amazon Bedrock: 利用するモデル（今回はClaude 3.7 Sonnet）のモデルアクセスを事前に有効化
AWS Lambda: 関数作成・更新権限
Amazon S3: バケット作成・オブジェクトアップロード権限
AWS IAM: ロール作成・ポリシーアタッチ権限
Amazon CloudWatch Logs: ログ閲覧権限

使用リージョン

今回は us-east-1（バージニア北部） を使用しました。他のリージョンを利用される場合は、そのリージョンで利用可能なモデルを確認してください。

開発環境

筆者は以下の環境で実施しました。

Python: 3.13.5
boto3 （AWS SDK for Python）: 1.42.8
AWS CLI: 2.27.57 ※boto3 の認証情報としてのみの利用です。AWS側の設定はほぼすべてマネジメントコンソールから行います。

実装手順

今回は社内ヘルプデスクのチャットボットのイメージで実装を進めます。

Phase 1: S3バケットの準備

S3バケット作成

AWSマネジメントコンソールにログイン
サービス → S3 を選択
バケットを作成 ボタンをクリック
以下の設定を入力：
- バケット名: your-bedrock-test-bucket（任意の一意な名前）
- AWSリージョン: 米国東部 (バージニア北部) us-east-1
- オブジェクト所有者: ACL 無効（推奨）
- パブリックアクセスをすべてブロック: 有効（デフォルト）
バケットを作成 をクリック

フォルダ作成

作成したバケットを選択
フォルダの作成 ボタンをクリック
フォルダ名: agent-uploads と入力
フォルダの作成 をクリック

テスト画像のアップロード

今回はナレッジベースを用意していないこともあり、簡単(？)なMicrosoft 365 Apps（Office）のインストールエラー画面でテストしてみます。

agent-uploads フォルダを開く
アップロード ボタンをクリック
ファイルを追加 をクリックし、テスト用画像（JPEG/PNG）を選択
アップロード をクリック

Phase 2: Lambda関数の作成

IAM実行ロールの作成

Lambda関数が必要とする権限を持つIAMロールを作成します。

AWSマネジメントコンソール → IAM → ロール を選択
ロールを作成 をクリック
信頼されたエンティティタイプ: AWS のサービス
ユースケース: Lambda を選択
次へをクリック

許可ポリシーの追加:
以下のポリシーを検索して追加：

AWSLambdaBasicExecutionRole（CloudWatch Logs用）
AmazonBedrockFullAccess（Bedrock InvokeModel用）
AmazonS3ReadOnlyAccess（S3読み取り用）

今回はテスト目的のため、以下の管理ポリシーを使用しています。本番運用では最小権限のカスタムポリシーを作成することを強く推奨します。

次へをクリック
ロール名: BedrockImageAnalyzerFromS3-ExecutionRole
ロールを作成 をクリック

Lambda関数の作成

AWSマネジメントコンソール → Lambda → 関数を選択
関数の作成 をクリック
一から作成 を選択
以下の設定を入力：
- 関数名: BedrockImageAnalyzerFromS3
- ランタイム: Python 3.14
- アーキテクチャ: x86_64
- 実行ロール: 既存のロールを使用する → BedrockImageAnalyzerFromS3-ExecutionRole
関数の作成 をクリック

Lambda関数コードの設定

コードソースエディタで、以下のコードを lambda_function.py に貼り付けます：

import json
import boto3
import base64
from botocore.exceptions import ClientError
import re

# クライアント初期化
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')
s3 = boto3.client('s3', region_name='us-east-1')

def lambda_handler(event, context):
    """
    S3 URIから画像を取得してClaude 3.7 Sonnetで分析
    """
    print(f"Event received: {json.dumps(event, ensure_ascii=False)}")
    
    try:
        # パラメータ取得
        parameters = event.get('parameters', [])
        params_dict = {p['name']: p['value'] for p in parameters}
        
        s3_uri = params_dict.get('s3_image_uri', '').strip()
        user_prompt = params_dict.get('prompt', '画像の内容を詳しく説明してください。')
        
        # S3 URI検証
        if not s3_uri or not s3_uri.startswith('s3://'):
            return create_agent_response(
                event,
                {'error': 's3_image_uri パラメータが必要です（形式: s3://bucket/key）'}
            )
        
        # S3 URIをパース
        bucket, key = parse_s3_uri(s3_uri)
        print(f"S3から画像取得: Bucket={bucket}, Key={key}")
        
        # S3から画像取得
        try:
            s3_response = s3.get_object(Bucket=bucket, Key=key)
            image_data = s3_response['Body'].read()
            content_type = s3_response.get('ContentType', 'image/jpeg')
            
            # Base64エンコード
            image_base64 = base64.b64encode(image_data).decode('utf-8')
            print(f"画像取得成功: サイズ={len(image_data)} bytes, ContentType={content_type}")
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'NoSuchKey':
                return create_agent_response(
                    event,
                    {'error': f'画像が見つかりません: {s3_uri}'}
                )
            elif error_code == 'NoSuchBucket':
                return create_agent_response(
                    event,
                    {'error': f'バケットが見つかりません: {bucket}'}
                )
            else:
                raise
        
        # メディアタイプマッピング
        media_type_map = {
            'image/jpeg': 'image/jpeg',
            'image/jpg': 'image/jpeg',
            'image/png': 'image/png',
            'image/gif': 'image/gif',
            'image/webp': 'image/webp'
        }
        media_type = media_type_map.get(content_type.lower(), 'image/jpeg')
        
        # Claude 3.7 Sonnetを呼び出し
        model_id = 'us.anthropic.claude-3-7-sonnet-20250219-v1:0'
        print(f"モデルに画像分析をリクエスト")
        print(f"Model ID: {model_id}")
        print(f"プロンプト: {user_prompt[:100]}...")
        
        bedrock_response = bedrock.invoke_model(
            modelId=model_id,
            contentType='application/json',
            accept='application/json',
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 4096,
                "messages": [{
                    "role": "user",
                    "content": [
                        {
                            "type": "image",
                            "source": {
                                "type": "base64",
                                "media_type": media_type,
                                "data": image_base64
                            }
                        },
                        {
                            "type": "text",
                            "text": user_prompt
                        }
                    ]
                }]
            })
        )
        
        # レスポンス解析
        response_body = json.loads(bedrock_response['body'].read())
        analysis_result = response_body['content'][0]['text']
        
        print(f"分析完了: {len(analysis_result)} 文字")
        
        # エージェントにレスポンスを返す
        return create_agent_response(
            event,
            {
                'analysis': analysis_result,
                'model_used': model_id,
                'image_source': s3_uri,
                'media_type': media_type,
                'image_size_bytes': len(image_data)
            }
        )
        
    except Exception as e:
        print(f"Error: {str(e)}")
        import traceback
        traceback.print_exc()
        
        return create_agent_response(
            event,
            {'error': f'画像分析中にエラーが発生しました: {str(e)}'}
        )

def parse_s3_uri(s3_uri):
    """
    S3 URIをbucketとkeyに分解
    """
    match = re.match(r's3://([^/]+)/(.+)', s3_uri)
    if not match:
        raise ValueError(f"無効なS3 URI形式: {s3_uri}")
    
    bucket = match.group(1)
    key = match.group(2)
    
    return bucket, key


def create_agent_response(event, body):
    """
    Bedrock エージェント用のレスポンス作成
    """
    return {
        'messageVersion': '1.0',
        'response': {
            'actionGroup': event.get('actionGroup', ''),
            'function': event.get('function', ''),
            'functionResponse': {
                'responseBody': {
                    'TEXT': {
                        'body': json.dumps(body, ensure_ascii=False)
                    }
                }
            }
        }
    }

Deploy ボタンをクリックしてコードを保存します。

Lambda関数の設定変更

一般設定の変更:

設定タブ → 一般設定 を選択
編集をクリック
以下の設定を変更：
- タイムアウト: 1分 0秒（Bedrock API呼び出しに時間がかかるため）
- メモリ: 256 MB（画像のBase64エンコード処理のため）
保存をクリック

Lambda関数のテスト

テスト タブを選択
新しいイベントを作成 を選択
イベント名: TestS3Image
イベント JSON に以下を入力（バケット名とキーは自身の環境に合わせて変更）：

{
  "messageVersion": "1.0",
  "function": "analyze_s3_image",
  "parameters": [
    {
      "name": "s3_image_uri",
      "type": "string",
      "value": "s3://your-bedrock-test-bucket/agent-uploads/test_image.jpg"
    },
    {
      "name": "prompt",
      "type": "string",
      "value": "この画像について説明してください"
    }
  ],
  "actionGroup": "ImageAnalysisFromS3"
}

5. テスト ボタンをクリック
6. 実行結果が成功することを確認

画像内の文字を読めているようですね

Phase 3: Bedrock エージェントの設定

Bedrock エージェントの作成

AWSマネジメントコンソール → Amazon Bedrock を選択
左側メニューの構築 → エージェント を選択
エージェントを作成 をクリック
以下の設定を入力：
- エージェント名: ImageAnalyzerAgent（任意）
- エージェントの説明: S3画像を分析するマルチモーダルエージェント
作成をクリック

エージェントの詳細設定

モデルの選択:

モデルの詳細 セクションで編集をクリック
モデル: Anthropic Claude 3.5 Sonnetを選択
保存をクリック

エージェントの指示:

エージェントへの指示 セクションで以下を入力：

あなたは画像分析が可能なAIアシスタントです。

ユーザーが画像について質問した場合:
1. 入力テキストから「画像パス: s3://...」または「画像パス: https://...」の形式でS3 URIを抽出
2. analyze_s3_image 関数を呼び出し
   - s3_image_uri パラメータ: 抽出したS3 URI
   - prompt パラメータ: ユーザーの質問内容
3. 分析結果をユーザーに分かりやすく日本語で説明

例:
入力: "この画像について説明してください。
画像パス: s3://my-bucket/uploads/test.jpg"
→ analyze_s3_image(
    s3_image_uri="s3://my-bucket/uploads/test.jpg",
    prompt="この画像について説明してください。"
  )

S3 URIが見つからない場合は、画像パスを確認するようユーザーに依頼してください。

保存をクリック

アクショングループの追加

アクショングループ セクションの追加をクリック
以下の設定を入力：
- アクショングループ名: ImageAnalysisFromS3
- 説明: S3画像分析アクション
アクショングループタイプ: 関数の詳細を定義 を選択
アクショングループの呼び出し: 既存の Lambda 関数を選択してくださいを選択
- Lambda関数: BedrockImageAnalyzerFromS3 を選択
関数を追加 をクリック

関数の詳細:

関数名: analyze_s3_image
説明: S3に保存された画像を分析します

パラメータを追加:

パラメータ名	説明	タイプ	必須
`s3_image_uri`	S3画像のURI (s3://bucket/key または HTTPS URL)	String	True
`prompt`	画像分析のためのプロンプト	String	False

作成をクリック

Lambdaリソースベースポリシーの追加

Bedrock エージェントがLambda関数を呼び出せるよう、リソースベースポリシーを追加します。

AWSマネジメントコンソール → Lambda → 関数 → BedrockImageAnalyzerFromS3 を選択
設定タブ → アクセス権限 を選択
リソースベースのポリシーステートメント セクションの アクセス許可を追加 をクリック
AWSのサービス を選択
以下の設定を入力：
- サービス: Other
- ステートメント ID: AllowBedrockAgentInvocation
- プリンシパル: bedrock.amazonaws.com
- ソース ARN: arn:aws:bedrock:us-east-1:<AWSアカウントID>:agent/<エージェントID>
  - エージェントIDは、Bedrock エージェントの詳細画面で確認できます
- アクション: lambda:InvokeFunction
保存をクリック

エージェントの準備（Prepare）

Amazon Bedrock → エージェント → 該当エージェントを選択
画面上部の準備ボタンをクリック
ステータスが「準備完了」になるまで待機

アクショングループを追加・変更した後は、必ず準備を実行してください。

Phase 4: テストと検証

Bedrockコンソールでのテスト

Amazon Bedrock → エージェント → 該当エージェントを選択
右側の テスト パネルを開く
チャットに以下を入力

この画像について説明してください。
画像パス: s3://your-bedrock-test-bucket/agent-uploads/test_image.jpg

期待される出力:

画像内容の詳細な説明
エラーコード（0-2032）の読み取り

いい感じですね。

トレースの確認

エージェントが内部で期待どおりの動作をしているか、テストパネルで トレースを表示 を有効にして、以下を確認します。

アクショングループ呼び出し（actionGroupInvocationInput）: analyze_s3_image 関数が呼び出されているか
パラメータ（Parameters）: s3_image_uri と prompt が正しく渡されているか

Pythonクライアントからのテスト

疑似フロントエンドとして以下のスクリプトを作成し、ローカル環境からテストします。

invoke_agent_with_s3_image.py:

import boto3
import json
import os
from pathlib import Path
import mimetypes

# 設定（自身の環境に合わせて変更）
AWS_REGION = 'us-east-1'
S3_BUCKET_NAME = 'your-bedrock-test-bucket'  # 実際のバケット名に変更
S3_UPLOAD_PREFIX = 'agent-uploads/'  # アップロード先プレフィックス
AGENT_ID = 'XXXXXXXXXX'  # 実際のAgent IDに変更
AGENT_ALIAS_ID = 'XXXXXXXXXX'  # 実際のAlias IDに変更

# クライアント作成
s3_client = boto3.client('s3', region_name=AWS_REGION)
bedrock_client = boto3.client('bedrock-agent-runtime', region_name=AWS_REGION)


def upload_image_to_s3(image_path):
    """
    画像をS3にアップロード
    
    Args:
        image_path (str): ローカル画像ファイルパス
        
    Returns:
        str: S3 URI (s3://bucket/key)
    """
    if not os.path.exists(image_path):
        raise FileNotFoundError(f"画像ファイルが見つかりません: {image_path}")
    
    # ファイル名とメディアタイプ取得
    file_name = Path(image_path).name
    content_type, _ = mimetypes.guess_type(image_path)
    if not content_type or not content_type.startswith('image/'):
        content_type = 'image/jpeg'  # デフォルト
    
    # S3キー生成（タイムスタンプ付きでユニークに）
    from datetime import datetime
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    s3_key = f"{S3_UPLOAD_PREFIX}{timestamp}_{file_name}"
    
    print(f"   画像をS3にアップロード中...")
    print(f"   ローカル: {image_path}")
    print(f"   S3: s3://{S3_BUCKET_NAME}/{s3_key}")
    
    # アップロード実行
    with open(image_path, 'rb') as file:
        s3_client.put_object(
            Bucket=S3_BUCKET_NAME,
            Key=s3_key,
            Body=file,
            ContentType=content_type,
            Metadata={
                'original-filename': file_name,
                'uploaded-by': 'invoke_agent_with_s3_image'
            }
        )
    
    s3_uri = f"s3://{S3_BUCKET_NAME}/{s3_key}"
    print(f"   アップロード完了: {s3_uri}\n")
    
    return s3_uri

def invoke_agent_with_image(question, s3_image_uri, session_id='test-session'):
    """
    Bedrock エージェントを呼び出して画像分析を実行
    
    Args:
        question (str): ユーザーの質問
        s3_image_uri (str): S3画像URI (s3://bucket/key)
        session_id (str): セッションID
        
    Returns:
        str: エージェントからの応答
    """
    print(f"   Bedrock Agentを呼び出し中...")
    print(f"   質問: {question}")
    print(f"   画像: {s3_image_uri}")
    print(f"   Session ID: {session_id}\n")
    
    # 質問文に画像パスを含める
    input_text = f"{question}\n\n画像パス: {s3_image_uri}"
    
    # InvokeAgent API呼び出し
    response = bedrock_client.invoke_agent(
        agentId=AGENT_ID,
        agentAliasId=AGENT_ALIAS_ID,
        sessionId=session_id,
        inputText=input_text,
        enableTrace=True  # トレース有効化
    )
    
    # レスポンス処理
    result = ""
    print("   応答受信中...\n")
    print("=" * 70)
    
    for event in response['completion']:
        # チャンクデータ
        if 'chunk' in event:
            chunk = event['chunk']
            if 'bytes' in chunk:
                chunk_text = chunk['bytes'].decode('utf-8')
                result += chunk_text
        
        # トレース情報（デバッグ用）
        if 'trace' in event:
            trace = event['trace'].get('trace', {})
            
            # オーケストレーショントレース
            if 'orchestrationTrace' in trace:
                orch_trace = trace['orchestrationTrace']
                
                # アクション呼び出し
                if 'invocationInput' in orch_trace:
                    inv_input = orch_trace['invocationInput']
                    if 'actionGroupInvocationInput' in inv_input:
                        action_input = inv_input['actionGroupInvocationInput']
                        print(f"\n[アクション呼び出し]")
                        print(f"  関数: {action_input.get('function', 'N/A')}")
                        print(f"  パラメータ: {json.dumps(action_input.get('parameters', []), ensure_ascii=False)}")
    
    print("\n" + "=" * 70)
    print("\n   応答完了\n")
    
    return result


def main():
    """
    メイン処理
    """
    print("\n" + "=" * 70)
    print("Amazon Bedrock Agent - S3画像分析テスト")
    print("=" * 70 + "\n")
    
    # 画像ファイルパス（実際のパスに変更）
    image_path = input("画像ファイルのパスを入力してください: ").strip()
    
    # パスのクォート除去（ドラッグ&ドロップ対応）
    image_path = image_path.strip('"').strip("'")
    
    if not os.path.exists(image_path):
        print(f"   エラー: ファイルが見つかりません: {image_path}")
        return
    
    # 質問入力
    question = input("質問を入力してください: ").strip()
    if not question:
        question = "この画像について詳しく説明してください。"
    
    try:
        # Step 1: S3にアップロード
        s3_uri = upload_image_to_s3(image_path)
        
        # Step 2: Bedrock Agent呼び出し
        result = invoke_agent_with_image(question, s3_uri)
        
        # 結果表示
        print("\n" + "=" * 70)
        print("最終結果")
        print("=" * 70)
        print(result)
        print("=" * 70 + "\n")
        
    except Exception as e:
        print(f"\n   エラーが発生しました: {str(e)}")
        import traceback
        traceback.print_exc()


if __name__ == '__main__':
    main()

実行例:

> python invoke_agent_with_s3_image.py

トラブルシューティング

よくあるエラーと解決策です。

エラー1: Lambdaの "Unhandled" エラー

症状:

Your request couldn't be completed. Lambda function encountered a problem.
The error message from the Lambda function is Unhandled.

原因と解決策:

原因	確認方法	解決策
タイムアウト	CloudWatch Logsで `Status: timeout` を確認	Lambda関数の設定で、タイムアウトを延長
メモリ不足	CloudWatch Logsで `Memory Size` と `Max Memory Used` を確認	Lambda関数の設定で、メモリを増加
Bedrock権限不足	ログで `AccessDeniedException` を確認	IAMロールに `bedrock:InvokeModel` 権限を追加
S3権限不足	ログで `AccessDenied` を確認	IAMロールに `s3:GetObject` 権限を追加

エラー2: リソースベースポリシー未設定

症状: エージェントからLambda関数が呼び出せない

解決策:
Lambda関数のリソースベースポリシーにBedrock エージェントの呼び出し許可を追加（Lambdaリソースベースポリシーの追加を参照）

エラー3: ValidationException (InvokeModel)

症状:

ValidationException: Malformed input request: On-demand throughput isn't supported for this model

原因: モデルIDが不正（Inference Profileが必要など）

解決策: 正しいモデルIDを指定

エラー4: NoSuchKey (S3)

症状:

botocore.exceptions.ClientError: An error occurred (NoSuchKey) when calling the GetObject operation

原因: 指定されたS3オブジェクトが存在しない

解決策:

S3コンソールでバケットの内容を確認
(Lambda関数テスト時など)画像が存在しない場合はアップロード

エラー5: Agent not prepared

症状:

ValidationException: Agent is not prepared

原因: アクショングループ追加後に「準備」（Prepare）を実行していない

解決策: Bedrock エージェントの詳細画面で準備ボタンをクリック

本番運用に向けて

エージェント機能の拡張

今回構築したエージェントは単一画像・単一ユーザを前提とした検証用の構成です。本番運用では以下の機能拡張を検討してください。

ナレッジベースの実装
複数画像やテキストのみメッセージへの対応
フロントエンドの認証／認可
マルチセッション・マルチユーザ対応
S3画像ファイルのライフサイクル管理
CloudWatch Alarmsの設定

まぁ、本格的な業務用途はBedrock AgentCore を選択されると思いますが...

IAM権限の最小化

テスト環境では AmazonBedrockFullAccess や AmazonS3ReadOnlyAccess などの広範な権限を使用していますが、本番利用時は必ず最小権限のカスタムポリシーに変更してください。

モデル選択の最適化

今回は比較的利用コストが安いClaude 3.7 Sonnet を利用しましたが、用途に応じて最適なモデルを選択してください。

参考資料

AWS公式ドキュメント

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up