在 Amazon Bedrock 上运行 OpenAI 模型（58分钟阅读）

# 在 Amazon Bedrock 上开始使用 OpenAI 模型 来源: https://developers.openai.com/cookbook/examples/partners/aws/openai_models_with_amazon_bedrock Amazon Bedrock 上的 OpenAI 模型提供了兼容 OpenAI 的 Responses API 接口，适用于需要文本生成、结构化输出、应用程序工具、直接文件输入、响应状态、提示缓存和后台处理的生产工作流。本指南通过构建一个虚构零售商 **BrightCart** 的客服助手工作流来保持实操性，该工作流处理延迟和损坏订单的更换请求。你将使用 OpenAI Python SDK 进行常规应用调用，并在需要检查确切请求体时，使用一个轻量的原始 HTTPS 辅助函数。流程从设置和最小的预检开始，然后逐步涉及响应生命周期、模型控制、结构化 JSON、应用程序工具、文件输入、状态管理、缓存、后台处理、上下文压缩、操作检查及清理。你将学会如何： 1. 使用 Bedrock 特定的环境变量配置 Bedrock 托管的 OpenAI 模型。 2. 验证 Responses 端点，并检查响应模式、使用元数据和标准化错误。 3. 使用原始 HTTPS 和 OpenAI SDK 发送文本请求。 4. 生成符合 Schema 约束的 JSON 和更轻量的 JSON 模式交接。 5. 调用应用程序管理的函数工具、并行工具和自定义文本工具。 6. 发送直接的 PDF 输入，继续有状态和无状态的对话，并携带加密的推理上下文。 7. 使用提示缓存、后台模式、压缩、操作冒烟检查以及存储响应的清理。 **前提条件**：一个用于 Amazon Bedrock 上 OpenAI 模型的 Bearer Token，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` 模型。 ```bash 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，`requests` 用于对 Responses 端点的原始 HTTPS 调用，`pandas` 和 IPython 显示辅助工具则用于在 Cookbook 渲染器中保持请求和响应摘要的可读性。检查单元格输出，确认包已安装或已存在。 ```bash %pip install -U "openai>=2.28.0" requests pandas ipython --quiet print("Dependencies installed or already available: openai, requests, pandas, ipython") ``` ``` [1m[[0m[34;49mnotice[0m[1;39;49m][0m[39;49m A new release of pip is available: [0m[31;49m24.0[0m[39;49m -> [0m[32;49m26.1.1[0m [1m[[0m[34;49mnotice[0m[1;39;49m][0m[39;49m To update, run: [0m[32;49mpip install --upgrade pip[0m Note: you may need to restart the kernel to use updated packages. Dependencies installed or already available: openai, requests, pandas, ipython ``` ### 1.2 导入库和默认值 导入笔记本中使用的标准库、SDK、HTTP 客户端和显示工具。此单元格还设置了当环境变量未设置时使用的默认 Bedrock 区域和模型。检查打印的默认值，确认笔记本将从 `us-west-2` 和 `openai.gpt-5.4` 启动，除非你覆盖它们。 ```python 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', '<table style="white-space: nowrap; table-layout: auto; width: auto;"') wrapped_html = f'<div style="max-width: {max_col_width_px}px; overflow-x: auto;"><div style="display: flex; flex-direction: column; max-width: 100%;">{table_html}</div></div>' display(HTML(wrapped_html)) def print_label(label: str) -> None: display(HTML(f"### {html.escape(label)}")) def print_labeled_text(label: str, body: str, max_width_px: int = 520) -> None: safe_text = html.escape(body) safe_label = html.escape(label) display(HTML(( f'<div style="max-width: {max_width_px}px; overflow-x: auto;">' f'<div><strong>{safe_label}:</strong></div>' f'<div style="margin-top: 0.15em; white-space: pre-wrap; word-break: break-word;">{safe_text}</div>' f'</div>' ))) def print_labeled_json(label: str, data: Any, max_width_px: int = 520) -> None: if isinstance(data, (builtins.dict, builtins.list)): body = json.dumps(data, indent=2, default=builtins.str) else: body = builtins.str(data) print_labeled_text(label, body, max_width_px) # ... (continues with other helper functions) print("Helpers ready.") ``` ``` Helpers ready. ``` ### 1.6 验证端点 第一个实况调用故意做得很小。它发送一个最小的 Responses 请求，设置 `store=false` 和一条简短的文本指令，这样你可以在运行更丰富的示例之前捕捉到设置问题。