在亚马逊 Bedrock 中使用 Converse API 入门

在这个笔记本中，我们将探索亚马逊 Bedrock 中 Converse API 的基础知识。Converse 或 ConverseStream API 是一个统一的结构化文本 API 操作，允许我们简化对 Bedrock LLM 的调用，使用通用语法和消息结构化提示来支持任何受支持的模型提供商。

让我们从安装或更新 boto3 开始。我们只需要在第一次运行这个单元格。

!pip3 install -qU boto3

import boto3
import sys
print('Running boto3 version:', boto3.__version__)

让我们定义要使用的区域和模型。我们还可以设置我们的 boto3 客户端。

boto3_session = boto3.session.Session()
region = boto3_session.region_name

print('Using region: ', region)

bedrock = boto3.client(
    service_name = 'bedrock-runtime',
    region_name = region,
    )

MODEL_IDS = [
    "amazon.titan-tg1-large",
    "anthropic.claude-3-haiku-20240307-v1:0",
    "anthropic.claude-3-sonnet-20240229-v1:0",
]

我们现在准备在 Bedrock 中设置我们的 Converse API 操作。请注意，我们对任何模型使用相同的语法，包括消息格式化提示和推理参数。还要注意，我们以相同的方式读取输出，与使用的模型无关。

可选地，我们可以定义不通用于所有提供商的其他特定于模型的请求字段。有关更多信息，请查看 Bedrock Converse API 文档。

一次性调用的 Converse

def invoke_bedrock_model(client, id, prompt, max_tokens=2000, temperature=0, top_p=0.9):
    response = ""
    try:
        response = client.converse(
            modelId=id,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "text": prompt
                        }
                    ]
                }
            ],
            inferenceConfig={
                "temperature": temperature,
                "maxTokens": max_tokens,
                "topP": top_p
            }
            #additionalModelRequestFields={
            #}
        )
    except Exception as e:
        print(e)
        result = "Model invocation error"
    try:
        result = response['output']['message']['content'][0]['text'] \
        + '\n--- Latency: ' + str(response['metrics']['latencyMs']) \
        + 'ms - Input tokens:' + str(response['usage']['inputTokens']) \
        + ' - Output tokens:' + str(response['usage']['outputTokens']) + ' ---\n'
        return result
    except Exception as e:
        print(e)
        result = "Output parsing error"
    return result

最后，我们可以测试我们的调用。

在这个例子中，我们在撰写这个示例时 Bedrock 支持的所有文本模型上运行相同的提示。

prompt = ("What is the capital of Italy?")
print(f'Prompt: {prompt}\n')

for i in MODEL_IDS:
    response = invoke_bedrock_model(bedrock, i, prompt)
    print(f'Model: {i}\n{response}')

用于流式调用的 ConverseStream

我们也可以使用 Converse API 进行流式调用。在这种情况下，我们依赖于 ConverseStream 操作。

MODEL_IDS = [
    "amazon.titan-tg1-large",
    "anthropic.claude-3-haiku-20240307-v1:0",
    "anthropic.claude-3-sonnet-20240229-v1:0",
]

def invoke_bedrock_model_stream(client, id, prompt, max_tokens=2000, temperature=0, top_p=0.9):
    response = ""
    response = client.converse_stream(
        modelId=id,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "text": prompt
                    }
                ]
            }
        ],
        inferenceConfig={
            "temperature": temperature,
            "maxTokens": max_tokens,
            "topP": top_p
        }
    )
    # Extract and print the response text in real-time.
    for event in response['stream']:
        if 'contentBlockDelta' in event:
            chunk = event['contentBlockDelta']
            sys.stdout.write(chunk['delta']['text'])
            sys.stdout.flush()
    return

prompt = ("What is the capital of Italy?")
print(f'Prompt: {prompt}\n')

for i in MODEL_IDS:
    print(f'\n\nModel: {i}')
    invoke_bedrock_model_stream(bedrock, i, prompt)

正如我们所看到的，Converse API 允许我们使用相同的语法跨所有模型轻松运行调用。