Amazon Bedrock で OpenAI モデルを実行する方法（58 分読了）

Amazon Bedrock 上の OpenAI モデルは、テキスト生成、構造化出力、アプリケーションツール、直接ファイル入力、レスポンス状態、プロンプトキャッシング、およびバックグラウンド処理を必要とする本番ワークフロー向けに、OpenAI 互換の Responses API サフェースを提供します。このクックブックでは、遅延および破損した注文の交換リクエストを取り扱う架空の小売業者「BrightCart」向けのサポートアシスタントワークフローを構築することで、具体例を示します。

通常アプリケーション呼び出しには OpenAI Python SDK を使用し、正確なリクエストボディを検証する必要がある場合にのみ、小さな生 HTTPS ヘルパーを使用します。フローはセットアップと最小限の事前チェックから始まり、レスポンスライフサイクル、モデル制御、構造化 JSON、アプリケーションツール、ファイル入力、状態管理、キャッシング、バックグラウンド処理、コンテキスト圧縮、操作チェック、そしてクリーンアップへと段階的に積み重ねられていきます。

以下を学ぶことができます:

• Bedrock 固有の環境変数を使用して、Bedrock ホスト型の OpenAI モデルを設定する方法。

• Responses エンドポイントを検証し、レスポンススキーマ、使用状況メタデータ、および正規化されたエラーを確認する方法。

• 生 HTTPS と OpenAI SDK の両方を使用してテキストリクエストを送信する方法。

• スキーマ制約付きの JSON を生成し、軽量な JSON モードハンドオフを行う方法。

• アプリケーション管理型の関数ツール、並列ツール、およびカスタムテキストツールを呼び出す方法。

• 直接 PDF 入力を送信し、状態あり・状態なしの会話を継続し、暗号化された推論コンテキストを引き継ぐ方法。

• プロンプトキャッシング、バックグラウンドモード、圧縮、操作用スモークチェック、および保存済みレスポンスのクリーンアップを使用する方法。

前提条件：Amazon Bedrock 上の OpenAI モデル用のベアータン、Python 3.9 以降、および Bedrock OpenAI 互換エンドポイントへのネットワークアクセスが必要です。

このガイドでは、デフォルトで us-west-2 リージョンにおいて openai.gpt-5.4 を実行します。他のサポートされている組み合わせを使用する場合は、セットアップセルを実行する前に AWS_REGION、BEDROCK_MODEL、および BEDROCK_BASE_URL を同時に変更してください。

AWS リージョン | サポートされるモデル ID

us-west-2 | openai.gpt-5.4

us-east-2 | openai.gpt-5.5, openai.gpt-5.4

このセクションでは、ノートブックのランタイムを準備します。ここでは軽量な Python スタックをインストールし、Bedrock 固有の環境変数を読み込み、生粋の HTTPS セッションと OpenAI SDK クライアントの両を作成し、エンドポイントが提供する場合にモデルメタデータを発見し、後の例で使用される共通ヘルパー関数を定義します。

ノートブックを実行する前に、これらの環境変数を設定してください。デフォルトの組み合わせは us-west-2 リージョンと openai.gpt-5.4 です。

export AWS_BEARER_TOKEN_BEDROCK="YOUR_BEDROCK_BEARER_TOKEN"

export AWS_REGION="us-west-2"

export BEDROCK_MODEL="openai.gpt-5.4"

export BEDROCK_BASE_URL="https://bedrock-mantle.${AWS_REGION}.api.aws/openai/v1"

ベアータン（Bearer Token）は AWS_BEARER_TOKEN_BEDROCK から読み込まれます。これが欠落している場合、セットアップセルはパスワード形式のプロンプトでその入力を求めますが、値自体は表示しません。

1.1 依存関係のインストール

