Amazon Bedrock 代理现在支持记忆保留和代码解释（预览版）



借助 Amazon Bedrock 代理，生成式人工智能（AI）应用程序可以在不同的系统和数据来源上运行多步任务。几个月前，我们简化了代理的创建和配置。目前，我们在预览版中引入了两项新的完全托管功能：

在多次互动间保留记忆 – 代理现在可以保留与每位用户的对话摘要，并能够提供流畅的自适应体验，特别是对于复杂的多步骤任务，例如面向用户的互动和企业自动化解决方案（例如预订航班或处理保险索赔）。

支持代码解释 – 代理现在可以在安全的沙盒环境中动态生成和运行代码片段，并能够处理复杂的用例，例如数据分析、数据可视化、文本处理、求解方程和优化问题。为了便于使用此功能，我们还增加了直接向代理上传文档的功能。

让我们更详细地看看这些新功能是如何工作的。

多次互动间的记忆保留
借助记忆保留，您可以构建能够随着时间的推移学习和适应每位用户的独特需求和偏好的代理。通过保持持久记忆，代理可以从用户中断的地方继续工作，从而提供顺畅的对话和工作流程，尤其是对于复杂的多步骤任务。

想象一下用户正在预订航班。由于能够保留记忆，代理可以了解他们的旅行偏好，并利用这些知识来简化后续的预订请求，从而创造个性化和高效的体验。例如，它可以自动向用户推荐合适的座位或与用户之前选择的餐食相似的餐食。

使用记忆保留来提高上下文感知能力还可以简化业务流程自动化。例如，企业用来处理客户反馈的代理现在可以了解与同一客户之前进行的互动和正在进行的互动，而无需处理自定义集成。

每个用户的对话历史记录和上下文都安全地存储在唯一的记忆标识符（ID）下，确保用户之间的完全隔离。借助记忆保留，可以更轻松地构建提供无缝、自适应和个性化体验的代理，体验也会随着时间的推移不断改进。我们来看看该功能的实际应用。

在 Amazon Bedrock 代理中使用记忆保留
在 Amazon Bedrock 控制台中，我从导航窗格的生成器工具部分选择代理，然后开始创建代理。

对于代理，我使用 agent-book-flight 作为名称，并使用以下内容作为描述：

帮助预订航班。

然后，在代理生成器中，我选择 Anthropic 的 Claude 3 Sonnet 模型并输入以下指令：

要预订航班，您应该知道出发地和目的地机场以及航班起飞的日期和时间。

在“其他设置”中，我启用了用户输入，允许代理提出澄清性问题以获取必要的输入。当预订航班的请求遗漏了一些必要的信息（例如出发地和目的地或航班的日期和时间）时，这将有所帮助。

在新的“记忆”部分中，我允许记忆在每次会话结束时生成和存储会话摘要，并使用默认的 30 天作为记忆持续时间。

然后，我添加了一个操作组来搜索和预订航班。我使用 search-and-book-flights 作为名称，并使用以下描述：

搜索给定日期两个目的地之间的航班，并预订特定的航班。

然后，我选择通过函数详细信息定义操作组，然后创建新的 Lambda 函数。Lambda 函数将为该操作组中的所有函数实现业务逻辑。

我为该操作组添加了两个函数：一个用于搜索航班，另一个用于预订航班。

第一个函数是 search-for-flights，其描述如下：

搜索给定日期在两个目的地之间的航班。

此函数的所有参数都是必需的，类型为字符串。以下是参数的名称和描述：

origin_airport – 出发地 IATA 机场代码
destination_airport – 目的地 IATA 机场代码
date – 航班的日期，格式为 YYYYMMDD

第二个函数是 book-flight，其使用以下描述：

预定给定日期和时间两个目的地之间的航班。

同样，所有参数都是必填的，类型为字符串。以下是参数的名称和描述：

origin_airport – 出发地 IATA 机场代码
destination_airport – 目的地 IATA 机场代码
date – 航班的日期，格式为 YYYYMMDD
time – 航班的时间，格式为 HHMM

要完成代理的创建，我选择“创建”。

要访问 Lambda 函数的源代码，我选择 search-and-book-flights 操作组，然后选择“查看”（在“选择 Lambda 函数”设置附近）。通常，我会使用该 Lambda 函数与现有系统（例如旅行预订平台）集成。在这种情况下，我使用以下代码来模拟代理的预订平台。

import json
import random
from datetime import datetime, time, timedelta


def convert_params_to_dict(params_list):
    params_dict = {}
    for param in params_list:
        name = param.get("name")
        value = param.get("value")
        if name is not None:
            params_dict[name] = value
    return params_dict


