AWS Bedrockで生成AIアプリを実装する際の実践的な統合パターン

本記事では、AWS Bedrockを使用してLLMベースのアプリケーションを開発する際の、実装パターンと実践的なコード例を紹介します。API呼び出しの最適化、エラーハンドリング、コスト管理を含めた、本番環境で使える手法を学べます。

AWS Bedrockとは

AWS Bedrockは、フルマネージド型のサービスで、ClaudeやLlama、Mistralなどの複数のLLMモデルにAPIを通じてアクセスできます。インフラ管理が不要で、スケーラビリティに優れており、エンタープライズグレードのセキュリティを備えています。

従来のLLM統合と異なり、Bedrockでは以下が実現できます：

• 複数のモデルを同一のAPIで切り替え可能
• データは暗号化され、AWS内で保持される
• リージョン単位でのコンプライアンス要件対応が容易

Bedrockアプリケーション開発の基本フロー

前提条件とセットアップ

以下の環境で動作確認しています：macOS 14 / Python 3.11 / boto3 1.34以上 / AWS CLI 2.15以上

まず、AWSアカウントでBedrockの利用を有効化し、使用するモデルへのアクセスをリクエストする必要があります。AWSマネジメントコンソールから「Bedrock」→「Model access」で対象モデルを選択し、「Request access」をクリックしてください。

# 必要なパッケージをインストール
pip install boto3 python-dotenv

# AWS認証情報の確認
aws sts get-caller-identity

基本的なテキスト生成の実装

AWS SDKを使用して、Claudeモデルにアクセスする最もシンプルな例から始めましょう。

import boto3
import json
from typing import Optional

