AWS下满足数据居留需求的RAG实现

主要重点

在这篇文章中，我们探讨了如何在满足数据居留要求的情况下，使用AWS的混合和边缘服务扩展Amazon BedrockAgents，以构建分布式的检索增强生成（RAG）应用程序。我们介绍了完全本地RAG和混合RAG的解决方案，并强调了这些架构如何支持受数据隐私法规约束的行业使用生成AI。

随著的正式推出，企业能够迅速开发在多个系统和数据源中运行的生成AI应用。然而，在某些地理区域和受监管行业中，由于数据保护和隐私法规的限制，企业希望将云中的生成AI服务与本地的受管数据结合使用。本文展示了如何将AmazonBedrock Agents扩展到混合和边缘服务，比如和，以便构建使用本地数据的分布式RAG应用，从而提升模型的性能表现。

解决方案概述

对于处理或存储敏感信息（例如个人识别信息，PII）的组织，客户要求能够满足这些特定要求，并提供确保数据按照当地法律和法规存储及处理的机制。通过AWS的混合和边缘服务，例如LocalZones和Outposts，您可以利用AWS云的可扩展性和灵活性，以及本地基础设施的低延迟和处理能力。这种混合方法使组织能够在数据源附近运行应用程序，减少延迟，提升对于时间敏感型工作负载的响应能力，并遵守数据法规。

尽管针对数据居留架构的和的讨论已经广泛存在，但生成AI和基础模型（FM）引入了一系列额外的架构考量。随著生成AI模型日益强大和普及，客户询问我们如何在更接近创建和使用数据的设备、传感器和最终用户的位置部署这些模型。此外，小型语言模型（SLM）在资源受限的设备上执行自然语言处理和预测自动化等复杂功能的兴趣也在增长。要了解客户使用SLM的更多机会，见我们AWS行业博客的。

除了SLM，边缘生成AI的兴趣主要源于以下两个因素：

在接下来的部分，我们将深入探讨两种主要的架构模式：完全本地RAG和混合RAG。

完全本地RAG

在Outposts机架上部署大型语言模型（LLM）以应用于RAG时，LLM将在上自我托管，知识库将使用Amazon Elastic Block Storage (AmazonEBS)或在Outpost机架上创建。上传到Rack中的知识库的文档可能是私密和敏感的，因此不会转移到AWS区域，而是完全保留在Outpost机架上。您可以使用在AmazonElastic Compute Cloud (AmazonEC2)上托管的本地向量数据库，或使用在Outpost机架上的Amazon RelationalDatabase Service (Amazon RDS) forPostgreSQL及扩展来存储嵌入。以下图例展示了例子。

删除)

混合RAG

某些客户受数据保护或隐私法规要求将数据保留在特定的州界内。为了符合这些要求并保持使用生成AI所需的数据，拥有混合和边缘环境的客户需要在Region和边缘同时托管其FM。这种设置使您能够将数据用于生成目的并保持合规性。为了协调这种分布式系统的行为，您需要一个能理解提示的细微差别的系统，并指引您到正确的在合规环境下运行的FM。AmazonBedrock Agents使这种混合系统中的分布式系统成为可能。

Amazon BedrockAgents使您能够在应用中构建和配置自动代理。代理协同FM、数据源、软件应用程序和用户对话之间的互动。这种协同作用包括调用函数来调用其他FM的能力，这使得在边缘运行自我管理的FM成为可能。通过这种机制，您可以为高度受管行业建构分布式RAG应用，以符合数据居留要求。在混合部署场景中，根据客户的提示，可以在指定的Region执行某些操作，并将其他操作延迟到在LocalZone中自我托管的FM。以下示例说明了混合RAG的高级架构。

删除)

接下来的部分，我们将深入探讨这两种解决方案及其实施方法。

完全本地RAG：解决方案深入剖析

要开始，您需要在Outpost机架上配置虚拟私有云（VPC）及其边缘子网。要在Outpost上创建边缘子网，您需要查找想要创建子网的OutpostAmazon资源名称（ARN）及其可用区。创建互联网网关、路由表和子网关联后，启动一系列EC2实例在Outpost中运行您的RAG应用程序，包括以下组件。

• 向量存储 – 为了支持RAG（检索增强生成），在AWS Outposts的C5系列EC2实例上部署开源向量数据库，例如ChromaDB或Faiss。这个向量数据库将存储文档的向量表示，作为您本地知识库的关键组件。您选择的嵌入模型将用于将文本（包括文档和查询）转换为这些向量表示，以便有效地存储和检索。实际的知识库由原始文本文档及其对应的向量表示组成，这些表示存储在向量数据库中。为了查询这个知识库并生成基于检索结果的响应，您可以使用LangChain将向量搜索检索到的相关文档链接到供给LLM的提示，这种方法可以提升LLM的生成过程，融入当地的领域知识。
• 聊天机器人应用 – 在第二台EC2实例（C5系列）上部署以下两个组件：一个后端服务负责接收提示并将请求传回运行在Outpost上的LLM，以及一个简单的React应用，使用户能够向本地生成AI聊天机器人提问。
• LLM或SLM – 在第三台EC2实例（G4系列）上，部署一个LLM或SLM，以通过受欢迎的框架（如）进行边缘推理。此外，您可以使用将模型部署到本地端点，例如运行在边缘的EC2实例。