def generate_random_times(date_str, num_flights, min_hours, max_hours):
    # 根据输入日期设置种子
    seed = int(date_str)
    random.seed(seed)

    # 将 min_hours 和 max_hours 转换为分钟
    min_minutes = min_hours * 60
    max_minutes = max_hours * 60

    # 生成随机时间
    random_times = set()
    while len(random_times) < num_flights:
        minutes = random.randint(min_minutes, max_minutes)
        hours, mins = divmod(minutes, 60)
        time_str = f"{hours:02d}{mins:02d}"
        random_times.add(time_str)

    return sorted(random_times)


def get_flights_for_date(date):
    num_flights = random.randint(1, 6) # 每天 1 到 6 个航班之间
    min_hours = 6 # 6am
    max_hours = 22 # 10pm
    flight_times = generate_random_times(date, num_flights, min_hours, max_hours)
    return flight_times
    
    
def get_days_between(start_date, end_date):
    # 将字符串日期转换为日期时间对象
    start = datetime.strptime(start_date, "%Y%m%d")
    end = datetime.strptime(end_date, "%Y%m%d")
    
    # 计算两个日期之间的天数
    delta = end - start
    
    # 生成开始时间和结束时间（包括在内）之间的所有日期的列表
    date_list = [start + timedelta(days=i) for i in range(delta.days + 1)]
    
    # 将日期时间对象转换回 “YYYYMMDD” 字符串格式
    return [date.strftime("%Y%m%d") for date in date_list]


def lambda_handler(event, context):
    print(event)
    agent = event['agent']
    actionGroup = event['actionGroup']
    function = event['function']
    param = convert_params_to_dict(event.get('parameters', []))

    if actionGroup == 'search-and-book-flights':
        if function == 'search-for-flights':
            flight_times = get_flights_for_date(param['date'])
            body = f"On {param['date']} (YYYYMMDD), these are the flights from {param['origin_airport']} to {param['destination_airport']}:\n{json.dumps(flight_times)}"
        elif function == 'book-flight':
            body = f"Flight from {param['origin_airport']} to {param['destination_airport']} on {param['date']} (YYYYMMDD) at {param['time']} (HHMM) booked and confirmed."
        elif function == 'get-flights-in-date-range':
            days = get_days_between(param['start_date'], param['end_date'])
            flights = {}
            for day in days:
                flights[day] = get_flights_for_date(day)
            body = f"These are the times (HHMM) for all the flights from {param['origin_airport']} to {param['destination_airport']} between {param['start_date']} (YYYYMMDD) and {param['end_date']} (YYYYMMDD) in JSON format:\n{json.dumps(flights)}"
        else:
            body = f"Unknown function {function} for action group {actionGroup}."
    else:
        body = f"Unknown action group {actionGroup}."
    
    # 按照代理的预期确定输出的格式
    responseBody =  {
        "TEXT": {
            "body": body
        }
    }

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

    }

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

    return function_response

我准备好代理以在控制台中对其进行测试，并问下面的问题：

2024 年 7 月 20 日从伦敦希思罗机场飞往罗马菲乌米奇诺机场的航班有哪些？

代理回复了时间清单。我选择 “显示跟踪” 以获取有关代理如何处理我的指令的详细信息。

在 “跟踪” 选项卡中，我浏览了跟踪步骤，以了解代理编排所使用的思维链。例如，我在这里看到代理在调用 Lambda 函数之前处理了机场名称到代码的转换（伦敦希思罗机场为 LHR，罗马菲乌米奇诺机场为 FCO）。

在新的 “记忆” 选项卡中，我看到了记忆的内容。控制台使用特定的测试记忆 ID。在应用程序中，为了保持每个用户的记忆分离，我可以为每个用户使用不同的记忆 ID。

我查看了航班清单并要求预订一个航班：

预定下午 6:02 的航班。

代理回复正在确认预订。

几分钟后，在会话到期后，我在 “记忆” 选项卡中看到我的对话摘要。

我选择扫把图标以从新的对话开始，然后问一个本身并不能为代理提供完整上下文的问题：

我的航班起飞当天还有哪些其他航班？

代理回忆起我们上次谈话时我预订的航班。为了给我一个答案，代理要求我确认航班详情。请注意，Lambda 函数只是一个模拟，没有将预订信息存储在任何数据库中。航班详情是从代理的记忆中找回的。

我确认这些值，并获得当天出发地和目的地相同的其他航班的清单。

是的。

为了更好地展示记忆保留的优点，让我们使用适用于 Python 的 Amazon SDK (Boto3) 来调用代理。为此，我首先需要创建代理别名和版本。我记下代理 ID 和别名 ID，因为调用代理时需要它们。

在代理调用中，我添加了新的 memoryID 选项来使用记忆。加入这个选项会带来两点好处：

• 代理使用为该 memoryID 保留的记忆（如果有）来改善其响应。
• 将为该 MemoryID 保留当前会话的对话摘要，以便可以在另一个会话中使用。

