使用 Amazon SageMaker Data Wrangler 简化生成 AI 的数据准备

作者：Ajjay Govindaram 和 Nikita Ivkin | 2023 年 11 月 27 日
在 高级 (300) | | | |

关键要点

• 生成人工智能模型需要大量清晰结构化的训练数据才能发挥其潜力。
• 大多数现实世界数据以非结构化格式存在，处理这些数据是有效利用 AI 的关键。
• 使用 Amazon SageMaker Data Wrangler，可以高效地进行数据准备，从而加速从数据到商业洞察的转化。
• 通过本示例，用户可以了解如何将 PDF 数据集预处理到为 RAG（检索增强生成）模型服务的最终效果。

生成人工智能（）模型在生成高质量文本、图像和其他内容方面展现了惊人的能力。然而，这些模型需要大量清洁和结构化的训练数据才能充分发挥其潜力。大多数真实世界的数据以非结构化格式存在，如 PDF，这需要预处理才能有效使用。

根据 的统计，当前超过 80% 的商业数据都是非结构化数据。这些数据包括电子邮件、PDF、扫描文档、图像、音频和视频等多种格式。虽然这些数据包含宝贵的见解，但其非结构化特性使得 AI算法难以解释和学习。根据 ，仅有 18% 的企业报告能够有效利用非结构化数据。

随着 AI采用率的持续上升，开发高效机制来消化和学习非结构化数据显得愈加重要。这可能涉及更好的预处理工具、半监督学习技术以及自然语言处理的进步。有效利用其非结构化数据的公司将从 AI 中获得显著的竞争优势。清洗数据对良好的模型表现至关重要，提取的文本中仍然包含大量无意义的和通用的文本（例如，阅读 HTML）。从互联网抓取的数据通常有大量重复内容，社交媒体、评论或任何用户生成的内容也可能包含毒性和偏见内容，因此需要通过某些预处理步骤进行过滤。低质量内容或机器人生成的文本也会出现，可以使用相关元数据进行过滤（例如，过滤出客户反馈评价低的客户服务响应）。

在检索增强生成（）模型的多个阶段，数据准备至关重要。知识源文档需要经过预处理，例如清洗文本和生成语义嵌入，以便能够有效地进行索引和检索。用户的自然语言查询同样需要预处理，以便能够编码为向量，并与文档嵌入进行比较。检索到相关上下文后，可能需要进一步的预处理，例如截断，然后与用户的查询拼接，以创建基础模型的最终提示。现在， 支持全面的数据准备功能，由 提供支持。通过这项集成，SageMaker Canvas为客户提供了一个从数据到商业洞察的端到端无代码工作区。用户可以轻松发现和聚合来自50多个数据源的数据，并使用 SageMaker Canvas的可视化界面探索和准备数据，借助超过300个内置分析和转换功能。

解决方案概述

在本篇文章中，我们将处理一个 PDF 文档数据集—— 。此外，我们将展示如何为 RAG预处理数据集。具体而言，我们将清洗数据并创建 RAG文档以回答有关数据集内容的问题。考虑以下机器学习（ML）问题：用户向大型语言模型（LLM）提问：“如何在 Amazon Bedrock中过滤和搜索模型？”。LLM 在训练或微调阶段未见过该文档，因此无法回答这个问题，并且很可能会出现幻觉。我们在本文的目标是从 PDF 中找到相关文本（即 RAG），并将其附加到提示中，从而使 LLM 能够回答与该文档相关的问题。

以下是您可以通过 （由 提供支持）完成的主要预处理步骤：

1. 从 PDF 文档中提取文本（由 Textract 提供支持）
2. 删除敏感信息（由 Comprehend 提供支持）
3. 将文本分割成块
4. 为每块创建嵌入（由 Bedrock 提供支持）
5. 将嵌入上传到矢量数据库（由 OpenSearch 提供支持）

前提条件

在本次操作演示中，您应具备以下条件：

• 一个具有创建 （AWS IAM）策略和角色权限的 AWS 账户
• 访问 、一个 实例，以及 Studio 用户。有关前提条件的更多信息，请参见 。
• 访问 模型。请遵循 指南。
• 访问 。Amazon SageMaker Studio 执行角色必须有权调用 Amazon Comprehend 动作。
• 访问 。Amazon SageMaker Studio 执行角色必须有权调用 Amazon Textract。
• 对 Amazon Simple Storage Service（Amazon ）桶的读写访问。
• 作为矢量数据库访问 。矢量数据库的选择是重要的架构决策，要考虑多个良好的选项，每个选项各有优缺点。在本示例中，我们选择了 Amazon OpenSearch 作为我们的矢量数据库。

注意 ： 根据创建 OpenSearch Service域。为简单起见，选择具有主用户名和密码的选项以实现细粒度访问控制。创建域后，使用以下映射创建矢量索引，矢量维度 1536 对应 Amazon Titan嵌入：

