如何异步调用AWS Bedrock？Python SDK调用Claude3并将结果发送至S3

完全可以通过AWS Python SDK（boto3）异步调用Bedrock上的Claude3模型，并且支持将调用结果直接发送到指定的S3存储桶，无需额外手动上传。以下是具体实现步骤和代码示例：

1. 配置IAM权限 #

确保执行代码的IAM实体（用户/角色）拥有以下权限：

• Bedrock异步调用权限：bedrock:InvokeModelAsync 和 bedrock:GetModelInvocationJob
• S3写入权限：s3:PutObject（针对目标存储桶）

示例IAM权限策略（JSON格式）：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModelAsync",
                "bedrock:GetModelInvocationJob"
            ],
            "Resource": "arn:aws:bedrock:*::foundation-model/anthropic.claude-3-*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::your-target-bucket/*"
        }
    ]
}

2. 异步调用Claude3并指定S3输出 #

使用boto3的Bedrock客户端发起异步调用，直接在请求中配置S3输出位置，模型完成推理后会自动将结果写入指定S3路径。

代码示例 #

import boto3
import time

# 初始化Bedrock客户端（根据你的AWS区域调整）
bedrock = boto3.client('bedrock', region_name='us-east-1')

# 构造Claude3请求体（符合Anthropic官方格式）
request_body = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 1000,
    "messages": [
        {
            "role": "user",
            "content": "请解释异步编程的核心概念"
        }
    ]
}

# 发起异步调用，指定S3输出配置
response = bedrock.invoke_model_async(
    modelId="anthropic.claude-3-sonnet-20240229-v1:0",  # 替换为目标Claude3模型ID
    contentType="application/json",
    accept="application/json",
    body=bytes(str(request_body), 'utf-8'),
    outputConfig={
        "S3OutputConfig": {
            "S3Uri": "s3://your-target-bucket/bedrock-output/"  # 替换为你的S3桶路径
        }
    }
)

# 获取异步任务ID
job_id = response['jobId']
print(f"异步任务已启动，Job ID: {job_id}")

# 轮询任务状态，直到完成/失败
while True:
    job_status = bedrock.get_model_invocation_job(jobId=job_id)
    status = job_status['status']
    print(f"当前任务状态: {status}")
    
    if status in ['Completed', 'Failed', 'Stopped']:
        break
    
    time.sleep(10)  # 每10秒查询一次状态

if status == 'Completed':
    print(f"任务完成，结果已存储到S3: {job_status['outputDataConfig']['S3OutputConfig']['S3Uri']}")
else:
    print(f"任务失败，原因: {job_status.get('failureReason', '未知错误')}")

3. 查看S3中的结果 #

任务完成后，你可以直接在指定的S3路径下找到JSON格式的结果文件。文件内容示例：

{
    "content": [
        {
            "type": "text",
            "text": "异步编程的核心概念是让程序在等待耗时操作（如IO、网络请求）时，不阻塞主线程，转而执行其他任务..."
        }
    ],
    "role": "assistant",
    "stop_reason": "end_turn",
    "usage": {
        "input_tokens": 22,
        "output_tokens": 187
    }
}

内容的提问来源于stack exchange，提问作者Chedva