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('', '") display(HTML(f""" .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; }} {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) str: rendered = " ".join(builtins.str(text).split()) if len(rendered) None: if not condition: raise ValueError(message) def warn_or_raise(condition: bool, message: str) -> bool: if condition: return True display(HTML(f"**Warning:** {html.escape(message)} ")) 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""" {safe_label} {safe_text} """)) 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"{html.escape(label)} ")) 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', '')}") 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"**Key takeaway:** {html.escape(text)} ")) def redact_payload(payload: dict[str, Any]) -> dict[str, Any]: def redact(value: Any) -> Any: if isinstance(value, builtins.dict): return

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

背景や根拠まで確認しますか？

調べる

選ぶ

サイト