亚马逊 Bedrock 代理 - 测试代理调用

本笔记本提供了使用 bedrock-agent-runtime 的 boto3 客户端测试代理调用的示例代码。

用例

现在我们已经创建了代理并将知识库附加到它上面，我们将使用运行时客户端使用不同的查询调用代理。Boto3 SDK for agent 分为两个客户端: bedrock-agent 和 bedrock-agent-runtime。bedrock-agent 客户端负责创建、更新、删除和/或准备代理或知识库的功能。而 bedrock-agent-runtime 负责调用代理(使用 <code>invoke_agent</code> API)并从知识库中检索文档(使用 <code>retrieve</code> 和 <code>retrieve_and_generate</code> API)。本笔记本将重点关注代理的运行时调用。我们将使用 invoke_agent API。

笔记本演练

在本笔记本中，我们将:

  • 检索从上一个笔记本中保存的变量
  • 使用知识库和操作组调用新会话中的代理
  • 调用现有会话中的代理
  • 使用提示属性调用代理
  • 启用跟踪调用代理
  • 使用不同语言的请求调用代理

下一步:

在下一个实验室中，我们将清理创建的资源。

先决条件

在开始这个实验室之前，我们需要加载在上一个笔记本中存储的变量。

%store -r

导入所需的包并初始化变量

让我们也导入必要的包，设置日志记录，并初始化 bedrock-agent-runtime 的 boto3 客户端。我们还初始化了 dynamodb 客户端作为支持功能，以检查我们的代理正确执行了任务。