ノートブックで使用されるパッケージをインストールします。アプリケーションの例には OpenAI SDK を使用し、Responses エンドポイントへの生 HTTPS 呼び出しには requests を使用し、pandas と IPython の表示ヘルパーを使用して、Cookbook レンダラー内でリクエストとレスポンスの要約を読みやすく保ちます。パッケージがインストールされたか、または既に存在するかを確認するためにのみセル出力を検査してください。

%pip install -U "openai>=2.28.0" requests pandas ipython --quiet

print("Dependencies installed or already available: openai, requests, pandas, ipython")

[notice] A new release of pip is available: 24.0 -> 26.1.1

[notice] To update, run: pip install --upgrade pip

Note: you may need to restart the kernel to use updated packages.

Dependencies installed or already available: openai, requests, pandas, ipython

1.2 Import Libraries and Defaults

ノートブック全体で使用される標準ライブラリ、SDK、HTTP クライアント、表示ユーティリティをインポートします。このセルではまた、環境変数が既に設定されていない場合に使用されるデフォルトの Bedrock リージョンとモデルも設定されます。ノートブックが us-west-2 および openai.gpt-5.4 から開始されることを確認するために、またはそれらを上書きしない限り、印刷されたデフォルト値を検査してください。

from __future__ import annotations

import base64

import builtins

import html

import json

import os

import shlex

import textwrap

import time

from datetime import date, timedelta

from getpass import getpass

from typing import Any, Callable, Iterable

import pandas as pd

import requests

from IPython.display import HTML, Markdown, display

from openai import OpenAI

DEFAULT_REGION = "us-west-2"

DEFAULT_MODEL = "openai.gpt-5.4"

PREFERRED_MODELS = [DEFAULT_MODEL]

def gpt_version_tuple(model_id: str) -> tuple[int, int] | None:

normalized = model_id.lower().removeprefix("openai.")

if not normalized.startswith("gpt-"):

return None

version = normalized.removeprefix("gpt-").split("-")[0]

parts = version.split(".")

try:

major = builtins.int(parts[0])

minor = builtins.int(parts[1]) if len(parts) > 1 else 0

except ValueError:

return None

return major, minor

def prompt_cache_retention_for_model(model_id: str) -> str:

version = gpt_version_tuple(model_id)

if version and version >= (5, 5):

return "24h"

return "in_memory"

pd.set_option("display.max_columns", None)

pd.set_option("display.max_rows", 200)

pd.set_option("display.max_colwidth", None)

pd.set_option("display.width", 160)

def display_wrapped_table(df: pd.DataFrame, *, max_col_width_px: int = 520, index: bool = False) -> None:

if df.empty:

display(Markdown("_No rows to display._"))

return

table_html = df.to_html(index=index, escape=True, border=0)

