欢迎来到雲闪世界。全面介绍如何利用 AWS Lambda、Bedrock 和 S3 创建总结会议记录的工作流程
免责声明：本文中使用的会议记录纯属虚构，仅用于作为本文说明和教育目的。它并不反映任何实际的对话、事件或个人。任何与实际人物或事件的相似之处纯属巧合。本文概述的项目涉及使用多种 Amazon Web Services (AWS)，具体来说包括 Amazon S3、Amazon API Gateway、AWS Lambda 和 Amazon Bedrock。
您是否曾在工作场所参加过看似多余甚至毫无意义的会议？对于我们大多数企业员工来说，答案无疑是肯定的。我每周都会参加至少一次这样的会议。有些星期，我的日程安排得满满的，几乎没有时间完成任务。最后，我只能在静音状态下开始处理任务。偶尔，我会错过一些与我有关的重要事情，所以我不得不重新开始全神贯注。这可能会让人非常沮丧，因为在整整一个小时内，可能只有几秒钟的对话与您有关。
随着生成式 AI 应用程序的出现，我再也不用担心自己会陷入这些情况了！在本文中，我将引导读者了解如何利用 AWS Bedrock、AWS Lambda、AWS API Gateway 和 AWS S3 构建可扩展的管道，利用生成式 AI 产品来总结会议记录。
为什么选择 AWS
假设您熟悉 ChatGPT 等生成式 AI 应用程序。既然您可以将文字记录复制并粘贴到 ChatGPT 中并在几秒钟内生成摘要，为什么还要使用 AWS？此外，一些应用程序已经可以做到这一点，如果没有包含在内，您可以将其添加到桌面。这样做的原因有很多，但我将重点介绍其中最重要的两个：可扩展性和安全性。
可扩展性
作为一个云平台，AWS 允许用户、企业、实体等大规模构建应用程序和存储数据。对于我们在本文中讨论的用例，AWS 中的会议记录摘要应用程序可以扩展为在公司内拥有几乎无限数量的最终用户，从会议记录的来源处直接输入到 AWS 的内部管道，并优化记录和生成的摘要的存储成本和容量。此外，您可以自定义提示、输出令牌和许多其他参数。
安全
之前，我提到过，人们可以通过复制和粘贴文字记录来使用 ChatGPT 生成摘要；这会违反许多安全最佳实践，如果关键数据泄露，你甚至可能在某些情况下承担责任。借助 AWS，可以设置必要的权限和防护措施，这样人们就可以充分利用生成式 AI 应用程序，而不必担心公司内部数据穿越公共网络。现在，事不宜迟，让我们直接进入这个项目。
第 1 部分：创建 S3 存储桶
我们将从创建 S3 存储桶开始。S3 是 AWS 提供的对象存储服务，允许用户快速大规模存储数据。它还可以与 AWS 的许多其他服务和第三方应用程序集成。
要导航到 S3，请在搜索栏中搜索它，单击旁边的服务菜单，或者如果您以前使用过它，请在最近访问的部分中找到它。
进入 S3 页面后，单击创建存储桶。
为存储桶命名，其他选项保持不变。如上图所示，我将存储桶命名为bedrock-text-summarization。请注意，所有存储桶名称都必须是全局唯一的，这意味着任何 AWS 账户中都不能有两个存储桶具有相同的名称。
第 2 部分：创建 boto3 层
截至撰写本文时，AWS Lambda 还没有最新版本的 boto3（适用于 Python 的 AWS SDK）。因此，我们必须创建一个 boto3 层并将其上传到 AWS。这确保我们拥有适当的库来调用 AWS Bedrock。这个过程相对简单；但是，如果您从未使用过 Windows 命令提示符，则可能需要一些时间。请注意，以下命令不适用于 Linux 系统。
首先，打开 Windows 命令提示符并分别运行以下命令。如果这是您第一次这样做，安装 boto3 包可能需要一些时间。
## 为您的 boto3 层创建一个新目录
mkdir boto3_layer ## 将其设置为主目录
cd boto3_layer ## 创建一个名为“python”的目录
mkdir python ## 在 python 目录中创建一个虚拟环境
python3 -m venv venv ## 激活虚拟环境
venv/bin/activate ## 将 boto3 安装到环境中
pip install boto3 -t ./python ## 停用虚拟环境
deactivate完成后，我选择按照以下步骤手动压缩 Python 文件：
1. 打开文件资源管理器并导航到包含要压缩的文件夹的目录（在本例中，是我们创建的 boto3_layer 内的 python 文件夹）。
2. 右键单击要压缩的 python 文件夹。
3.从上下文菜单中选择“发送到” 。
4. 点击“压缩文件夹”
获得压缩文件夹后，导航到 Lambda 主页并单击左侧菜单中的“层” 。
您必须命名您的层并从此时上传您的压缩文件夹。此外，请确保选择正确的兼容运行时。在本例中，我将选择最新版本的 Python；但是，如果您计划在脚本中包含早期版本的 Python 中的库/函数/等，您还必须确保已启用它们。
图片由作者提供
第 3 部分：请求 Bedrock 访问权限
Bedrock 是 AWS 内的一个独特服务。简而言之，它是一个 API，允许用户连接到来自领先 AI 公司的许多基础模型。要使用其中任何模型，您都需要请求访问权限。导航到 Bedrock 主页并选择左下角的“模型访问” ：
进入模型访问页面后，您必须单击要访问的特定模型。在这个项目中，我们使用Anthropic 的Claude 。导航到此部分，在 Anthropic 提供的每个产品旁边，您应该会看到一个链接，上面写着：可供请求。请注意，在下面的屏幕截图中，我已经拥有访问权限，因此您的屏幕可能看起来不是这样的。单击此处并完成必要的步骤。实际的批准过程只需要几分钟。我建议通读您使用的任何模型的文档和定价。出于我们的目的以及在撰写本文时，我们将使用的 Claude 版本每 1,000 个输入令牌花费 0.008 美元，每 1,000 个输出令牌花费 0.024 美元。我们的整个工作流程一次使用最多只需几美分，但要知道，如果不小心，任何重复使用都可能很快增加成本。
图片由作者提供
第 4 部分：创建 Lambda 函数
AWS Lambda 是 AWS 提供的无服务器计算服务。它与 AWS 的许多其他服务集成在一起。我将其视为 AWS 的集中式编排工具。例如，每当将新数据上传到特定的 AWS S3 存储桶（云存储服务） 时，您都可以配置 lambda 函数以将训练作业发送到 AWS SageMaker（机器学习训练和模型托管服务）以启动模型训练作业。这只是 AWS Lambda 无数可能性之一。
对于这个项目，我们将使用 Lambda 向 AWS Bedrock 发送一个带有会议摘要记录的提示，该提示将生成会议摘要以及记录中分配给特定利益相关者的后续行动。然后，Lambda 会将生成的摘要发送到我们之前创建的 S3 存储桶。我将在本节末尾发布完整的 Lambda 函数代码。
首先，导航到 Lambda 函数页面并单击“函数”。从那里，您只需命名您的函数并选择必要的运行时，在我们的例子中，它将是最新版本的 Python。
图片由作者提供
您的屏幕现在应该如下图所示：
接下来，我们将添加 boto3 层。导航到页面底部查看“层”部分。单击“添加层”。这将带您进入以下页面：
从这里，选择自定义层，然后选择我们上传的 boto3 层。您还将看到另一个下拉菜单，指定哪个版本；选择1，然后单击添加按钮。
现在我们已经添加了层，您可以导航到代码源窗口开始编写我们的函数。我将在本文末尾链接完整 Lambda 函数的 GitHub 要点；但是，我将在这里逐一分解。
讽刺的是，我们的 Lambda 函数将结合一系列较小的函数来协调我们的完整工作流程。我们的第一个函数将从文档中提取纯文本。换句话说，任何可读的文本都将从文档中提取。这是必要的，因为文档可能充斥着我们不一定需要的噪音。
def extract_text_from_multipart (数据): msg = message_from_bytes(数据) text_content = ""如果msg.is_multipart():对于msg.walk()中的一部分:如果part.get_content_type() == "text/plain" : text_content += part.get_payload(decode= True ).decode( 'utf-8' ) + "\n"其他:如果msg.get_content_type() == "text/plain" : text_content = msg.get_payload(decode= True ).decode( 'utf-8' )如果text_content则返回text_content.strip()否则无 接下来，我们将创建一个函数来发送文本和调用 AWS bedrock 的提示。对于实际模型，我们将使用 Anthropic 的 Claude V2。请注意，您可以轻松编辑提示文本以满足您的特定需求。记下字典主体。在这里我们可以为调用的 AI 模型指定参数。不熟悉这些参数？别担心！我将在下面分解它们：
- max_tokens_to_sample：这是应用程序将采样的最大标记数，或者更实际地说，这是模型将读取的最大单个单词和/或数字数。
- top_k：对于模型生成的每个标记，它会考虑最有可能的“下一个”标记列表。较低的值将缩小标记的选择范围，而较高的值将允许模型考虑更广泛的选择范围。
- top_p：与 top_k 非常相似；但是，它使用概率分布逻辑。例如，如果我们选择 0.1，模型将考虑最有可能的前 10% 个标记。
- stop_sequences：本质上是模型的停止点。如果模型遇到您指定的停止点，它将在该序列之后停止生成。
您还可以在此处查看正式文档！
def generate_summary_from_bedrock ( content：str ) -> str：prompt_text = f“”“Human：总结以下会议记录并列出具体的后续行动以及谁应该采取这些行动：{content}Assistant：“”“body = { “prompt”：prompt_text，“max_tokens_to_sample”：5000，“top_k”：250，“top_p”：0.2，“stop_sequences”：[ “\n\nHuman：” ] } try：bedrock = boto3.client（“bedrock-runtime”，region_name = “us-east-1”，config = botocore.config.Config（read_timeout = 300，retries = { 'max_attempts'：3 }））response = bedrock.invoke_model(body=json.dumps(body),modelId= "anthropic.claude-v2:1" ) response_content = response.get( 'body' ).read().decode( 'utf-8' ) response_data = json.loads(response_content) summary = response_data[ "completion" ].strip() return summary except Exception as e: print ( f"生成摘要时出错：{e} " ) return ""我们调用的模型的响应需要一个位置来发送会议摘要。因此，我们将构建一个将其发送到 S3 存储桶的函数。
## 将输出保存到给定的 S3 存储桶
def save_summary_to_s3_bucket ( summary, s3_bucket, s3_key ): s3 = boto3.client( 's3' ) try : s3.put_object(Bucket = s3_bucket, Key = s3_key, Body = summary) print ( "摘要已保存到 s3" ) except Exception as e: print ( "将摘要保存到 s3 时出错" )最后，我们将创建Lambda Handler函数，它是 Lambda 函数的核心。Lambda Handler充当调用 Lambda 函数时的入口点；因此，我们将利用我们在处理程序中创建的所有函数。
def lambda_handler（event，context）：decoded_body = base64.b64decode（event [ 'body' ]）text_content = extract_text_from_multipart（decoded_body）if not text_content：return { 'statusCode'：400，'body'：json.dumps（“无法提取内容”）} summary = generate_summary_from_bedrock（text_content）if summary：current_time = datetime.now（）。strftime（'％H％M％S'）#UTC 时间，不一定是您的时区s3_key = f'summary-output/ {current_time} .txt's3_bucket= 'bedrock-text-summarization'save_summary_to_s3_bucket（summary，s3_bucket，s3_key）else：print（“无摘要已生成”）return { 'statusCode' : 200，'body' :json.dumps（“摘要生成完成”）}我们的函数可能已经完成；但是，我们必须进行一些小配置。当我们调用我们的基岩模型时，可能需要几秒钟甚至一分钟才能生成摘要，具体取决于我们发送的成绩单的大小。因此，我们将 Lambda 函数的超时设置为最多四分钟，以确保我们的函数不会超时。为此，请导航到底部的配置菜单，单击左侧的常规配置，然后单击窗口中的编辑。之后，您应该会看到如下所示的屏幕。确保超时设置为四分钟。
图片由作者提供
第 5 部分：创建 API
正如AWS API Gateway 主页上所述，我们将在这里构建会议记录摘要生成器的前门。使用 API Gateway，我们将构建一个具有 POST 方法的 API，允许我们将数据发送到 Lambda 函数。
首先，导航到 AWS API Gateway 主页并单击创建 API。
接下来，单击HTTP API窗口中的“构建”，因为这种类型的 API 最适合我们的特定用例。
命名并单击“下一步”。
继续单击下一步，直到进入“查看和创建”页面。单击“创建”。您现在应该位于“路线”页面，该页面应如下所示。单击左侧窗口中的“创建” 。
我们将在这里创建 POST 路由，以便我们使用 API 将数据发送到我们的应用程序。从下拉列表中选择 POST 并为其命名。
现在您应该回到“路线”页面。单击右上角的“部署” 。将出现如下所示的弹出窗口。单击“创建新阶段”，以蓝色突出显示。
我们将在此创建开发阶段。请注意，我们不一定需要执行此步骤；这只是一个实际项目。但是，在开发生命周期中为应用程序部署设置不同的阶段是一种很好的做法。在应用程序面向客户之前，这些阶段充当制衡系统。将此阶段命名为dev（或您喜欢的任何名称），并确保未启用自动部署。
创建开发阶段后，我们还没有完成。您应该在Stages页面上。在单击部署之前，我们仍然需要集成我们的 Lambda 函数。单击左侧窗口中的Integrations 。
单击“POST”，然后单击“创建并附加集成”。
图片由作者提供
在集成类型下选择Lambda 函数，然后在下一个下拉菜单中选择我们的 Lambda 函数。确保 AWS 区域与您的 Lambda 函数相同，否则您在调用 API 时会遇到错误。
现在，单击右上角的部署，然后在弹出窗口中选择开发。此时，API 已正式部署，并带有我们可以调用的 URL。单击左侧菜单中开发部分上方的链接，它将导航到我们可以找到调用 URL 的页面。请注意，这些 URL 是实时的，因此只要有 URL，任何人都可以使用它们，并且我们没有添加任何需要附加的权限/令牌（因此，我在发布本文之前删除了我的 URL）。话虽如此，我还是要谨慎，因为它会调用需要花钱的服务。
第 6 部分：将所有内容整合在一起！
现在到了简单而又令人兴奋的部分！是时候将我们的项目付诸实践了。我们只需调用一次 API 即可获得会议摘要。要调用我们的应用程序，我们将使用Postman。请注意，我们可以使用许多技术来调用我们的 API，但我发现 Postman 最适合我们的目的。这在现实世界中可能会通过与贵公司的会议软件无缝集成的前端应用程序来完成。
在 Postman 主页上，点击Workspaces并创建一个新的工作区。当您看到下面的下拉窗口时，选择HTTP 。
您现在应该看到一个如下图所示的窗口。
我们正在使用 API 发送数据；因此，我们必须选择GET旁边的下拉菜单并将其更改为POST。要获取 API 的 URL，请导航回我们创建开发阶段 API 后进入的页面。复制并粘贴我们开发阶段的调用 URL，然后将其粘贴到 Postman 工作区中的 POST 旁边。以下屏幕截图可作为获取 URL 的参考：
除了复制粘贴我们的 API 之外，我们还需要添加我们之前创建的会议摘要路由。在我们的 Postman 工作区中，将 /meeting-summary 添加到 URL 末尾。以下屏幕截图供参考：
接下来，单击Body，然后选择form-data。在 Key下，键入document，然后在键入document 的位置右侧的下拉菜单中选择file。然后，您将在此处上传会议记录。对于这个项目，我决定找点乐子，让 Chat GPT 生成一份关于一家 AI 公司的虚假会议记录，该公司的 Gen AI 应用程序具有自我意识。点击发送之前，我们的 Postman 工作区应该是什么样子的屏幕截图，以及我们发送的记录的简要预览如下！
现在是时候点击发送了！如果我们的申请成功，我们应该在 Postman 中看到以下响应：
关于调试 Lambda 函数的说明
当我最初构建此应用程序时，第一次运行失败了。您的应用程序很可能在第一次尝试时失败，所以如果是这种情况，请不要担心。我还没有遇到过在第一次测试运行时就构建成功的模型、产品、应用程序等的开发人员或数据科学家。
如果存在问题，它很可能会在 Lambda 函数中显示出来。要调试它，请导航到 AWS Cloudwatch，单击日志组，然后单击我们的 Lambda 函数。导航到日志流部分，它将显示我们 Lambda 函数的任何运行的响应。这是错误处理派上用场的地方。我在第一次运行时遇到的一些问题包括我的代码中的小拼写错误以及 API、Lambda 和 S3 之间的不匹配区域。如果您有任何问题，它们可能与我的不同；但是，这些可能是值得注意的良好起点。下面的屏幕截图供参考：
图片由作者提供
成功运行应用程序后，您的 S3 存储桶中应该会有一个 txt 文件。导航到 S3，选择我们最初创建的存储桶，然后单击以查看新生成的文件。选择下载并查看输出！以下是我生成的内容，供参考：
令人印象深刻！你不这么认为吗？几秒钟内，该应用程序就生成了一份摘要和具体后续行动，并分配了关键利益相关者。我知道这是一份假的会议记录，但想象一下，使用这样的应用程序可以节省多少时间。
结论
我必须大力感谢 Patrik Szepesi 和他的Udemy 课程，该课程介绍了 AWS 中的生成式 AI，它激励我继续这个项目并撰写了这篇文章。它还激励我开始研究一些个人用例，我希望这篇文章可以鼓励读者也这样做！我希望您发现这些内容对您的 AI 工程之旅很有价值。如果您有任何问题或想给我发送反馈，请随时发表评论！
下面是完整的 Lambda 函数
import boto3
import botocore.config
import json
import base64
from datetime import datetime
from email import message_from_bytes## Extract text of text/plain type from the given document and decode it
def extract_text_from_multipart(data):msg = message_from_bytes(data)text_content = ""if msg.is_multipart():for part in msg.walk():if part.get_content_type() == "text/plain":text_content += part.get_payload(decode=True).decode('utf-8') + "\n"else:if msg.get_content_type() == "text/plain":text_content = msg.get_payload(decode=True).decode('utf-8')return text_content.strip() if text_content else None## Calls on AWS bedrock with a given prompt and parameters and returns the output from bedrock. Note that the function expect a string and returns a string
def generate_summary_from_bedrock(content:str) ->str:prompt_text = f"""Human: Summarize the following meeting transcript and list the specific follow up actions along with who should take those actions: {content}Assistant:"""body = {"prompt":prompt_text,"max_tokens_to_sample":5000,"top_k":250,"top_p":0.2,"stop_sequences": ["\n\nHuman:"]}try:bedrock = boto3.client("bedrock-runtime",region_name="us-east-1",config = botocore.config.Config(read_timeout=300, retries = {'max_attempts':3}))response = bedrock.invoke_model(body=json.dumps(body),modelId="anthropic.claude-v2:1")response_content = response.get('body').read().decode('utf-8')response_data = json.loads(response_content)summary = response_data["completion"].strip()return summaryexcept Exception as e:print(f"Error generating the summary: {e}")return ""## Saves the output to a given S3 bucket
def save_summary_to_s3_bucket(summary, s3_bucket, s3_key):s3 = boto3.client('s3')try:s3.put_object(Bucket = s3_bucket, Key = s3_key, Body = summary)print("Summary saved to s3")except Exception as e:print("Error when saving the summary to s3")def lambda_handler(event,context):decoded_body = base64.b64decode(event['body'])text_content = extract_text_from_multipart(decoded_body)*******
*******
*******
*******
*******
*******
完整代码可联系博主感谢关注雲闪世界。（Aws解决方案架构师vs开发人员&GCP解决方案架构师vs开发人员）
订阅频道(https://t.me/awsgoogvps_Host)
TG交流群(t.me/awsgoogvpsHost)