import boto3
import logging
import pprint
import json
import pandas as pd
from agent import invoke_agent_helper
logging.basicConfig(format='[%(asctime)s] p%(process)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
bedrock_agent_runtime_client = boto3.client('bedrock-agent-runtime')
dynamodb = boto3.resource('dynamodb')

创建支持 invokeAgent 函数

让我们再次使用来自 agents.py 的相同支持函数 invoke_agent_helper，以允许我们在启用或禁用跟踪的情况下调用代理，并使用或不使用会话状态。

此函数允许用户使用 session_id 向代理发送 query。用户可以决定使用 enable_trace 布尔变量启用跟踪，并通过 session_state 变量传递会话状态作为字典。

如果提供了新的 session_id，代理将创建一个没有先前上下文的新对话。如果重复使用相同的 session_id，与该会话 id 相关的对话历史记录将可供代理使用。

如果将 enable_trace 设置为 True，每个来自代理的响应都会附带一个跟踪，详细说明代理正在编排的步骤。它允许我们跟踪代理的推理(通过思维链提示)，从而得出当前对话时的最终响应。

最后，我们还可以使用 session_state 参数传递会话上下文。会话状态允许我们与代理共享以下信息:

  • sessionAttributes: 在用户和代理之间的会话中持续存在的属性。所有使用相同 session_id 的 invokeAgent 调用都属于同一个会话，并将与它们共享 sessionAttributes，只要会话时间限制未超过且用户未结束会话。sessionAttributes 在 lambda 函数中可用，但不会添加到代理的提示中。因此，我们只能在 lambda 函数可以处理它们的情况下使用会话属性。我们可以在此处 找到使用会话属性的更多示例。使用 lambda 函数集成实现细粒度访问控制也是一个很好的模式。我们可以在此处 找到一个示例。
  • promptSessionAttributes: 在单个 invokeAgent 调用中持续存在的属性。提示属性被添加到提示和 lambda 函数中。我们还可以在编辑编排基础提示时使用 $prompt_session_attributes$ 占位符。
  • invocationId: 代理在 ReturnControlPayload 对象的 returnControl 字段中返回的 id。如果传递 Return of Control 调用的答案，则需要此字段。我们可以在此处 找到如何使用它的示例。
  • returnControlInvocationResults: 在 Amazon Bedrock 代理之外调用操作获得的结果。如果传递 Return of Control 调用的答案，则需要此字段。我们可以在此处 找到如何使用它的示例。

创建支持 selectAllFromDynamoDB 函数

我们还将创建名为 selectAllFromDynamoDB 的支持函数，以从 dynamoDB 表 restaurant_bookings 中选择所有数据。此函数将用于验证我们代理的行为。

def selectAllFromDynamodb():
    # Get the table object
    table = dynamodb.Table(table_name)

    # Scan the table and get all items
    response = table.scan()
    items = response['Items']

    # Handle pagination if necessary
    while 'LastEvaluatedKey' in response:
        response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
        items.extend(response['Items'])

    items = pd.DataFrame(items)
    return items

# test function invocation
items = selectAllFromDynamodb()
items

使用新会话 ID 调用代理

让我们首先使用 InvokeAgent 函数查询没有先前上下文的知识库。

%%time
import uuid
session_id:str = str(uuid.uuid1())
query = "What is in the childrens menu?"
response = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(response)

使用现有会话 ID 调用代理

接下来，我们可以使用上下文来问一个后续问题。为此，我们使用相同的 session_id 与 invokeAgent 函数。

%%time
query = "Which of those options are vegetarian?"
response = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(response)

如我们所见，代理知道我们正在谈论儿童菜单。

使用操作组调用代理

现在让我们使用我们的代理来预订。通过这样做，我们将要求代理从我们的操作组中执行一个操作来创建一个新的预订。

%%time
query = "Hi, I am Maria. I want to create a booking for 4 people, at 9pm on the 5th of May 2024."
response = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(response)

让我们再次检查数据是否正确添加到 dynamoDB 表中。

selectAllFromDynamodb()

太棒了!我们使用代理创建了一个预订。但是，当在餐厅预订桌子时，我们通常已经登录到一个知道我们名字的系统。如果我们的代理也知道就太好了!

为此，我们可以使用会话上下文向我们的提示提供一些属性。在这种情况下，我们将直接使用 promptSessionAttributes 参数提供它。让我们也启动一个新的会话 id，这样我们的代理就不会记住我们的名字。

%%time
session_id:str = str(uuid.uuid1())
query = "I want to create a booking for 2 people, at 8pm on the 6th of May 2024."
session_state = {
    "promptSessionAttributes": {
        "name": "John"
    }
}
response = invoke_agent_helper(query, session_id, agent_id, alias_id, session_state=session_state)
print(response)

再次让我们验证正确的数据是否添加到 dynamoDB 中。

selectAllFromDynamodb()

我们还可以使用代理的会话信息来验证数据，因为代理知道预订 id 来调用获取预订详细信息功能。

%%time
query = "Get the details for the last booking created"
booking_id = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(booking_id)

删除创建的预订

让我们也测试一下删除预订功能，通过删除最后创建的预订 id。

%%time
query = f"I want to delete the booking."
response = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(response)

我们确认预订也已从 dynamoDB 表中删除。

selectAllFromDynamodb()

使用 promptSessionAttributes 调用代理处理时间信息

在现实生活应用中，上下文非常重要。我们希望在考虑当前日期和周围几天的情况下进行预订。Amazon Bedrock 代理还允许我们使用提示属性为代理提供时间上下文。让我们用预订明天来测试一下。

# retrieving today
from datetime import datetime
today = datetime.today().strftime('%b-%d-%Y')
today

%%time
# reserving a table for tomorrow
session_id:str = str(uuid.uuid1())
query = "I want to create a booking for 2 people, at 8pm tomorrow."
session_state = {
    "promptSessionAttributes": {
        "name": "John",
        "today": today
    }
}
response = invoke_agent_helper(query, session_id, agent_id, alias_id, session_state=session_state)
print(response)

现在，为了确认一切都正确添加，让我们检索最后一个预订的详细信息。

%%time
query = "Could you get the details for the last booking created?"
booking_id = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(booking_id)

让代理提供食物推荐

我们代理的另一个很好的用例是使用其推理能力来请求一些食物推荐。让我们看几个例子。

%%time
session_id:str = str(uuid.uuid1())
query = "What do you have for kids that don't like fries?"
response = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(response)

%%time
session_id:str = str(uuid.uuid1())
query = "I am allergic to shrimps. What can I eat at this restaurant?"
response = invoke_agent_helper(query, session_id, agent_id, alias_id)
print(response)

启用跟踪调用代理

我们还可以在启用跟踪的情况下调用我们的代理，以生成每个步骤的详细信息，这些步骤正在由代理编排。让我们看看在检查知识库中的文档时执行的所有步骤。

%%time
session_id:str = str(uuid.uuid1())
query = "What are the desserts on the adult menu?"
response = invoke_agent_helper(query, session_id, agent_id, alias_id, enable_trace=True)
print(response)

使用不同的提示语言调用代理

我们在本笔记本中突出的最后一个功能是 LLM 模型处理不同语言输入的能力。我们只在英语中实现了我们的代理，但如果它也能处理其他语言就太好了!

好消息是它可以!这是由于 LLM 的多语言功能。最好的部分是，我们不需要在代理中做任何改变，它甚至会用请求的语言进行响应。

让我们试试几个查询!

%%time
# Invoking agents in spanish
session_id:str = str(uuid.uuid1())
query = "¿Podrías reservar una mesa para dos 25/07/2024 a las 19:30"
session_state = {
    "promptSessionAttributes": {
        "Nombre": "Gabriela"
    }
}
response = invoke_agent_helper(query, session_id, agent_id, alias_id, session_state=session_state)
print(response)

%%time
# Invoking agents in german
session_id: