LambdaによるBedrockの応用操作を身につけよう

この章では、Bedrockを使ってみよう で扱ったLambdaからの1問1答呼び出しを発展させて、マルチターン会話・ツール呼び出しループ・システムプロンプトなど、LambdaでBedrockを扱う際の応用操作をハンズオン形式で手を動かしながら体験します。

1. 事前準備

この章では、以下のツールやアカウントが必要です。まだ準備できていない場合は、リンク先の手順に沿って準備をお願いします。

• AWSアカウント
• Python

Bedrockを使ってみよう を先に完了し、LambdaからConverse APIをシンプルに呼び出せる状態から始めます。本章ではその上に、Bedrock を Lambda から扱う際の応用操作を1つずつ試していきます。

2. ハンズオンの準備

以降の各セクションで共通して使う Lambda 関数と実行ロールを CloudFormation で用意し、Bedrock の推論プロファイル ID を環境変数として設定します。ここでの準備が整えば、以降のセクションでは index.py の書き換えとテスト実行だけで各テーマを試せます。

2.1 ハンズオン全体の流れ

以降のセクションでは、bedrock-agent 関数の index.py を各テーマごとに書き換えながら、以下の順で応用操作を試していきます。各セクションのコードは原則として独立して読める形にしていて、「続き」と書いてあるセクションだけは前セクションのコードを土台にした差分になります。

2.2 Lambda関数の作成

Lambda 関数、Bedrock 呼び出し権限を持たせた実行ロール、そしてマルチターン会話の履歴を保存する DynamoDB テーブルを、CloudFormation テンプレートで一括作成します。関数の器作りをテンプレートに任せることで、以降のセクションでは index.py のコードそのものに集中できます。

以下のテンプレートをダウンロードしてください。

テンプレートには、以下の3つのリソースが定義されています。

CloudFormationのダッシュボードでスタックの作成 > 新しいリソースを使用（標準）をクリックし、スタック作成ウィザードを開きます。

ステップ1: スタックの作成

「前提条件 - テンプレートの準備」で既存のテンプレートを選択を選び、「テンプレートの指定」の「テンプレートソース」でテンプレートファイルのアップロードを選択します。「テンプレートファイルのアップロード」のファイルの選択からダウンロードした bedrock-agent.yaml を指定し、次へをクリックします。

ステップ2: スタックの詳細を指定

「スタック名」に以下の値を入力します。

入力できたら次へをクリックします。

ステップ3: スタックオプションの設定

そのまま画面を下にスクロールし、「機能」セクションのAWS CloudFormation によって IAM リソースがカスタム名で作成される場合があることを承認します。にチェックを入れて次へをクリックします。

ステップ4: 確認して作成

内容を確認して、最下部の送信をクリックしてスタックを作成します。

スタックの作成完了確認

スタックのステータスが CREATE_COMPLETE になったら、スタック詳細画面の「リソース」タブに、BedrockAgentSessionsTable（bedrock-agent-sessions テーブル）、BedrockAgentLambdaRole（bedrock-agent-lambda-role ロール）、BedrockAgentFunction（bedrock-agent 関数）の3つが CREATE_COMPLETE で並んでいることを確認します。

2.3 環境変数の設定

作成した Lambda 関数に、回答生成に使うモデルの推論プロファイル ID を、環境変数として渡します。DynamoDB テーブル名の TABLE_NAME は CloudFormation で自動的にセット済みなので、ここでは MODEL_ID だけを追加します。

推論プロファイル ID は、Bedrockを使ってみよう で控えたものと同じ値（例: jp.anthropic.claude-sonnet-4-6）を使います。

Lambda コンソールで bedrock-agent 関数を開き、「設定」タブ→「環境変数」を選択して「編集」をクリックします。既に TABLE_NAME が入っているのが確認できます。この状態を保ったまま、「環境変数の追加」から以下を追加して「保存」します。

「環境変数」欄に TABLE_NAME と MODEL_ID の2つが並んで表示されていれば、設定完了です。

3. システムプロンプト

ここでは、Converse API の system パラメータを使って、エージェントに役割や応答スタイルを持たせる仕組みを扱います。マルチターン会話やツール呼び出しに入る前に、まずは Converse API に system パラメータを1つ足すだけで応答が変わる体験から始めます。

3.1 システムプロンプト

システムプロンプトの概要

デフォルトのモデル呼び出しでは、モデルはBedrockの標準的な応答スタイル（無難で汎用的な口調・観点）で回答してきます。実運用のエージェントでは、「カスタマー対応窓口として丁寧に答える」「セキュリティ観点でコードをレビューする」「応答は必ずJSONで返す」「対応範囲外の質問には答えない」のように、モデルにどう振る舞ってほしいか（役割・観点・応答形式・制約）を細かく指定したい場面が大半です。

システムプロンプトは、Converse APIの system パラメータでモデルに渡す「エージェントの役割・振る舞い・制約」を宣言する仕組みです。ユーザ発言（messages）とは別枠で扱われ、どのターンでも常に効く前提として参照されます。

Lambdaのコード変更

index.pyを以下の内容に置き換えます。システムプロンプト付きの1問1答に絞った最小構成です。

今回設定する SYSTEM_PROMPT は、料理の作り方を答えるアシスタントを想定したものです。AI はデフォルトでは「ですます」調の文章で回答してくるので、以下の3点を明確にコントロールする狙いで書いています。

• 「ですます」調ではなく体言止めで書かせる
• 箇条書きで簡潔に答えさせる
• 応答に使うカテゴリを「材料」「工程」「栄養」の3つに固定する

import os
import boto3

MODEL_ID = os.environ["MODEL_ID"]
REGION = "ap-northeast-1"

SYSTEM_PROMPT = """あなたは料理の作り方を答えるアシスタントです。
以下のルールを守って応答してください。

- 「材料」「工程」「栄養」の3つのカテゴリに分けて答える
- 各カテゴリの中身は箇条書きで示す
- 箇条書きの各項目は体言止めで書く（「〜です」「〜ます」は使わない）
"""

