在Amazon Bedrock上开始使用OpenAI模型：完整指南
本教程面向有编程基础的开发者，通过虚构零售商BrightCart的客服助手场景，系统讲解如何在Amazon Bedrock上使用OpenAI模型。内容涵盖环境配置、原始HTTPS与SDK调用、响应生命周期、结构化输出、函数调用、文件输入、状态管理、提示缓存、后台任务、上下文压缩及运维检查。教程强调工程实践，提供可复现代码、常见错误排查与生产优化建议，帮助开发者快速构建可靠、高效的AI应用。
技术背景
随着生成式AI的普及，企业需要在云平台上安全、合规地使用大模型。Amazon Bedrock作为托管服务，提供对多种基础模型的统一访问，包括OpenAI模型。通过Bedrock，开发者可以利用OpenAI的Responses API，同时享受AWS的生态集成、安全性和可扩展性。本教程解决的核心问题是：如何在Bedrock上配置和调用OpenAI模型，并处理生产环境中的关键工程挑战，如结构化输出、工具调用、状态管理和性能优化。适用场景包括客服助手、内容生成、数据分析等。主要限制包括：模型可用性因区域而异，需使用Bearer令牌认证，且部分高级功能（如提示缓存）可能受模型版本影响。
核心原理
本教程的核心原理是OpenAI Responses API与Amazon Bedrock的集成。Responses API提供统一的接口，支持文本生成、结构化输出、工具调用、文件输入、状态管理和后台任务。其数据流为：客户端发送请求（包含模型ID、输入、参数）到Bedrock端点，Bedrock将请求路由到指定的OpenAI模型，模型处理后返回响应对象，包含输出文本、状态、令牌使用情况和元数据。关键设计决策包括：使用环境变量管理配置，支持原始HTTPS和SDK两种调用方式，通过辅助函数封装重复逻辑，以及提供错误规范化、重试机制和响应清理。
架构分析
系统架构分为三层：客户端层（OpenAI SDK或原始HTTPS）、服务层（Amazon Bedrock端点）和模型层（OpenAI模型）。客户端层通过环境变量配置认证和端点，使用辅助函数统一请求构建、响应解析和错误处理。服务层负责认证、路由和负载均衡。模型层执行推理并返回结构化响应。设计思想强调模块化和可测试性：辅助函数（如create_response、output_text）隔离了API细节，便于单元测试；配置通过环境变量注入，支持多环境部署；错误处理统一规范化，便于日志和监控。
完整实战教程
OpenAI模型在Amazon Bedrock上提供了兼容OpenAI的Responses API接口，适用于需要文本生成、结构化输出、应用工具、直接文件输入、响应状态、提示缓存和后台工作的生产工作流。本教程通过为虚构零售商BrightCart构建一个客服助手工作流来具体演示，该零售商处理延迟和损坏订单的更换请求。
你将使用OpenAI Python SDK进行常规应用调用，并在需要检查确切请求体时使用一个简单的原始HTTPS辅助函数。流程从设置和最小预检开始，然后逐步涵盖响应生命周期、模型控制、结构化JSON、应用工具、文件输入、状态管理、缓存、后台处理、上下文压缩、运维检查和清理。
你将学习如何：
- 使用Bedrock特定的环境变量配置Bedrock托管的OpenAI模型。
- 验证Responses端点，并检查响应模式、使用元数据和规范化错误。
- 使用原始HTTPS和OpenAI SDK发送文本请求。
- 生成受模式约束的JSON和更轻量的JSON模式交接。
- 调用应用管理的函数工具、并行工具和自定义文本工具。
- 发送直接PDF输入，继续有状态和无状态对话，并携带加密的推理上下文。
- 使用提示缓存、后台模式、压缩、运维冒烟检查和存储响应清理。
先决条件：用于Amazon Bedrock上OpenAI模型的Bearer令牌、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 |
1. 配置 Amazon Bedrock
本节准备笔记本运行时。它安装小型 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 中读取。如果缺失，设置单元格会以密码样式提示输入，并且不会打印该 token。
1.1 安装依赖
安装笔记本所使用的包。OpenAI SDK 用于应用程序示例，requests 用于对 Responses 端点的原始 HTTPS 调用，pandas 和 IPython 显示辅助工具用于在 Cookbook 渲染器中保持请求和响应摘要的可读性。检查单元格输出仅用于确认包已安装或已存在。
%pip install -U "openai>=2.28.0" requests pandas ipython --quiet
print("Dependencies installed or already available: openai, requests, pandas, ipython")
1.2 导入库和默认设置
导入本笔记本中使用的标准库、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)
1.3 配置 Bedrock 凭证和客户端
从环境中读取 Bedrock 配置并构建客户端。BEDROCK_BASE_URL 被规范化一次，原始的 requests.Session 在其请求头中携带 bearer 令牌，而 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)
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 and later use 24h extended prompt caching; earlier GPT-5 models can use 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)
1.5 辅助函数设置
定义工作流中共享的辅助函数。这些辅助函数用于渲染请求形状、规范化 API 错误、发送原始 HTTPS 请求、包装 SDK 调用并支持可选重试、提取 output_text、汇总 token 使用情况、跟踪存储的响应 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[Display truncated for readability. Inspect the Python variable for the full value.]"
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 {
key: ("<inline file data redacted for display>" if key == "file_data" else redact(item))
for key, item in value.items()
}
if isinstance(value, builtins.list):
return [redact(item) for item in value]
return value
return json.loads(json.dumps(redact(payload), default=builtins.str))
def compact_detail(detail: Any) -> str:
if isinstance(detail, (builtins.dict, builtins.list)):
return compact_text(json.dumps(detail, default=builtins.str), 500)
return compact_text(detail, 500)
def record_check(name: str, status: str, detail: Any = "") -> None:
RESULTS_SUMMARY.append({"name": name, "status": status, "detail": compact_detail(detail)})
def record_response(example: str, response_type: str, content: Any, limit: int = 900) -> None:
if isinstance(content, pd.DataFrame):
rendered = content.to_json(orient="records", indent=2)
elif isinstance(content, (builtins.dict, builtins.list)):
rendered = json.dumps(content, indent=2, default=builtins.str)
else:
rendered = builtins.str(content)
rendered = rendered.strip()
if len(rendered) > limit:
rendered = rendered[:limit].rstrip() + chr(10) + "..."
EXAMPLE_RESPONSES.append({
"example": example,
"response_type": response_type,
"response": rendered,
})
def print_response_gallery() -> pd.DataFrame:
gallery = pd.DataFrame(EXAMPLE_RESPONSES)
if gallery.empty:
gallery = pd.DataFrame(columns=["example", "response_type", "response"])
display_wrapped_table(gallery, max_col_width_px=620)
return gallery
def normalize_error(response: requests.Response, body: Any) -> dict[str, Any]:
return {
"exception_class": "HTTPError",
"status_code": response.status_code,
"retryable": response.status_code in TRANSIENT_STATUS_CODES,
"request_id": response.headers.get("x-request-id"),
"body": body,
}
def describe_api_error(exc: Exception) -> dict[str, Any]:
try:
parsed = json.loads(builtins.str(exc))
if isinstance(parsed, builtins.dict) and "status_code" in parsed:
return {
"exception_class": type(exc).__name__,
"status_code": parsed.get("status_code"),
"retryable": parsed.get("retryable"),
"request_id": parsed.get("request_id"),
"message": compact_text(parsed.get("body", parsed), 500),
}
except Exception:
pass
status_code = getattr(exc, "status_code", None)
response = getattr(exc, "response", None)
request_id = None
if response is not None:
headers = getattr(response, "headers", {})
request_id = headers.get("x-request-id") if hasattr(headers, "get") else None
return {
"exception_class": type(exc).__name__,
"status_code": status_code,
"retryable": status_code in TRANSIENT_STATUS_CODES,
"request_id": request_id,
"message": builtins.str(exc)[:500],
}
def request_json(method: str, path: str, *, payload: dict[str, Any] | None = None) -> dict[str, Any]:
response = http.request(
method,
endpoint(path),
json=payload,
timeout=API_TIMEOUT_SECONDS,
)
try:
body = response.json() if response.text else {}
except json.JSONDecodeError:
body = {"raw_text": response.text}
if response.status_code >= 400:
raise RuntimeError(json.dumps(normalize_error(response, body), indent=2, default=builtins.str))
return body
def to_dict(value: Any) -> Any:
if hasattr(value, "model_dump"):
return value.model_dump(mode="json")
if isinstance(value, builtins.list):
return [to_dict(item) for item in value]
if isinstance(value, builtins.dict):
return {key: to_dict(item) for key, item in value.items()}
return value
def output_text(response: Any) -> str:
direct = getattr(response, "output_text", None)
if direct:
return direct
data = to_dict(response)
pieces: list[str] = []
for item in data.get("output", []) or []:
for content in item.get("content", []) or []:
if content.get("type") == "output_text":
pieces.append(content.get("text", ""))
return "".join(pieces)
def response_items(response: Any) -> list[dict[str, Any]]:
data = to_dict(response)
return builtins.list(data.get("output", []) or [])
def first_output_item(response: Any, item_type: str) -> dict[str, Any] | None:
for item in response_items(response):
if item.get("type") == item_type:
return item
return None
def summarize_response(response: Any) -> dict[str, Any]:
data = to_dict(response)
usage = data.get("usage") or {}
input_details = usage.get("input_tokens_details") or {}
output_details = usage.get("output_tokens_details") or {}
return {
"id": data.get("id"),
"model": data.get("model"),
"status": data.get("status"),
"output_item_types": [item.get("type") for item in data.get("output", []) or []],
"input_tokens": usage.get("input_tokens"),
"output_tokens": usage.get("output_tokens"),
"total_tokens": usage.get("total_tokens"),
"cached_input_tokens": input_details.get("cached_tokens"),
"reasoning_output_tokens": output_details.get("reasoning_tokens"),
"service_tier": data.get("service_tier"),
}
def call_with_retries(label: str, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
kwargs.setdefault("timeout", API_TIMEOUT_SECONDS)
last_exc: Exception | None = None
for attempt in range(1, MAX_RETRIES + 2):
try:
return func(*args, **kwargs)
except Exception as exc:
last_exc = exc
error = describe_api_error(exc)
should_retry = bool(error["retryable"] and attempt <= MAX_RETRIES)
if not should_retry:
raise
time.sleep(min(2 ** (attempt - 1), 8))
raise RuntimeError(f"{label} failed after retries") from last_exc
def create_response(**kwargs: Any) -> Any:
kwargs.setdefault("model", MODEL_ID)
return call_with_retries("responses.create", client.responses.create, **kwargs)
def retrieve_response(response_id: str) -> Any:
return call_with_retries("responses.retrieve", client.responses.retrieve, response_id)
def delete_response(response_id: str) -> Any:
return call_with_retries("responses.delete", client.responses.delete, response_id)
def remember_stored_response(response: Any) -> None:
response_id = getattr(response, "id", None) or to_dict(response).get("id")
if response_id:
STORED_RESPONSE_IDS.append(response_id)
def handle_example_error(features: str | list[str], exc: Exception) -> None:
feature_list = [features] if isinstance(features, builtins.str) else features
error = describe_api_error(exc)
for feature in feature_list:
record_check(feature, "warn", error)
print_labeled_text("Result", "This live call did not complete in this environment.")
print_labeled_json("Response summary", error)
def build_curl_command(payload: dict[str, Any]) -> str:
body = json.dumps(payload)
return " ".join([
"curl", "-sS", shlex.quote(RESPONSES_URL),
"-H", shlex.quote("Content-Type: application/json"),
"-H", shlex.quote("Authorization: Bearer $AWS_BEARER_TOKEN_BEDROCK"),
"-d", shlex.quote(body),
])
def run_raw_http_request(payload: dict[str, Any]) -> dict[str, Any]:
return request_json("POST", "/responses", payload=payload)
print("Helpers ready.")
1.6 验证端点
第一次实际调用故意做得非常小。它发送一个最小的 Responses 请求，设置 store=false 和一条简短的文本指令，以便在运行更丰富的示例之前发现设置问题。检查请求结构、返回的文本、状态、模型、输出项类型和令牌使用情况。
from __future__ import annotations
preflight_payload = {
"model": MODEL_ID,
"input": "Reply with exactly: ok",
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(preflight_payload)
try:
preflight_response = create_response(**preflight_payload)
require(output_text(preflight_response).strip(), "Preflight response did not return output text.")
record_check("Endpoint shape", "pass", RESPONSES_URL)
model_selection_detail = f"{len(AVAILABLE_MODELS)} models discovered" if AVAILABLE_MODELS else "Using configured model; model-list metadata is not required for requests."
record_check("Model selection", "pass", model_selection_detail)
preflight_text = output_text(preflight_response).strip()
record_response("Endpoint verification", "text", preflight_text)
print_labeled_text("Result", preflight_text)
print_response_summary(preflight_response)
print_key_takeaway('A tiny response confirms that the endpoint, key, model, and request shape are working.')
except Exception as exc:
handle_example_error(["Endpoint shape", "Model selection"], exc)
1.7 规范化 API 错误
生产环境集成需要一致的状态码、重试决策、请求 ID 和响应体的错误日志。此单元格记录了笔记本使用的规范化错误格式，而不会故意发出失败的请求。后续单元格在实时调用失败或返回非 2xx 状态时使用相同的格式。
from __future__ import annotations
error_taxonomy_example = {
"normalized_fields": ["exception_class", "status_code", "retryable", "request_id", "message"],
"retryable_status_codes": sorted(TRANSIENT_STATUS_CODES),
"notes": "call_with_retries(...) uses this taxonomy for transient retry handling.",
}
record_check("Error handling", "pass", error_taxonomy_example)
print_json(error_taxonomy_example)
2. 发出你的第一个响应请求
本节从两个角度展示响应请求的接口。首先，你检查并运行一个原始的 HTTPS 请求，以便端点、请求头和 JSON 主体可见。然后，你使用 OpenAI SDK 实现相同类型的应用工作流，一旦配置正确，这应该是大多数生产代码应优先选择的路径。
2.1 检查原始 HTTPS 请求形态
为 BrightCart 支持助手的回复构建一个最小的 Responses 负载，并渲染一个可复制粘贴的 curl 命令。该命令引用 $AWS_BEARER_TOKEN_BEDROCK 而不是嵌入令牌，并且笔记本不会执行将 bearer 令牌放入进程参数中的 shell 命令。检查 model、input、max_output_tokens 和 store 字段。
from __future__ import annotations
basic_curl_payload = {
"model": MODEL_ID,
"input": "BrightCart customer Maya asks why replacement order ORDER-8831 is delayed. Write two labeled plain-text lines for the support agent. Do not use leading hyphens or bold text.",
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(basic_curl_payload)
print_labeled_text("Result", build_curl_command(basic_curl_payload))
print_key_takeaway('The curl command shows the raw HTTPS shape behind the SDK call.')
2.2 发送原始 HTTPS 请求
通过原始 HTTPS 辅助函数发送相同的请求。此单元格演示了底层 POST /responses 路径，并通过遍历输出项从响应体中提取文本。检查返回的响应 ID、模型、状态和文本输出，以了解应用程序接收到的模式。
from __future__ import annotations
print_request_shape(basic_curl_payload)
try:
basic_http_response = run_raw_http_request(basic_curl_payload)
record_check("Text generation", "pass", basic_http_response.get("id"))
response_text_parts = []
for item in basic_http_response.get("output", []):
for content in item.get("content", []):
if content.get("type") == "output_text":
response_text_parts.append(content.get("text", ""))
raw_http_output = "".join(response_text_parts).strip()
record_response("First raw HTTPS request", "text", raw_http_output)
print_labeled_text("Result", raw_http_output)
print_labeled_json("Response summary", {
"id": basic_http_response.get("id"),
"model": basic_http_response.get("model"),
"status": basic_http_response.get("status"),
})
print_key_takeaway("The response body contains message output that application code can extract as text.")
except Exception as exc:
handle_example_error("Text generation", exc)
2.3 使用 OpenAI SDK
OpenAI SDK 在显式传入 Bedrock 的 bearer token 和 base URL 时，可以调用兼容 OpenAI 的 API。此单元格通过 client.responses.create 发送文本生成请求，将 reasoning.effort 设置为 low，并打印简洁的响应摘要。检查输出文本、token 计数、输出项类型以及端点返回的任何推理 token 元数据。
官方文档：推理模型 描述了如何在 Responses API 中使用推理努力。
from __future__ import annotations
sdk_text_payload = {
"model": MODEL_ID,
"input": "Write a three-sentence overview for a developer building a BrightCart support assistant with the Responses API.",
"reasoning": {"effort": "low"},
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(sdk_text_payload)
try:
text_response = create_response(**sdk_text_payload)
sdk_text = output_text(text_response).strip()
require(sdk_text, "SDK text response did not return output text.")
record_check("Text generation", "pass", summarize_response(text_response))
record_check("Reasoning effort", "pass", summarize_response(text_response))
record_response("SDK text generation", "text", sdk_text)
print_labeled_text("Result", sdk_text)
print_response_summary(text_response)
print_key_takeaway('The SDK returns a response object with text, status, token usage, and output item metadata.')
except Exception as exc:
handle_example_error(["Text generation", "Reasoning effort"], exc)
2.4 创建并检索响应
Responses API 可以存储响应，并稍后通过 ID 检索。这种模式对于审计跟踪、调试以及引用先前上下文的后续轮次非常有用。此单元格创建一个存储的响应，跟踪用于清理的 ID，检索它，并比较检索到的文本和使用情况元数据。
from __future__ import annotations
lifecycle_payload = {
"model": MODEL_ID,
"input": (
"BrightCart is building a support assistant for delayed replacement orders. "
"Return exactly three labeled plain-text lines: goal, data needed, and human-review rule. Do not use leading hyphens or bold text."
),
"max_output_tokens": 1024,
"store": True,
}
print_request_shape(lifecycle_payload)
try:
lifecycle_response = create_response(**lifecycle_payload)
remember_stored_response(lifecycle_response)
retrieved_response = retrieve_response(lifecycle_response.id)
retrieved_summary = summarize_response(retrieved_response)
retrieved_text = output_text(retrieved_response).strip()
require(retrieved_text, "Retrieved response did not contain text output.")
lifecycle_status = "pass" if retrieved_summary.get("status") in {None, "completed"} else "warn"
record_check("Responses lifecycle", lifecycle_status, retrieved_response.id)
record_check("Response schema", "pass", retrieved_summary)
record_check("Usage metadata", "pass" if retrieved_summary.get("total_tokens") is not None else "warn", retrieved_summary)
record_response("Create and retrieve response", "text", retrieved_text)
print_labeled_text("Result", retrieved_text)
print_labeled_json("Created response summary", summarize_response(lifecycle_response))
print_response_summary(retrieved_summary)
print_key_takeaway('store=True lets an application retrieve the response later by ID with usage metadata intact.')
except Exception as exc:
handle_example_error(["Responses lifecycle", "Response schema", "Usage metadata"], exc)
2.5 添加推理力度、服务层级和提示缓存参数
模型控制与正常输入一起传递。此请求结合了 reasoning.effort、service_tier、prompt_cache_key 和 prompt_cache_retention，以便您了解操作控制和提示缓存元数据如何与普通文本输出出现在同一响应模式中。检查 service_tier、cached_input_tokens、推理令牌元数据和总令牌使用情况。
注意：此笔记本使用 PROMPT_CACHE_RETENTION 而不是硬编码 prompt_cache_retention。对于 openai.gpt-5.4，该值为 in_memory；对于 openai.gpt-5.5 及更高版本，该值为 24h，因为这些模型需要扩展提示缓存。
from __future__ import annotations
control_payload = {
"model": MODEL_ID,
"input": (
"For the BrightCart support assistant, explain prompt caching in exactly two labeled plain-text lines: "
"one latency benefit and one consistency benefit."
),
"reasoning": {"effort": "low"},
"prompt_cache_key": "brightcart-support-policy-guide",
"prompt_cache_retention": PROMPT_CACHE_RETENTION,
"service_tier": "auto",
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(control_payload)
try:
control_response = create_response(**control_payload)
control_summary = summarize_response(control_response)
control_text = output_text(control_response).strip()
require(control_text, "Control response did not return text.")
status = "pass" if control_summary.get("status") in {None, "completed"} else "warn"
record_check("Prompt caching", "pass" if control_summary.get("cached_input_tokens") is not None else "warn", control_summary)
record_check("Service tier", "pass" if control_summary.get("service_tier") is not None else "warn", control_summary)
record_check("Reasoning effort", status, control_summary)
record_response("Service tier and prompt cache request", "text", control_text)
print_labeled_text("Result", control_text)
print_response_summary(control_summary)
print_key_takeaway('Model controls travel with the same request as normal input, while returned metadata can vary by endpoint.')
except Exception as exc:
handle_example_error(["Prompt caching", "Service tier", "Reasoning effort"], exc)
3. 生成结构化 JSON
结构化 JSON 将模型输出转换为应用程序代码可以解析、验证和路由的数据。本节比较严格的模式约束输出与更轻量的 JSON 模式。当您的应用程序需要契约时，请使用结构化输出；当有效的 JSON 就足够且确切模式可以保持灵活时，请使用 JSON 模式。
3.1 定义结构化输出模式
定义下一个实时请求所使用的支持工单模式。该模式列出了应用程序期望的确切字段，包括类别、优先级、情感、摘要、所需操作和升级状态。检查请求结构，了解 text.format.type="json_schema"、strict=true 以及 JSON Schema 是如何附加到普通 Responses 请求中的。
from __future__ import annotations
support_triage_schema = {
"type": "object",
"properties": {
"ticket_id": {"type": "string"},
"category": {"type": "string", "enum": ["delivery_delay", "return_exchange", "damaged_item", "billing", "account"]},
"priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]},
"customer_sentiment": {"type": "string"},
"summary": {"type": "string"},
"required_actions": {"type": "array", "items": {"type": "string"}, "minItems": 2},
"escalation_needed": {"type": "boolean"},
},
"required": ["ticket_id", "category", "priority", "customer_sentiment", "summary", "required_actions", "escalation_needed"],
"additionalProperties": False,
}
structured_payload = {
"model": MODEL_ID,
"input": (
"Support ticket TICKET-7429: Maya Chen says ORDER-8831 is a replacement for a damaged standing desk. "
"The replacement is two days late, the carrier scan has not moved, and she needs the desk before Monday. "
"She asks for a supervisor callback and refund options. Triage this ticket for the next support agent."
),
"text": {"format": {"type": "json_schema", "name": "support_ticket_triage", "strict": True, "schema": support_triage_schema}},
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(structured_payload)
print_key_takeaway('The schema is part of the request and defines the fields the next cell validates.')
3.2 验证受模式约束的输出
使用上一个单元格中的模式调用模型，将返回的文本解析为 JSON，并在 Python 中验证重要字段。API 请求要求遵循模式，而应用端验证仍会检查返回的对象是否适合下游路由。检查解析后的对象和响应摘要。
from __future__ import annotations
def validate_support_triage(payload: dict[str, Any]) -> dict[str, Any]:
require("ticket_id" in payload, "Missing key: ticket_id")
require(payload.get("ticket_id") == "TICKET-7429", "Ticket ID did not match expected value.")
require("required_actions" in payload, "Missing key: required_actions")
require(isinstance(payload.get("required_actions"), builtins.list), "required_actions must be a list.")
require(len(payload["required_actions"]) >= 2, "required_actions should contain at least two actions.")
return payload
print_request_shape(structured_payload)
try:
structured_response = create_response(**structured_payload)
raw_structured_text = output_text(structured_response).strip()
try:
structured_payload_result = validate_support_triage(json.loads(raw_structured_text))
record_check("Structured Outputs", "pass", structured_payload_result)
record_response("Structured ticket triage", "json", structured_payload_result)
print_labeled_json("Result", structured_payload_result)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON: {e}")
except Exception as parse_exc:
record_check("Structured Outputs", "warn", {"message": "Response did not match the expected schema shape.", "text_sample": raw_structured_text[:600], "error": builtins.str(parse_exc)})
print_labeled_text("Result", "The request completed, but the returned text did not match the expected schema shape.")
print_wrapped(raw_structured_text[:1200])
print_response_summary(structured_response)
print_key_takeaway('Schema-constrained output gives application code a predictable JSON object to parse and validate.')
except Exception as exc:
handle_example_error("Structured Outputs", exc)
3.3 使用 JSON 模式
JSON 模式要求模型返回一个有效的 JSON 对象，而不强制严格的模式。这对于轻量级的交接非常有用，你仍然希望得到可解析的输出，但可以容忍较宽松的契约。此单元格请求一个支持交接对象，解析它，并检查预期的键。
from __future__ import annotations
json_mode_payload = {
"model": MODEL_ID,
"input": (
"Return JSON for a support chat handoff with keys customer_name, order_id, issue_summary, next_step, "
"and metrics_to_watch. Context: Maya Chen asks about delayed replacement order ORDER-8831; the carrier scan is stale. "
"metrics_to_watch should be an array."
),
"text": {"format": {"type": "json_object"}},
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(json_mode_payload)
try:
json_mode_response = create_response(**json_mode_payload)
payload = json.loads(output_text(json_mode_response).strip())
require({"customer_name", "order_id", "issue_summary", "next_step", "metrics_to_watch"}.issubset(payload), "JSON mode response missed required keys.")
record_check("JSON mode", "pass", payload)
record_response("JSON support handoff", "json", payload)
print_labeled_json("Result", payload)
print_response_summary(json_mode_response)
print_key_takeaway('JSON mode is useful when valid JSON is enough and a strict schema is not required.')
except Exception as exc:
handle_example_error("JSON mode", exc)
3.4 通过推理努力控制冗长程度
冗长控制有助于调整生成散文的形态，而推理努力控制模型在回答前花费多少推理工作。笔记本在上述 SDK 和模型控制单元格中演示了 reasoning.effort；本单元格通过发送同一政策主题的简洁和详细版本，聚焦于 text.verbosity。检查并排的文本和令牌摘要，以比较风格和用法。
from __future__ import annotations
verbosity_prompt = "Explain BrightCart's delayed-replacement policy to a new support agent."
compact_payload = {
"model": MODEL_ID,
"input": verbosity_prompt + " Reply in one sentence under 35 words.",
"text": {"verbosity": "low"},
"max_output_tokens": 1024,
"store": False,
}
detailed_payload = {
"model": MODEL_ID,
"input": verbosity_prompt + " Reply in exactly three numbered plain-text lines, each under 18 words. Do not use leading hyphens or bold text.",
"text": {"verbosity": "high"},
"max_output_tokens": 1024,
"store": False,
}
print_labeled_json("Request shape", {
"compact": redact_payload(compact_payload),
"detailed": redact_payload(detailed_payload),
})
try:
compact_response = create_response(**compact_payload)
detailed_response = create_response(**detailed_payload)
compact_guidance_text = output_text(compact_response).strip()
detailed_guidance_text = output_text(detailed_response).strip()
require(compact_guidance_text and detailed_guidance_text, "Verbosity responses did not return text.")
compact_summary = summarize_response(compact_response)
detailed_summary = summarize_response(detailed_response)
status = "pass" if compact_summary.get("status") in {None, "completed"} and detailed_summary.get("status") in {None, "completed"} else "warn"
record_check("Verbosity", status, {"compact_chars": len(compact_guidance_text), "detailed_chars": len(detailed_guidance_text)})
record_response("Compact policy guidance", "text", compact_guidance_text)
record_response("Detailed policy guidance", "text", detailed_guidance_text)
print_labeled_text("Result: compact guidance", compact_guidance_text)
print_labeled_text("Result: detailed guidance", detailed_guidance_text)
verbosity_summary = pd.DataFrame([
{"request": "compact", **compact_summary},
{"request": "detailed", **detailed_summary},
])
print_label("Response summary")
display_wrapped_table(verbosity_summary, max_col_width_px=420)
print_key_takeaway('Verbosity controls tune the answer style while the prompt still bounds the output.')
except Exception as exc:
handle_example_error("Verbosity", exc)
4. 添加应用管理的工具
函数调用让模型可以向你的应用请求数据或操作，但你的代码仍然负责执行工具并返回结果。本节定义本地 BrightCart 工具，然后逐步讲解单个函数调用、多个独立调用以及自定义文本工具。示例保持工具输出确定性，以便轻松检查请求循环。
4.1 定义本地工具模式和函数
定义用于订单状态和客户资料查询的本地示例工具。工具模式描述了模型在决定调用什么时可以使用的名称、描述、参数形状、必填字段和严格性。Python 函数代表应用程序系统，如订单管理、CRM 或策略服务。
from __future__ import annotations
function_tools = [
{
"type": "function",
"name": "get_order_status",
"description": "Look up a sample BrightCart order status.",
"parameters": {
"type": "object",
"properties": {"order_id": {"type": "string", "description": "An order ID such as ORDER-8831."}},
"required": ["order_id"],
"additionalProperties": False,
},
"strict": True,
},
{
"type": "function",
"name": "get_customer_profile",
"description": "Look up sample customer context for a BrightCart support interaction.",
"parameters": {
"type": "object",
"properties": {"customer_id": {"type": "string", "description": "A customer ID such as CUST-1042."}},
"required": ["customer_id"],
"additionalProperties": False,
},
"strict": True,
},
]
def get_order_status(order_id: str) -> dict[str, Any]:
orders = {
"ORDER-8831": {
"order_id": "ORDER-8831",
"customer_id": "CUST-1042",
"item": "standing desk replacement",
"status": "delayed",
"carrier_scan": "No movement for 36 hours at Denver sort center",
"promised_delivery": (date.today() + timedelta(days=2)).isoformat(),
"recommended_policy": "If delay exceeds 48 hours, offer expedited replacement or 15% concession with agent approval.",
},
"ORDER-2044": {
"order_id": "ORDER-2044",
"customer_id": "CUST-1042",
"item": "ergonomic chair",
"status": "delivered",
"carrier_scan": "Delivered yesterday at front desk",
"promised_delivery": (date.today() - timedelta(days=1)).isoformat(),
"recommended_policy": "Confirm delivery details before opening a replacement request.",
},
}
return orders.get(order_id, {"order_id": order_id, "status": "unknown", "customer_id": None})
def get_customer_profile(customer_id: str) -> dict[str, Any]:
profiles = {
"CUST-1042": {
"customer_id": "CUST-1042",
"name": "Maya Chen",
"loyalty_tier": "Gold",
"region": "California",
"recent_issue": "Damaged standing desk replacement",
"contact_preference": "email with SMS updates for shipping changes",
}
}
return profiles.get(customer_id, {"customer_id": customer_id, "loyalty_tier": "unknown"})
def dispatch_tool_call(call: dict[str, Any]) -> dict[str, Any]:
name = call["name"]
args = json.loads(call["arguments"])
if name == "get_order_status":
output = get_order_status(**args)
elif name == "get_customer_profile":
output = get_customer_profile(**args)
else:
raise ValueError(f"Unsupported tool: {name}")
return {"type": "function_call_output", "call_id": call["call_id"], "output": json.dumps(output)}
print("Sample function tools:")
print_json([tool["name"] for tool in function_tools])
print("\nSample order lookup:")
print_json(get_order_status("ORDER-8831"))
4.2 调用函数工具
此单元格运行基本的函数调用循环。第一个请求为模型提供了一个订单状态工具，并要求其选择参数。应用程序解析返回的 function_call，运行本地 Python 函数，将 function_call_output 项发送回去，并请求最终的基于事实的答案。检查工具参数、本地工具输出、最终模型文本和响应元数据。
from __future__ import annotations
function_input = [{"role": "user", "content": "Use get_order_status for ORDER-8831, then explain the next best action for the support agent in two labeled plain-text lines. Do not use leading hyphens or bold text."}]
order_status_tool = [tool for tool in function_tools if tool["name"] == "get_order_status"]
function_request = {
"model": MODEL_ID,
"input": function_input,
"tools": order_status_tool,
"tool_choice": "required",
"max_output_tokens": 1024,
"store": False,
}
def create_tool_plan_with_auto_fallback(request: dict[str, Any]) -> tuple[Any, str]:
try:
return create_response(**request), builtins.str(request.get("tool_choice"))
except Exception as first_exc:
fallback_request = {**request, "tool_choice": "auto"}
try:
return create_response(**fallback_request), "auto"
except Exception:
raise first_exc
print_request_shape(function_request)
try:
function_plan, tool_choice_used = create_tool_plan_with_auto_fallback(function_request)
function_calls = [item for item in response_items(function_plan) if item.get("type") == "function_call"]
if function_calls:
function_call = function_calls[0]
function_args = json.loads(function_call["arguments"])
require(function_args.get("order_id") == "ORDER-8831", f"Unexpected function arguments: {function_args}")
tool_output = dispatch_tool_call(function_call)
final_response = create_response(
model=MODEL_ID,
input=function_input + response_items(function_plan) + [tool_output],
tools=order_status_tool,
max_output_tokens=1024,
store=False,
)
final_answer = output_text(final_response).strip()
tool_output_payload = json.loads(tool_output["output"])
record_check("Function calling", "pass", {"tool_choice_used": tool_choice_used, "arguments": function_args})
record_response("Order-status tool answer", "text", final_answer)
print_labeled_json("Result: tool arguments", function_args)
print_labeled_json("Result: tool output", tool_output_payload)
print_labeled_text("Result: final model answer", final_answer)
print_response_summary(final_response)
print_key_takeaway('Function calling separates model-selected arguments from application-executed business logic.')
else:
fallback_order = get_order_status("ORDER-8831")
fallback_prompt = (
"The model response did not include a function_call item. Use this application lookup result "
"to answer in two labeled plain-text lines without leading hyphens or bold text: " + json.dumps(fallback_order)
)
final_response = create_response(
model=MODEL_ID,
input=function_input + [{"role": "user", "content": fallback_prompt}],
max_output_tokens=1024,
store=False,
)
final_answer = output_text(final_response).strip()
returned_item_types = [item.get("type") for item in response_items(function_plan)]
record_check("Function calling", "warn", {"tool_choice_used": tool_choice_used, "returned_item_types": returned_item_types})
record_response("Order-status local fallback answer", "text", final_answer)
print_labeled_json("Result: returned output item types", returned_item_types)
print_labeled_json("Result: local tool output", fallback_order)
print_labeled_text("Result: final model answer", final_answer)
print_response_summary(final_response)
print_key_takeaway('The local lookup keeps the function-calling pattern understandable even when the model returns text.')
except Exception as exc:
handle_example_error("Function calling", exc)
4.3 处理多个工具调用
并行工具调用允许模型在单轮中请求多个独立的查询。此单元格允许两次订单状态查询，执行每个本地函数调用，并在要求模型比较当前发货问题之前将两个输出都发送回去。检查返回的订单 ID 和最终答案，以确认响应基于应用程序数据而非模型记忆。
from __future__ import annotations
parallel_input = [{"role": "user", "content": "Use get_order_status for ORDER-8831 and ORDER-2044, then summarize whether Maya has one shipping problem or multiple active shipping problems in two labeled plain-text lines. Do not use leading hyphens or bold text."}]
parallel_request = {
"model": MODEL_ID,
"input": parallel_input,
"tools": function_tools,
"tool_choice": "auto",
"parallel_tool_calls": True,
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(parallel_request)
try:
parallel_plan = create_response(**parallel_request)
parallel_calls = [item for item in response_items(parallel_plan) if item.get("type") == "function_call"]
parallel_outputs = [dispatch_tool_call(call) for call in parallel_calls]
parallel_order_ids = [json.loads(call["arguments"]).get("order_id") for call in parallel_calls if call.get("name") == "get_order_status"]
expected_order_ids = ["ORDER-8831", "ORDER-2044"]
missing_order_ids = [order_id for order_id in expected_order_ids if order_id not in builtins.set(parallel_order_ids)]
if not missing_order_ids:
parallel_final = create_response(
model=MODEL_ID,
input=parallel_input + response_items(parallel_plan) + parallel_outputs,
tools=function_tools,
max_output_tokens=1024,
store=False,
)
parallel_answer = output_text(parallel_final).strip()
record_check("Parallel tool calls", "pass", {"tool_call_count": len(parallel_calls), "order_ids": parallel_order_ids})
record_response("Parallel order lookup answer", "text", parallel_answer)
print_labeled_json("Result: tool calls", {"tool_call_count": len(parallel_calls), "order_ids": parallel_order_ids})
print_labeled_text("Result: final model answer", parallel_answer)
print_response_summary(parallel_final)
print_key_takeaway('Parallel tool calls let the model request multiple lookups, while the application still controls execution.')
else:
fallback_orders = [get_order_status(order_id) for order_id in expected_order_ids]
fallback_prompt = (
"The model did not request every expected order lookup. Use these application lookup results "
"to answer in two labeled plain-text lines without leading hyphens or bold text: " + json.dumps(fallback_orders)
)
parallel_final = create_response(
model=MODEL_ID,
input=parallel_input + [{"role": "user", "content": fallback_prompt}],
max_output_tokens=1024,
store=False,
)
parallel_answer = output_text(parallel_final).strip()
record_check("Parallel tool calls", "warn", {"returned_order_ids": parallel_order_ids, "missing_order_ids": missing_order_ids})
record_response("Parallel order lookup fallback answer", "text", parallel_answer)
print_labeled_json("Result: returned tool calls", {"tool_call_count": len(parallel_calls), "order_ids": parallel_order_ids})
print_labeled_json("Result: local tool outputs", fallback_orders)
print_labeled_text("Result: final model answer", parallel_answer)
print_response_summary(parallel_final)
print_key_takeaway('Local lookup outputs keep the parallel-tool pattern understandable if not every call is returned.')
except Exception as exc:
handle_example_error("Parallel tool calls", exc)
4.4 使用自定义文本工具
自定义工具将自由格式的文本传递给应用程序拥有的逻辑，而不是要求结构化的 JSON 参数对象。此单元格定义了一个支持笔记规范化器，请求自定义工具调用，并在端点返回普通文本而不是自定义调用时包含本地回退。检查输出项类型和规范化后的笔记。
from __future__ import annotations
custom_tools = [
{
"type": "custom",
"name": "normalize_support_note",
"description": "Normalize a freeform support note written by an agent. Input is plain text.",
"format": {"type": "text"},
}
]
def normalize_support_note_text(note: str) -> str:
fields = [part.strip().upper() for part in note.split("|")]
labels = ["ORDER_ID", "CUSTOMER_ID", "ISSUE", "CUSTOMER_REQUEST", "POLICY_OPTION"]
return "\n".join(
f"{label}: {value}"
for label, value in zip(labels, fields)
if value
)
support_note = "order-8831 | cust-1042 | replacement delayed | customer wants supervisor | offer expedited replacement or 15% concession"
custom_input = [{
"role": "user",
"content": (
"Call normalize_support_note with this exact note. Do not answer directly; "
f"send the note to the custom tool: {support_note}"
),
}]
custom_request = {
"model": MODEL_ID,
"input": custom_input,
"tools": custom_tools,
"tool_choice": {"type": "custom", "name": "normalize_support_note"},
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(custom_request)
print_labeled_text("Result: local fallback normalization", normalize_support_note_text(support_note))
try:
custom_plan = create_response(**custom_request)
returned_item_types = [item.get("type") for item in response_items(custom_plan)]
try:
custom_call = first_output_item(custom_plan, "custom_tool_call")
if custom_call is None:
raise LookupError("No custom_tool_call item returned.")
tool_input = custom_call.get("input", "").strip()
normalized_note = normalize_support_note_text(tool_input)
record_check("Custom tools", "pass", {"output_item_types": returned_item_types, "normalized_note": normalized_note})
record_response("Normalized support note", "text", normalized_note)
print_labeled_json("Result: returned output item types", returned_item_types)
print_labeled_text("Result: custom tool input", tool_input)
print_labeled_text("Result: application-owned normalized output", normalized_note)
except LookupError:
fallback_text = output_text(custom_plan).strip() or "No text content was returned."
normalized_note = normalize_support_note_text(support_note)
record_check("Custom tools", "warn", {
"expected": "custom_tool_call item named normalize_support_note",
"actual_output_item_types": returned_item_types,
"meaning": "The model response did not include a custom-tool invocation, so the application fallback normalization is shown for teaching.",
})
record_response("Custom tool text fallback", "text", fallback_text)
record_response("Application-owned normalization fallback", "text", normalized_note)
print_labeled_json("Result: returned output item types", returned_item_types or ["no typed output items returned"])
print_labeled_text("Result: model text response", fallback_text)
print_labeled_text("Result: application-owned normalization", normalized_note)
print_response_summary(custom_plan)
print_key_takeaway('Custom tools are useful when the application owns a freeform parsing or execution step.')
except Exception as exc:
handle_example_error("Custom tools", exc)
5. 发送直接文件输入
直接文件输入与应用程序管理的工具是分开的。文件可以作为 input_file 项与文本指令一起包含在当前 Responses 请求中，这在模型需要在本轮读取文件而无需设置检索索引时非常有用。
5.1 将 PDF 作为 input_file 附加
此单元格在内存中生成一个微型 PDF 转录，将其作为 base64 文件数据附加，并要求从文档中提取精确的 JSON 字段。检查 PDF 预览、预期字段、解析后的响应和用法摘要。
from __future__ import annotations
def make_simple_pdf(lines: list[str]) -> bytes:
def pdf_escape(text: str) -> str:
return text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
stream_lines = ["BT", "/F1 11 Tf", "72 740 Td", "15 TL"]
for idx, line in enumerate(lines):
if idx:
stream_lines.append("T*")
stream_lines.append(f"({pdf_escape(line)}) Tj")
stream_lines.append("ET")
stream = "\n".join(stream_lines).encode("latin-1", "replace")
objects = [
b"<< /Type /Catalog /Pages 2 0 R >>",
b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
b"<< /Length " + builtins.str(len(stream)).encode("ascii") + b" >>\nstream\n" + stream + b"\nendstream",
]
pdf = b"%PDF-1.4\n"
offsets = [0]
for idx, obj in enumerate(objects, start=1):
offsets.append(len(pdf))
pdf += f"{idx} 0 obj\n".encode("ascii") + obj + b"\nendobj\n"
xref_offset = len(pdf)
pdf += f"xref\n0 {len(objects) + 1}\n0000000000 65535 f \n".encode("ascii")
for offset in offsets[1:]:
pdf += f"{offset:010d} 00000 n \n".encode("ascii")
pdf += f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\nstartxref\n{xref_offset}\n%%EOF\n".encode("ascii")
return pdf
file_lines = [
"BrightCart support transcript",
"Ticket: TICKET-7429",
"Customer: Maya Chen",
"Order: ORDER-8831",
"Product: Standing desk replacement",
"Issue: Replacement for a damaged item is delayed and carrier scan has not moved",
"Customer request: Supervisor callback and refund options",
"Policy options: expedited replacement or 15% concession with agent approval after 48-hour delay",
]
file_text = "\n".join(file_lines)
pdf_data = base64.b64encode(make_simple_pdf(file_lines)).decode("utf-8")
expected_direct_file_fields = {
"ticket_id": "TICKET-7429",
"customer": "Maya Chen",
"order_id": "ORDER-8831",
"product": "Standing desk replacement",
}
direct_file_request = {
"model": MODEL_ID,
"input": [
{
"role": "user",
"content": [
{
"type": "input_file",
"filename": "brightcart-support-transcript.pdf",
"file_data": f"data:application/pdf;base64,{pdf_data}",
},
{
"type": "input_text",
"text": (
"Read the attached PDF support transcript and return JSON with keys "
"ticket_id, customer, order_id, product, issue, requested_resolution, and policy_options. "
"Use exact values from the file. Do not return null for fields that are present in the file."
),
},
],
}
],
"text": {"format": {"type": "json_object"}},
"max_output_tokens": 1024,
"store": False,
}
print_labeled_text("Result: PDF transcript preview", file_text)
print_request_shape(direct_file_request)
print_labeled_json("Result: expected fields", expected_direct_file_fields)
try:
direct_file_response = create_response(**direct_file_request)
raw_direct_file_output = output_text(direct_file_response).strip()
try:
direct_file_payload = json.loads(raw_direct_file_output)
missing_or_empty = [
key for key, expected in expected_direct_file_fields.items()
if builtins.str(direct_file_payload.get(key, "")).strip().lower() != expected.lower()
]
null_fields = [key for key, value in direct_file_payload.items() if value in {None, "", []}]
if missing_or_empty or null_fields:
record_check("Direct file inputs", "warn", {
"message": "The request completed, but the model did not extract the expected values from the attached PDF.",
"missing_or_unexpected_fields": missing_or_empty,
"empty_fields": null_fields,
"payload": direct_file_payload,
})
record_response("Support transcript extraction returned by model", "json", direct_file_payload)
print_labeled_text("Result", "The request completed, but the model did not extract the expected values from the attached PDF.")
print_labeled_json("Result: returned JSON", direct_file_payload)
else:
record_check("Direct file inputs", "pass", direct_file_payload)
record_response("Support transcript extraction", "json", direct_file_payload)
print_labeled_json("Result", direct_file_payload)
except Exception as parse_exc:
record_check("Direct file inputs", "warn", {
"message": "The request completed, but the response was not valid JSON.",
"text_sample": raw_direct_file_output[:600],
"error": builtins.str(parse_exc),
})
record_response("Support transcript extraction text", "text", raw_direct_file_output[:1200])
print_labeled_text("Result", raw_direct_file_output[:1200])
print_response_summary(direct_file_response)
print_key_takeaway('Direct file input is useful when the file should be read in the current request context.')
except Exception as exc:
handle_example_error("Direct file inputs", exc)
6. 管理对话状态
对话状态决定了后续轮次如何获取先前的上下文。Responses API 通过 previous_response_id 支持存储的连续性，应用程序也可以自行管理状态，重新发送相关的输入历史。本节比较这两种模式，然后在支持的情况下展示加密的推理上下文。
6.1 使用 previous_response_id 继续对话
使用 previous_response_id 可以从已存储的响应继续对话，而无需重新发送完整的先前提示。第一个请求存储 BrightCart 案例详情；第二个请求仅传递新的后续指令以及之前的响应 ID。检查后续对话是否保留了订单、客户、问题和下一步操作。
from __future__ import annotations
promised_delivery = (date.today() + timedelta(days=2)).isoformat()
stateful_seed_input = (
f"Customer Maya Chen opened ticket TICKET-4812 about order ORDER-8831. "
"The item is a standing desk replacement for a damaged delivery. "
f"The promised delivery date is {promised_delivery}, but the carrier scan has not moved in 36 hours. "
"Customer sentiment is frustrated because this is the second attempt. "
"Support policy says to offer expedited replacement or a 15% concession if the delay exceeds 48 hours. "
"Escalation owner is Tier 2 Returns."
)
stateful_followup_input = "Return five labeled lines: ticket ID, order ID, customer name, issue, and next best action."
stateful_request_shape = {
"model": MODEL_ID,
"input": stateful_followup_input,
"previous_response_id": "<response-id-from-prior-stored-turn>",
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(stateful_request_shape)
try:
stateful_turn_1 = create_response(model=MODEL_ID, input=stateful_seed_input, max_output_tokens=1024, store=True)
remember_stored_response(stateful_turn_1)
stateful_turn_2 = create_response(model=MODEL_ID, input=stateful_followup_input, previous_response_id=stateful_turn_1.id, max_output_tokens=1024, store=False)
text = output_text(stateful_turn_2).strip()
require("order-8831" in text.lower() or "maya" in text.lower(), "Stateful continuation response missed expected support context.")
record_check("Stateful continuation", "pass", stateful_turn_1.id)
record_response("Stateful support handoff", "text", text)
print_labeled_text("Result", text)
print_response_summary(stateful_turn_2)
print_key_takeaway('previous_response_id lets a follow-up use stored context without resending the full prior turn.')
except Exception as exc:
handle_example_error("Stateful continuation", exc)
6.2 重建无状态上下文
无状态续接意味着应用在每次请求时都发送相关的历史记录。当你的产品已经拥有对话存储、保留策略或审计要求时，这种方式非常合适。此单元格发送一段简短的聊天历史以及一条新的交接指令，并检查摘要和令牌使用情况。
from __future__ import annotations
stateless_history = [
{"role": "user", "content": "Support chat TICKET-3920: Customer Jordan Lee says ORDER-7718 arrived with a cracked monitor stand."},
{"role": "assistant", "content": "Captured damaged-item issue for ORDER-7718 and asked for preferred resolution."},
{"role": "user", "content": "Jordan wants a replacement shipped this week and asks whether the damaged item must be returned first."},
]
stateless_payload = {
"model": MODEL_ID,
"input": stateless_history + [{"role": "user", "content": "Summarize this support chat for the next agent in five labeled plain-text lines. Do not use leading hyphens or bold text."}],
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(stateless_payload)
try:
stateless_response = create_response(**stateless_payload)
stateless_text = output_text(stateless_response).strip()
require(stateless_text, "Stateless continuation response did not return text.")
record_check("Stateless continuation", "pass", summarize_response(stateless_response))
record_response("Stateless support handoff", "text", stateless_text)
print_labeled_text("Result", stateless_text)
print_response_summary(stateless_response)
print_key_takeaway('Stateless continuation sends the relevant history with each request when the application owns conversation storage.')
except Exception as exc:
handle_example_error("Stateless continuation", exc)
6.3 携带加密推理上下文
具备推理能力的模型在收到请求时，可能会返回推理条目和加密的推理内容。此单元格请求加密的推理元数据，将之前的响应条目带入后续请求，并检查是否返回了加密内容。隐藏的推理文本不会暴露；应用程序仅在支持的情况下携带不透明的上下文向前传递。
官方文档：推理模型 描述了 Responses 工作流中的推理模型和推理努力。
from __future__ import annotations
encrypted_history = [
{"role": "user", "content": "For a customer-support assistant handling names, order IDs, and refund context, compare stateful and stateless continuation in two sentences."}
]
encrypted_turn_payload = {
"model": MODEL_ID,
"input": encrypted_history,
"reasoning": {"effort": "medium"},
"include": ["reasoning.encrypted_content"],
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(encrypted_turn_payload)
try:
encrypted_turn_1 = create_response(**encrypted_turn_payload)
encrypted_turn_2 = create_response(
model=MODEL_ID,
input=encrypted_history + response_items(encrypted_turn_1) + [
{"role": "user", "content": "Based on the prior reasoning context, recommend one approach for a regulated support workflow in two labeled plain-text lines. Do not use leading hyphens or bold text."}
],
max_output_tokens=1024,
store=False,
)
reasoning_items = [item for item in response_items(encrypted_turn_1) if item.get("type") == "reasoning"]
has_encrypted_content = any(item.get("encrypted_content") for item in reasoning_items)
record_check("Encrypted reasoning", "pass", {"encrypted_content_returned": has_encrypted_content, "reasoning_item_count": len(reasoning_items)})
encrypted_answer = output_text(encrypted_turn_2).strip()
record_response("State strategy recommendation", "text", encrypted_answer)
print_labeled_json("Result: reasoning metadata", {
"returned_item_types": [item.get("type") for item in response_items(encrypted_turn_1)],
"encrypted_reasoning_content_returned": has_encrypted_content,
})
print_labeled_text("Result: follow-up answer", encrypted_answer)
print_response_summary(encrypted_turn_2)
print_key_takeaway('Encrypted reasoning content can be carried forward where supported without exposing hidden reasoning text.')
except Exception as exc:
handle_example_error("Encrypted reasoning", exc)
7. 使用提示缓存
当请求共享完全相同的静态前缀时，提示缓存可改善延迟和成本。
7.1 比较两个缓存键控请求
此单元格将稳定的 BrightCart 策略文本放在输入的开头，使用 prompt_cache_key 发送两次相同的请求，并比较令牌元数据。当端点返回缓存详细信息时，检查第二个响应中的 cached_input_tokens。
注意：PROMPT_CACHE_RETENTION 从活动的 MODEL_ID 中选择。对于 openai.gpt-5.5 及更高版本模型，它使用 24h；对于 openai.gpt-5.4，则使用 in_memory。
from __future__ import annotations
base_support_policy = [
"BrightCart support policy:",
"1. Be empathetic, concise, and specific about the customer's order.",
"2. Do not promise refunds, credits, or delivery dates unless the policy context supports it.",
"3. For damaged-item replacements, check replacement status before offering concessions.",
"4. If a replacement delay exceeds 48 hours, offer expedited replacement or a 15% concession subject to agent approval.",
]
policy_reference_paragraph = (
"Expanded cacheable policy context: BrightCart agents should identify the customer, order ID, replacement status, "
"carrier scan age, promised delivery window, item category, prior concessions, and supervisor approval needs before "
"drafting a customer-facing answer. The assistant should preserve a calm tone, avoid unsupported promises, separate "
"confirmed facts from assumptions, recommend one clear next action, and document why any escalation, expedited "
"replacement, or concession is appropriate. Repeated policy context like this is intentionally stable across many "
"requests so prompt caching can reuse the prefix when the same cache key is supplied."
)
expanded_policy_context = "\n".join(
f"Policy reference paragraph {idx + 1}: {policy_reference_paragraph}"
for idx in range(32)
)
stable_support_policy = "\n".join(base_support_policy + [expanded_policy_context])
cache_input = [
{"role": "system", "content": stable_support_policy},
{"role": "user", "content": "Draft a two-sentence agent reply for Maya Chen about delayed replacement order ORDER-8831."},
]
estimated_cache_input_words = len(json.dumps(cache_input).split())
require(estimated_cache_input_words > 2048, f"Prompt-cache input should be over 2048 words; found {estimated_cache_input_words}.")
cache_payload = {
"model": MODEL_ID,
"input": cache_input,
"prompt_cache_key": "brightcart-support-policy-v1",
"prompt_cache_retention": PROMPT_CACHE_RETENTION,
"max_output_tokens": 1024,
"store": False,
}
print_request_shape(cache_payload)
print_labeled_json("Prompt-cache input size", {"estimated_input_words": estimated_cache_input_words, "target_minimum_tokens": 2048})
try:
cache_response_1 = create_response(**cache_payload)
cache_response_2 = create_response(**cache_payload)
cache_summary_1 = summarize_response(cache_response_1)
cache_summary_2 = summarize_response(cache_response_2)
cache_comparison = pd.DataFrame([
{
"request": "first",
"input_tokens": cache_summary_1.get("input_tokens"),
"cached_input_tokens": cache_summary_1.get("cached_input_tokens"),
"output_tokens": cache_summary_1.get("output_tokens"),
"total_tokens": cache_summary_1.get("total_tokens"),
},
{
"request": "second",
"input_tokens": cache_summary_2.get("input_tokens"),
"cached_input_tokens": cache_summary_2.get("cached_input_tokens"),
"output_tokens": cache_summary_2.get("output_tokens"),
"total_tokens": cache_summary_2.get("total_tokens"),
},
])
record_check("Prompt caching", "pass" if cache_summary_2.get("cached_input_tokens") is not None else "warn", {"first": cache_summary_1, "second": cache_summary_2})
cache_reply = output_text(cache_response_2).strip()
record_response("Prompt-cache token comparison", "table", cache_comparison)
record_response("Cached support-policy reply", "text", cache_reply)
print_labeled_text("Result", cache_reply)
print_labeled_json("First request summary", cache_summary_1)
print_labeled_json("Second request summary", cache_summary_2)
print_label("Response summary")
display_wrapped_table(cache_comparison, max_col_width_px=260)
print_key_takeaway('cached_input_tokens is the metadata field to inspect for prompt-cache reuse.')
except Exception as exc:
handle_example_error("Prompt caching", exc)
8. 运行后台工作
后台模式会异步启动响应，并允许应用程序轮询终端状态。
8.1 提交并轮询后台响应
此单元格发送 background=true，存储响应 ID，在状态为排队或进行中时进行轮询，然后打印最终的管理器摘要。检查状态历史、最终状态、响应 ID 和令牌摘要。
from __future__ import annotations
backlog = """
Same-day BrightCart support backlog:
1. 18 delayed-order contacts, mostly from the West Coast distribution lane.
2. 7 damaged-item replacement contacts; 3 mention replacement delays.
3. 5 return-window exception requests after holiday promotions.
""".strip()
background_payload = {
"model": MODEL_ID,
"input": f"Return exactly three labeled plain-text lines for a support-manager summary: theme, risk, next action. Keep each line under 12 words. Do not use leading hyphens or bold text.\n\n{backlog}",
"background": True,
"max_output_tokens": 1024,
"store": True,
}
print_request_shape(background_payload)
try:
background_response = create_response(**background_payload)
remember_stored_response(background_response)
status_history = [getattr(background_response, "status", None)]
for _ in range(15):
if getattr(background_response, "status", None) not in {"queued", "in_progress"}:
break
time.sleep(2)
background_response = retrieve_response(background_response.id)
status_history.append(getattr(background_response, "status", None))
background_summary = summarize_response(background_response)
manager_summary = output_text(background_response).strip()
require(manager_summary, "Background response did not return text.")
status = "pass" if background_summary.get("status") in {None, "completed"} else "warn"
record_check("Background mode", status, {"status_history": status_history, "id": getattr(background_response, "id", None), "final_status": background_summary.get("status")})
record_response("Background manager summary", "text", manager_summary)
print_labeled_json("Result: status history", status_history)
print_labeled_text("Result: manager summary", manager_summary)
print_response_summary(background_summary)
print_key_takeaway('Background mode starts work asynchronously and lets the application poll by response ID.')
except Exception as exc:
handle_example_error("Background mode", exc)
9. 压缩长时间运行的上下文
压缩会将长时间的对话状态缩减为持久的事实、未解决的问题、约束条件和后续行动。此单元格将应用端的压缩模式记录为一个小型 JSON 对象，以便在不增加另一个实时功能路径的情况下明确概念。在下一轮对话之前，检查保留了哪些事实以及省略了哪些细节。
from __future__ import annotations
compaction_note = {
"feature": "Compaction",
"how_to_apply": "Summarize older support turns into durable facts, open questions, policy constraints, and next actions before continuing the workflow.",
"brightcart_example": {
"durable_facts": ["Customer Maya Chen", "ORDER-8831", "replacement delayed", "carrier scan stale"],
"policy_constraints": ["Do not promise refund without eligibility", "Offer expedited replacement or 15% concession after 48-hour delay with approval"],
"next_action": "Check latest carrier scan and supervisor callback status.",
},
}
record_check("Compaction", "documented", compaction_note)
record_response("Compacted support context", "json", compaction_note)
print_json(compaction_note)
10. 运行操作冒烟检查
操作冒烟检查是轻量级的设置检查，不是负载测试或服务级别测量。此单元格发送三个短请求，测量本地耗时，汇总成功率和令牌使用情况，并从配置的 Bedrock 基础 URL 推断区域。检查延迟、完成状态、示例输出和令牌总数。
from __future__ import annotations
def infer_region_from_base_url(base_url: str) -> str | None:
host = normalize_base_url(base_url).replace("https://", "").split("/")[0]
for part in host.split("."):
if part.count("-") >= 2 and any(char.isdigit() for char in part):
return part
return None
def percentile(values: list[float], pct: float) -> float | None:
if not values:
return None
ordered = sorted(values)
index = min(len(ordered) - 1, max(0, round((pct / 100) * (len(ordered) - 1))))
return round(ordered[index], 3)
operations_features = ["Latency runtime example", "Throughput runtime example", "Reliability runtime example", "Region check"]
operations_payload = {"model": MODEL_ID, "input": "Reply with one short customer-support sentence.", "service_tier": "auto", "max_output_tokens": 1024, "store": False}
print_request_shape(operations_payload)
if not RUN_RESPONSIVENESS_CHECK:
record_check("Endpoint responsiveness", "skipped", "BEDROCK_RESPONSIVENESS_CHECK is disabled.")
print_labeled_text("Result", "Responsiveness check disabled.")
else:
prompts = [
"Reply in one short sentence: apologize for a delayed replacement order.",
"Reply with one metric name for support-assistant quality.",
"Reply in one short sentence: hand off a return exception to a supervisor.",
]
samples = []
for idx, prompt in enumerate(prompts):
started = time.perf_counter()
try:
response = create_response(model=MODEL_ID, input=prompt, service_tier="auto", max_output_tokens=1024, store=False)
elapsed = time.perf_counter() - started
summary = summarize_response(response)
text = output_text(response).strip()
samples.append({
"ok": bool(text),
"latency_seconds": round(elapsed, 3),
"output_tokens": summary.get("output_tokens") or 0,
"total_tokens": summary.get("total_tokens") or 0,
"status": summary.get("status"),
"sample_output": text[:140],
})
except Exception as exc:
elapsed = time.perf_counter() - started
samples.append({"ok": False, "latency_seconds": round(elapsed, 3), "error": describe_api_error(exc)})
successes = [sample for sample in samples if sample["ok"]]
completed = [sample for sample in successes if sample.get("status") in {None, "completed"}]
latencies = [sample["latency_seconds"] for sample in successes]
responsiveness_summary = {
"region_hint": infer_region_from_base_url(BASE_URL),
"base_url_host": normalize_base_url(BASE_URL).replace("https://", "").split("/")[0],
"sample_count": len(samples),
"success_rate": len(successes) / len(samples) if samples else 0,
"completed_rate": len(completed) / len(samples) if samples else 0,
"avg_latency_seconds": round(sum(latencies) / len(latencies), 3) if latencies else None,
"p50_latency_seconds": percentile(latencies, 50),
"p90_latency_seconds": percentile(latencies, 90),
"total_output_tokens": sum(sample.get("output_tokens", 0) for sample in samples),
"total_tokens": sum(sample.get("total_tokens", 0) for sample in samples),
}
status = "pass" if len(successes) == len(samples) and len(completed) == len(samples) else "warn"
for feature in operations_features:
record_check(feature, status, responsiveness_summary)
record_response("Endpoint responsiveness summary", "json", {**responsiveness_summary, "samples": samples})
print_labeled_json("Result", responsiveness_summary)
print_label("Response summary")
display_wrapped_table(pd.DataFrame(samples), max_col_width_px=360)
print_key_takeaway('Responsiveness samples are setup checks, not a load test or service-level measurement.')
11. 清理并查看结果
由生命周期、有状态延续和后台示例创建的存储响应记录在 STORED_RESPONSE_IDS 中。此最终单元格在启用清理时尝试删除存储的响应，然后打印运行摘要和示例响应画廊。请先检查警告；它们通常标识端点配置、模型可用性或功能支持差异。
from __future__ import annotations
cleanup_rows = []
tracked_ids = list(dict.fromkeys(STORED_RESPONSE_IDS))
if not tracked_ids:
cleanup_rows.append({"response_id": "none", "status": "no stored responses tracked", "detail": ""})
elif not CLEAN_UP_STORED_RESPONSES:
for stored_id in tracked_ids:
cleanup_rows.append({"response_id": stored_id, "status": "skipped", "detail": "BEDROCK_CLEANUP_STORED_RESPONSES is disabled"})
record_check("Stored response cleanup", "skipped", cleanup_rows)
else:
for stored_id in tracked_ids:
try:
delete_result = delete_response(stored_id)
cleanup_rows.append({"response_id": stored_id, "status": "deleted", "detail": compact_text(to_dict(delete_result), 240)})
except Exception as exc:
cleanup_rows.append({"response_id": stored_id, "status": "warn", "detail": describe_api_error(exc)})
cleanup_status = "pass" if all(row["status"] == "deleted" for row in cleanup_rows) else "warn"
record_check("Stored response cleanup", cleanup_status, cleanup_rows)
print_label("Stored response cleanup")
display_wrapped_table(pd.DataFrame(cleanup_rows), max_col_width_px=520)
summary_df = pd.DataFrame(RESULTS_SUMMARY)
print_label("Run summary")
display_wrapped_table(summary_df, max_col_width_px=620)
print_label("Example responses")
print_response_gallery()
常见问题
常见错误包括：1. 认证失败（Bearer令牌缺失或无效），检查环境变量AWS_BEARER_TOKEN_BEDROCK是否设置。2. 模型不可用（区域不支持），确认AWS_REGION和BEDROCK_MODEL配对。3. 网络超时，调整BEDROCK_REQUEST_TIMEOUT_SECONDS。4. 请求格式错误，使用原始HTTPS辅助函数检查请求体。5. 工具调用失败，确保工具模式严格且函数实现正确。排查方法：使用预检请求验证端点，查看响应状态码和错误体，利用x-request-id追踪请求。生产优化：启用重试机制（设置BEDROCK_MAX_RETRIES），使用提示缓存减少延迟，采用后台模式处理长任务，定期清理存储的响应。
企业应用扩展
本教程可扩展的方向包括：1. 集成更多AWS服务（如Lambda、S3）构建完整应用。2. 实现更复杂的工具调用链，支持多轮对话。3. 利用Bedrock的模型评估和监控功能。4. 结合LangChain等框架简化开发。5. 探索其他模型（如Claude、Llama）的迁移。6. 增加安全措施（如IAM策略、VPC端点）。7. 实现自动伸缩和成本优化。开发者可根据实际需求，在现有架构上灵活扩展。
总结
本教程面向有编程基础的开发者，通过虚构零售商BrightCart的客服助手场景，系统讲解如何在Amazon Bedrock上使用OpenAI模型。内容涵盖环境配置、原始HTTPS与SDK调用、响应生命周期、结构化输出、函数调用、文件输入、状态管理、提示缓存、后台任务、上下文压缩及运维检查。教程强调工程实践，提供可复现代码、常见错误排查与生产优化建议，帮助开发者快速构建可靠、高效的AI应用。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：A
- 难度：中级
- 验证等级：V1
- 原文作者：shikhar-cyber, sudee-oai
- 原文日期：2026-05-31
- 来源许可：MIT
- 来源版本：
1b59df8d09b6159a164743b653a62b7236216054 - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员