通过使用 AWS 开发工具包，我还可以获取内容或删除特定 memoryId 的记忆内容。

import random
import string
import boto3
import json

DEBUG = False # 启用调试以查看所有跟踪步骤
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"

AGENT_ID = 'URSVOGLFNX'
AGENT_ALIAS_ID = 'JHLX9ERCMD'

SESSION_ID_LENGTH = 10
SESSION_ID = "".join(
    random.choices(string.ascii_uppercase + string.digits, k=SESSION_ID_LENGTH)
)

# 每个用户的唯一标识符
MEMORY_ID = 'danilop-92f79781-a3f3-4192-8de6-890b67c63d8b' 
bedrock_agent_runtime = boto3.client('bedrock-agent-runtime', region_name='us-east-1')


def invoke_agent(prompt, end_session=False):
    response = bedrock_agent_runtime.invoke_agent(
        agentId=AGENT_ID,
        agentAliasId=AGENT_ALIAS_ID,
        sessionId=SESSION_ID,
        inputText=prompt,
        memoryId=MEMORY_ID,
        enableTrace=DEBUG,
        endSession=end_session,
    )

    completion = ""

    for event in response.get('completion'):
        if DEBUG:
            print(event)
        if 'chunk' in event:
            chunk = event['chunk']
            completion += chunk['bytes'].decode()

    return completion


def delete_memory():
    try:
        response = bedrock_agent_runtime.delete_agent_memory(
            agentId=AGENT_ID,
            agentAliasId=AGENT_ALIAS_ID,
            memoryId=MEMORY_ID,
        )
    except Exception as e:
        print(e)
        return None
    if DEBUG:
        print(response)


def get_memory():
    response = bedrock_agent_runtime.get_agent_memory(
        agentId=AGENT_ID,
        agentAliasId=AGENT_ALIAS_ID,
        memoryId=MEMORY_ID,
        memoryType='SESSION_SUMMARY',
    )
    memory = ""
    for content in response['memoryContents']:
        if 'sessionSummary' in content:
            s = content['sessionSummary']
            memory += f"Session ID {s['sessionId']} from {s['sessionStartTime'].strftime(DATE_FORMAT)} to {s['sessionExpiryTime'].strftime(DATE_FORMAT)}\n"
            memory += s['summaryText'] + "\n"
    if memory == "":
        memory = "<no memory>"
    return memory


def main():
    print("Delete memory? (y/n)")
    if input() == 'y':
        delete_memory()

    print("Memory content:")
    print(get_memory())

    prompt = input('> ')
    if len(prompt) > 0:
        print(invoke_agent(prompt, end_session=False)) # 启动新会话
        invoke_agent('end', end_session=True) # 结束会话

if __name__ == "__main__":
    main()

我从笔记本电脑运行 Python 脚本。我选择删除当前记忆（即使记忆目前应该是空的），然后要求预定特定日期早上的航班。

删除记忆？（y/n）
y
记忆内容：
<无记忆>
> 为我预订 2024 年 7 月 20 日从 LHR 到 FCO 的早间航班。
我已经为你预订了 2024 年 7 月 20 日 06:44 从伦敦希思罗机场（LHR）飞往罗马菲乌米奇诺机场（FCO）的早间航班。

我等了几分钟，然后再次运行脚本。该脚本每次运行时都会创建一个新会话。这次，我不删除记忆，并看到具有同一 memoryid 的之前互动的摘要。然后，我问我的航班预定在哪一天。尽管这是一个新的会话，代理会在记忆的内容中找到之前的预订。

删除记忆？（y/n）
n
记忆内容：
Session ID MM4YYW0DL2 from 2024-07-09 15:35:47 to 2024-07-09 15:35:58
该用户的目标是预定 2024 年 7 月 20 日从 LHR 到 FCO 的早间航班。该助理预订了要求的日期（2024 年 7 月 20 日）当天从 LHR 飞往 FCO 的 0644 次早间航班。助手成功为用户预订了所要求的早间航班。该用户要求预订 2024 年 7 月 20 日从伦敦希思罗机场（LHR）飞往罗马菲乌米奇诺机场（FCO）的早间航班。助手预订了指定路线和日期的 0644 次航班。

> 我的航班在哪一天？
我记得我们之前的谈话中，你预订了 2024 年 7 月 20 日从伦敦希思罗机场（LHR）飞往罗马菲乌米奇诺机场（FCO）的早间航班。请确认 2024 年 7 月 20 日这个日期是否是您要求的正确航班日期？

是的，那是我的航班！

根据您的使用情况，记忆保留可以帮助跟踪来自同一用户的先前互动和偏好，并提供跨会话的无缝体验。

会话摘要包括一般概述以及用户和助手的观点。对于像这样的简短会话来说，这可能会导致一些重复。