bedrock = boto3.client("bedrock-runtime", region_name=REGION)


def lambda_handler(event, context):
    question = event["question"]

    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=[{"role": "user", "content": [{"text": question}]}],
        system=[{"text": SYSTEM_PROMPT}],
    )

    output_message = response["output"]["message"]
    answer = "".join(
        block["text"] for block in output_message["content"] if "text" in block
    )
    return {"answer": answer}

貼り付けたら「Deploy」を押して反映します。

コードの解説

SYSTEM_PROMPT = """あなたは料理の作り方を答えるアシスタントです。
以下のルールを守って応答してください。

- 「材料」「工程」「栄養」の3つのカテゴリに分けて答える
- 各カテゴリの中身は箇条書きで示す
- 箇条書きの各項目は体言止めで書く（「〜です」「〜ます」は使わない）
"""

SYSTEM_PROMPT としてエージェントの役割と応答スタイルを定義しています。特別なテンプレートやDSLはなく、普通の自然言語（日本語でも英語でも）で書いた文章をそのままモデルが解釈してくれる仕組みです。今回のように「〜のアシスタントです」で役割を宣言し、その下に守ってほしいルールを箇条書きで並べる、というのが一般的な書き方です。

response = bedrock.converse(
    modelId=MODEL_ID,
    messages=[{"role": "user", "content": [{"text": question}]}],
    system=[{"text": SYSTEM_PROMPT}],
)

converse に system 引数を追加しました。system はテキストブロックのリストで、複数ブロックに分けても、1つのブロックに全て詰めてもどちらでも構いません。ここでは1つのブロックに SYSTEM_PROMPT をまるごと渡しています。

動作確認

「テスト」タブで、以下のテストイベントを実行します。

{
  "question": "オムライスの作り方を教えて"
}

AI の応答は毎回まったく同じにはならないので、以下は一例です。以下のポイントが押さえられていれば成功です。

• 応答が「材料」「工程」「栄養」の3つのカテゴリに分かれている
• 各カテゴリの中身が箇条書きになっている
• 箇条書きの各項目が体言止めで書かれている（「〜です」「〜ます」で終わっていない）

{
  "answer": "【材料】\n- 卵 2個\n- ご飯 200g\n- 玉ねぎ 1/4個\n- ケチャップ 大さじ2\n- 塩コショウ 少々\n\n【工程】\n- 玉ねぎのみじん切り\n- ご飯と玉ねぎをバターで炒める\n- ケチャップ、塩コショウで味付け\n- 薄焼き卵の作成\n- ケチャップライスを卵で包む\n\n【栄養】\n- タンパク質 約20g\n- 炭水化物 約80g\n- 脂質 約15g"
}

システムプロンプトを外したときの応答（普通の丁寧な文章で作り方を説明してくれる形）と見比べると、応答の内容（3つのカテゴリ）と形式（体言止めの箇条書き）が明確にコントロールできていることが分かります。

このように、通常のAIの応答スタイルを特定の形に寄せたい場合や、エージェントに必ず守らせたい基準（役割・応答形式・禁止事項など）を宣言したい場合に、システムプロンプトを設定するのが有効です。

4. マルチターン会話

ここでは、DynamoDB に会話履歴を保存して文脈を保つマルチターン会話を扱います。合わせて、履歴が伸び続けたときにコンテキスト長・料金が膨らむのを防ぐ長さ管理についても扱います。

4.1 マルチターン会話

マルチターン会話の概要

「Bedrockを使ってみよう」で扱ったLambdaのコードは、質問1件をモデルに投げて応答1件を受け取る「1問1答」の呼び出しでした。この方式ではモデルは前の会話を覚えていないため、「今何時？」→「その3時間後は？」のように前の応答を踏まえた質問には答えられません。

マルチターン会話は、過去のやり取りをすべてモデルに渡すことで、会話の文脈を保持できるようにする仕組みです。

Lambda 関数はステートレスなので、呼び出し間で会話履歴を保持するには外部ストレージが必要です。今回は session_id をキーに DynamoDB へ履歴を保存し、次の呼び出しで同じ session_id を指定するだけで会話を続けられる形にします。

sequenceDiagram
    participant U as ユーザ
    participant L as Lambda
    participant D as DynamoDB
    participant M as モデル（Bedrock）

    U->>L: 質問1 (session_id)
    L->>D: セッションから履歴を取得
    D-->>L: 空
    L->>M: 履歴 + 質問1
    M-->>L: 応答1
    L->>D: 履歴 (質問1 + 応答1) を保存
    L-->>U: 応答1
    U->>L: 質問2 (同じ session_id)
    L->>D: セッションから履歴を取得
    D-->>L: 質問1 + 応答1
    L->>M: 履歴 + 質問2
    M-->>L: 応答2（文脈を踏まえた回答）
    L->>D: 履歴 (前 + 質問2 + 応答2) を保存
    L-->>U: 応答2

Lambdaのコード変更

bedrock-agent 関数の「コード」タブを開き、index.py を以下の内容に置き換えます。

import os
import boto3

MODEL_ID = os.environ["MODEL_ID"]
TABLE_NAME = os.environ["TABLE_NAME"]
REGION = "ap-northeast-1"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)
table = boto3.resource("dynamodb", region_name=REGION).Table(TABLE_NAME)


def lambda_handler(event, context):
    session_id = event["session_id"]
    question = event["question"]

    result = table.get_item(Key={"session_id": session_id})
    messages = result.get("Item", {}).get("messages", [])

    messages.append({"role": "user", "content": [{"text": question}]})

    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=messages,
    )

    output_message = response["output"]["message"]
    messages.append(output_message)

    table.put_item(Item={"session_id": session_id, "messages": messages})

    answer = "".join(
        block["text"] for block in output_message["content"] if "text" in block
    )
    return {"answer": answer}

貼り付けたら「Deploy」を押して反映します。

