如何对Amazon Bedrock发起异步调用？Python Lambda层实现方案

你当前代码用了async/await但仍串行执行，核心原因是boto3的客户端是同步的，bedrock.invoke_model是阻塞调用，asyncio.gather无法发挥并行作用。以下是基于aioboto3的可行解决思路及代码实现：

1. 依赖准备 #

在Lambda层中安装aioboto3和aiobotocore，注意匹配Lambda运行时的Python版本（比如Python 3.11），避免环境兼容问题。

2. 异步客户端改造 #

用aioboto3创建异步的Bedrock Runtime客户端，替换原有的boto3同步客户端，通过async with管理客户端生命周期，确保资源正确释放。

3. 完整并行调用代码示例 #

import asyncio
import json
import aioboto3

# 提前定义好AWS认证信息及模型ID
aws_access_key = "你的密钥ID"
aws_secret_access_key = "你的密钥"
aws_session_token = "你的会话令牌"
aws_region = "你的区域"
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

async def _into_comment(segments: list[str]):
    # 创建异步Bedrock Runtime客户端
    async with aioboto3.client(
        service_name="bedrock-runtime",
        aws_access_key_id=aws_access_key,
        aws_secret_access_key=aws_secret_access_key,
        aws_session_token=aws_session_token,
        region_name=aws_region
    ) as bedrock:
        async def sum_up(segment: str):
            body = json.dumps({
                "max_tokens": 256,
                "messages": [{"role": "user", "content": f"Summarize this: {segment}"}],
                "anthropic_version": "bedrock-2023-05-31"
            })
            # 调用异步版本的invoke_model
            response = await bedrock.invoke_model(body=body, modelId=model_id)
            # 解析响应（按需调整处理逻辑）
            return json.loads(response['body'].read())
        
        # 并行执行所有总结任务
        summaries = await asyncio.gather(*[sum_up(segment) for segment in segments])
        return summaries

# Lambda环境若需要解决事件循环嵌套问题，可保留以下代码
# import nest_asyncio
# nest_asyncio.apply()

# 执行调用（Lambda Handler中可直接调用asyncio.run）
# segments = ["文本片段1", "文本片段2", "文本片段3"]
# summaries = asyncio.run(_into_comment(segments))

4. Lambda层部署注意事项 #

• 打包依赖时，需针对Lambda运行时环境编译（可通过Docker模拟Lambda环境打包），避免本地与云端依赖不兼容。
• 确保Lambda执行角色拥有bedrock:InvokeModel权限。
• 注意Bedrock的调用配额，根据自身配额调整并行任务数量，避免触发限流。

5. 替代方案：线程池并行 #

如果不想引入异步依赖，也可以用线程池实现并行调用，适合简单场景：

import json
import boto3
from concurrent.futures import ThreadPoolExecutor

bedrock = boto3.client(
    service_name="bedrock-runtime",
    aws_access_key_id=aws_access_key,
    aws_secret_access_key=aws_secret_access_key,
    aws_session_token=aws_session_token,
    region_name=aws_region
)

def sum_up(segment: str):
    body = json.dumps({
        "max_tokens": 256,
        "messages": [{"role": "user", "content": f"Summarize this: {segment}"}],
        "anthropic_version": "bedrock-2023-05-31"
    })
    response = bedrock.invoke_model(body=body, modelId=model_id)
    return json.loads(response['body'].read())

def _into_comment(segments: list[str]):
    # 控制线程数，避免超过Bedrock配额
    with ThreadPoolExecutor(max_workers=5) as executor:
        summaries = list(executor.map(sum_up, segments))
    return summaries

# summaries = _into_comment(segments)

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