Go to Qiita Advent Calendar Top
LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@syukan3

Amazon Bedrock × AWS Lambda で “Claude‑Sonnet” を呼び出す

Posted at

――サーバーレスで LLM API を組み込む最小実装を読み解く

本記事では、下に掲げた Python 3.12 用 Lambda 関数を題材に、コードの意図と設計思想を詳しく解説します。
「とりあえず動く」だけではなく 環境変数による設定切替・責務分割・ロギング／例外ハンドリング・テスト容易性 まで押さえた実装になっているので、プロダクションへの横展開や学習のベースラインに最適です。

完全版コード

以下のコードブロックは一切改行やコメントを削らずそのまま掲載します。コピペで SAM／CDK／Serverless Framework などに組み込めます。

"""
Lambda function to generate text via Amazon Bedrock (Anthropic Claude-Sonnet).

主な特徴
--------
1. **設定値は環境変数で管理**  
   - リージョン、モデル ID、トークン数、温度を `os.getenv()` で取得  
   - 本番・検証・開発をコード無改変で切り替え可能

2. **関数を細分化してハンドラを極小化**  
   - build_payload()     : Bedrock 送信用の JSON ペイロードを組み立て  
   - invoke_bedrock()    : Bedrock へリクエスト、レスポンスの検証＆整形  
   - build_response()    : API Gateway 互換の HTTP レスポンスを生成  

3. **ロギング／例外ハンドリング**  
   - CloudWatch Logs に詳細を残しつつ、クライアントには汎用的なメッセージを返却  
   - boto3 例外と JSON パース例外を個別に捕捉してわかりやすく再送出

4. **ユニットテスト容易性**  
   - 主要ロジックは純粋関数化し、`boto3` Stubber でテストしやすい形に
"""

import json
import logging
import os
from typing import Any, Dict

import boto3
from botocore.exceptions import BotoCoreError, ClientError

# ───────────────────────────────────────────────────────────────
# 1. 環境変数の読み込み（デフォルト値を設けて安全にフェイルオーバー）
# ───────────────────────────────────────────────────────────────
REGION: str = os.getenv("BEDROCK_REGION", "ap-northeast-1")
MODEL_ID: str = os.getenv(
    "BEDROCK_MODEL_ID", "apac.anthropic.claude-sonnet-4-20250514-v1:0"
)
MAX_TOKENS: int = int(os.getenv("MAX_TOKENS", "100"))        # 文字数ベースではなくトークン数
TEMPERATURE: float = float(os.getenv("TEMPERATURE", "0.7"))  # 0=決定的、1=多様

# ───────────────────────────────────────────────────────────────
# 2. boto3 クライアントとロガーの初期化
#    - boto3.client() はコールドスタート時に 1 回だけ実行されるため効率的
# ───────────────────────────────────────────────────────────────
br = boto3.client("bedrock-runtime", region_name=REGION)

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)  # INFO ログ以上を出力（必要なら環境変数で変更可)


# ───────────────────────────────────────────────────────────────
# 3. ユーティリティ関数群
# ───────────────────────────────────────────────────────────────
def build_payload(prompt: str) -> Dict[str, Any]:
    """
    Bedrock に送るリクエストボディを生成する。

    Parameters
    ----------
    prompt : str
        ユーザーから受け取った入力文章

    Returns
    -------
    dict
        Bedrock API 仕様に沿った JSON ペイロード
    """
    return {
        "anthropic_version": "bedrock-2023-05-31",
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": MAX_TOKENS,
        "temperature": TEMPERATURE,
    }


def invoke_bedrock(prompt: str) -> str:
    """
    Bedrock Claude-Sonnet モデルを呼び出し、生成テキストを取得する。

    Parameters
    ----------
    prompt : str
        ユーザー入力

    Returns
    -------
    str
        生成されたテキスト

    Raises
    ------
    RuntimeError
        Bedrock 呼び出し自体が失敗した場合（ネットワーク、権限など）
    ValueError
        Bedrock からのレスポンス形式が想定外だった場合
    """
    try:
        # Step 1: Bedrock へリクエスト送信
        resp = br.invoke_model(
            modelId=MODEL_ID,
            body=json.dumps(build_payload(prompt)),
            accept="application/json",
            contentType="application/json",
        )

        # Step 2: レスポンス body は StreamingBody → bytes
        body_text = resp["body"].read().decode("utf-8")

        # Step 3: JSON にデシリアライズ & 正常系バリデーション
        result = json.loads(body_text)
        generated_text: str = result["content"][0]["text"]
        return generated_text

    # boto3 由来のエラー（認証失敗・API 仕様変更など）
    except (BotoCoreError, ClientError) as e:
        logger.exception("Bedrock API 呼び出しに失敗しました")
        # ここで再送出することで上位ハンドラが共通処理できる
        raise RuntimeError("Bedrock invocation failed") from e

    # 期待しない JSON 形式やキー欠損の場合
    except (KeyError, IndexError, json.JSONDecodeError) as e:
        logger.exception("Bedrock レスポンスの解析に失敗しました: %s", e)
        raise ValueError("Invalid Bedrock response") from e