コードの解説

table = boto3.resource("dynamodb", region_name=REGION).Table(TABLE_NAME)

DynamoDB を扱うため、テーブルオブジェクトを取得しています。boto3.resource を使うと、Python の辞書やリストをそのまま put_item に渡せるので、{"S": ...} などの型指定を書かなくて済みます。

session_id = event["session_id"]
question = event["question"]

呼び出し側から session_id（会話を識別するキー）と question（今回の質問）を受け取ります。会話履歴そのものを受け取る必要はなく、session_id があれば DynamoDB から取り出せる設計です。

result = table.get_item(Key={"session_id": session_id})

session_id をキーに DynamoDB からアイテムを取得します。取得結果は result に入ります。

messages = result.get("Item", {}).get("messages", [])

result の中から過去の会話履歴（messages 属性）を取り出し、Python の messages 変数にセットします。これで、このあと今回の質問を追記して Bedrock に渡すための「土台となる会話履歴」が messages に入った状態になります。初回は DynamoDB にアイテムが存在しないので、.get("Item", {}) で空辞書にフォールバックし、その中の messages も空リストで扱います。

messages.append({"role": "user", "content": [{"text": question}]})

取り出した履歴に、今回のユーザ発言を追記します。

response = bedrock.converse(
    modelId=MODEL_ID,
    messages=messages,
)

output_message = response["output"]["message"]
messages.append(output_message)

converse に履歴を丸ごと渡してモデルの応答を受け取り、次のターンに備えて messages にモデル応答も追記します。ここまでは基本的な Converse 呼び出しの流れです。

table.put_item(Item={"session_id": session_id, "messages": messages})

更新した履歴（ユーザ発言 + モデル応答まで含む）を DynamoDB に保存し直します。同じ session_id に対する put_item は上書きになるので、常に最新状態が保たれます。

answer = "".join(
    block["text"] for block in output_message["content"] if "text" in block
)
return {"answer": answer}

モデル応答からテキスト部分を取り出し、answer として返します。履歴は DynamoDB に保存済みなので、レスポンスに含めて呼び出し側に返す必要はありません。

動作確認

「テスト」タブから、以下の内容で最初のテストイベントを作成して実行します。イベント名は任意（例: turn1）です。

{
  "session_id": "test-session-1",
  "question": "こんにちは。私の名前はTaroです。"
}

AI の応答は毎回まったく同じにはならないので、以下は一例です。以下のポイントが押さえられていれば成功です。

• 応答に Taro への挨拶や、名前を認識したことに触れる内容が含まれる

{
  "answer": "こんにちは、Taroさん！お名前を教えていただきありがとうございます。今日はどのようなお手伝いができるでしょうか？"
}

続いて、session_id はそのままで質問だけを書き換えた2ターン目を試します。

{
  "session_id": "test-session-1",
  "question": "私の名前を覚えていますか？"
}

このターンでは、以下のポイントが押さえられていれば成功です。

• 応答に Taro という名前が含まれており、モデルが前ターンの内容を覚えていることが分かる

{
  "answer": "はい、覚えていますよ。あなたのお名前はTaroさんですね。"
}

1問1答の呼び出しでは絶対に返ってこない応答なので、DynamoDB に履歴を保存することで文脈が保たれていることが確認できます。

DynamoDBに保存された履歴を確認する

Lambda の応答だけでなく、DynamoDB 側で「実際に会話履歴が保存されているか」も直接確認してみます。

DynamoDBコンソールを開き、左メニューの項目を探索をクリックします。テーブル一覧から bedrock-agent-sessions を選び、右側の「項目のスキャンまたはクエリ」で「スキャン」を選択したまま、実行するをクリックします。返された項目の欄に session_id が test-session-1 のアイテムが1件並んでいれば、DynamoDB に会話履歴が保存されている状態です。

messages 列の値をコピーして展開すると、以下のように DynamoDB のマップ／リスト形式で保存されていることが分かります。

[
  {
    "M": {
      "content": {
        "L": [
          { "M": { "text": { "S": "こんにちは。私の名前はTaroです。" } } }
        ]
      },
      "role": { "S": "user" }
    }
  },
  {
    "M": {
      "content": {
        "L": [
          { "M": { "text": { "S": "こんにちは、Taroさん!はじめまして😊\n\n何かお手伝いできることはありますか?お気軽にどうぞ!" } } }
        ]
      },
      "role": { "S": "assistant" }
    }
  },
  {
    "M": {
      "content": {
        "L": [
          { "M": { "text": { "S": "私の名前を覚えていますか?" } } }
        ]
      },
      "role": { "S": "user" }
    }
  },
  {
    "M": {
      "content": {
        "L": [
          { "M": { "text": { "S": "はい、もちろんです!あなたのお名前は **Taro** さんですね😊\n\n何かご質問やお手伝いできることはありますか?" } } }
        ]
      },
      "role": { "S": "assistant" }
    }
  }
]

DynamoDB は保存する値ごとに型を付ける仕様で、M はマップ（辞書）、L はリスト、S は文字列を表します。Lambda 側の Python コード上では [{"role": "user", "content": [{"text": "..."}]}, ...] という素朴な形で扱っていた履歴が、DynamoDB 側ではこのように型注釈で包まれた形で保存されています（boto3.resource("dynamodb") がこの変換を自動でやってくれるので、Python コード側では意識せずに書けます）。

role: user と role: assistant のブロックが交互に4件並んでいるのは、1ターン目・2ターン目のやり取り（ユーザ発言→モデル応答×2ターン）がそのまま蓄積されているためです。次の呼び出しで Lambda が同じ session_id を使って get_item すると、この履歴が復元され、モデルが Taro という名前を覚えているように振る舞える、という一連の流れをここで実際に目で見て確認できます。

4.2 会話履歴の長さ管理

会話履歴の長さ管理の概要

前セクションのマルチターン実装をそのまま運用に載せると、DynamoDB に保存する messages は会話を重ねるたびに際限なく伸びていきます。すると、次の2つの問題が起きます。

