创建带有代码解释器的代理
在这个笔记本中,我们将使用新的代码解释器功能为Amazon Bedrock创建一个代理。代码解释器是一个特殊的预定义工具(动作组),它为模型提供了一个沙盒环境,在这个环境中它可以执行代码(目前是Python),使用一组可用的预定义库。
这个例子首先将使用代码解释器来回答数学问题。LLM通常在数学精度方面存在困难,但擅长编写代码,所以代理将编写代码来执行数学计算,使用代码解释器执行,并将结果传回给用户。我们还将展示如何将文件传递给代理,以供聊天处理或使用代码解释进行分析。最后,代理将使用代码解释器编写代码来创建通常无法创建的文件类型,如图表。
示例:
- 创建带有代码解释的代理
- 调用代理询问一些数学问题
- 调用代理传递文件进行聊天
- 调用代理传递文件进行代码解释
- 调用代理绘制图表
- 调用代理创建文档
将构建以下架构:
先决条件
在开始之前,让我们更新botocore和boto3包以确保我们有最新版本
!python3 -m pip install --upgrade -q boto3
!python3 -m pip install --upgrade -q botocore
!python3 -m pip install --upgrade -q awscli
让我们现在检查boto3版本,以确保已安装正确的版本。我们的版本应该大于或等于1.34.139。
import boto3
import botocore
import awscli
print(boto3.__version__)
print(botocore.__version__)
print(awscli.__version__)
接下来,我们想导入支持包并设置logger对象
import json
import time
from io import BytesIO
import uuid
import pprint
import logging
# 设置logger
logging.basicConfig(format='[%(asctime)s] p%(process)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
让我们现在为所需的AWS服务创建boto3客户端
# 获取boto3客户端以供所需的AWS服务
sts_client = boto3.client('sts')
iam_client = boto3.client('iam')
lambda_client = boto3.client('lambda')
bedrock_agent_client = boto3.client('bedrock-agent')
bedrock_agent_runtime_client = boto3.client('bedrock-agent-runtime')
接下来,我们可以为代理和要创建的lambda函数设置一些配置变量
session = boto3.session.Session()
region = session.region_name
account_id = sts_client.get_caller_identity()["Account"]
region, account_id
# 配置变量
suffix = f"{region}-{account_id}"
agent_name = "assistant-w-code-interpret"
agent_bedrock_allow_policy_name = f"{agent_name}-ba-{suffix}"
agent_role_name = f'AmazonBedrockExecutionRoleForAgents_{agent_name}'
agent_foundation_model = "anthropic.claude-3-sonnet-20240229-v1:0"
agent_description = "具有代码解释器的助手,可以编写和执行代码来回答问题"
agent_instruction = """
你是一个助手,帮助客户回答问题和创建文档。
你可以访问代码解释器来执行Python代码,所以当任务最好通过Python代码来处理时,
编写所需的代码并将其传递给代码解释器执行,然后将结果返回给用户。
"""
agent_alias_name = f"{agent_name}-alias"
创建合成股票价格数据
我们将使用一个CSV文件,其中包含虚构公司’FAKECO’的历史价格数据;我们在这里创建它。
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def make_synthetic_stock_data(filename):
# 定义开始和结束日期
start_date = datetime(2023, 6, 27)
end_date = datetime(2024, 6, 27)
# 创建日期范围
date_range = pd.date_range(start_date, end_date, freq='D')
# 初始化列表以存储数据
symbol = []
dates = []
open_prices = []
high_prices = []
low_prices = []
close_prices = []
adj_close_prices = []
volumes = []
# 设置初始股价
initial_price = 100.0
# 生成合理的股价
for date in date_range:
symbol.append('FAKECO')
dates.append(date)
open_price = np.round(initial_price + np.random.uniform(-1, 1), 2)
high_price = np.round(open_price + np.random.uniform(0, 5), 2)
low_price = np.round(open_price - np.random.uniform(0, 5), 2)
close_price = np.round(np.random.uniform(low_price, high_price), 2)
adj_close_price = close_price
volume = np.random.randint(1000, 10000000)
open_prices.append(open_price)
high_prices.append(high_price)
low_prices.append(low_price)
close_prices.append(close_price)
adj_close_prices.append(adj_close_price)
volumes.append(volume)
initial_price = close_price
# 创建一个DataFrame
data = {
'Symbol': symbol,
'Date': dates,
'Open': open_prices,
'High': high_prices,
'Low': low_prices,
'Close': close_prices,
'Adj Close': adj_close_prices,
'Volume': volumes
}
stock_data = pd.DataFrame(data)
# 保存数据帧
stock_data.to_csv(filename, index=False)
# 确保输出目录存在
import os
if not os.path.exists('output'):
os.makedirs('output')
stock_file = os.path.join('output', 'FAKECO.csv')
if not os.path.exists(stock_file):
make_synthetic_stock_data(stock_file)
创建代理
我们现在将创建代理。为此,我们首先需要创建允许特定基础模型的Bedrock模型调用的代理策略,以及与该策略关联的代理IAM角色。
# 为代理创建IAM策略
bedrock_agent_bedrock_allow_policy_statement = {
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AmazonBedrockAgentBedrockFoundationModelPolicy",
"Effect": "Allow",
"Action": "bedrock:InvokeModel",
"Resource": [
f"arn:aws:bedrock:{region}::foundation-model/{agent_foundation_model}"
]
}
]
}
bedrock_policy_json = json.dumps(bedrock_agent_bedrock_allow_policy_statement)
agent_bedrock_policy = iam_client.create_policy(
PolicyName=agent_bedrock_allow_policy_name,
PolicyDocument=bedrock_policy_json
)
# 为代理创建IAM角色并附加IAM策略
assume_role_policy_document = assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {
"Service": "bedrock.amazonaws.com"
},
"Action": "sts:AssumeRole"
}]
}
assume_role_policy_document_json = json.dumps(assume_role_policy_document)
agent_role = iam_client.create_role(
RoleName=agent_role_name,
AssumeRolePolicyDocument=assume_role_policy_document_json
)
# 暂停以确保角色已创建
time.sleep(10)
iam_client.attach_role_policy(
RoleName=agent_role_name,
PolicyArn=agent_bedrock_policy['Policy']['Arn']
)
创建Bedrock代理
一旦创建了所需的IAM角色,我们就可以使用Bedrock Agent客户端创建一个新的代理。为此,我们使用create_agent函数。它需要代理名称、基础模型和说明。我们还可以提供代理描述。请注意,创建的代理尚未准备就绪。稍后,我们将准备并使用该代理。
我们无法在创建时将代理设置为使用代码解释器;因为代码解释器是一个特殊的动作组,所以在下面创建动作组时完成。
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
让我们现在将代理ID存储在一个本地变量中,以便在后续步骤中使用。
agent_id = response['agent']['agentId']
agent_id
创建代理动作组
在Bedrock代理中,动作组定义了代理可以使用的工具。我们现在将创建一个代理动作组,为代理提供代码解释器,这是一个用于评估代码的运行时环境。动作组还可以定义其他工具,如lambda函数,并可以定义一个渠道,让模型向用户征求澄清输入(将用户视为模型可以调用的工具)。但是,我们的动作组只定义了代码解释器。这是通过一个特殊的访问参数parentActionGroupSignature来完成的(参见boto3文档 )
要允许我们的代理生成、运行和调试代码以完成任务,请设置parentActionGroupSignature=AMAZON.CodeInterpreter。对于此动作组,我们必须将description、apiSchema和actionGroupExecutor字段留空。
请注意,我们还可以定义一个动作组,其parentActionGroupSignature设置为特殊值AMAZON.UserInput。如果设置了这个,那么在编排过程中,如果我们的代理确定需要调用动作组中的API,但没有足够的信息来完成API请求,它将改为调用这个动作组,并返回一个Observation来重新提示用户提供更多信息。如果我们知道交互有一个人在循环中,用户输入就是合适的。我们这里不这样做。
# 暂停以确保代理已创建
time.sleep(30)
# 现在,我们可以在这里配置和创建一个动作组:
# 为代理启用代码解释
agent_action_group_response = bedrock_agent_client.create_agent_action_group(
agentId=agent_id,
agentVersion='DRAFT',
actionGroupName='code-interpreter',
parentActionGroupSignature='AMAZON.CodeInterpreter',
actionGroupState='ENABLED'
)
agent_action_group_response
准备代理
让我们创建一个可用于内部测试的代理的DRAFT版本。
response = bedrock_agent_client.prepare_agent(
agentId=agent_id
)
print(response)
# 暂停以确保代理已准备就绪
time.sleep(30)
# 从响应中提取agentAliasId
agent_alias_id = "TSTALIASID"
调用代理
我们现在将定义一个助手函数来调用代理并解析其响应,然后调用它来查看它使用代码调用。
定义代理调用助手函数
这个助手函数可以调用我们的代理并解析返回的响应流。
注意:这个助手函数与类似示例中使用的助手函数有所不同,它还定义了一个show_code_use参数,如果代理调用了代码解释器,这个参数会导致助手打印一条消息,并且还将事件流解析分离到另一个助手函数process_response中。后来在笔记本中,我们将用一个更丰富的版本替换process_response,它可以捕获返回的文件并更丰富地格式化数据
invoke_agent_helper函数允许用户使用session_id向代理发送query。会话定义了用户与代理之间的一轮来回对话。代理可以记住会话内的全部上下文。一旦用户结束会话,这个上下文就会被删除。
用户然后可以决定是否使用enable_trace布尔变量启用跟踪,并通过session_state变量传递会话状态作为字典。
如果提供了新的session_id,代理将创建一个没有先前上下文的新对话。如果重复使用相同的session_id,代理就会知道与该会话相关的对话上下文。
如果将enable_trace设置为True,代理的每个响应都会附带一个跟踪,详细说明代理正在编排的步骤。它允许我们跟踪代理的(通过思维链提示)推理,从而得出最终响应。
为了处理内存能力,使用了memory_id参数。一旦会话结束,它将将内容总结为新的会话ID作为memory_id的一部分。
我们还可以使用session_state参数传递会话上下文。会话状态允许我们与代理共享以下信息:
sessionAttributes: 在用户和代理之间