GenAI > Dify >

餐厅助手 - 创建代理

本笔记本提供了为 Amazon Bedrock 构建附有操作组的代理的示例代码。

用例

我们将创建一个餐厅助手，允许客户创建、删除或获取预订信息。架构如下所示:

代理架构

笔记本演练

在本笔记本中，我们将:

选择代理的基础模型
创建一个 DynamoDB 表来存储预订详细信息
创建一个处理餐厅预订的 Lambda 函数
创建一个代理
创建一个操作组并将其与代理关联
测试代理调用

下一步:

在下一个实验室中，我们将为代理添加知识库，并使用提示属性为代理调用提供额外信息。

先决条件

本笔记本需要以下权限:

创建和删除 Amazon IAM 角色
创建 Lambda 函数
创建 DynamoDB 表
访问 Amazon Bedrock

如果在 SageMaker Studio 上运行，我们应该将以下托管策略添加到我们的角色:

IAMFullAccess
AWSLambda_FullAccess
AmazonBedrockFullAccess
AmazonDynamoDBFullAccess

请确保在 Amazon Bedrock 控制台中启用 Anthropic Claude 3 Sonnet 模型访问，因为笔记本将使用 Anthropic Claude 3 Sonnet 模型来测试创建的代理。

设置

在运行本笔记本的其余部分之前，我们需要运行下面的单元格来确保安装了必要的库。

!pip install --upgrade -q -r requirements.txt

现在让我们导入必要的库并初始化所需的 boto3 客户端。

import time
import boto3
import logging
import ipywidgets as widgets
import uuid

from agent import create_agent_role, create_lambda_role
from agent import create_dynamodb, create_lambda, invoke_agent_helper