• コンテキスト長の上限: モデルが1リクエストで受け取れるトークン数には上限があり、超えるとエラーで拒否される
• 料金とレイテンシの増加: Bedrockはリクエスト・レスポンスのトークン数で課金されるため、履歴が長いほど1ターンあたりのコストと応答時間が増える

実運用では、履歴のサイズをコントロールする仕組みを入れておくのが定石です。今回はもっともシンプルな「最新N件だけを残す」アプローチで対応します。要約による圧縮やトークン数実測による打ち切りといった高度なアプローチもありますが、まずは動きを体感しやすい方式から始めます。

Lambdaのコード変更

前セクションの index.py に、MAX_MESSAGES 定数と、DynamoDB から取得した messages を直近 MAX_MESSAGES 件に切り詰める処理を追加します。index.py を以下の内容に置き換えます。

import os
import boto3

MODEL_ID = os.environ["MODEL_ID"]
TABLE_NAME = os.environ["TABLE_NAME"]
REGION = "ap-northeast-1"
MAX_MESSAGES = 6

bedrock = boto3.client("bedrock-runtime", region_name=REGION)
table = boto3.resource("dynamodb", region_name=REGION).Table(TABLE_NAME)


def lambda_handler(event, context):
    session_id = event["session_id"]
    question = event["question"]

    result = table.get_item(Key={"session_id": session_id})
    messages = result.get("Item", {}).get("messages", [])

    if len(messages) > MAX_MESSAGES:
        messages = messages[-MAX_MESSAGES:]

    messages.append({"role": "user", "content": [{"text": question}]})

    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=messages,
    )

    output_message = response["output"]["message"]
    messages.append(output_message)

    table.put_item(Item={"session_id": session_id, "messages": messages})

    answer = "".join(
        block["text"] for block in output_message["content"] if "text" in block
    )
    return {"answer": answer}

貼り付けたら「Deploy」を押して反映します。

コードの解説

MAX_MESSAGES = 6

DynamoDB から取得した messages に残す件数の上限を定義しています。今回は挙動が観察しやすいよう 6 にしていますが、実運用ではトークン上限や1ターンあたりの平均長を踏まえて数十〜数百のオーダーで調整します。

if len(messages) > MAX_MESSAGES:
    messages = messages[-MAX_MESSAGES:]

DynamoDB から取得した履歴が上限を超えていたら、直近の MAX_MESSAGES 件だけに切り詰めます。今回の質問を積む前に切っているのがポイントで、これから加える発言 + モデル応答が上限内に収まる形にしてから会話を進めます。切り詰めた履歴はモデル呼び出しの後で put_item によって DynamoDB に上書き保存されるため、次のターン以降も履歴が肥大化することはありません。

動作確認

前セクションで使ったセッションは履歴が残っているので、新しい session_id（例: test-session-length）で試します。まず1ターン目のテストイベントを実行します。

{
  "session_id": "test-session-length",
  "question": "私の名前はTaroです。覚えておいてください。"
}

続いて、同じ session_id のまま質問だけを書き換えて、複数ターン会話を重ねます。合計7〜8ターンほど投げてみてください。

{
  "session_id": "test-session-length",
  "question": "好きな食べ物は何ですか？"
}

{
  "session_id": "test-session-length",
  "question": "おすすめの本を教えてください"
}

{
  "session_id": "test-session-length",
  "question": "今日の天気について教えてください"
}

ある程度ターンを重ねたところで、最初の会話に登場した名前を覚えているかを確認します。

{
  "session_id": "test-session-length",
  "question": "私の名前を覚えていますか？"
}

AI の応答は毎回まったく同じにはならないので、以下は一例です。以下のポイントが押さえられていれば成功です。

• 応答に Taro という名前が含まれていない
• 「お名前をまだうかがっていません」「お名前を教えてください」など、名前を認識していない趣旨の応答が返る

{
  "answer": "申し訳ありませんが、私はあなたのお名前を知りません。😊\n\nこの会話の中で、お名前を教えていただいていないためです。\n\nまた、私はAIですので、**会話をまたいで情報を記憶する機能がありません**。\n\nぜひお名前を教えていただければ、この会話の中では覚えておきます！😊"
}

MAX_MESSAGES を超えた古いメッセージが切り詰められているため、モデルが Taro を忘れる挙動が確認できます。前セクション（切り詰めなしのマルチターン会話）では覚えていた Taro を、履歴管理の切り詰めによって忘れる形になっています。

MAX_MESSAGES の値を大きくすれば Taro を覚え続けられるので、上限値と「モデルが覚えていられる範囲」のトレードオフを実感するには、値を変えながら同じシナリオを試してみるとよいです（session_id も変えて履歴をリセットしてから試すのがおすすめです）。なお MAX_MESSAGES は必ず偶数を指定してください。マルチターン会話は毎ターン user + assistant の2件を積む設計のため、奇数で切り詰めると直近履歴の先頭が assistant になり、Converse API のバリデーション（履歴は user から始まる必要がある）を通らなくなります。

5. ツール呼び出し

ここでは、モデルが外部のツール（関数）を呼び出しながら回答を組み立てるエージェント的な仕組みを扱います。基本のツール呼び出しループを実装したうえで、ツール実行が失敗したときのハンドリングを追加します。ツール呼び出しループの実装自体は、モデルが1ターンで複数ツールを同時に要求してくる並列呼び出しにも最初から対応しています。

このグループの各セクションでは、Lambda 関数として以下2つのツールを持たせた実装を組み立てます。

例えば「田中太郎（開発部 エンジニア、tanaka@example.com）のメール署名を作って」と聞くと、モデルはcreate_signatureを呼び出し、Lambda 側の会社名（DevOps株式会社）と代表電話番号（03-1234-5678）が付いた署名を返してくれます。同じように「1990年5月10日生まれの人の運勢を占って」と聞くと、モデルはread_fortuneを呼び出し、Lambda 側の占いロジックの結果をそのまま渡してくれます。どちらも「Lambda 側が持っている値」がモデル自身には見えないため、応答にその値が含まれていることが「ツールが実際に呼ばれた証拠」になります。

