关于 bedrock
面向使用 Amazon Bedrock 构建生成式 AI 应用的开发者,介绍基础模型调用、嵌入和图像生成等功能。内容还涵盖模型访问、知识库、推理类型、CLI 及常见运维问题。
主要功能
- 01调用多家厂商的基础模型
- 02文本生成、嵌入与图像生成
- 03流式响应与多轮对话
- 04知识库检索与 RAG
- 05模型访问和生命周期参考
- 06CLI 操作及故障排查
适用场景
- 01在应用中生成文本或图像
- 02为文本生成嵌入向量
- 03构建多轮模型对话
- 04通过知识库检索并生成回答
安装
复制下面的命令,在终端中运行。
npx skills add https:/ / github.com/ itsmostafa/ aws-agent-skills --skill bedrock安装后在 Claude Code 中输入/bedrock确认已加载。
附带文件model-invocation.md
详细介绍
AWS Bedrock
Amazon Bedrock 通过统一 API 提供对 AI 公司基础模型(FM)的访问。借助文本生成、嵌入和图像生成能力,构建生成式 AI 应用。
目录
核心概念
基础模型
通过 Bedrock 提供的预训练模型:
- Claude(Anthropic):文本生成、分析、编码
- Nova / Titan(Amazon):文本、多模态、嵌入
- GPT / gpt-oss(OpenAI):文本生成、推理
- Llama(Meta):开放权重文本生成
- Mistral:高效文本生成
- Stable Image(Stability AI):图像生成与编辑
模型访问
在商业区域,默认启用对所有无服务器模型的访问(无需在控制台中选择启用)。在 GovCloud (US) 中,仍需在 Model access 页面手动启用模型(第三方模型也需在关联的商业账户中启用):
- 首次调用第三方模型时,会通过 AWS Marketplace 自动订阅(最多需要 15 min);调用方需要
aws-marketplace:Subscribe、Unsubscribe、ViewSubscriptions权限 - 在
bedrock-runtime上使用 Anthropic 模型时,每个账户/组织需填写一次使用场景表单(put-use-case-for-model-access) - 调用即表示接受 EULA;若要阻止某个模型,需通过 SCP/IAM 拒绝该模型的
bedrock:InvokeModel和bedrock:InvokeModelWithResponseStream权限(例如ConverseStream等流式 API 会使用后者)。仅拒绝aws-marketplace:Subscribe并不能阻止首次使用
端点
| 端点 | API | 用途 |
|---|---|---|
bedrock-runtime.{region}.amazonaws.com(推荐) | InvokeModel、Converse、Anthropic Messages(/anthropic)、OpenAI Responses/Chat Completions(/openai/v1) | Guardrails、跨区域推理、提示词路由、应用推理配置文件 |
bedrock-mantle.{region}.api.aws | OpenAI Responses/Chat Completions(/openai/v1)、Anthropic Messages | 服务端工具(Web Search)、background=true 异步处理、Projects/Workspaces、仅适用于 CRIS 模型的单区域访问 |
- 两者的每个 Token 价格相同;通过 SigV4 或 Bedrock API 密钥(
AWS_BEARER_TOKEN_BEDROCK)进行身份验证 - IAM:
bedrock:InvokeModel(runtime)与bedrock-mantle:CreateInference(mantle) bedrock-runtime上的 Responses API 仅支持同步调用,且没有服务端工具
推理配置文件和模型生命周期
- 较新的模型(例如 Claude Sonnet 5)在
bedrock-runtime上没有区域内按需 ID:请使用地理范围(us.、eu.、au.)或global.推理配置文件 ID 作为modelId - 生命周期为
Active->Legacy->EOL(参见get-foundation-model中的modelLifecycle)。Legacy:不再支持新增 Provisioned Throughput、微调或提高配额;EOL:请求将失败 - 模型卡片会列出“不早于此日期 EOL”的日期;固定模型 ID 前请先确认
知识库和 Agent
- 托管知识库(
type: MANAGED):由 Bedrock 负责存储、索引和检索。只有此类型支持AgenticRetrieveStream(查询分解、迭代检索,以及可选的 AgentCore Memory,通过memoryConfiguration配置) - 原生多模态托管知识库使用 TwelveLabs Marengo Embed 3.0(
twelvelabs.marengo-embed-3-0-v1:0)直接嵌入视频/音频/图像;仅支持通过Retrieve使用文本进行查询(不支持RetrieveAndGenerate) - Bedrock Agents Classic 处于维护模式:自 July 30, 2026 起不再向新账户开放(此前 12 个月内未使用过的账户调用
CreateAgent/InvokeInlineAgent会返回 403),模型目录已冻结。新建 Agent 请使用 Amazon Bedrock AgentCore
推理类型
| 类型 | 使用场景 | 定价 |
|---|---|---|
| On-Demand | 工作负载有波动 | 按 Token 计费 |
| Provisioned Throughput | 稳定的高流量 | 按小时承诺用量 |
| Batch Inference | 大规模异步处理 | 按 Token 计费,价格优惠 |
常见模式
调用模型(文本生成)
AWS CLI:
# Invoke Claude
aws bedrock-runtime invoke-model \
--model-id us.anthropic.claude-sonnet-5 \
--content-type application/json \
--accept application/json \
--cli-binary-format raw-in-base64-out \
--body '{
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [
{"role": "user", "content": "Explain AWS Lambda in 3 sentences."}
]
}' \
response.json
# Claude Sonnet 5/Opus 5 think by default: content may start with a thinking block
cat response.json | jq -r '.content[] | select(.type=="text") | .text'
boto3:
import boto3
import json
bedrock = boto3.client('bedrock-runtime')
def invoke_claude(prompt, max_tokens=4096):
response = bedrock.invoke_model(
modelId='us.anthropic.claude-sonnet-5',
contentType='application/json',
accept='application/json',
body=json.dumps({
'anthropic_version': 'bedrock-2023-05-31',
'max_tokens': max_tokens,
'messages': [
{'role': 'user', 'content': prompt}
]
})
)
result = json.loads(response['body'].read())
# Skip thinking blocks (adaptive thinking is on by default for Sonnet 5).
# max_tokens caps thinking + text, so a truncated response may have no text block.
if result['stop_reason'] == 'max_tokens':
print('Truncated at max_tokens: raise it or lower output_config.effort')
return next((b['text'] for b in result['content'] if b['type'] == 'text'), '')
# Usage
response = invoke_claude('What is Amazon S3?')
print(response)
流式响应
import boto3
import json
bedrock = boto3.client('bedrock-runtime')
def stream_claude(prompt):
response = bedrock.invoke_model_with_response_stream(
modelId='us.anthropic.claude-sonnet-5',
contentType='application/json',
accept='application/json',
body=json.dumps({
'anthropic_version': 'bedrock-2023-05-31',
'max_tokens': 4096,
'messages': [
{'role': 'user', 'content': prompt}
]
})
)
for event in response['body']:
chunk = json.loads(event['chunk']['bytes'])
if chunk['type'] == 'content_block_delta':
yield chunk['delta'].get('text', '')
# Usage
for text in stream_claude('Write a haiku about cloud computing.'):
print(text, end='', flush=True)
生成嵌入
import boto3
import json
bedrock = boto3.client('bedrock-runtime')
def get_embedding(text):
response = bedrock.invoke_model(
modelId='amazon.titan-embed-text-v2:0',
contentType='application/json',
accept='application/json',
body=json.dumps({
'inputText': text,
'dimensions': 1024,
'normalize': True
})
)
result = json.loads(response['body'].read())
return result['embedding']
# Usage
embedding = get_embedding('AWS Lambda is a serverless compute service.')
print(f'Embedding dimension: {len(embedding)}')
带历史记录的对话
import boto3
import json
bedrock = boto3.client('bedrock-runtime')
class Conversation:
def __init__(self, system_prompt=None):
self.messages = []
self.system = system_prompt
def chat(self, user_message):
self.messages.append({
'role': 'user',
'content': user_message
})
body = {
'anthropic_version': 'bedrock-2023-05-31',
'max_tokens': 4096,
'messages': self.messages
}
if self.system:
body['system'] = self.system
response = bedrock.invoke_model(
modelId='us.anthropic.claude-sonnet-5',
contentType='application/json',
accept='application/json',
body=json.dumps(body)
)
result = json.loads(response['body'].read())
if result['stop_reason'] == 'max_tokens':
# max_tokens caps thinking + text; don't store a truncated/empty turn
self.messages.pop()
raise RuntimeError('Truncated at max_tokens: raise it or lower output_config.effort')
assistant_message = next(
(b['text'] for b in result['content'] if b['type'] == 'text'), ''
)
self.messages.append({
'role': 'assistant',
'content': assistant_message
})
return assistant_message
# Usage
conv = Conversation(system_prompt='You are an AWS solutions architect.')
print(conv.chat('What database should I use for a chat application?'))
print(conv.chat('What about for time-series data?'))
列出可用模型
# List all foundation models
aws bedrock list-foundation-models \
--query 'modelSummaries[*].[modelId,modelName,providerName]' \
--output table
# Filter by provider
aws bedrock list-foundation-models \
--by-provider anthropic \
--query 'modelSummaries[*].modelId'
# Get model details (includes modelLifecycle.status)
aws bedrock get-foundation-model \
--model-identifier anthropic.claude-sonnet-5
检查模型访问权限
# agreementAvailability.status AVAILABLE / NOT_AVAILABLE, authorizationStatus
aws bedrock get-foundation-model-availability \
--model-id anthropic.claude-sonnet-5
# Anthropic one-time use case form (base64-encoded JSON:
# companyName, companyWebsite, intendedUsers, industryOption, otherIndustryOption, useCases)
aws bedrock put-use-case-for-model-access --form-data <base64-json>
# Programmatic agreement for third-party models
aws bedrock list-foundation-model-agreement-offers --model-id <model-id>
aws bedrock create-foundation-model-agreement --model-id <model-id> --offer-token <token>
统计令牌数
# Free; returns inputTokens. Not supported for every model (e.g. CRIS-only Claude models)
aws bedrock-runtime count-tokens \
--model-id anthropic.claude-3-5-haiku-20241022-v1:0 \
--input '{"converse": {"messages": [{"role": "user", "content": [{"text": "Hello"}]}]}}'
CLI 参考
Bedrock(控制平面)
| 命令 | 说明 |
|---|---|
aws bedrock list-foundation-models | 列出可用模型 |
aws bedrock get-foundation-model | 获取模型详情 |
aws bedrock list-custom-models | 列出微调模型 |
aws bedrock create-model-customization-job | 开始微调 |
aws bedrock list-provisioned-model-throughputs | 列出预置容量 |
aws bedrock get-foundation-model-availability | 检查模型的访问/协议状态 |
aws bedrock put-use-case-for-model-access | 提交 Anthropic 首次使用场景表单 |
aws bedrock list-inference-profiles | 列出系统/应用推理配置文件 |
aws bedrock create-model-invocation-job | 开始批处理作业(--model-invocation-type InvokeModel|Converse) |
Bedrock Runtime(数据平面)
| 命令 | 说明 |
|---|---|
aws bedrock-runtime invoke-model | 同步调用模型 |
aws bedrock-runtime converse | 多轮对话 API |
aws bedrock-runtime count-tokens | 统计输入令牌数(--input 使用 invokeModel 或 converse) |
aws bedrock-runtime apply-guardrail | 根据护栏评估内容 |
InvokeModelWithResponseStream 和 ConverseStream 仅支持 SDK(不在 AWS CLI v2 中)。
Bedrock Agent Runtime
| 命令 | 说明 |
|---|---|
aws bedrock-agent-runtime retrieve | 查询知识库 |
aws bedrock-agent-runtime retrieve-and-generate | RAG 查询 |
InvokeAgent、RetrieveAndGenerateStream 和 AgenticRetrieveStream 仅支持 SDK(事件流)。
最佳实践
成本优化
- 使用合适的模型:简单任务使用较小的模型
- 设置 max_tokens:尽可能限制输出长度
- 缓存响应:用于重复的相同查询
- 尽可能批处理:批量处理时使用批量推理
- 监控用量:设置 CloudWatch 成本告警
- 全局推理配置文件:在数据驻留要求允许时,比地理区域配置文件便宜约 10%
- 思考令牌按输出计费:Claude Sonnet 5/Opus 5 默认启用思考;如果不需要,可传入
"thinking": {"type": "disabled"}或降低output_config.effort,并重新考虑max_tokens(它限制思考内容和文本的总量) - Converse 批处理格式:
--model-invocation-type Converse可在不同模型间保持相同的请求格式 - 成本归属:将 IAM 主体标记为成本分配标签(两个端点均适用)
性能
- 使用流式传输:长输出时可改善用户体验
- 连接池:重复使用 boto3 客户端
- 区域部署:使用最近的区域以降低延迟
- 预置吞吐量:适用于稳定的高流量工作负载
- 端点选择:默认使用
bedrock-runtime;仅在需要 mantle 专属功能时使用bedrock-mantle
安全性
- IAM 最小权限:仅授予所需的模型访问权限
- VPC 端点:保持流量私有
- 护栏:实施内容过滤
- 使用 CloudTrail 审计:跟踪模型调用
- Web Search(mantle):设置
external_web_access: false,使 Fetch 保持在 AWS 边界内;AmazonBedrockFullAccess不包含bedrock-websearch:ExternalWebAccess,因此默认值true会导致 Fetch 静默失败 - 固定活动模型:检查
modelLifecycle,并在 EOL 前迁移出Legacy模型
IAM 权限
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": [
"arn:aws:bedrock:us-east-1:123456789012:inference-profile/us.anthropic.claude-sonnet-5",
"arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
]
},
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": "arn:aws:bedrock:*::foundation-model/anthropic.claude-sonnet-5",
"Condition": {
"StringEquals": {
"bedrock:InferenceProfileArn": "arn:aws:bedrock:us-east-1:123456789012:inference-profile/us.anthropic.claude-sonnet-5"
}
}
}
]
}
推理配置文件需要授予对配置文件 ARN 以及每个目标区域中的基础模型的访问权限(使用 aws bedrock get-inference-profile --inference-profile-identifier <id> 列出,查看 models 字段)。拒绝区域访问的 SCP 必须允许这些目标区域(或通过 bedrock:InferenceProfileArn 进行豁免)。
故障排查
AccessDeniedException
原因:
- 首次使用第三方模型时缺少
aws-marketplace:Subscribe权限(自动订阅失败;修复后可能需要约 2 分钟) - 尚未提交 Anthropic 用例表单
- IAM 策略缺少
bedrock:InvokeModel,或推理配置文件缺少目标区域的基础模型 ARN - 模型 ID 或区域错误
- Bedrock Agents Classic:对于此前未使用过的账户,在
CreateAgent/InvokeInlineAgent上出现“Bedrock Agents is in Maintenance Mode” 403 错误(请使用 AgentCore)
调试:
# Check model access status
aws bedrock get-foundation-model-availability \
--model-id anthropic.claude-sonnet-5
# Test IAM permissions
aws iam simulate-principal-policy \
--policy-source-arn arn:aws:iam::123456789012:role/my-role \
--action-names bedrock:InvokeModel \
--resource-arns "arn:aws:bedrock:us-east-1:123456789012:inference-profile/us.anthropic.claude-sonnet-5"
# The profile ARN can pass while cross-Region routing is still denied: also simulate each
# destination-Region model ARN (models field of get-inference-profile) with the profile context
aws iam simulate-principal-policy \
--policy-source-arn arn:aws:iam::123456789012:role/my-role \
--action-names bedrock:InvokeModel \
--resource-arns "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-sonnet-5" \
"arn:aws:bedrock:us-west-2::foundation-model/anthropic.claude-sonnet-5" \
--context-entries '[{"ContextKeyName":"bedrock:InferenceProfileArn","ContextKeyType":"string","ContextKeyValues":["arn:aws:bedrock:us-east-1:123456789012:inference-profile/us.anthropic.claude-sonnet-5"]}]'
ModelNotReadyException
**原因:**模型仍在配置中,或暂时不可用。
**解决方案:**使用指数退避实现重试:
import time
from botocore.exceptions import ClientError
def invoke_with_retry(bedrock, body, max_retries=3):
for attempt in range(max_retries):
try:
return bedrock.invoke_model(
modelId='us.anthropic.claude-sonnet-5',
body=json.dumps(body)
)
except ClientError as e:
if e.response['Error']['Code'] == 'ModelNotReadyException':
time.sleep(2 ** attempt)
else:
raise
raise Exception('Max retries exceeded')
ThrottlingException
原因:
- 超出每个模型的每分钟令牌配额(在
bedrock-runtime上合并计算输入和输出),或每日令牌配额 - 超出 RPM 配额(因模型而异;有些模型没有此配额)
- 并发请求过多
解决方案:
- 申请提高配额(申请“Cross-Region InvokeModel tokens per minute for ”,以同时覆盖 TPM/TPD;Legacy 模型不提供此项配额)
- 降低
max_tokens:它会影响配额扣减 - 使用跨区域推理配置文件以提高吞吐量
- 实现指数退避
- 考虑使用预置吞吐量
ValidationException
常见问题:
- 模型 ID 无效,或模型已结束生命周期
- 错误提示该模型 ID 不支持按需吞吐量:请使用推理配置文件 ID(以
us./global.为前缀) - 请求正文格式错误
- max_tokens 超出模型限制
- 对只接受
adaptive/disabled的模型(例如 Claude Sonnet 5)使用thinking.type: "enabled"和budget_tokens - 将
output_config.format(结构化输出)发送到bedrock-mantle(请在bedrock-runtime上使用 Converse/InvokeModel)
调试:
# Check model-specific requirements
aws bedrock get-foundation-model \
--model-identifier anthropic.claude-sonnet-5 \
--query 'modelDetails.[inferenceTypesSupported,modelLifecycle.status]'