#客户端
s3_client = boto3.client('s3')
sts_client = boto3.client('sts')
session = boto3.session.Session()
region = session.region_name
account_id = sts_client.get_caller_identity()["Account"]
bedrock_agent_client = boto3.client('bedrock-agent')
bedrock_agent_runtime_client = boto3.client('bedrock-agent-runtime')
logging.basicConfig(format='[%(asctime)s] p%(process)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
region, account_id

设置代理信息

我们现在将设置定义我们代理的变量:

agent_name: 提供要创建的代理的名称，在本例中为 booking-agent
agent_description: 代理的描述，用于在控制台上显示代理列表。此描述不是代理提示的一部分
agent_instruction: 代理应该和不应该做什么的说明。此描述是代理提示的一部分，在代理调用期间使用
agent_action_group_name: 定义代理操作时使用的操作组名称，在本例中为 TableBookingsActionGroup。
agent_action_group_description: 在 UI 上列出操作组时使用的操作组名称描述。此描述不会被代理提示使用。

suffix = f"{region}-{account_id}"
agent_name = 'booking-agent'
agent_bedrock_allow_policy_name = f"{agent_name}-ba"
agent_role_name = f'AmazonBedrockExecutionRoleForAgents_{agent_name}'

agent_description = "负责餐厅预订的代理"
agent_instruction = """
你是一个餐厅代理,帮助客户检索预订信息、创建新预订或删除现有预订
"""

agent_action_group_description = """
获取餐桌预订信息、创建新预订或删除现有预订的操作"""

agent_action_group_name = "TableBookingsActionGroup"

选择基础模型

使用此下拉菜单选择代理的基础模型。我们可以在此处找到有关支持的基础模型的更多信息。

agent_foundation_model_selector = widgets.Dropdown(
    options=[
        ('Claude 3 Sonnet', 'anthropic.claude-3-sonnet-20240229-v1:0'),
        ('Claude 3 Haiku', 'anthropic.claude-3-haiku-20240307-v1:0')
    ],
    value='anthropic.claude-3-sonnet-20240229-v1:0',
    description='FM:',
    disabled=False,
)
agent_foundation_model_selector

让我们确认模型已正确选择。

agent_foundation_model = agent_foundation_model_selector.value
agent_foundation_model

创建 DynamoDB 表

现在让我们创建一个名为 restaurant_bookings 的 Amazon DynamoDB 表。此表将存储有关预订的信息，包括 booking_id、预订 date、预订人的 name、预订的 hour 和预订人数 num_guests。为此，我们使用 agent.py 文件中的 create_dynamodb 函数。此函数将支持表的创建及其要求(IAM 角色和权限)。

table_name = 'restaurant_bookings'
create_dynamodb(table_name)

创建 Lambda 函数

接下来，我们将创建 AWS Lambda 函数，该函数执行我们代理的操作。此 Lambda 函数将有 3 个操作:

get_booking_details(booking_id): 根据预订 ID 返回预订的详细信息
create_booking(date, name, hour, num_guests): 为餐厅创建新预订
delete_booking(booking_id): 根据预订 ID 删除现有预订

lambda_handler 从代理接收 event，event 包含有关要执行的 function 及其 parameters 的信息。

Lambda 函数返回一个 functionResponse，响应主体包含一个 TEXT 字段。

我们可以在此处找到有关如何设置代理 Lambda 函数的更多信息。

让我们首先将 Lambda 函数的代码写入 lambda_function.py 文件。

%%writefile lambda_function.py
import json
import uuid
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('restaurant_bookings')

def get_named_parameter(event, name):
    """
    从 lambda 事件中获取参数
    """
    return next(item for item in event['parameters'] if item['name'] == name)['value']


def get_booking_details(booking_id):
    """
    检索餐厅预订的详细信息
    
    参数:
        booking_id (string): 要检索的预订 ID
    """
    try:
        response = table.get_item(Key={'booking_id': booking_id})
        if 'Item' in response:
            return response['Item']
        else:
            return {'message': f'没有找到 ID 为 {booking_id} 的预订'}
    except Exception as e:
        return {'error': str(e)}


def create_booking(date, name, hour, num_guests):
    """
    创建新的餐厅预订
    
    参数:
        date (string): 预订日期
        name (string): 预订人姓名
        hour (string): 预订时间
        num_guests (integer): 预订人数
    """
    try:
        booking_id = str(uuid.uuid4())[:8]
        table.put_item(
            Item={
                'booking_id': booking_id,
                'date': date,
                'name': name,
                'hour': hour,
                'num_guests': num_guests
            }
        )
        return {'booking_id': booking_id}
    except Exception as e:
        return {'error': str(e)}


def delete_booking(booking_id):
    """
    删除现有的餐厅预订
    
    参数:
        booking_id (str): 要删除的预订 ID
    """
    try:
        response = table.delete_item(Key={'booking_id': booking_id})
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return {'message': f'ID 为 {booking_id} 的预订已成功删除'}
        else:
            return {'message': f'无法删除 ID 为 {booking_id} 的预订'}
    except Exception as e:
        return {'error': str(e)}
    

def lambda_handler(event, context):
    # 获取在调用 lambda 函数期间使用的操作组
    actionGroup = event.get('actionGroup', '')
    
    # 应该调用的函数名称
    function = event.get('function', '')
    
    # 调用函数的参数
    parameters = event.get('parameters', [])

    if function == 'get_booking_details':
        booking_id = get_named_parameter(event, "booking_id")
        if booking_id:
            response = str(get_booking_details(booking_id))
            responseBody = {'TEXT': {'body': json.dumps(response)}}
        else:
            responseBody = {'TEXT': {'body': '缺少 booking_id 参数'}}

    elif function == 'create_booking':
        date = get_named_parameter(event, "date")
        name = get_named_parameter(event, "name")
        hour = get_named_parameter(event, "hour")
        num_guests = get_named_parameter(event, "num_guests")

        if date and hour and num_guests:
            response = str(create_booking(date, name, hour, num_guests))
            responseBody = {'TEXT': {'body': json.dumps(response)}}
        else:
            responseBody = {'TEXT': {'body': '缺少必需参数'}}

    elif function == 'delete_booking':
        booking_id = get_named_parameter(event, "booking_id")
        if booking_id:
            response = str(delete_booking(booking_id))
            responseBody = {'TEXT': {'body': json.dumps(response)}}
        else:
            responseBody = {'TEXT': {'body': '缺少 booking_id 参数'}}

    else:
        responseBody = {'TEXT': {'body': '无效函数'}}

    action_response = {
        'actionGroup': actionGroup,
        'function': function,
        'functionResponse': {
            'responseBody': responseBody
        }
    }

    function_response = {'response': action_response, 'messageVersion': event['messageVersion']}
    print("Response: {}".format(function_response))

    return function_response

接下来，我们使用 agent.py 文件中的支持函数 create_lambda_role 和 create_lambda 创建 Lambda 函数的要求 IAM 角色和策略，并创建 Lambda 函数。

lambda_iam_role = create_lambda_role(agent_name, table_name)

lambda_function_name = f'{agent_name}-lambda'

lambda_function = create_lambda(lambda_function_name, lambda_iam_role)

创建代理

现在我们已经创建了 DynamoDB 表和 Lambda 函数，让我们创建我们的代理。

为此，我们首先需要创建一个代理角色及其所需的策略。让我们使用 agent.py 文件中的 create_agent_role 函数来实现。

agent_role = create_agent_role(agent_name, agent_foundation_model)

agent_role

创建了代理 IAM 角色后，我们现在可以使用 boto3 函数 <code>create_agent</code> 来创建我们的代理。

在代理创建时，我们只需提供代理名称、基础模型和说明。我们将在创建代理后将操作组与其关联。

response = bedrock_agent_client.create_agent(
    agentName=agent_name,
    agentResourceRoleArn=agent_role['Role']['Arn'],
    description=agent_description,
    idleSessionTTLInSeconds=1800,
    foundationModel=agent_foundation_model,
    instruction=agent_instruction,
)
response

现在我们的代理已经创建好了，我们将检索 agentId。它将用于在下一步中将操作组与代理关联。

agent_id = response['agent']['agentId']
print("The agent id is:",agent_id)

创建代理操作组

现在我们已经创建了代理，让我们创建一个操作组并将其与代理关联。操作组将允许我们的代理执行预订任务。为此，我们将使用定义在 JSON 格式中的函数模式来"告知"我们的代理现有的功能。

函数模式要求提供函数 name、description 和 parameters。每个参数都有一个参数名称、描述、类型和一个布尔标志，指示该参数是否为必需。

让我们将函数 JSON 定义为 agent_functions。

agent_functions = [
    {
        'name': 'get_booking_details'，
        'description': '检索餐厅预订的详细信息',
        'parameters': {
            "booking_id": {
                "description": "要检索的预订 ID"，
                "required": True,
                "type": "string"
            }