5.1 ツール呼び出し

ツール呼び出しの概要

AIエージェントは、モデルが外部のツール（関数）を呼び出しながら回答を組み立てる仕組みです。

sequenceDiagram
    participant U as ユーザ
    participant A as エージェント（Lambda）
    participant M as モデル（Bedrock）
    participant T as ツール（関数）

    U->>A: 質問
    A->>M: 質問 + 使えるツール一覧
    M-->>A: 「このツールを呼びたい」
    A->>T: ツールを実行
    T-->>A: 実行結果
    A->>M: ツール実行結果を渡す
    M-->>A: 最終回答
    A-->>U: 最終回答

モデルは「今の情報だけでは答えられない」と判断すると、使えるツールの中から適切なものを選び、引数と一緒に呼び出しを要求してきます。エージェント側は要求されたツールを実行し、結果をモデルに返します。このツール呼び出し要求→実行→結果を返すというやり取りが完了するまでループを回すのが、エージェントの中核ロジックです。

Lambdaのコード変更

index.pyを以下の内容に置き換えます。1回のLambda呼び出しの中で完結する形にしていて、ツール定義と run_tool 関数、そして「モデルがツール呼び出しを要求してきたら実行して結果を返す」を繰り返すループを組んだ形です。

import datetime
import os
import boto3

MODEL_ID = os.environ["MODEL_ID"]
REGION = "ap-northeast-1"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)

COMPANY_NAME = "DevOps株式会社"
COMPANY_PHONE = "03-1234-5678"

TOOLS = [
    {
        "toolSpec": {
            "name": "create_signature",
            "description": "氏名・部署名・役職・メールアドレスを受け取り、所定のフォーマットのメール署名を返します。会社名と代表電話番号は Lambda 側で自動的に付与されます。",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string", "description": "氏名"},
                        "department": {"type": "string", "description": "部署名"},
                        "title": {"type": "string", "description": "役職"},
                        "email": {"type": "string", "description": "メールアドレス"},
                    },
                    "required": ["name", "department", "title", "email"],
                }
            },
        }
    },
    {
        "toolSpec": {
            "name": "read_fortune",
            "description": "生年月日から、その人の今日の運勢・ラッキーカラー・ラッキーナンバーを所定のフォーマットで返します。",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "birthday": {
                            "type": "string",
                            "description": "生年月日（YYYY-MM-DD 形式）",
                        }
                    },
                    "required": ["birthday"],
                }
            },
        }
    },
]

FORTUNES = ["大吉", "中吉", "吉", "末吉", "小吉"]
LUCKY_COLORS = ["ゴールド", "エメラルドグリーン", "ロイヤルブルー", "ラベンダー", "コーラルピンク"]
LUCKY_NUMBERS = [7, 3, 8, 2, 5]


def run_tool(name: str, params: dict) -> str:
    if name == "create_signature":
        return (
            "━━━━━━━━━━━━━━━━━━━━━━\n"
            f"【{params.get('name', '')}】{COMPANY_NAME} {params.get('department', '')} {params.get('title', '')}\n"
            f"Email: {params.get('email', '')}\n"
            f"Tel:   {COMPANY_PHONE}\n"
            "━━━━━━━━━━━━━━━━━━━━━━"
        )
    if name == "read_fortune":
        birthday = params.get("birthday", "")
        try:
            year, month, day = (int(x) for x in birthday.split("-"))
            datetime.date(year, month, day)
        except ValueError:
            return "存在しない日付、または無効な形式です（YYYY-MM-DD 形式で指定してください）"
        index = (year + month + day) % 5
        return (
            "★★★ 誕生日占い ★★★\n"
            f"生年月日: {birthday}\n"
            f"今日の運勢: {FORTUNES[index]}\n"
            f"ラッキーカラー: {LUCKY_COLORS[index]}\n"
            f"ラッキーナンバー: {LUCKY_NUMBERS[index]}\n"
            "★★★★★★★★★★★★"
        )
    return f"Unknown tool: {name}"


def lambda_handler(event, context):
    question = event["question"]
    messages = [{"role": "user", "content": [{"text": question}]}]

    for _ in range(5):
        response = bedrock.converse(
            modelId=MODEL_ID,
            messages=messages,
            toolConfig={"tools": TOOLS},
        )
        output_message = response["output"]["message"]
        messages.append(output_message)

        if response["stopReason"] != "tool_use":
            answer = "".join(
                block["text"] for block in output_message["content"] if "text" in block
            )
            return {"answer": answer}

        tool_results = []
        for block in output_message["content"]:
            if "toolUse" not in block:
                continue
            tool_use = block["toolUse"]
            result = run_tool(tool_use["name"], tool_use.get("input", {}))
            tool_results.append(
                {
                    "toolResult": {
                        "toolUseId": tool_use["toolUseId"],
                        "content": [{"text": result}],
                    }
                }
            )
        messages.append({"role": "user", "content": tool_results})

    return {"answer": "ツール呼び出しが上限に達しました"}

貼り付けたら「Deploy」を押して反映します。

コードの解説

TOOLS = [
    {
        "toolSpec": {
            "name": "create_signature",
            "description": "氏名・部署名・役職・メールアドレスを受け取り、所定のフォーマットのメール署名を返します。会社名と代表電話番号は Lambda 側で自動的に付与されます。",
            ...
        }
    },
    ...
]

TOOLSはモデルに提示するツール定義の配列です。nameはモデルがツールを呼び出すときに指定する識別子、descriptionはモデルがそのツールをいつ使うか判断する材料になります。inputSchemaはJSON Schema形式で、モデルが渡してくる引数の型を宣言します。

