使用 Amazon SageMaker Data Wrangler 简化生成 AI 的数据准备
作者：Ajjay Govindaram 和 Nikita Ivkin 2023 年 11 月 27 日在 高级 (300) Amazon SageMaker Amazon SageMaker Data Wrangler 人工智能 生成 AI
关键要点
生成人工智能模型需要大量清晰结构化的训练数据才能发挥其潜力。大多数现实世界数据以非结构化格式存在，处理这些数据是有效利用 AI 的关键。使用 Amazon SageMaker Data Wrangler，可以高效地进行数据准备，从而加速从数据到商业洞察的转化。通过本示例，用户可以了解如何将 PDF 数据集预处理到为 RAG检索增强生成模型服务的最终效果。生成人工智能生成 AI模型在生成高质量文本、图像和其他内容方面展现了惊人的能力。然而，这些模型需要大量清洁和结构化的训练数据才能充分发挥其潜力。大多数真实世界的数据以非结构化格式存在，如 PDF，这需要预处理才能有效使用。
根据 IDC 的统计，当前超过 80 的商业数据都是非结构化数据。这些数据包括电子邮件、PDF、扫描文档、图像、音频和视频等多种格式。虽然这些数据包含宝贵的见解，但其非结构化特性使得 AI 算法难以解释和学习。根据 Deloitte 在 2019 年的一项调查，仅有 18 的企业报告能够有效利用非结构化数据。
随着 AI 采用率的持续上升，开发高效机制来消化和学习非结构化数据显得愈加重要。这可能涉及更好的预处理工具、半监督学习技术以及自然语言处理的进步。有效利用其非结构化数据的公司将从 AI 中获得显著的竞争优势。清洗数据对良好的模型表现至关重要，提取的文本中仍然包含大量无意义的和通用的文本例如，阅读 HTML。从互联网抓取的数据通常有大量重复内容，社交媒体、评论或任何用户生成的内容也可能包含毒性和偏见内容，因此需要通过某些预处理步骤进行过滤。低质量内容或机器人生成的文本也会出现，可以使用相关元数据进行过滤例如，过滤出客户反馈评价低的客户服务响应。
在检索增强生成RAG模型的多个阶段，数据准备至关重要。知识源文档需要经过预处理，例如清洗文本和生成语义嵌入，以便能够有效地进行索引和检索。用户的自然语言查询同样需要预处理，以便能够编码为向量，并与文档嵌入进行比较。检索到相关上下文后，可能需要进一步的预处理，例如截断，然后与用户的查询拼接，以创建基础模型的最终提示。现在，Amazon SageMaker Canvas 支持全面的数据准备功能，由 Amazon SageMaker Data Wrangler 提供支持。通过这项集成，SageMaker Canvas 为客户提供了一个从数据到商业洞察的端到端无代码工作区。用户可以轻松发现和聚合来自50多个数据源的数据，并使用 SageMaker Canvas 的可视化界面探索和准备数据，借助超过300个内置分析和转换功能。
解决方案概述
在本篇文章中，我们将处理一个 PDF 文档数据集Amazon Bedrock 用户指南。此外，我们将展示如何为 RAG 预处理数据集。具体而言，我们将清洗数据并创建 RAG 文档以回答有关数据集内容的问题。考虑以下机器学习ML问题：用户向大型语言模型LLM提问：“如何在 Amazon Bedrock 中过滤和搜索模型？”。LLM 在训练或微调阶段未见过该文档，因此无法回答这个问题，并且很可能会出现幻觉。我们在本文的目标是从 PDF 中找到相关文本即 RAG，并将其附加到提示中，从而使 LLM 能够回答与该文档相关的问题。
以下是您可以通过 Amazon SageMaker Canvas由 Amazon SageMaker Data Wrangler 提供支持完成的主要预处理步骤：
从 PDF 文档中提取文本由 Textract 提供支持删除敏感信息由 Comprehend 提供支持将文本分割成块为每块创建嵌入由 Bedrock 提供支持将嵌入上传到矢量数据库由 OpenSearch 提供支持前提条件
在本次操作演示中，您应具备以下条件：
一个具有创建 AWS 身份和访问管理AWS IAM策略和角色权限的 AWS 账户访问 Amazon SageMaker、一个 Amazon SageMaker Studio 实例，以及 Studio 用户。有关前提条件的更多信息，请参见 使用 Amazon SageMaker Canvas 开始。访问 Amazon Bedrock 模型。请遵循 模型访问 指南。访问 Amazon Comprehend。Amazon SageMaker Studio 执行角色必须有权调用 Amazon Comprehend DetectPiiEntities 动作。访问 Amazon Textract。Amazon SageMaker Studio 执行角色必须有权调用 Amazon Textract。对 Amazon Simple Storage ServiceAmazon S3桶的读写访问。作为矢量数据库访问 Amazon OpenSearch。矢量数据库的选择是重要的架构决策，要考虑多个良好的选项，每个选项各有优缺点。在本示例中，我们选择了 Amazon OpenSearch 作为我们的矢量数据库。注意：根据这里的指南创建 OpenSearch Service 域。为简单起见，选择具有主用户名和密码的选项以实现细粒度访问控制。创建域后，使用以下映射创建矢量索引，矢量维度 1536 对应 Amazon Titan 嵌入：
jsonPUT knowledgebaseindex{ settings { indexknn True } mappings { properties { textcontent { type text fields { keyword { type keyword } } } textcontentv { type knnvector dimension 1536 } } }}
操作演示
构建数据流
在本节中，我们将介绍如何构建数据流以从 PDF 中提取文本和元数据，清洗和处理数据，使用 Amazon Bedrock 生成嵌入，并在 Amazon OpenSearch 中对数据进行索引。
启动 SageMaker Canvas
要启动 SageMaker Canvas，请完成以下步骤：
魔方加速器免费下载安装在 Amazon SageMaker 控制台中，选择导航面板中的 Domains。选择您的域。在启动菜单上，选择 Canvas。创建数据流
完成以下步骤以在 SageMaker Canvas 中创建数据流：
在 SageMaker Canvas 首页，选择 Data Wrangler。在页面右侧选择 Create，然后给数据流命名，选择 Create。 这将跳转到数据流页面。选择 Import data，选择表格数据。现在让我们从 Amazon S3 桶导入数据：
选择 Import data，从下拉列表中选择 Tabular。在 Data Source 中，从下拉列表中选择 Amazon S3。 导航到包含 PDF 文件位置的元数据文件，选择该文件。 现在，元数据文件已经加载到数据准备数据流中，我们可以继续添加下一步以转换数据并索引到 Amazon OpenSearch。在这个例子中，文件具有以下元数据，包含每个文件在 Amazon S3 目录中的位置。要添加新的转换，请执行以下步骤：
选择加号并选择 Add Transform。
选择 Add Step 并选择 Custom Transform。
您可以使用 Pandas、PySpark、Python 自定义函数和 SQL PySpark 创建自定义转换。对于这个用例，选择 Python (PySpark)。输入步骤的名称。在示例代码片段中，浏览并选择 从 PDF 中提取文本。根据需要修改代码片段并选择 Add。 接下来，让我们添加一个步骤，以利用 Amazon Comprehend 从提取的数据中删除个人识别信息PII。选择 Add Step 并选择 Custom Transform，然后选择 Python (PySpark)。从示例代码片段中，浏览并选择 屏蔽 PII。根据需要修改代码片段并选择 Add。
下一步是对文本内容进行分块。选择 Add Step 并选择 Custom Transform，然后选择 Python (PySpark)。从示例代码片段中，浏览并选择 分块文本。根据需要修改代码片段并选择 Add。
让我们使用 Amazon Bedrock 的 Titan 嵌入模型将文本内容转换为向量嵌入。选择 Add Step 并选择 Custom Transform，然后选择 Python (PySpark)。
从示例代码片段中，浏览并选择 使用 Bedrock 生成文本嵌入。根据需要修改代码片段并选择 Add。
现在我们已经获得了 PDF 文件内容的矢量嵌入。让我们继续将数据索引到 Amazon OpenSearch 中。选择 Add Step 并选择 Custom Transform，然后选择 Python (PySpark)。您可以重新编写以下代码以使用您首选的矢量数据库。为简单起见，我们使用主用户名和密码访问 OpenSearch API，对于生产工作负载，请根据您的组织政策选择选项。pythonfrom pysparksqlfunctions import col udffrom pysparksqltypes import StringTypeimport jsonimport requests
textcolumn = textredactedchunksembeddingoutputcolumn = textcolumn response
headers = {ContentType application/json kbnxsrf true osdxsrf true securitytenant global}indexname = s3vectordatav1
def indexdata(textredactedchunks textredactedchunksembedding) inputjson = jsondumps({textcontent textredactedchunks[1] textcontentv textredactedchunksembedding[1]}) response = requestsrequest(method=POST url=fhttps//searchcanvasvectordbdomaindt3yq3b4cykwuvc6t7rnkvmnkauswest2esamazonawscom/{indexname}/doc headers=headers json=inputjson auth=(masteruser masterpass) timeout=30) return responsecontent
indexingudf = udf(indexdata StringType())df = dfwithColumn(indexresponse indexingudf(col(textredactedchunks) col(textredactedchunksembedding)))
最终，创建的数据流如下所示：