代码解释支持
Amazon Bedrock 代理现在支持代码解释，因此代理可以在安全的沙盒环境中动态生成和运行代码片段，从而显著扩展了他们可以处理的用例，包括数据分析、可视化、文本处理、方程求解和优化问题等复杂任务。

代理现在能够处理具有不同数据类型和格式的输入文件，包括 CSV、XLS、YAML、JSON、DOC、HTML、MD、TXT 和 PDF。代码解释还支持代理生成图表，从而增强用户体验并使数据解释更易于使用。

当大型语言模型（LLM）确定代码解释可以帮助更准确地解决特定问题并且有意不支持用户请求生成任意代码的场景时，代理就会使用代码解释。为了安全起见，为每个用户会话提供了一个隔离的、沙盒化的代码运行时环境。

让我们快速测试一下，看看它如何帮助代理处理复杂的任务。

在 Amazon Bedrock 代理中使用代码解释
在 Amazon Bedrock 控制台中，我选择与之前的演示（agent-book-flight）中的代理相同的代理，然后选择在代理生成器中编辑。在代理生成器中，启用“其他设置”下的“代码解释器”并保存。

我准备好代理并直接在控制台中对其进行测试。首先，我问一个数学问题。

计算前 10 个素数的总和。

几秒钟后，我从代理那里得到答案：

前 10 个素数的总和为 129。

答案是准确的。查看跟踪，代理生成并运行了下面的 Python 程序来计算我的问题：

import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

primes = []
n = 2
while len(primes) < 10:
    if is_prime(n):
        primes.append(n)
    n += 1
    
print(f"前 10 个素数是： {primes}")
print(f"前 10 个素数的总和是：{sum(primes)}")

现在，让我们回过头来谈谈 agent-book-flight 代理。我想要更多的了解一个较长时间段内可以预定的全部航班。为此，我首先向同一个操作组添加一个新函数，以获取日期范围内可以预定的所有航班。

我将新函数命名为 get-flights-in-date-range，并使用以下描述：

获取日期范围内每天两个目的地之间的所有航班。

所有参数都是必需的，类型为字符串。以下是参数名称和描述：

origin_airport – 出发地 IATA 机场代码
destination_airport – 目的地 IATA 机场代码
start_date – 航班的开始日期，格式为 YYYYMMDD
end_date – 航班的结束日期，格式为 YYYYMMDD

如果你看一下我之前分享的 Lambda 函数代码，你会发现它已经支持该代理函数。

既然代理有办法通过单个函数调用提取更多信息，我让代理在图表中可视化航班信息数据：

绘制一张图表，列出 2024 年 8 月的前十天每天从 JFK 飞到 SEA 的航班数量。

代理回复中包括一个图表：

我选择链接以将图像下载到我的计算机上：

没错。实际上，Lambda 函数中的模拟器每天生成一到六个航班，如图表中所示。

对附加文件使用代码解释
由于代码解释允许代理处理和提取数据中的信息，因此我们引入了在调用代理时包含文档的功能。例如，我有一个 Excel 文件，其中包含针对不同航班预订的航班数量：

我使用测试界面中的剪辑图标附上文件并提问（代理以粗体回复）：

最受欢迎的航线是什么？ 最不受欢迎的航线又是什么？

根据分析，最受欢迎的航线是 JFK -> SEA，有 921 个预定，最不受欢迎的航线是 FCO -> LHR，有 456 个预订。

总共预订了多少个航班？

所有航线的预订航班总数为 2557 个。

绘制一张图表，比较下这些航线的航班预定数与预定总数对比的百分比。

我可以查看跟踪以了解用于从文件中提取信息并将信息传递给代理的 Python 代码。我可以附加多个文件并使用不同的文件格式。AWS SDK 中提供了这些选项，可让代理在您的应用程序中使用文件。

注意事项
在提供了 Amazon Bedrocks 代理和 Anthropic 的 Claude 3 Sonnet 或 Haiku（预览期间支持的型号）的所有 AWS 区域中，记忆保留以预览版的形式提供。代码解释在美国东部（弗吉尼亚州北部）、美国西部（俄勒冈州）和欧洲地区（法兰克福）区域以预览版的形式提供。

在预览期间，将记忆保留和代码解释用于代理不会产生额外费用。使用包含这些特性的代理时，将收取普通模型使用费。启用记忆保留后，您需要为用于汇总会话的模型付费。有关更多信息，请参阅 Amazon Bedrock 定价页面。

要了解更多信息，请参阅《用户指南》中的“Amazon Bedrock 代理”部分。要深入了解技术内容并了解其他人如何在其解决方案中使用生成式人工智能，请访问 community.aws。

— Danilo

*前述特定亚马逊云科技生成式人工智能相关的服务仅在亚马逊云科技海外区域可用，亚马逊云科技中国仅为帮助您了解行业前沿技术和发展海外业务选择推介该服务。