モデルは description に書かれた自然言語の説明を読んで「ユーザの質問がこのツールに合うか」を判断するため、description の書き方がツール呼び出しの精度に直結します。例えば create_signature の description に「氏名・部署名・役職・メールアドレスを受け取り、所定のフォーマットのメール署名を返します」と書いてあることで、モデルは「田中太郎のメール署名を作って」のような質問に対してこのツールを選ぶようになります。ツールを増やしていくときは、それぞれの description が「どんなときに呼ばれるべきか」を過不足なく説明できているかを意識するのがポイントです。

COMPANY_NAME = "DevOps株式会社"
COMPANY_PHONE = "03-1234-5678"

会社名と代表電話番号は、Lambda 側の定数として持っています。実運用では「同じ会社の従業員なら共通で使う情報」に該当し、ユーザ側（呼び出し元）から毎回渡してもらう必要のない値をここで固定します。モデルはこの定数を見ることができないので、応答にこの値が含まれていれば create_signature ツールが実際に呼ばれた証拠になります。

def run_tool(name: str, params: dict) -> str:
    if name == "create_signature":
        return (
            "━━━━━━━━━━━━━━━━━━━━━━\n"
            f"【{params.get('name', '')}】{COMPANY_NAME} {params.get('department', '')} {params.get('title', '')}\n"
            ...
            f"Tel:   {COMPANY_PHONE}\n"
            ...
        )
    if name == "read_fortune":
        ...

run_toolは、モデルが呼び出しを要求してきたツールを実行するローカル関数です。関数名や書き方は自由ですが、以下のように TOOLS 定義とペアになるようにインタフェースを組みます。

• 第1引数 name: モデルが呼び出しを要求したツールの名前。TOOLS[].toolSpec.name の値（今回なら "create_signature" または "read_fortune"）がそのまま渡ってくる
• 第2引数 params: モデルが渡してきた引数の辞書。TOOLS[].toolSpec.inputSchema で宣言した JSON Schema に沿った形（例: create_signature なら {"name": "田中太郎", "department": "開発部", ...}）で渡ってくる
• 返り値: ツール実行の結果を文字列で返す。この文字列がそのまま後続の toolResult.content[].text としてモデルに戻される

関数本体では name で分岐して、TOOLS で定義した各ツールに対応する処理を実装します。TOOLS に新しいツールを追加したら、run_tool にも同じ name の分岐を1つ足す、というのが基本的なメンテナンスの流れになります。create_signature の COMPANY_NAME／COMPANY_PHONE や、占いの独自ロジックのように「モデルには内容が推測できない値」を持たせておくと、後で動作確認するときに「モデルが自力で答えたのか、ツールを呼んだのか」が判別しやすくなります。

question = event["question"]
messages = [{"role": "user", "content": [{"text": question}]}]

このセクションでは Lambda 呼び出しごとに新しい会話を始める前提にしていて、messages はユーザ発言だけを含んだ状態から開始します。呼び出しをまたぐ会話履歴を扱いたい場合は、「マルチターン会話」セクションのように session_id を受け取って DynamoDB から履歴を復元する形に変えます。

for _ in range(5):

エージェントの中核であるツール呼び出しループの入口です。range(5) で上限回数を設けているのは、モデルが延々とツールを呼び出し続けて課金・実行時間が想定外に膨らむことを防ぐためです。実運用では、要件に応じて上限値を調整します。

response = bedrock.converse(
    modelId=MODEL_ID,
    messages=messages,
    toolConfig={"tools": TOOLS},
)

converse の呼び出しで toolConfig に TOOLS を渡すことで、モデルに使えるツール一覧を提示します。

output_message = response["output"]["message"]
messages.append(output_message)

モデル応答を取り出し、messages に積みます。ループ内でツール呼び出しと結果のやり取りを続けるため、次のループの converse 呼び出しでも同じ messages を参照できるようにしておきます。

if response["stopReason"] != "tool_use":
    answer = "".join(
        block["text"] for block in output_message["content"] if "text" in block
    )
    return {"answer": answer}

Converse のレスポンスに含まれる stopReason が tool_use 以外（end_turn など）のときは、モデルが最終回答を返した合図です。テキストを取り出して呼び出し側に返し、ループを抜けます。

tool_results = []
for block in output_message["content"]:
    if "toolUse" not in block:
        continue
    tool_use = block["toolUse"]
    result = run_tool(tool_use["name"], tool_use.get("input", {}))
    tool_results.append(
        {
            "toolResult": {
                "toolUseId": tool_use["toolUseId"],
                "content": [{"text": result}],
            }
        }
    )

stopReason が tool_use のときは、モデルがツール実行を要求している状態です。応答に含まれる toolUse ブロックごとに run_tool を呼び出し、実行結果を toolResult ブロックとして tool_results に詰めていきます。toolUseId はモデルの要求と実行結果を紐づけるためのIDで、モデル側の要求に付いているものをそのまま返します。この for block in output_message["content"]: の書き方は、モデルが1つの応答に複数の toolUse ブロックを詰めて返してきた場合（＝並列ツール呼び出し）にもそのまま対応します。

messages.append({"role": "user", "content": tool_results})

まとめた tool_results を user ロールの新しいメッセージとして messages に積みます。次のループでこの実行結果をモデルに渡すことで、モデルはツールの実行結果を踏まえた次のアクション（別のツール呼び出し、または最終回答）を決められます。

動作確認

「テスト」タブを開き、以下の内容で新しいテストイベントを作成します。イベント名は任意（例: signature）です。質問文にはユーザ側で持っている情報（氏名・部署名・役職・メール）だけを渡し、会社名や代表電話番号は Lambda 側に任せます。

{
  "question": "田中太郎（開発部 エンジニア、tanaka@example.com）のメール署名を作って"
}

AI の応答は毎回まったく同じにはならないので、以下は一例です。以下のポイントが押さえられていれば成功です（モデルは ━━━ の罫線や 【】 の飾りを絵文字入りの Markdown に置き換えて返すことがあるので、値そのもの で判定します）。