选择性地，您的底层专有数据源可以存储在Outposts上的Amazon Simple Storage Service (AmazonS3)或使用在拥有EBS卷的Amazon EC2实例上运行的兼容Amazon S3的解决方案中。

各个组件之间的通讯通过以下图例所示的流量流进行。


工作流程包括以下步骤：

1. 通过前端应用，用户上传将用作知识库的文档，并将其存储在Outpost机架的Amazon EBS中。这些文档由应用进行分片并发送给嵌入模型。
2. 嵌入模型（托管在与本地LLM API推理服务器相同的EC2实例上）将文本片段转换为向量表示。
3. 生成的嵌入被发送到向量数据库并存储，以完成知识库的创建。
4. 通过前端应用，用户向聊天机器人接口发问。
5. 提示被转发到本地LLM API推理服务器实例，该提示被分词并使用本地嵌入模型转换为向量表示。
6. 该问题的向量表示被发送到向量数据库，在那里执行相似性搜索以获取来自知识库的匹配数据。
7. 在本地LLM获得查询和知识库的相关上下文后，将处理该提示，生成响应并将其发回聊天机器人应用。
8. 聊天机器人应用通过其界面向用户展示LLM的响应。

要了解有关完全本地RAG应用程序的更多信息或亲自体验示例应用程序，请参见我们公共AWS工作坊的第2模块：。

混合RAG：解决方案深入剖析

要开始，您需要配置一个具有边缘子网的VPC，这可以是为Outpost机架或LocalZone对应的，具体取决于使用场景。创建互联网网关、路由表和子网关联后，在Outpost机架（或LocalZone）上启动一个EC2实例以运行您的混合RAG应用。对于EC2实例本身，您可以重用完全本地RAG中的相同组件：向量存储、后端API服务器、嵌入模型和本地LLM。

在此架构中，我们高度依赖于管理服务，如Lambda和AmazonBedrock，因为只有少数选定的FM和知识库需要部署在边缘，而不是协调器本身。为此，我们将扩展现有的Amazon BedrockAgents工作流到边缘，使用一个示例。

在这个示例客户服务机器人中，我们是一家提供鞋类购买客户服务的鞋类零售商机器人，通过类似人类的对话提供选择。我们还假设围绕鞋类制作的知识库是专有的，因此存储在边缘。因此，围绕鞋类制作的问题将由知识库中的数据和在边缘运行的本地FM进行回复。

为了确保用户提示有效地转发到正确的FM，我们依赖于Amazon BedrockAgents的行动组。行动组定义了代理可以执行的操作，例如place_order或check_inventory。在我们的示例中，我们可以在现有的行动组中定义一个名为hybrid_rag或learn_shoemaking的额外动作组，专门处理只能由AWS混合和边缘位置处理的提示。

作为代理的InvokeAgent API的一部分，代理解释提示（例如“皮革在鞋制作中如何使用？”）并以FM生成其应采取的下一步逻辑，包括对行动组中最审慎行动的预测。在这个例子中，我们希望将提示“您好，我想要一些鞋的推荐。”指向/check_inventory行动组，而将提示“皮革在鞋制作中如何使用？”指向/hybrid_rag行动组。

以下图例显示了这种协同作用，该协同作用由Amazon Bedrock代理的协调阶段实现。


为了创建额外的边缘特定行动组，新的OpenAPI架构必须反映新的行动hybrid_rag，并提供详细的描述、结构和参数，定义该行动在行动组中的API操作，专注于仅在特定边缘位置可用的数据域。

在使用OpenAPI规范定义行动组后，您可以定义一个Lambda函数来编程该行动组的业务逻辑。下面的这段Lambda处理器代码可能包括支持函数（如queryEdgeModel），用于每个行动组对应的业务逻辑。

logger.info('API Path') logger.info(api_path)

    
    
    if api_path == '/customer/{CustomerName}':
        parameters = event['parameters']
        for parameter in parameters:
            if parameter["name"] == "CustomerName":
                cName = parameter["value"]
        body = return_customer_info(cName)
    elif api_path == '/place_order':
        parameters = event['parameters']
        for parameter in parameters:
            if parameter["name"] == "ShoeID":
                id = parameter["value"]
            if parameter["name"] == "CustomerID":
                cid = parameter["value"]
        body = place_shoe_order(id, cid)
    elif api_path == '/check_inventory':
        body =