# 亚马逊基岩推理 ## 概述 Amazon Bedrock Runtime 提供用于调用基础模型的 API，包括 Claude (Opus、Sonnet、Haiku)、Nova (Amazon)、Titan (Amazon) 和第三方模型（Cohere、AI21、Meta）。支持具有流功能的同步和异步推理。 **目的**：通过所有 Bedrock 模型的统一 API 进行生产级模型推理 **Pattern** ：基于任务（不同推理模式独立运行） **关键能力**： 1. **模型调用** \- 使用本机或 Converse API 直接模型调用 2. **流式传输** \- 低延迟的实时令牌流式传输 3. **异步调用** \- 长达 24 小时的长时间运行任务 4. **令牌计数** \- 推理前的成本估算 5. **Guardrails** \- 运行时内容过滤和安全 6. **推理概要** \-跨区域路由和成本优化 **质量目标**： * 延迟：< 1 秒用于流式传输的第一个令牌 * 吞吐量：高达 4,000 个令牌/秒 * 可用性：99.9% SLA，具有跨区域配置文件 ** ** ## 何时使用在以下情况下使用基岩推理： * 调用克劳德、新星、泰坦或其他基岩模型 * 构建对话式人工智能应用程序 * 实现流响应以获得更好的用户体验 * 运行长时间运行的异步推理任务 * 应用运行时护栏以确保内容安全 * 通过推理配置文件优化成本 * 在模型调用之前计算令牌 * 实现多轮对话 **何时不使用**： * 构建复杂的代理（使用 bedrock-agentcore） *知识库RAG（使用基岩知识库） * 模型定制（使用基岩微调） ** ** ## 先决条件 ### 必填 * 具有 Bedrock 访问权限的 AWS 账户 * 在 AWS 控制台中启用模型访问 * Bedrock 运行时的 IAM 权限 ### 推荐 * `boto3 >= 1.34.0`（适用于最新的 Converse API） * 了解特定于模型的输入格式 * CloudWatch用于监控 ### 安装 pip 安装 boto3 botocore ### 启用模型访问 # 检查可用型号 aws bedrock list-foundation-models --region us-east-1 # 通过控制台请求模型访问： # AWS 控制台 → Bedrock → 模型访问 → 管理模型访问 ** ** ## 模型 ID 和推理配置文件 ### 克劳德模型（人类）型号|型号 ID |推理配置文件 ID |地区 |最大代币数 ---|---|---|---|--- **克劳德作品 4.5** | `anthropic.claude-opus-4-5-20251101-v1:0` | `global.anthropic.claude-opus-4-5-20251101-v1:0` |全球| 20万 **克劳德十四行诗 4.5** | `anthropic.claude-sonnet-4-5-20250929-v1:0` | `us.anthropic.claude-sonnet-4-5-20250929-v1:0` |美国 | 20万 **克劳德俳句 4.5** | `anthropic.claude-haiku-4-5-20251001-v1:0` | `us.anthropic.claude-haiku-4-5-20251001-v1:0` |美国 | 20万 **克劳德十四行诗 3.5 v2** | `anthropic.claude-3-5-sonnet-20241022-v2:0` | `us.anthropic.claude-3-5-sonnet-20241022-v2:0` |美国 | 20万 **克劳德俳句 3.5** | `anthropic.claude-3-5-haiku-20241022-v1:0` | `us.anthropic.claude-3-5-haiku-20241022-v1:0` |美国 | 20万 ### 亚马逊 Nova 模型型号|型号 ID |推理配置文件 ID |地区 |最大代币数 ---|---|---|---|--- **Nova Pro** | `amazon.nova-pro-v1:0` | `us.amazon.nova-pro-v1:0` |美国 | 30万 **Nova Lite** | `amazon.nova-lite-v1:0` | `us.amazon.nova-lite-v1:0` |美国 | 30万 **新星微** | `amazon.nova-micro-v1:0` | `us.amazon.nova-micro-v1:0` |美国 | 128K ### 亚马逊泰坦模型型号|型号 ID |地区 |最大代币数 ---|---|---|--- **泰坦文本总理** | `amazon.titan-text-premier-v1:0` |全部 | 32K **泰坦短信快车** | `amazon.titan-text-express-v1` |全部 | 8K ### 推理配置文件前缀 * `us.` \- 仅限美国的路由（美国流量的延迟较低） * `global.` \- 全球跨区域路由（最高可用性） * `apac.` \- 亚太路由（亚太地区流量延迟较低） ** ** ## 快速参考 ### 客户端初始化导入boto3 从输入 import 可选 def get_bedrock_client(region_name: str = 'us-east-1', profile_name：可选[str] =无）： """初始化基岩运行时客户端""" 会话 = boto3.Session( 区域名称=区域名称, 配置文件名称=配置文件名称）返回 session.client('bedrock-runtime') # 用法 bedrock = get_bedrock_client(region_name='us-west-2') ** ** ## 操作 ### 1\.调用模型（原生API）使用特定于模型的请求格式直接调用模型。 **基本调用**：导入 json def invoke_claude(提示: str, model_id: str = 'us.anthropic.claude-sonnet-4-5-20250929-v1:0'): """使用原生 API 调用 Claude""" 基岩 = get_bedrock_client() # 克劳德特有的请求格式请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， “消息”：[ { “角色”：“用户”， “内容”：提示 } ], “温度”：0.7， “顶部_p”：0.9 } 响应 = bedrock.invoke_model( modelId=model_id, 正文=json.dumps(request_body) ） # 解析响应 response_body = json.loads(response['body'].read()) 返回response_body['内容'][0]['文本'] # 用法 result = invoke_claude("用简单的术语解释量子计算") 打印（结果） **有系统提示**：请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， "system": "你是一位擅长技术文档的人工智能助手。", “消息”：[ { “角色”：“用户”， "content": "为 REST 端点编写 API 文档" } ] } **使用工具**：请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：4096， “消息”：[ { “角色”：“用户”， "content": "旧金山的天气怎么样？" } ], “工具”：[ { “名称”：“获取天气”， "description": "获取某个位置的当前天气", “输入模式”：{ “类型”：“对象”， “属性”：{ “位置”：{ “类型”：“字符串”， "description": "城市名称" } }, “必需”：[“位置”] } } ] } ** ** ### 2\.匡威 API（统一接口）与模型无关的 API，适用于所有基岩模型，具有一致的接口。 **基本对话**： def converse_with_model( 消息：列表， model_id: str = 'us.anthropic.claude-sonnet-4-5-20250929-v1:0', system_prompts: 可选[列表] = 无，最大令牌数：int = 2048 ）： """统一模型交互的 Converse API""" 基岩 = get_bedrock_client() 推理配置 = { 'maxTokens'：max_tokens， “温度”：0.7， “顶部P”：0.9 } 请求参数 = { 'modelId': model_id, “消息”：消息， 'inferenceConfig'：inference_config } 如果系统提示： request_params['系统'] = 系统提示响应 = bedrock.converse(**request_params) 返回响应 # 用法消息 = [ { '角色'：'用户'， '内容'：[ {'text': '微服务架构有哪些好处？'} ] } ] 系统提示= [ {'text': '您是软件架构专家。'} ] 响应 = converse_with_model(消息, system_prompts=system_prompts) Assistant_message = 响应['输出']['消息'] 打印(assistant_message['内容'][0]['文本']) **多轮对话** : def multi_turn_conversation(): """带有上下文的多轮对话""" 基岩 = get_bedrock_client() 消息 = [] model_id = 'us.anthropic.claude-sonnet-4-5-20250929-v1:0' # 第 1 回合消息.append({ '角色'：'用户'， 'content': [{'text': '我的名字是 Alice，我在医疗保健行业工作。'}] }) 响应 = bedrock.converse( modelId=model_id, 消息=消息， inferenceConfig={'maxTokens': 1024} ） # 添加助手回复历史记录 messages.append(响应['输出']['消息']) # 第 2 轮（模型记住上下文）消息.append({ '角色'：'用户'， 'content': [{'text': '我的领域有哪些人工智能应用？'}] }) 响应 = bedrock.converse( modelId=model_id, 消息=消息， inferenceConfig={'maxTokens': 1024} ）返回响应['输出']['消息']['内容'][0]['文本'] **使用工具（Converse API）**： def converse_with_tools(): """通过工具使用Convert API""" 基岩 = get_bedrock_client() 工具=[ { '工具规范'：{ '名称': '获取股票价格', 'description': '获取符号的当前股票价格', '输入模式'：{ 'json'：{ '类型'：'对象'， “属性”：{ '符号'：{ '类型'：'字符串'， 'description': '股票代码' } }, '必需'：['符号'] } } } } ] 消息 = [ { '角色'：'用户'， 'content': [{'text': "AAPL 股票的价格是多少？"}] } ] 响应 = bedrock.converse( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 消息=消息， toolConfig={'工具': 工具}, inferenceConfig={'maxTokens': 2048} ） # 检查模型是否想要使用工具如果响应['stopReason'] == 'tool_use'： tool_use = 响应['输出']['消息']['内容'][0]['toolUse'] print(f"请求的工具：{tool_use['name']}") print(f"工具输入：{tool_use['input']}") # 执行工具并返回结果 # (将工具结果添加到消息中并再次调用 converse) 返回响应 ** ** ### 3\.流响应（实时令牌）在生成令牌时对其进行流式传输，以降低感知延迟。 **使用本机 API 进行流式传输**： defstream_claude_response（提示：str）： """实时流响应令牌""" 基岩 = get_bedrock_client() 请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， “消息”：[ { “角色”：“用户”， “内容”：提示 } ] } 响应 = bedrock.invoke_model_with_response_stream( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 正文=json.dumps(request_body) ） # 处理事件流流=响应['正文'] 全文=“” 对于流中的事件：块 = event.get('块') 如果块： chunk_obj = json.loads(chunk['bytes'].decode()) 如果 chunk_obj['type'] == 'content_block_delta': 增量 = chunk_obj['增量'] 如果 delta['type'] == 'text_delta': 文本 = 增量['文本'] 打印（文本，结束=''，刷新= True）全文 += 文本 elif chunk_obj['type'] == 'message_stop': print() # 末尾换行返回全文 # 用法 response = stream_claude_response("写一个关于机器人的短篇故事") **使用 Converse API 进行流式传输**： defstream_converse（消息：列表，model_id：str）： """使用 Converse API 传输响应""" 基岩 = get_bedrock_client() 响应 = bedrock.converse_stream( modelId=model_id, 消息=消息， inferenceConfig={'maxTokens': 2048} ）流 = 响应['流'] 全文=“” 对于流中的事件：如果事件中有“contentBlockDelta”：增量=事件['contentBlockDelta']['增量'] 如果 'text' 在 delta 中：文本 = 增量['文本'] 打印（文本，结束=''，刷新= True）全文 += 文本事件中的 elif 'messageStop'：打印（）打破返回全文 # 用法 messages = [{'角色': '用户', '内容': [{'文本': '解释神经网络'}]}] Stream_converse(消息, 'us.anthropic.claude-sonnet-4-5-20250929-v1:0') **带有错误处理的流式传输**： def safe_streaming（提示：str）： """具有全面错误处理功能的流式传输""" 基岩 = get_bedrock_client() 请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， "messages": [{"角色": "用户", "内容": 提示}] } 尝试：响应 = bedrock.invoke_model_with_response_stream( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 正文=json.dumps(request_body) ）全文=“” 对于响应中的事件['body']：块 = event.get('块') 如果块： chunk_obj = json.loads(chunk['bytes'].decode()) 如果 chunk_obj['type'] == 'content_block_delta': 文本 = chunk_obj['delta'].get('文本', '') 打印（文本，结束=''，刷新= True）全文 += 文本 elif chunk_obj['类型'] == '错误'： print(f"\n流媒体错误: {chunk_obj['error']}") 打破返回全文除了异常 e： print(f"流失败：{e}") 提高 ** ** ### 4\.计数代币在调用模型之前估计令牌使用情况和成本。 **逆向令牌计数**： def count_tokens（消息：列表，model_id：str）： """计算代币以估算成本""" 基岩 = get_bedrock_client() # 可选的系统提示系统提示= [ {'text': '你是一个有用的助手。'} ] # 可选工具工具=[ { '工具规范'：{ '名称'：'示例工具'， 'description': '示例工具', '输入模式'：{ 'json'：{ '类型'：'对象'， “属性”：{} } } } } ] 响应 = bedrock.converse_count( modelId=model_id, 消息=消息，系统=系统提示， toolConfig={'工具': 工具} ） # 获取代币数量用法 = 响应['用法'] print(f"输入标记：{usage['inputTokens']}") print(f"系统令牌: {usage.get('systemTokens', 0)}") print(f"工具令牌: {usage.get('toolTokens', 0)}") print(f"总输入：{usage['totalTokens']}") 返回用法 # 用法消息 = [ {'role': 'user', 'content': [{'text': '这是一条测试消息'}]} ] 令牌 = count_tokens(消息, 'us.anthropic.claude-sonnet-4-5-20250929-v1:0') **成本估算**： defestimate_cost（消息：列表，model_id：str，estimated_output_tokens：int = 1000）： """调用前估计推理成本""" 基岩 = get_bedrock_client() # 计算输入的标记 token_response = bedrock.converse_count( modelId=model_id, 消息=消息） input_tokens = token_response['使用情况']['totalTokens'] # 定价（截至 2024 年 12 月，价格因地区而异）定价={ 'us.anthropic.claude-opus-4-5-20251101-v1:0': { 'input': 15.00 / 1_000_000, # 每 100 万个输入代币 15 美元 'output': 75.00 / 1_000_000 # 每 1M 输出代币 75 美元 }, 'us.anthropic.claude-sonnet-4-5-20250929-v1:0': { “输入”：3.00 / 1_000_000， “输出”：15.00 / 1_000_000 }, 'us.anthropic.claude-haiku-4-5-20251001-v1:0': { “输入”：0.80 / 1_000_000， “输出”：4.00 / 1_000_000 } } 如果定价中包含 model_id： input_cost = input_tokens * 定价[model_id]['input'] 输出成本 = 估计输出令牌 * 定价[model_id]['输出'] 总成本 = 输入成本 + 输出成本 print(f"输入令牌：{input_tokens:,} (${input_cost:.6f})") print(f"预计输出：{estimated_output_tokens:,} (${output_cost:.6f})") print(f"预计总计：${total_cost:.6f}") 返回{ '输入令牌'：输入令牌， 'estimated_output_tokens'：estimated_output_tokens， '输入成本'：输入成本， '输出成本'：输出成本， '总成本'：总成本 } 其他： print("该型号的定价不可用") 返回无 ** ** ### 5\.异步调用（长时间运行的任务）适用于耗时超过 60 秒（最多 24 小时）的推理任务。 **启动异步调用**： def async_invoke_model（提示：str，s3_output_uri：str）： """为长任务启动异步模型调用""" 基岩 = get_bedrock_client() 请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：10000， “消息”：[ { “角色”：“用户”， “内容”：提示 } ] } 响应 = bedrock.invoke_model_async( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', modelInput=json.dumps(request_body), 输出数据配置={ 's3OutputDataConfig'：{ 's3Uri'：s3_output_uri } } ） invoking_arn = 响应['invokingArn'] print(f"异步调用已开始：{inspiration_arn}") 返回调用_arn # 用法 s3_output = 's3://my-bucket/bedrock-outputs/result.json' arn = async_invoke_model("写一篇10000字的技术指南", s3_output) **检查异步状态**： def check_async_status(inplication_arn: str): """检查异步调用的状态""" 基岩 = get_bedrock_client() 响应 = bedrock.get_async_invoke( invotionArn=involution_arn ）状态 = 响应['状态'] print(f"状态：{status}") 如果状态=='已完成'： output_uri = 响应['outputDataConfig']['s3OutputDataConfig']['s3Uri'] print(f"输出位于：{output_uri}") # 下载并解析结果 # (使用boto3 S3客户端检索) elif 状态 == '失败': print(f"失败原因：{response.get('failureMessage', '未知')}") 返回响应 # 用法状态 = check_async_status(arn) **列出异步调用**： def list_async_incalls(status_filter: 可选[str] = None): """列出所有异步调用""" 基岩 = get_bedrock_client() 参数 = {} 如果状态过滤器： params['statusEquals'] = status_filter # '进行中', '已完成', '失败' 响应 = bedrock.list_async_invokes(**params) 用于在response.get('asyncInvokeSummaries', [])中调用：打印（f“ARN：{调用['invokingArn']}”） print(f"状态：{调用['status']}") print(f"提交时间：{调用['submitTime']}") 打印（“---”）返回响应 ** ** ### 6\.应用护栏（运行时安全）在运行时应用内容过滤和安全策略。 **使用 Guardrail 调用**： def invoke_with_guardrail( 提示：str，护栏_id：str， Guardrail_version: str = '草稿' ）： """调用带有运行时护栏的模型""" 基岩 = get_bedrock_client() 请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， “消息”：[ { “角色”：“用户”， “内容”：提示 } ] } 响应 = bedrock.invoke_model( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 正文=json.dumps(request_body), GuardrailIdentifier=guardrail_id, GuardrailVersion=guardrail_version ） # 检查内容是否被阻止 response_body = json.loads(response['body'].read()) 如果响应中的“amazon-bedrock-guardrailAction”['ResponseMetadata']['HTTPHeaders']：操作 = 响应['ResponseMetadata']['HTTPHeaders']['amazon-bedrock-guardrailAction'] 如果操作=='GUARDRAIL_INTERVENED'： print("内容被护栏挡住") 返回无返回response_body['内容'][0]['文本'] # 用法结果 = invoke_with_guardrail( “告诉我有关量子计算的事情”， Guardrail_id='abc123xyz', Guardrail_version='1' ） **与护栏对话**： def converse_with_guardrail（消息：列表，guardrail_config：字典）： """带有护栏配置的 Converse API""" 基岩 = get_bedrock_client() 响应 = bedrock.converse( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 消息=消息， inferenceConfig={'maxTokens': 2048}, GuardrailConfig=guardrail_config ） # 检查护栏干预的踪迹如果响应为“trace”：跟踪 = 响应['跟踪']['护栏'] 如果trace.get('action') == 'GUARDRAIL_INTERVENED': print("护栏遮挡的内容") 用于 Trace.get('assessments', []) 中的评估： print(f"政策：{评估['topicPolicy']}") 返回响应 # 用法护栏配置 = { '护栏标识符'：'abc123xyz'， '护栏版本': '1', '跟踪'：'启用' } messages = [{'角色': '用户', '内容': [{'文本': '测试消息'}]}] converse_with_guardrail（消息，guardrail_config） ** ** ## 错误处理模式 ### 全面的错误处理从 botocore.exceptions 导入 ClientError, BotoCoreError 导入时间 defRobust_invoke（提示：str，max_retries：int = 3）： """使用重试逻辑和错误处理调用模型""" 基岩 = get_bedrock_client() 请求正文 = { “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， "messages": [{"角色": "用户", "内容": 提示}] } 对于范围内的尝试（max_retries）：尝试：响应 = bedrock.invoke_model( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 正文=json.dumps(request_body) ） response_body = json.loads(response['body'].read()) 返回response_body['内容'][0]['文本'] 除了 ClientError 为 e： error_code = e.response['错误']['代码'] 如果 error_code == 'ThrotdlingException'： wait_time = (2 ** attempts) + 1 # 指数退避 print(f"节流。重试之前等待 {wait_time} 秒 {attempt + 1}/{max_retries}") 时间.睡眠（等待时间）继续 elif error_code == 'ModelTimeoutException'： print("模型超时 - 请求花费的时间太长") 如果尝试 < max_retries - 1：时间.睡眠(2) 继续提高 elif error_code == 'ModelErrorException'： print("模型错误 - 检查输入格式") 提高 elif error_code == 'ValidationException'： print("参数无效") 提高 elif error_code == 'AccessDeniedException'： print("访问被拒绝 - 检查 IAM 权限和模型访问") 提高 elif error_code == 'ResourceNotFoundException'： print("未找到模型 - 检查模型 ID") 提高其他： print(f"意外错误：{error_code}") 提高除了 BotoCoreError 为 e： print(f"连接错误：{e}") 如果尝试 < max_retries - 1：时间.睡眠(2) 继续提高引发异常（f“在 {max_retries} 次尝试后失败”） ### 具体错误场景 def handler_model_errors(): """常见错误场景及解决方案""" 基岩 = get_bedrock_client() 尝试： # 尝试调用响应 = bedrock.invoke_model( modelId='us.anthropic.claude-sonnet-4-5-20250929-v1:0', 正文=json.dumps({ “anthropic_version”：“基岩-2023-05-31”， “最大令牌”：2048， “消息”：[{“角色”：“用户”，“内容”：“测试”}] }) ）除了 ClientError 为 e： error_code = e.response['错误']['代码'] 如果 error_code == 'ModelNotReadyException'： # 模型仍在加载中 print("模型未准备好，等待 30 秒并重试") elif error_code == 'ServiceQuotaExceededException'： # 达到服务配额 print("超出配额-请求增加或使用不同区域") elif error_code == 'ModelStreamErrorException'： # 串流时出错 print("流中断 - 重新启动流") ** ** ## 最佳实践 ### 1\.成本优化 def cost_optimized_inference(提示: str, require_high_accuracy: bool = False): """根据任务复杂度和成本选择模型""" # 简单任务→俳句（最便宜） # 中等任务 → Sonnet（平衡） # 复杂任务 → Opus（最有能力）如果不要求高准确度： model_id = 'us.anthropic.claude-haiku-4-5-20251001-v1:0' print("使用俳句提高成本效率") elif require_high_accuracy： model_id = 'global.anthropic.claude-opus-4-5-20251101-v1:0' print("使用 Opus 获得最大准确度") 其他： model_id = 'us.anthropic.claude-sonnet-4-5-20250929-v1:0' print("使用 Sonnet 来平衡性能") 返回invoke_claude（提示，model_id） ### 2\.使用推理配置文件 def use_inference_profiles(): """利用推理配置文件来节省成本""" # 跨区域配置可节省 30-50% 的成本 # 具有自动区域故障转移功能配置文件= { 'global_opus': 'global.anthropic.claude-opus-4-5-20251101-v1:0', 'us_sonnet': 'us.anthropic.claude-sonnet-4-5-20250929-v1:0', 'us_haiku': 'us.anthropic.claude-haiku-4-5-20251001-v1:0' } # 使用全局配置文件来实现高可用性 # 使用区域配置文件来降低延迟返回配置文件 ### 3\.实施缓存从 functools 导入 lru_cache 导入哈希库 @lru_cache(最大大小=100) def cached_inference（提示：str，model_id：str）： """缓存相同提示的响应""" 返回invoke_claude（提示，model_id） def cache_key(提示: str) -> str: """生成提示的缓存密钥""" 返回 hashlib.sha256(prompt.encode()).hexdigest() ### 4\.监控令牌使用情况 def track_token_usage（消息：列表，model_id：str）： """跟踪并记录令牌使用情况""" 基岩 = get_bedrock_client() # 调用前计数 token_count = bedrock.converse_count( modelId=model_id, 消息=消息） input_tokens = token_count['使用情况']['totalTokens'] # 调用响应 = bedrock.converse( modelId=model_id, 消息=消息， inferenceConfig={'maxTokens': 2048} ） # 获取实际输出token output_tokens = 响应['用法']['outputTokens'] Total_tokens = 响应['使用情况']['totalInputTokens'] + output_tokens # 登录到CloudWatch或数据库 print(f"输入：{input_tokens}，输出：{output_tokens}，总计：{total_tokens}") 返回响应 ### 5\.使用流媒体获得更好的用户体验 defstream_for_user_experience（提示：str）： """交互式应用程序始终使用流式传输""" # 流媒体减少感知延迟 # 用户立即看到令牌而不是等待返回stream_claude_response（提示） ### 6\.长任务的异步 def use_async_for_batch（提示：列表，s3_bucket：str）： """使用异步调用进行批处理""" 调用_arns = [] 对于 idx，枚举中提示（提示）： s3_uri = f's3://{s3_bucket}/outputs/result-{idx}.json' arn = async_invoke_model（提示，s3_uri） invoice_arns.append(arn) 返回调用_arns ** ** ## IAM 权限 ### 最低运行时权限 { "版本": "2012-10-17", “声明”：[ { "效果": "允许", “行动”：[ “基岩：调用模型”， “基岩：InvokeModelWithResponseStream” ], “资源”：[ "arn:aws:bedrock:*::foundation-model/anthropic.claude-*", "arn:aws:bedrock:*::foundation-model/amazon.nova-*", “arn:aws:bedrock:*::foundation-model/amazon.titan-*” ] }, { "效果": "允许", “行动”：[ “基石：匡威”， “基石：ConverseStream” ], “资源”：“*” } ] } ### 使用异步调用 { "版本": "2012-10-17", “声明”：[ { "效果": "允许", “行动”：[ “基岩：调用模型”， “基岩：InvokeModelWithResponseStream”， “基岩：InvokeModelAsync”， “基岩：GetAsyncInvoke”， “基岩：ListAsyncInvokes” ], “资源”：“*” }, { "效果": "允许", “行动”：[ “s3：放置对象”， “s3：获取对象” ], “资源”：“arn：aws：s3 ::：my-bedrock-bucket / *” } ] } ** ** ## 渐进式披露 ### 快速入门（此文件） * 客户端初始化 * 模型 ID 和推理配置文件 * 基本调用（本机和 Converse API） * 流式响应 * 令牌计数 * 异步调用 * 护栏应用 * 错误处理模式 * 最佳实践 ### 详细参考资料 * **[高级调用模式]()**：批处理、并行请求、自定义重试逻辑、响应解析 * **[多模态支持]()**：Claude 和 Nova 的图像输入、文档解析、视觉功能 * **[工具使用和函数调用]()** : 完整的工具使用模式、多轮工具对话、错误处理 * **[性能优化]()**：延迟优化、吞吐量调整、成本降低策略 * **[监控和可观察性]()**：CloudWatch 集成、自定义指标、成本跟踪、使用情况分析 ** ** ## 相关技能 * **bedrock-agentcore** ：使用托管基础设施构建生产 AI 代理 * **基岩护栏**：配置内容过滤器和安全策略 * **基础知识库**：具有向量存储和检索的 RAG * **bedrock-prompts** ：管理和版本提示 * **人类专家**：Claude API 模式和最佳实践 * **claude-cost-optimization** ：Claude 的成本跟踪和优化 * **boto3-eks** ：适用于集装箱基岩应用 ** ** ## 来源 * [Amazon Bedrock 运行时 API]() * [Boto3 Bedrock 运行时]() * [Converse API 文档]() * [基岩上的克劳德]() * [推理配置文件]()

基岩推理

包含技能