• 質問で渡していない DevOps株式会社（会社名）が応答に含まれる（Lambda 側の定数から取り出したもの）
• 質問で渡していない 03-1234-5678（代表電話番号）が応答に含まれる（Lambda 側の定数から取り出したもの）

{
  "answer": "以下のメール署名を作成しました。\n\n━━━━━━━━━━━━━━━━━━━━━━\n【田中太郎】DevOps株式会社 開発部 エンジニア\nEmail: tanaka@example.com\nTel:   03-1234-5678\n━━━━━━━━━━━━━━━━━━━━━━\n\nそのままメール本文の末尾に貼り付けてご利用ください。"
}

会社名と代表電話番号はユーザから渡していないので、モデルには「本来知りようがない値」です。それが応答に正しく DevOps株式会社 と 03-1234-5678 として含まれていれば、モデルが自力で作った署名ではなく、Lambda 上の create_signature ツールを実際に呼び出して結果を受け取っている、と明確に判断できます。

続けて、以下のテストイベントで占いツールも動作するかを確認します。

{
  "question": "1990年5月10日生まれの人の運勢を占って"
}

このターンでは、以下のポイントが押さえられていれば成功です（モデルは ★★★ の枠を絵文字入りの見た目に置き換えたり、独自の Markdown 装飾に整形することがあるので、値そのもの で判定します）。

• 応答に「大吉／中吉／吉／末吉／小吉」のいずれかが含まれる
• 応答に「ゴールド／エメラルドグリーン／ロイヤルブルー／ラベンダー／コーラルピンク」のいずれかが含まれる
• 応答に 7 / 3 / 8 / 2 / 5 のいずれかがラッキーナンバーとして含まれる
• 上の3つが Lambda 側の同一インデックスの値になっている（1990-05-10 なら (1990+5+10) % 5 = 0 なので 大吉 + ゴールド + 7 の組み合わせ）

Lambda 側の並びは任意で決めた組み合わせなので、モデルが自力で「大吉 + ゴールド + 7」といったピンポイントの組み合わせを正解として返すことはまずありません。この組み合わせで返ってきていれば、read_fortune ツールが実際に呼ばれた証拠になります。

{
  "answer": "1990年5月10日生まれの方の今日の運勢は以下の通りです。\n\n★★★ 誕生日占い ★★★\n生年月日: 1990-05-10\n今日の運勢: 大吉\nラッキーカラー: ゴールド\nラッキーナンバー: 7\n★★★★★★★★★★★★\n\n素敵な一日をお過ごしください!"
}

5.2 エラーハンドリング

エラーハンドリングの概要

前セクションの run_tool は、ツール実行が失敗しても文字列で「エラー」を返すだけで、モデル側にはそれが「エラーなのか正常な結果なのか」区別できません。実運用ではツールが外部APIやDBを叩くことが多く、ネットワーク断や無効なパラメータで失敗するのは日常的です。

Converse APIの toolResult ブロックには status フィールドがあり、"error" を指定するとモデルに「このツール実行は失敗した」と明示的に伝えられます。モデルはそれを踏まえて、別のパラメータで再挑戦したり、ユーザに事情を説明したりできます。

Lambdaのコード変更

前セクションの index.py に、run_tool を成功／失敗のペア（(status, content) タプル）で返す形にし、ループ側で toolResult の status に反映する変更を加えます。index.py を以下の内容に置き換えます。

import datetime
import os
import boto3

MODEL_ID = os.environ["MODEL_ID"]
REGION = "ap-northeast-1"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)

COMPANY_NAME = "DevOps株式会社"
COMPANY_PHONE = "03-1234-5678"

TOOLS = [
    {
        "toolSpec": {
            "name": "create_signature",
            "description": "氏名・部署名・役職・メールアドレスを受け取り、所定のフォーマットのメール署名を返します。会社名と代表電話番号は Lambda 側で自動的に付与されます。",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string", "description": "氏名"},
                        "department": {"type": "string", "description": "部署名"},
                        "title": {"type": "string", "description": "役職"},
                        "email": {"type": "string", "description": "メールアドレス"},
                    },
                    "required": ["name", "department", "title", "email"],
                }
            },
        }
    },
    {
        "toolSpec": {
            "name": "read_fortune",
            "description": "生年月日から、その人の今日の運勢・ラッキーカラー・ラッキーナンバーを所定のフォーマットで返します。",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "birthday": {
                            "type": "string",
                            "description": "生年月日（YYYY-MM-DD 形式）",
                        }
                    },
                    "required": ["birthday"],
                }
            },
        }
    },
]

FORTUNES = ["大吉", "中吉", "吉", "末吉", "小吉"]
LUCKY_COLORS = ["ゴールド", "エメラルドグリーン", "ロイヤルブルー", "ラベンダー", "コーラルピンク"]
LUCKY_NUMBERS = [7, 3, 8, 2, 5]


def run_tool(name: str, params: dict) -> tuple[str, str]:
    try:
        if name == "create_signature":
            signature = (
                "━━━━━━━━━━━━━━━━━━━━━━\n"
                f"【{params.get('name', '')}】{COMPANY_NAME} {params.get('department', '')} {params.get('title', '')}\n"
                f"Email: {params.get('email', '')}\n"
                f"Tel:   {COMPANY_PHONE}\n"
                "━━━━━━━━━━━━━━━━━━━━━━"
            )
            return ("success", signature)
        if name == "read_fortune":
            birthday = params.get("birthday", "")
            try:
                year, month, day = (int(x) for x in birthday.split("-"))
                datetime.date(year, month, day)
            except ValueError:
                return ("error", "存在しない日付、または無効な形式です（YYYY-MM-DD 形式で指定してください）")
            index = (year + month + day) % 5
            result = (
                "★★★ 誕生日占い ★★★\n"
                f"生年月日: {birthday}\n"
                f"今日の運勢: {FORTUNES[index]}\n"
                f"ラッキーカラー: {LUCKY_COLORS[index]}\n"
                f"ラッキーナンバー: {LUCKY_NUMBERS[index]}\n"
                "★★★★★★★★★★★★"
            )
            return ("success", result)
        return ("error", f"Unknown tool: {name}")
    except Exception as e:
        return ("error", f"ツール実行中にエラーが発生しました: {e}")