table_html = table_html.replace('<table border="0" class="dataframe">', '<table class="dataframe wrapped-output-table">")

display(HTML(f"""

<style>

.wrapped-output-table {{

border-collapse: collapse;

width: 100%;

table-layout: auto;

font-size: 13px;

}}

.wrapped-output-table th,

.wrapped-output-table td {{

border: 1px solid #d0d7de;

padding: 6px 8px;

text-align: left;

vertical-align: top;

white-space: pre-wrap;

overflow-wrap: anywhere;

word-break: break-word;

max-width: {max_col_width_px}px;

}}

.wrapped-output-table th {{

background: #f6f8fa;

font-weight: 600;

}}

</style>

{table_html}

"""))

print("Imports loaded.")

print("Default region:", DEFAULT_REGION)

print("Default model:", DEFAULT_MODEL)

Imports loaded.

Default region: us-west-2

Default model: openai.gpt-5.4

1.3 Configure Bedrock Credentials and Clients

環境変数から Bedrock の設定を読み取り、クライアントを構築します。BEDROCK_BASE_URL は一度正規化され、生の requests.Session にはベアータンがヘッダーとして付与され、OpenAI SDK クライアントは同じトークンとベース URL を明示的に指定して作成されます。本番呼び出しを行う前に、レンダリングされたテーブルを確認し、選択されたリージョン、モデル、エンドポイント、SDK クライアントの設定、および保存済みレスポンスのクリーンアップ動作を確認してください。

from __future__ import annotations

def env_value(*names: str) -> str | None:

for name in names:

value = os.environ.get(name)

if value:

return value

return None

def env_flag(name: str, default: bool = False) -> bool:

value = env_value(name)

if value is None:

return default

return value.strip().lower() in {"1", "true", "yes", "on"}

def normalize_base_url(url: str) -> str:

url = url.strip().rstrip("/")

if url.endswith("/responses"):

return url[: -len("/responses")]

return url

def endpoint(path: str) -> str:

return f"{BEDROCK_BASE_URL}/{path.lstrip('/')}

def responses_url(base_url: str) -> str:

return f"{normalize_base_url(base_url)}/responses"

API_TIMEOUT_SECONDS = float(env_value("BEDROCK_REQUEST_TIMEOUT_SECONDS") or "60")

MAX_RETRIES = builtins.int(env_value("BEDROCK_MAX_RETRIES") or "0")

CLEAN_UP_STORED_RESPONSES = env_flag("BEDROCK_CLEANUP_STORED_RESPONSES", True)

FAIL_ON_CHECK_FAILURE = env_flag("BEDROCK_FAIL_ON_CHECK_FAILURE", False)

RUN_RESPONSIVENESS_CHECK = env_flag("BEDROCK_RESPONSIVENESS_CHECK", True)

TRANSIENT_STATUS_CODES = {408, 409, 429, 500, 502, 503, 504}

AWS_REGION = (env_value("AWS_REGION") or DEFAULT_REGION).strip() or DEFAULT_REGION

BEDROCK_MODEL = (env_value("BEDROCK_MODEL") or DEFAULT_MODEL).strip() or DEFAULT_MODEL

BEDROCK_BASE_URL = normalize_base_url(

env_value("BEDROCK_BASE_URL") or f"https://bedrock-mantle.{AWS_REGION}.api.aws/openai/v1"

)

RESPONSES_URL = responses_url(BEDROCK_BASE_URL)

AWS_BEARER_TOKEN_BEDROCK = env_value("AWS_BEARER_TOKEN_BEDROCK")

if not AWS_BEARER_TOKEN_BEDROCK:

AWS_BEARER_TOKEN_BEDROCK = getpass("Paste your AWS Bedrock bearer token for this kernel session: ").strip()

if AWS_BEARER_TOKEN_BEDROCK:

os.environ["AWS_BEARER_TOKEN_BEDROCK"] = AWS_BEARER_TOKEN_BEDROCK

if not AWS_BEARER_TOKEN_BEDROCK:

raise RuntimeError("AWS_BEARER_TOKEN_BEDROCK is required to run the live examples.")

http = requests.Session()

http.headers.update({

"Authorization": f"Bearer {AWS_BEARER_TOKEN_BEDROCK}",

"Content-Type": "application/json",

})

client = OpenAI(api_key=AWS_BEARER_TOKEN_BEDROCK, base_url=BEDROCK_BASE_URL, max_retries=0)

BASE_URL = BEDROCK_BASE_URL

config_rows = [

{"setting": "AWS_REGION", "value": AWS_REGION},

{"setting": "BEDROCK_MODEL", "value": BEDROCK_MODEL},

{"setting": "BEDROCK_BASE_URL", "value": BEDROCK_BASE_URL},

{"setting": "SDK client", "value": "OpenAI(api_key=AWS_BEARER_TOKEN_BEDROCK, base_url=BEDROCK_BASE_URL)"},

{"setting": "cleanup stored responses", "value": CLEAN_UP_STORED_RESPONSES},

]

display_wrapped_table(pd.DataFrame(config_rows), max_col_width_px=680)

setting

value

AWS_REGION

us-west-2

BEDROCK_MODEL

openai.gpt-5.4

BEDROCK_BASE_URL

https://bedrock-mantle.us-west-2.api.aws/openai/v1

SDK client

OpenAI(api_key=AWS_BEARER_TOKEN_BEDROCK, base_url=BEDROCK_BASE_URL)

cleanup stored responses

True

1.4 利用可能なモデルの発見

選択されたエンドポイントがモデル一覧メタデータを公開している場合、利用可能なモデルを検索し、ノートブックの残りの部分で使用するモデルを選択します。BEDROCK_MODEL が設定されている場合はその値を使用しますが、そうでない場合は openai.gpt-5.4 を優先します。モデル一覧呼び出しはオプションです。一部の互換性のあるエンドポイントでは、モデルメタデータが利用できない場合でも推論が可能だからです。選択したモデルと返されたカタログ行を確認してください。

from __future__ import annotations

def list_openai_models(client: OpenAI) -> list[str]:

return sorted(model.id for model in client.models.list(timeout=API_TIMEOUT_SECONDS).data)

def resolve_model_id(client: OpenAI | None) -> tuple[str, list[str], str | None]:

configured_model = env_value("BEDROCK_MODEL")

available_models: list[str] = []

model_discovery_note: str | None = None

if client is not None:

try:

available_models = list_openai_models(client)

except Exception as exc:

status_code = getattr(exc, "status_code", None)

if status_code == 404:

model_discovery_note = "This endpoint did not expose model-list metadata. The guide will continue with the configured model."

else:

model_discovery_note = f"Model-list metadata could not be listed. The guide will continue with the configured model. Details: {builtins.str(exc)[:240]}"

if configured_model:

return configured_model, available_models, model_discovery_note

for candidate in PREFERRED_MODELS:

if candidate in available_models:

return candidate, available_models, model_discovery_note

for candidate in available_models:

if candidate.startswith("openai."):

return candidate, available_models, model_discovery_note

if available_models:

return available_models[0], available_models, model_discovery_note

return PREFERRED_MODELS[0], available_models, model_discovery_note

EXPLICIT_MODEL = env_value("BEDROCK_MODEL")

MODEL_ID, AVAILABLE_MODELS, MODEL_DISCOVERY_NOTE = resolve_model_id(client)

os.environ["BEDROCK_MODEL"] = MODEL_ID

PROMPT_CACHE_RETENTION = prompt_cache_retention_for_model(MODEL_ID)

PROMPT_CACHE_RETENTION_NOTE = (

"GPT-5.5 以降は 24 時間の拡張プロンプトキャッシュを使用します。それ以前の GPT-5 モデルでは in_memory（メモリ内）を使用可能です。"

)

config_rows = [{

"selected_model": MODEL_ID,

"model_was_explicit": bool(EXPLICIT_MODEL),

"model_catalog_status": "listed" if AVAILABLE_MODELS else "using configured model",

"discovered_model_count": len(AVAILABLE_MODELS),

"prompt_cache_retention": PROMPT_CACHE_RETENTION,

"prompt_cache_retention_note": PROMPT_CACHE_RETENTION_NOTE,

"note": MODEL_DISCOVERY_NOTE or "Model selection is ready.",

}]

display_wrapped_table(pd.DataFrame(config_rows), max_col_width_px=620)

if AVAILABLE_MODELS:

display_wrapped_table(pd.DataFrame({"available_models": AVAILABLE_MODELS[:25]}), max_col_width_px=520)

else:

print("Continuing with:", MODEL_ID)

selected_model

model_was_explicit

model_catalog_status

discovered_model_count

prompt_cache_retention

prompt_cache_retention_note

note

openai.gpt-5.4

False

using configured model

0

in_memory

GPT-5.5 およびそれ以降のバージョンでは 24 時間の拡張プロンプトキャッシュ（extended prompt caching）を使用可能ですが、以前の GPT-5 モデルでは in_memory を使用できます。

このエンドポイントはモデル一覧メタデータを公開していませんでした。ガイドでは設定されたモデルを継続して扱います。

Continuing with: openai.gpt-5.4

1.5 ヘルパー関数のセットアップ

ワークフローで共有するヘルパー関数を定義します。これらのヘルパーは、リクエスト形状のレンダリング、API エラーの正規化、生 HTTPS リクエストの送信、SDK 呼び出しへのオプションのリトライラップ、output_text の抽出、トークン使用量の要約、保存されたレスポンス ID の追跡、およびコンパクトなテーブルの表示を担当します。以下の例では各 API の概念に焦点を当てつつ、繰り返し発生するメカニクスはヘルパーが処理します。レスポンステキスト、使用量、エラー、クリーンアップがどのように処理されるかを理解したい場合は、このセルを検証してください。

from __future__ import annotations

RESULTS_SUMMARY: list[dict[str, Any]] = []

EXAMPLE_RESPONSES: list[dict[str, str]] = []

STORED_RESPONSE_IDS: list[str] = []

OUTPUT_WIDTH = 100

MAX_DISPLAY_TEXT_CHARS = builtins.int(env_value("BEDROCK_MAX_DISPLAY_CHARS") or "1200")

def truncate_display_text(text: Any, *, limit: int = MAX_DISPLAY_TEXT_CHARS) -> str:

rendered = builtins.str(text).strip()

if len(rendered) <= limit:

return rendered

return rendered[:limit].rstrip() + "\n[可読性の向上のため表示テキストを省略しました。完全な値は Python 変数を参照してください。]"

def compact_text(text: Any, limit: int = 220) -> str:

rendered = " ".join(builtins.str(text).split())

if len(rendered) <= limit:

return rendered

return rendered[:limit].rstrip() + "..."

def require(condition: Any, message: str) -> None:

if not condition:

raise ValueError(message)

def warn_or_raise(condition: bool, message: str) -> bool:

if condition:

return True

display(HTML(f"<div style=\"border-left:4px solid #d29922; padding:6px 10px; background:#fff8c5;\"><strong>Warning:</strong> {html.escape(message)}</div>"))

if FAIL_ON_CHECK_FAILURE:

raise AssertionError(message)

return False

def display_text_block(label: str, text: Any, *, limit: int = MAX_DISPLAY_TEXT_CHARS) -> None:

safe_label = html.escape(label)

safe_text = html.escape(truncate_display_text(text, limit=limit))

display(HTML(f"""

<div style="border:1px solid #d0d7de; border-radius:6px; margin:8px 0; overflow:hidden; font-size:13px;">

<div style="background:#f6f8fa; padding:6px 8px; font-weight:600;">{safe_label}</div>

<div style="padding:8px; white-space:pre-wrap; overflow-wrap:anywhere; line-height:1.45;">{safe_text}</div>

</div>

"""))

def print_wrapped(text: Any, *, width: int = OUTPUT_WIDTH) -> None:

print(textwrap.fill(builtins.str(text), width=width, break_long_words=True, break_on_hyphens=False))

def print_json(value: Any, *, width: int = OUTPUT_WIDTH) -> None:

display_json_block("JSON", value)

def print_label(label: str) -> None:

display(HTML(f"<div style=\"font-weight:600; margin:8px 0 4px;\">{html.escape(label)}</div>"))

def print_labeled_text(label: str, text: Any) -> None:

display_text_block(label, text)

def print_labeled_json(label: str, value: Any) -> None:

display_json_block(label, value)

def display_json_block(label: str, value: Any, *, limit: int = MAX_DISPLAY_TEXT_CHARS) -> None:

rendered = json.dumps(value, indent=2, default=builtins.str)

display_text_block(label, rendered, limit=limit)

def summarize_content(content: Any) -> str:

if isinstance(content, builtins.str):

return compact_text(content)

if isinstance(content, builtins.list):

parts: list[str] = []

for item in content:

if not isinstance(item, builtins.dict):

parts.append(compact_text(item, 80))

continue

item_type = item.get("type", "item")

if item_type == "input_text":

parts.append(f"input_text: {compact_text(item.get('text', ''), 120)}")

elif item_type == "input_file":

parts.append(f"input_file: {item.get('filename', '<inline file>')}")

else:

parts.append(item_type)

return "; ".join(parts)

return compact_text(content)

def summarize_input(input_value: Any) -> str:

if isinstance(input_value, builtins.str):

return compact_text(input_value, 260)

if isinstance(input_value, builtins.list):

messages: list[str] = []

for item in input_value[:4]:

if isinstance(item, builtins.dict):

role = item.get("role", item.get("type", "item"))

messages.append(f"{role}: {summarize_content(item.get('content', item))}")

else:

messages.append(compact_text(item, 120))

suffix = f"; +{len(input_value) - 4} more" if len(input_value) > 4 else ""

return f"{len(input_value)} item(s): " + "; ".join(messages) + suffix

return compact_text(input_value, 260)

def summarize_text_format(text_config: Any) -> str:

if not isinstance(text_config, builtins.dict):

return compact_text(text_config)

fmt = text_config.get("format")

if isinstance(fmt, builtins.dict):

fmt_type = fmt.get("type")

if fmt_type == "json_schema":

schema = fmt.get("schema") or {}

required = schema.get("required") or []

return f"json_schema: {fmt.get('name')} strict={fmt.get('strict')} required={len(required)} fields"

if fmt_type:

return builtins.str(fmt_type)

return compact_text(text_config)

def request_summary_rows(payload: dict[str, Any]) -> list[dict[str, str]]:

rows: list[dict[str, str]] = []

ordered_keys = [

"model", "max_output_tokens", "store", "background", "service_tier", "previous_response_id",

"parallel_tool_calls", "prompt_cache_key", "prompt_cache_retention",

]

for key in ordered_keys:

if key in payload:

rows.append({"field": key, "value": compact_text(payload[key], 180)})

if "reasoning" in payload:

rows.append({"field": "reasoning", "value": compact_text(payload["reasoning"], 180)})

if "text" in payload:

rows.append({"field": "text format", "value": summarize_text_format(payload["text"])})

if "include" in payload:

rows.append({"field": "include", "value": compact_text(payload["include"], 180)})

if "tools" in payload:

tool_names = [tool.get("name", tool.get("type", "tool")) for tool in payload.get("tools", [])]

rows.append({"field": "tools", "value": ", ".join(tool_names)})

if "tool_choice" in payload:

rows.append({"field": "tool_choice", "value": compact_text(payload["tool_choice"], 180)})

if "input" in payload:

rows.append({"field": "input", "value": summarize_input(payload["input"])})

return rows

def print_request_shape(payload: dict[str, Any]) -> None:

rows = request_summary_rows(redact_payload(payload))

print_label("Request shape")

display_wrapped_table(pd.DataFrame(rows), max_col_width_px=520)

def print_response_summary(response_or_summary: Any) -> None:

summary = response_or_summary if isinstance(response_or_summary, builtins.dict) and "output" not in response_or_summary else summarize_response(response_or_summary)

preferred = [

"id", "model", "status", "output_item_types", "input_tokens", "cached_input_tokens",

"output_tokens", "total_tokens", "reasoning_output_tokens", "service_tier",

]

rows = [{"field": key, "value": compact_text(summary.get(key), 220)} for key in preferred if key in summary]

print_label("Response summary")

display_wrapped_table(pd.DataFrame(rows), max_col_width_px=420)

def print_key_takeaway(text: str) -> None:

display(HTML(f"<div style=\"border-left:4px solid #1f6feb; padding:6px 10px; background:#f6f8fa; margin:8px 0;\"><strong>Key takeaway:</strong> {html.escape(text)}</div>"))

def redact_payload(payload: dict[str, Any]) -> dict[str, Any]:

def redact(value: Any) -> Any:

if isinstance(value, builtins.dict):

return