コンテンツにスキップ

第1章 生成AIの基本とAmazon Bedrock入門 — 学習メモ

書籍「Amazon Bedrock AgentCore実践入門」第1章のサンプルコード(このフォルダ)を読み解いた個人学習メモ。 AgentCore 全体像は ../../lectures/agentcore_basics/STUDY_NOTES.md 参照。 実行検証は伴わない(コードの構造から解説)。コードに現れた API 名だけ断定し、読めない挙動は「〜と推測」で明示する。

一言で

この章は「生の Bedrock API を素手で叩くとどれだけ面倒か」を体験する章。 converse 自体は1回呼ぶだけの薄いラッパーだが、ツール呼び出しになった途端「モデルの返答を見てツール要求かどうか判定し、実行し、結果をメッセージ履歴に手で追記し、また呼び直す」という while ループを自分で書く必要が出てくる。この面倒さこそが、第3章で学ぶ Strands Agents の Agent() が「ありがたい」理由の伏線になっている。対比の軸は「ループを誰が書くか=自分 vs フレームワーク」

全体像

3ファイルは互いに独立して実行できる(uv run 01_converse.py のように単体実行)。複雑さは 01 → 02 → 03 の順で段階的に増す。

  • 01_converse.py: 同期の1往復。もっとも単純な形。
  • 02_converse_stream.py: 同じ1往復だが、応答をトークン単位でストリーミング受信する。
  • 03_tool_use.py: 唯一「動的な処理フロー」を持つファイル。モデルがツール呼び出しを要求する限り while ループが回り続ける。

03_tool_use.py の実行フロー(file:56-110):

flowchart TD
    Start["converseを呼ぶ<br>messages + toolConfig"] --> Check{"レスポンスに<br>toolUseブロックが<br>含まれるか"}
    Check -->|"No(textのみ)"| Finish["textを出力してループ終了"]
    Check -->|"Yes"| AppendAssistant["response全体を<br>messagesに追加<br>(assistant発言として)"]
    AppendAssistant --> RunTool["tool_listから該当関数を<br>取得して実行"]
    RunTool --> AppendResult["toolResultを<br>messagesに追加<br>(role=user)"]
    AppendResult --> ReConverse["再度converseを呼ぶ"]
    ReConverse --> Check