def lambda_handler(event, context):
    question = event["question"]
    messages = [{"role": "user", "content": [{"text": question}]}]

    for _ in range(5):
        response = bedrock.converse(
            modelId=MODEL_ID,
            messages=messages,
            toolConfig={"tools": TOOLS},
        )
        output_message = response["output"]["message"]
        messages.append(output_message)

        if response["stopReason"] != "tool_use":
            answer = "".join(
                block["text"] for block in output_message["content"] if "text" in block
            )
            return {"answer": answer}

        tool_results = []
        for block in output_message["content"]:
            if "toolUse" not in block:
                continue
            tool_use = block["toolUse"]
            status, content = run_tool(tool_use["name"], tool_use.get("input", {}))
            tool_results.append(
                {
                    "toolResult": {
                        "toolUseId": tool_use["toolUseId"],
                        "content": [{"text": content}],
                        "status": status,
                    }
                }
            )
        messages.append({"role": "user", "content": tool_results})

    return {"answer": "ツール呼び出しが上限に達しました"}

貼り付けたら「Deploy」を押して反映します。

コードの解説

def run_tool(name: str, params: dict) -> tuple[str, str]:
    try:
        if name == "create_signature":
            signature = (...)
            return ("success", signature)
        if name == "read_fortune":
            birthday = params.get("birthday", "")
            try:
                year, month, day = (int(x) for x in birthday.split("-"))
                datetime.date(year, month, day)
            except ValueError:
                return ("error", "存在しない日付、または無効な形式です（YYYY-MM-DD 形式で指定してください）")
            ...
            return ("success", result)
        return ("error", f"Unknown tool: {name}")
    except Exception as e:
        return ("error", f"ツール実行中にエラーが発生しました: {e}")

run_tool の返り値を (status, content) のタプルに変えました。成功時は ("success", 結果文字列)、失敗時は ("error", エラーメッセージ) を返す形に統一しています。read_fortune の内側で日付の妥当性チェックを行い、想定外の例外は関数全体を囲む外側の try/except で捕まえて ("error", ...) で返す構造にしています。

status, content = run_tool(tool_use["name"], tool_use.get("input", {}))
tool_results.append(
    {
        "toolResult": {
            "toolUseId": tool_use["toolUseId"],
            "content": [{"text": content}],
            "status": status,
        }
    }
)

toolResult ブロックに status フィールドを追加しました。"success" か "error" を指定でき、"error" の場合はモデル側で「このツール呼び出しは失敗した」と認識してくれます。

動作確認

以下のテストイベントを実行します。read_fortune ツールに、あえて存在しない日付（2月30日）を含む質問を投げます。

{
  "question": "1990年2月30日生まれの人の運勢を占って"
}

AI の応答は毎回まったく同じにはならないので、以下のポイントが押さえられていれば成功です。

• モデルが「2月30日は存在しないので占えない」といった内容を返し、正しい生年月日を問い返す（失敗を踏まえてユーザに事情を説明するリカバリ）、あるいは近い日付（例: 1990-02-28）で再度 read_fortune を呼び出してリトライして通常の占い結果を返してくる
• もしモデルが 1990-02-30 を「そのまま解釈して普通に占ってくれた」ような応答を返した場合は、モデルが toolResult.status="error" を認識できずにエラー結果を無視している状態です（ツール呼び出しと error 返却自体は行われているものの、モデルがそれを踏まえた振る舞いに切り替えていない）

{
  "answer": "申し訳ありません。2月30日は実際には存在しない日付のため、占うことができませんでした。\n\nもしよろしければ、正しい生年月日（例: 1990年2月28日 など）をお教えいただけますか?あらためて占わせていただきます。"
}

toolResult.status に "error" を明示することで、モデルが失敗を認識して自ら別のアプローチ（ユーザへの問い返し／別パラメータでのリトライ）に切り替えていることが確認できます。

6. 不要リソースの削除

Lambda関数とIAMロールは放置してもリクエストがなければ課金は発生しませんが、今後のハンズオンで作り直すため、ここで一旦削除しておく方が整理された状態を保てます。

6.1 CloudFormationスタックを削除する

CloudFormationのダッシュボードで、ai-platform-bedrock-agent スタックのラジオボタンを選択し、右上のスタックを削除をクリックします。ステータスが DELETE_COMPLETE になれば、Lambda 関数・実行ロール・DynamoDB テーブル（会話履歴を含む）がまとめて削除されます。

7. まとめ

このセクションでは、シンプルなBedrock呼び出しを発展させて、LambdaでBedrockを扱う際の応用操作を5つのテーマに分けて体験しました。

• DynamoDB に session_id 単位で会話履歴を保存することで、Lambda のステートレス性を維持したままモデルが文脈を保った応答を返せることを確認した
• DynamoDB から取得した履歴を直近N件に切り詰めることで、履歴が伸び続けてコンテキスト長・料金が膨らむのを防ぐ設計にした
• system パラメータでシステムプロンプトを与えることで、エージェントに役割・応答スタイルを持たせられることを確認した
• ツール定義（toolSpec）でモデルに使えるツールを提示し、stopReason が tool_use のときにツール実行→結果を返すループを組んだ（1ターンで複数の toolUse ブロックが返る並列呼び出しにも同じ実装で対応できる形になっている）
• toolResult.status に "error" を明示することで、モデルがツール失敗を認識して自ら別のアプローチに切り替えられることを確認した

次のハンズオンでは、ここで手作りしてきたマルチターン会話・システムプロンプト・ツール呼び出しループ・履歴管理といったエージェントの共通部分を、Bedrock AgentCore に任せるアプローチを扱います。