class BedrockLLMClient:
    def __init__(self, region: str = "us-east-1", model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0"):
        """
        Bedrockクライアントの初期化
        region: AWSリージョン（モデルの利用可能性を確認してください）
        model_id: 使用するモデルID
        """
        self.bedrock_client = boto3.client("bedrock-runtime", region_name=region)
        self.model_id = model_id
        self.region = region

    def generate_text(self, prompt: str, max_tokens: int = 1024, temperature: float = 0.7) -> str:
        """
        シンプルなテキスト生成
        """
        try:
            request_body = {
                "anthropic_version": "bedrock-2023-06-01",
                "max_tokens": max_tokens,
                "messages": [
                    {
                        "role": "user",
                        "content": prompt
                    }
                ],
                "temperature": temperature
            }

            response = self.bedrock_client.invoke_model(
                modelId=self.model_id,
                contentType="application/json",
                accept="application/json",
                body=json.dumps(request_body)
            )

            response_body = json.loads(response["body"].read())
            return response_body["content"][0]["text"]

        except self.bedrock_client.exceptions.AccessDeniedException:
            raise Exception(f"モデル {self.model_id} へのアクセスが許可されていません。AWSコンソールでリクエストしてください。")
        except Exception as e:
            raise Exception(f"API呼び出しエラー: {str(e)}")

# 使用例
if __name__ == "__main__":
    client = BedrockLLMClient()
    result = client.generate_text("Pythonでの非同期プログラミングについて簡潔に説明してください。")
    print(result)

実践的な本番環境対応の実装

ストリーミング応答の処理

長時間かかるテキスト生成では、ストリーミングで段階的に結果を受け取ることで、ユーザー体験を向上させられます。

import json
import boto3

class BedrockStreamingClient:
    def __init__(self, region: str = "us-east-1", model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0"):
        self.bedrock_client = boto3.client("bedrock-runtime", region_name=region)
        self.model_id = model_id

    def generate_text_streaming(self, prompt: str, max_tokens: int = 2048) -> None:
        """
        ストリーミング応答でテキストを段階的に処理
        """
        request_body = {
            "anthropic_version": "bedrock-2023-06-01",
            "max_tokens": max_tokens,
            "messages": [
                {
                    "role": "user",
                    "content": prompt
                }
            ]
        }

        try:
            response = self.bedrock_client.invoke_model_with_response_stream(
                modelId=self.model_id,
                contentType="application/json",
                accept="application/json",
                body=json.dumps(request_body)
            )

            # イベントストリームを処理
            for event in response["body"]:
                if "chunk" in event:
                    chunk = json.loads(event["chunk"]["bytes"])
                    if chunk["type"] == "content_block_delta":
                        if chunk["delta"]["type"] == "text_delta":
                            print(chunk["delta"]["text"], end="", flush=True)

        except Exception as e:
            print(f"ストリーミングエラー: {str(e)}")

# 使用例
if __name__ == "__main__":
    client = BedrockStreamingClient()
    client.generate_text_streaming("次の数学問題を段階的に解いてください：12 × 13 = ?")
    print()  # 改行

複数ターンの会話管理

チャットボット機能を実装する際は、会話履歴を管理し、コンテキストを維持する必要があります。

from typing import List, Dict

class BedrockConversationManager:
    def __init__(self, model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0"):
        self.bedrock_client = boto3.client("bedrock-runtime", region_name="us-east-1")
        self.model_id = model_id
        self.conversation_history: List[Dict[str, str]] = []

    def add_message(self, role: str, content: str) -> None:
        """会話履歴にメッセージを追加"""
        self.conversation_history.append({
            "role": role,
            "content": content
        })

    def generate_response(self, user_message: str, system_prompt: Optional[str] = None) -> str:
        """ユーザーメッセージに応答を生成"""
        # ユーザーメッセージを履歴に追加
        self.add_message("user", user_message)

        # システムプロンプトを含めたリクエスト構築
        request_body = {
            "anthropic_version": "bedrock-2023-06-01",
            "max_tokens": 1024,
            "system": system_prompt or "あなたは親切で有用なアシスタントです。",
            "messages": self.conversation_history
        }

        try:
            response = self.bedrock_client.invoke_model(
                modelId=self.model_id,
                contentType="application/json",
                accept="application/json",
                body=json.dumps(request_body)
            )

            response_body = json.loads(response["body"].read())
            assistant_message = response_body["content"][0]["text"]

            # アシスタントの応答を履歴に追加
            self.add_message("assistant", assistant_message)

            return assistant_message

        except Exception as e:
            return f"エラーが発生しました: {str(e)}"

    def reset_conversation(self) -> None:
        """会話履歴をリセット"""
        self.conversation_history = []

# 使用例
if __name__ == "__main__":
    manager = BedrockConversationManager()
    system = "あなたはテクニカルサポートの専門家です。わかりやすく説明してください。"

    manager.generate_response("Pythonでのメモリリークについて教えてください", system)
    response2 = manager.generate_response("具体的な対策方法はありますか？", system)
    print(response2)

よくあるハマりポイントと対策

モデルアクセスエラー

症状： AccessDeniedException が発生する

原因： Bedrockコンソールでモデルへのアクセスをリクエストしていない、またはリクエストが承認されていない

対策：

• AWSマネジメントコンソール → Bedrock → Model accessでリクエスト状態を確認
• リージョン確認：モデルがそのリージョンで利用可能か確認
• IAM権限確認：bedrock:InvokeModel の権限があるか確認

リージョン不一致

症状： 「モデルが見つからない」というエラー

原因： モデルが選択したリージョンで利用不可

対策： 以下のコマンドで利用可能なリージョンとモデルを確認

aws bedrock list-foundation-models --region us-east-1

トークン数超過エラー

症状： 「Input tokens exceed maximum」というエラー

原因： 会話履歴が長すぎて、トークン数が上限を超えた

対策： 古い会話をサマライズするか、スライディングウィンドウで履歴を制限

def truncate_conversation(self, max_history: int = 10) -> None:
    """会話履歴を最新N件に制限"""
    if len(self.conversation_history) > max_history:
        self.conversation_history = self.conversation_history[-max_history:]

使うべき場面と使うべきでない場面

Bedrockが適している場面

• エンタープライズグレードのセキュリティが必要な場合
• データレジデンシー要件がある場合（特定のリージョンでのデータ保持）
• 複数のLLMモデルを同一UIで統合したい場合
• AWSインフラに既に深く統合されている場合

Bedrockが適さない場面

• 極低遅延（<100ms）が必須の場合：オンプレミスモデルの検討を
• 外部APIプロバイダー（OpenAI APIなど）が既に運用されている場合
• 開発段階で迅速にプロトタイプを試したい場合：Claude APIなどの直接使用も選択肢

コスト最適化のポイント

Bedrockの料金はモデル、入出力トークン数によって決まります。本番環境でのコスト削減には以下の対策が有効です：

• キャッシング活用： 繰り返し使用するシステムプロンプトは、Prompt Cachingを活用してコストを削減
• モデル選択： 要件に応じて、より小さなモデル（Haiku）を検討
• バッチ処理： 大量のリクエストはバッチAPI利用で割引を受ける

参考リソース

AWS Bedrock公式ドキュメント（日本語）