使用ライブラリ・原理

  • boto3.client("bedrock-runtime"): AWS SDK(boto3)で Bedrock Runtime エンドポイントへの署名付き(SigV4)HTTP クライアントを作る。認証情報は環境の AWS 認証情報チェーン(~/.aws/credentials や環境変数)から自動解決される。README では事前に aws sts get-caller-identity で疎通確認している。
  • Converse API(converse / converse_stream: モデルごとに異なる生の JSON ボディ形式を吸収し、Anthropic・Meta 等を横断する統一されたメッセージ形式で呼び出せる API。messages{"role": "user"|"assistant", "content": [...]} のリストで、content{"text": ...}{"toolUse": ...} 等のブロックの配列。
  • converse(同期): 応答が完成するまでブロックし、response["output"]["message"]["content"] に結果ブロック配列が入る(01_converse.py:20)。
  • converse_stream(非同期的にチャンクを流す): streaming_response["stream"] をイテレートし、"contentBlockDelta" in chunk を見て chunk["contentBlockDelta"]["delta"]["text"] を取り出す(02_converse_stream.py:20-24)。ストリーミングはトークン単位の差分(delta)が飛んでくるだけで、最終形の組み立ては呼び出し側の責務。
  • ツール利用(Function Calling): toolConfig={"tools": [tool_spec]} でツール定義(JSON Schema 形式の inputSchema)を渡す。モデルが「このツールを呼びたい」と判断すると、content{"toolUse": {"toolUseId", "name", "input"}} ブロックが混ざって返ってくる。API 自体はツールを実行しない。呼び出し側が toolUse を検出し、対応する関数を実行し、結果を {"toolResult": {"toolUseId", "content": [{"text": ...}]}} として messages に追記し、再度 converse を呼ぶ、という手続きを自分で書く必要がある(マニュアル ReAct ループ)。

ファイル別の役割

ファイル 役割
01_converse.py 最小の同期呼び出し。converse で1往復してテキストを出力するだけ
02_converse_stream.py converse_stream でトークン単位ストリーミング出力(end="" で改行抑制)
03_tool_use.py ツール呼び出し。while ループで toolUse 検出 → 実行 → toolResult 追加 → 再 converse を繰り返す
pyproject.toml 依存は boto3[crt]==1.42.96 のみ。Python >=3.14

中心コードの読み解き

03_tool_use.py:56-110(ツール呼び出しの while ループ本体):

while True:                                          # ①
    tool_request = []
    for content in response["output"]["message"]["content"]:
        if "text" in content:
            print(f"text: {content['text']}")
        if "toolUse" in content:
            tool_request.append(content)             # ②

    if len(tool_request) == 0:
        break                                        # ③

    messages.append(response["output"]["message"])   # ④

    tool_result = []
    for tool_use in tool_request:
        tool_use_id = tool_use["toolUse"]["toolUseId"]
        tool_name = tool_use["toolUse"]["name"]
        tool_input = tool_use["toolUse"]["input"]
        tool = tool_list[tool_name]                  # ⑤
        result = tool(**tool_input)
        tool_result.append({
            "toolResult": {
                "toolUseId": tool_use_id,             # ⑥
                "content": [{"text": result}],
            }
        })

    messages.append({"role": "user", "content": tool_result})  # ⑦

    response = client.converse(                       # ⑧
        modelId="us.anthropic.claude-sonnet-4-6",
        messages=messages,
        toolConfig={"tools": [tool_spec]},
    )
やってること なぜ
while True ループ継続の条件を「ツール要求が0件になるまで」に固定 モデルが何回ツールを呼ぶか事前に分からないため、終了条件を動的に判定する必要がある
toolUse ブロックを収集 1回の応答に複数のツール要求が混ざる可能性を考慮しリスト化 サンプルの入力「半径3センチと7センチ」は2回分のツール呼び出しを誘発する想定と推測
break toolUse が無ければテキスト回答のみ=完了とみなす Converse API はツール要求と最終回答を同じ content 配列の形で返すため、中身を見るまで完了かどうか分からない
messages.append(response["output"]["message"]) モデルの直前の発言(ツール要求を含む)を会話履歴に追加 Bedrock はステートレス API なので、次の converse 呼び出しに全履歴を毎回渡す必要がある
tool_list[tool_name] ツール名の文字列から実行すべき Python 関数を辞書引き モデルが返すのは「呼びたい関数名」の文字列だけで、実行主体は呼び出し側
toolUseId を結果に含める toolResult と対応する toolUse を ID で紐付ける 複数ツールを並行要求された場合に、どの結果がどの要求に対応するか API 側が判別するため
messages.append({"role": "user", ...}) ツール実行結果を「user」ロールとして履歴に追加 Converse API の規約上、toolResult は user メッセージの content として渡す形式になっている
⑧ 再度 converse を呼ぶ 更新された messages(ツール結果込み)で再問い合わせ モデルがツール結果を見て次の判断(さらにツールを呼ぶ/最終回答を返す)をできるようにする

学んだこと(要点)

  • Converse API のレスポンス構造は一貫して response["output"]["message"]["content"](ブロックのリスト)。text ブロックと toolUse ブロックが同じ配列に混在しうる。
  • ツール定義(tool_spec)は Bedrock 独自の toolSpec.inputSchema.json という入れ子構造で、中身自体は JSON Schema。
  • messages はただの Python リストで、呼び出し側が完全に手で管理する(フレームワークによる自動永続化・トリミングは一切ない)。
  • ストリーミング(converse_stream)は「テキストの差分だけ」流れてくるので、ツール呼び出しを含む場合の差分ハンドリングはこの章のサンプルには出てこない(03_tool_use.py は同期 converse のみ使用)。

落とし穴・現代版に移植するなら

  • modelId="us.anthropic.claude-sonnet-4-6" はクロスリージョン推論のための inference profile IDus. プレフィックス)。単一リージョンの生モデル ID とは別物で、移植時は現行の推奨 ID を確認する。
  • boto3[crt]==1.42.96 と厳密にバージョン固定(書籍の方針でバージョンドリフトによる破壊的変更を避けるため)。[crt] extra は AWS Common Runtime(C実装)を使い、署名処理などが高速化される。
  • サンプルにエラーハンドリングが一切ない(入門書の簡潔さ優先)。実運用では client.converse の例外(スロットリング・モデルアクセス未許可等)を捕捉する必要がある。
  • pyproject.tomlproject.name"handson" になっている(テンプレートの使い回しと推測。実体は第1章のコードで実害はない)。
  • 会話状態(messages)はプロセスのメモリ上にしかなく、スクリプト終了で消える。永続化・セッション管理は本章のスコープ外(第3章 08_session.pyFileSessionManager で初めて登場)。

記事参照

  • 書籍 第1章「生成AIの基本とAmazon Bedrock入門」。関連: ../../lectures/agentcore_basics/STUDY_NOTES.md(「1. 3階建ての全体像」「2. Strands Agents 入門」の生Bedrockとの対比表で本章のコードが引用されている)

作成: 2026-07-17 / 最終更新: 2026-07-17