json PUT knowledge-base-index { "settings": { "index.knn": True }, "mappings": { "properties": { "text_content": { "type": "text", "fields": { "keyword": { "type": "keyword" } } }, "text_content_v": { "type": "knn_vector", "dimension": 1536 } } } }

操作演示

构建数据流

在本节中，我们将介绍如何构建数据流以从 PDF 中提取文本和元数据，清洗和处理数据，使用 Amazon Bedrock 生成嵌入，并在 AmazonOpenSearch 中对数据进行索引。

启动 SageMaker Canvas

要启动 SageMaker Canvas，请完成以下步骤：

1. 在 Amazon 中，选择导航面板中的 Domains 。
2. 选择您的域。
3. 在启动菜单上，选择 Canvas 。

![Launch SageMaker删除)

创建数据流

完成以下步骤以在 SageMaker Canvas 中创建数据流：

1. 在 SageMaker Canvas 首页，选择 Data Wrangler 。
2. 在页面右侧选择 Create ，然后给数据流命名，选择 Create 。

![Create Data删除)

1. 这将跳转到数据流页面。
2. 选择 Import data ，选择表格数据。

![Import删除)

现在让我们从 Amazon S3 桶导入数据：

1. 选择 Import data ，从下拉列表中选择 Tabular 。
2. 在 Data Source 中，从下拉列表中选择 Amazon S3 。

![Select Amazon S3 as Data删除)

1. 导航到包含 PDF 文件位置的元数据文件，选择该文件。

![Select Metadata删除)

1. 现在，元数据文件已经加载到数据准备数据流中，我们可以继续添加下一步以转换数据并索引到 Amazon OpenSearch 。在这个例子中，文件具有以下元数据，包含每个文件在 Amazon S3 目录中的位置。

![Metadata删除)

要添加新的转换，请执行以下步骤：

1. 选择加号并选择 Add Transform 。
   ![Add删除)
2. 选择 Add Step 并选择 Custom Transform 。
3. 您可以使用 Pandas、PySpark、Python 自定义函数和 SQL PySpark 创建自定义转换。对于这个用例，选择 Python (PySpark) 。
4. 输入步骤的名称。在示例代码片段中，浏览并选择 从 PDF 中提取文本 。根据需要修改代码片段并选择 Add 。

![Extract Text删除)
![Extract Text删除)

1. 接下来，让我们添加一个步骤，以利用 从提取的数据中删除个人识别信息（PII）。选择 Add Step 并选择 Custom Transform ，然后选择 Python (PySpark) 。

从示例代码片段中，浏览并选择 屏蔽 PII 。根据需要修改代码片段并选择 Add 。

![Mask PII删除)

![Mask PII删除)

1. 下一步是对文本内容进行分块。选择 Add Step 并选择 Custom Transform ，然后选择 Python (PySpark) 。

从示例代码片段中，浏览并选择 分块文本 。根据需要修改代码片段并选择 Add 。

![Chunk Text删除)

![Chunk Text删除)

1. 让我们使用 的 Titan 嵌入模型将文本内容转换为向量嵌入。选择 Add Step 并选择 Custom Transform ，然后选择 Python (PySpark) 。

从示例代码片段中，浏览并选择 使用 Bedrock 生成文本嵌入 。根据需要修改代码片段并选择 Add 。

![Generate Text Embedding with Bedrock删除)

1. 现在我们已经获得了 PDF 文件内容的矢量嵌入。让我们继续将数据索引到 Amazon OpenSearch 中。选择 Add Step 并选择 Custom Transform ，然后选择 Python (PySpark) 。您可以重新编写以下代码以使用您首选的矢量数据库。为简单起见，我们使用主用户名和密码访问 OpenSearch API，对于生产工作负载，请根据您的组织政策选择选项。

text_column = "text_redacted_chunks_embedding" output_column = text_column +
"_response"

headers = {"Content-Type": "application/json", "kbn-xsrf": "true", "osd-xsrf":
"true", "security_tenant": "global"}; index_name = 's3_vector_data_v1'

def index_data(text_redacted_chunks, text_redacted_chunks_embedding):
input_json = json.dumps({"text_content": text_redacted_chunks[-1],
"text_content_v": text_redacted_chunks_embedding[-1]}) response =
requests.request(method="POST", url=f'https://search-canvas-vector-db-domain-
dt3yq3b4cykwuvc6t7rnkvmnka.us-west-2.es.amazonaws.com/{index_name}/_doc',
headers=headers, json=input_json, auth=(master_user, 'master_pass'),
timeout=30) return response.content

indexing_udf = udf(index_data, StringType()) df =
df.withColumn('index_response', indexing_udf(col("text_redacted_chunks"),
col("text_redacted_chunks_embedding"))) ```

![Index Data删除)

最终，创建的数据流如下所示：

![Data FlowDiagram](https://d2908q01vomqb2.cloudfront.net/f1f836cb4ea6efb2a0b1b99f41ad8b103eff4b59/2023/11/27/ML-15986