def build_response(
    prompt: str, generated: str, status: int = 200
) -> Dict[str, Any]:
    """
    API Gateway 互換の辞書型レスポンスを作成。

    Parameters
    ----------
    prompt : str
        クライアントから受け取ったプロンプト
    generated : str
        生成結果（もしくはエラーメッセージ）
    status : int, default=200
        HTTP ステータスコード

    Returns
    -------
    dict
        Lambda Proxy Integration 形式のレスポンス
    """
    return {
        "statusCode": status,
        "headers": {
            # ブラウザ/クライアントが UTF-8 と分かるよう charset を明示
            "Content-Type": "application/json; charset=utf-8"
        },
        "body": json.dumps(
            {
                "input_prompt": prompt,
                "generated_text": generated,
            },
            ensure_ascii=False,  # 日本語を \uXXXX エスケープさせない
        ),
    }


# ───────────────────────────────────────────────────────────────
# 4. Lambda エントリポイント
# ───────────────────────────────────────────────────────────────
def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    Lambda 関数のメインエントリ。

    Notes
    -----
    event : dict
        API Gateway から渡される入力。`prompt` キーに文字列が含まれる想定。
    context : LambdaContext
        実行コンテキスト（未使用だが request_id 等が必要なら活用可）

    Returns
    -------
    dict
        API Gateway 互換のレスポンス
    """
    # event.get() でデフォルトプロンプトを設定
    prompt: str = event.get("prompt", "Hello, how can I help you?")

    try:
        # Bedrock 呼び出し → 生成テキストを取得
        generated = invoke_bedrock(prompt)
        return build_response(prompt, generated)

    # ここで broad-except を使うのは「すべての上流例外を 500 で握り潰す」目的
    # → クライアントには stack trace を漏らさず、CloudWatch には残す
    except Exception as exc:  # pylint: disable=broad-except
        logger.error("Unhandled error: %s", exc, exc_info=True)
        return build_response(prompt, f"Error: {exc}", status=500)

コード全体像

レイヤー 主な責務 本コードでの実装
環境設定 リージョン・モデル ID・推論パラメータを環境変数で読み込む REGION / MODEL_ID / MAX_TOKENS / TEMPERATURE
外部依存 Bedrock ランタイムの呼び出し、ログ出力 boto3.client("bedrock-runtime") / logging
ユーティリティ ペイロード生成・API 呼び出し・HTTP レスポンス生成 build_payload / invoke_bedrock / build_response
入口 Lambda ハンドラ、上流に API Gateway を想定 lambda_handler

ワンポイント
Lambda を “API Gateway + Proxy” で公開するときは、lambda_handler が返す辞書を API Gateway がそのまま HTTP に変換します。statusCode / headers / body という３要素を揃えておくと、SDK 連携も cURL も楽になります。

1 環境変数で安全かつ柔軟に設定管理

フェイルオーバーを考慮したデフォルト値

REGION = os.getenv("BEDROCK_REGION", "ap-northeast-1")
  • getenv の第 2 引数にデフォルト値を置くことで、

    環境変数が存在しない ローカルテストでも Gracefully に動作。
  • int() や float() キャストを早めに行い、下流ロジックが “文字列か数値か” を気にしなくてよい構造に。

ステージ別切替

ステージ 設定例 切替方法
dev TEMPERATURE=1.0 sam local invoke -e event.json で環境変数を .env.dev から読ませる
stg MAX_TOKENS=50 CodePipeline で 別の Parameters セクション を適用
prod MODEL_ID=apac.anthropic.claude-sonnet-4-20250514-v1:0 マネコンで上書き or AWS Systems Manager Parameter Store

2 boto3 クライアントとロガーの初期化

  • コールドスタート時に 1 回だけ boto3.client() を実行。

    以降の呼び出しはコンテナ再利用で ウォームスタート になるため、
    低レイテンシ＆低料金（呼び出し回数削減）を両立。
  • logger.setLevel(logging.INFO) で “リリース後に DEBUG へ下げ忘れる事故” を防止。
    動的に変えたい場合は LOG_LEVEL を設定値に昇格させても良いでしょう。

3 ユーティリティ関数で責務を分割

build_payload()

  • Bedrock Claude 系の共通フォーマット anthropic_version を固定文字列で保持。
    将来のバージョンアップ時は 一箇所変更で済む ＝ 保守性◎。
  • messages は ChatML 準拠の構造。role に将来 "assistant" を追加すれば会話履歴にも拡張可。

invoke_bedrock()

  1. リクエスト送信
    accept / contentType を JSON に固定。

  2. StreamingBody → str 変換
    .read().decode("utf-8") でバイト列を明示的にデコード。

  3. 構造バリデーション

    • result["content"][0]["text"] を取り出す前に JSON ロード。
    • KeyError / IndexError / JSONDecodeError を個別捕捉し、
      「Bedrock 側？」 or 「自前のパース？」 を運用者が即判断できるログに。

★ Logger の粒度

  • logger.exception(...) は stack trace 付きで記録。
    障害調査で根本原因を突き止めるのに役立ちます。
  • 外部へは RuntimeError など抽象化した例外を再送出 → 内部構造の漏えいを防止。

build_response()

  • ensure_ascii=False で 日本語文字列を \u エスケープしない。
    クライアントサイドの再デコードが不要になり、ブラウザで直接可読。
  • Content‑Type に charset=utf-8 を付けることで、
    文字化け・ダウンロード挙動（application/json が text/plain になる等）を防止。

4 lambda_handler ―― 入口は極小、ビジネスロジックは外出し

  • 早期 return：生成成功なら 200、例外は一括捕捉して 500。
    これにより API Gateway 側の統合レスポンス設定がシンプルに。
  • デフォルトプロンプト "Hello, how can I help you?" を用意し、
    event が空でも動く ので CloudWatch Scheduled Event テスト も楽。

5 例外ハンドリングのベストプラクティス

どこで捕捉 捕捉する例外 クライアントに返すもの CloudWatch Logs
invoke_bedrock BotoCoreError / ClientError "Bedrock invocation failed" logger.exception() でスタックトレース
invoke_bedrock JSON/Key エラー "Invalid Bedrock response" 同上
lambda_handler broad‑except "Error: {exc}" + HTTP 500 logger.error(..., exc_info=True)

broad‑except はアンチパターンに見えますが、
“API としては 常に JSON を返し、スタックトレースを漏らさない” 目的なら有効です。

6 ユニットテスト容易性 ―― boto3 Stubber と純粋関数化

  • build_payload() / build_response() は 副作用ゼロ。引数＝入力、戻り値＝出力なので テーブル駆動テスト が簡単。

  • invoke_bedrock() は boto3.stub.Stubber で疑似レスポンスを注入可能。

    with Stubber(br) as stub:
    stub.add_response("invoke_model", expected_resp)
    assert invoke_bedrock("hi") == "hello"

  • lambda_handler() は event モックだけでテストでき、
    手間のかかる API Gateway / IAM / VPC セットアップが不要。

7 デプロイと環境切替（SAM 例）

Globals:
  Function:
    Runtime: python3.12
    MemorySize: 512
    Timeout: 5
    Environment:
      Variables:
        BEDROCK_REGION: !Ref AWS::Region
        BEDROCK_MODEL_ID: apac.anthropic.claude-sonnet-4-20250514-v1:0
        MAX_TOKENS: 100
        TEMPERATURE: 0.7
  • ステージ別 samconfig.toml を用意し、sam deploy --config-env dev などで上書き。
  • Bedrock はパブリックエンドポイントなので VPC 不要、
    NAT ゲートウェイ代が掛からない＝低コスト。

8 パフォーマンス & コスト最適化 Tips

課題 解決策
コールドスタート遅延 Lambda SnapStart（Java 限定 → Python 不可） / メモリ 512 MB 以上で CPU ブースト
トークン消費削減 MAX_TOKENS を API 側でガードし、
アプリ側の “ささやき声” プロンプトを短縮
同時実行スパイク 予約コンカレンシー を設定し “Cost‑explosion” を防止

9 セキュリティ考慮

  1. 最小権限 IAM

    {
  "Effect": "Allow",
  "Action": "bedrock:InvokeModel",
  "Resource": "arn:aws:bedrock:*::foundation-model/provider/anthropic/*"
}

  2. 環境変数暗号化（KMS）
    アクセスキーや秘匿パラメータを入れる際は aws:kms キーで暗号化を。

  3. API 認可

    • 公開 API → Amazon Cognito / JWT オーソライザー
    • 社内用途 → IAM 認証 (sigv4) or VPC PrivateLink

まとめ

  • 環境変数 × boto3 クライアント再利用 で “ステージ切替” と “低レイテンシ” を両立
  • 純粋関数＋Stubber により CI Pipeline での自動テスト が容易
  • 例外ハンドリングとロギング を徹底し、運用時の MTTR を最小化
  • Claude‑Sonnet など LLM 呼び出しを Lambda に閉じ込めることで、
    フロントや別マイクロサービスからは単なる REST API として扱える

次の一歩

  • Bedrock Streaming API への拡張（WebSocket or SSE）
  • 会話メモリを DynamoDB Table + TTL で外だしし、
    ChatBot UX をさらに向上

本記事が、みなさんの サーバーレス × 生成 AI プロジェクトの着実な第一歩になれば幸いです。

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?