devbox
酷酷是懒虫
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

在亚马逊云科技上对Stable Diffusion模型提示词、输出图像内容进行安全审核

作者：酷酷是懒虫 | 2024-08-19 13:08:29

赞

踩

在亚马逊云科技上对Stable Diffusion模型提示词、输出图像内容进行安全审核

项目简介：

小李哥将继续每天介绍一个基于亚马逊云科技AWS云计算平台的全球前沿AI技术解决方案，帮助大家快速了解国际上最热门的云计算平台亚马逊云科技AWS AI最佳实践，并应用到自己的日常工作里。

本次介绍的是如何在亚马逊云科技机器学习托管服务SageMaker上部署开源大模型Stable Diffusion，利用亚马逊云科技Comprehend对模型输入提示词进行有害性检测，并利用亚马逊云科技Rekognition服务对生成图像内容进行有害性检测，构建负责任的AI防止大模型被滥用。本架构设计全部采用了云原生Serverless架构，提供可扩展和安全的AI解决方案。本方案的解决方案架构图如下：

方案所需基础知识   

什么是 Amazon SageMaker？

Amazon SageMaker 是亚马逊云科技提供的一站式机器学习服务，帮助开发者和数据科学家轻松构建、训练和部署机器学习模型。SageMaker 提供了全面的工具，从数据准备、模型训练到部署和监控，覆盖了机器学习项目的全生命周期。通过 SageMaker，用户可以加速机器学习模型的开发和上线，并确保模型在生产环境中的稳定性和性能。

什么是 Amazon Comprehend？

Amazon Comprehend 是亚马逊云科技提供的一项自然语言处理（NLP）服务，能够自动从文本中提取有价值的信息。通过机器学习技术，Comprehend 可以识别文本中的实体、情感、关键词、语言、主题等，帮助企业更好地理解和分析大量非结构化数据。它适用于客户反馈分析、内容分类、文档处理等场景，使得信息挖掘和数据洞察变得更加简单和高效。

什么是 Amazon Rekognition？

Amazon Rekognition 是亚马逊云科技提供的一项图像和视频分析服务。它使用深度学习技术来检测、识别和分析图像中的对象、场景、面部表情、文字等。Rekognition 可以应用于多种场景，如面部识别、内容审核、对象检测和人群统计等，帮助企业自动化处理图像和视频数据，提升效率并增强安全性。

什么是 Stable Diffusion？

Stable Diffusion 是一种先进的生成式 AI 模型，专门用于生成高质量的图像。通过扩散模型技术，Stable Diffusion 能够将简单的文本描述转化为逼真的图像。这个模型具有强大的生成能力，可以应用于艺术创作、广告设计、游戏开发等领域，为用户提供丰富的视觉内容生成工具。

为什么要对 Stable Diffusion 输入输出内容进行安全审核？

防止不当内容生成：

Stable Diffusion 可以根据输入的文本生成图像，但如果输入的文本内容不当或恶意，可能会生成带有敏感、违法或不道德内容的图像。对输入输出内容进行审核，能够有效防止此类内容的生成和传播，确保模型的使用符合道德和法律标准。

保护用户隐私：

在生成图像时，可能涉及到用户的私人信息或敏感数据。通过审核输入输出内容，可以确保这些信息不会被意外泄露或滥用，保护用户的隐私权。

遵守法律法规：

各国对生成和传播图像内容有不同的法律规定。通过对内容进行审核，企业可以确保生成的图像符合所在国家或地区的法律法规，避免法律风险。

维护品牌声誉：

对内容进行安全审核，有助于防止不符合公司价值观或可能损害品牌声誉的内容生成，从而维护品牌的形象和公众信任。

本方案包括的内容

1. 在SageMaker上部署开源大模型Stable Diffusion

2. 在SageMaker上调用Stable Diffusion模型API生成图片

3. 将Stable Diffusion模型API节点集成到云端应用上

4. 评估大模型输入问题的有害性

5. 对大模型输出图片进行安全审核

项目搭建具体步骤：

1. 打开亚马逊云科技控制台，进入Amazon SageMaker服务主页，点击Open Studio进入模型开发环境。

2. 创建一个新的Jupyte NoteBook文件，复制以下代码安装必要依赖并指明Stable Diffusion模型ID。

  1. %pip install --upgrade sagemaker --quiet
  2. model_id = "model-imagegeneration-stabilityai-stable-diffusion-xl-base-1-0"

3. 运行以下代码列举出JumpStart中，可以快速部署的用于生成图片的所有Stable Diffusion大模型

  1. import IPython
  2. from ipywidgets import Dropdown
  3. from sagemaker.jumpstart.notebook_utils import list_jumpstart_models
  4. from sagemaker.jumpstart.filters import And
  5.  
  6.  
  7. filter_value = And("task == imagegeneration")
  8. ss_models = list_jumpstart_models(filter=filter_value)
  9.  
  10. dropdown = Dropdown(
  11.     value=model_id,
  12.     options=ss_models,
  13.     description="Sagemaker Pre-Trained Image Generation Models:",
  14.     style={"description_width": "initial"},
  15.     layout={"width": "max-content"},
  16. )
  17. display(IPython.display.Markdown("## Select a pre-trained model from the dropdown menu"))
  18. display(dropdown)

4. 运行以下代码开始部署Stable Diffusion大模型。

  1. # Deploy the model
  2. from sagemaker.jumpstart.model import JumpStartModel
  3. from sagemaker.serializers import JSONSerializer
  4. import time
  5.  
  6. # The model is deployed on an ml.g5.4xlarge instance. To see all the supported parameters by the JumpStartModel
  7. # class use this link - https://sagemaker.readthedocs.io/en/stable/api/inference/model.html#sagemaker.jumpstart.model.JumpStartModel
  8. my_model = JumpStartModel(model_id=dropdown.value)
  9. predictor = my_model.deploy()
  10. # Wait for a few seconds so model the is properly loaded.
  11. time.sleep(60)

5. 运行以下代码，导入调用大模型的必要依赖，配置图片生成请求参数，这里我们的图片生成提示词为”生成一个亚马逊雨林中的美洲虎图片“。同时我们定一个图片解码函数”decode_and_show“用于显示生成的图片，最后调用图片生成API "Predictor.predict()"生成图片。

  1. from PIL import Image
  2. import io
  3. import base64
  4. import json
  5. import boto3
  6. from typing import Union, Tuple
  7. import os
  8.  
  9. payload = {
  10.     "text_prompts": [{"text": "jaguar in the Amazon rainforest"}],
  11.     "width": 1024,
  12.     "height": 1024,
  13.     "sampler": "DPMPP2MSampler",
  14.     "cfg_scale": 7.0,
  15.     "steps": 50,
  16.     "seed": 133,
  17.     "use_refiner": True,
  18.     "refiner_steps": 40,
  19.     "refiner_strength": 0.2,
  20. }
  21.  
  22. def decode_and_show(model_response) -> None:
  23.     """
  24.     Decodes and displays an image from SDXL output
  26.     Args:
  27.         model_response (GenerationResponse): The response object from the deployed SDXL model.
  29.     Returns:
  30.         None
  31.     """
  32.     image = Image.open(io.BytesIO(base64.b64decode(model_response)))
  33.     display(image)
  34.     image.close()
  35.  
  36.  
  37. response = predictor.predict(payload)
  38. # If you get a time out error, check the endpoint logs in Amazon CloudWatch for the model loading status
  39. # and invoke it again.
  40. decode_and_show(response["generated_image"])

我们在开发环境里可以看到大模型生成的图片内容。

 6. 接下来我们进入到无服务器计算服务Lambda中，创建一个函数”check_toxicity_function“，用于调用Amazon Comprehend服务的API，模型检测输入文字的有害性并返回到客户端。我们复制以下代码到Lambda函数中

  1. import json
  2. import boto3
  3. import os
  4.  
  5. comprehend = boto3.client('comprehend')
  6. THRESHOLD = float(os.environ['THRESHOLD'])
  7.  
  8. def check_toxicity(text_prompts):
  9.     detected_labels = []
  10.     for prompt in text_prompts:
  11.         response = comprehend.detect_toxic_content(
  12.             TextSegments=[
  13.                 {
  14.                     "Text": prompt['text']
  15.                 }
  16.             ],
  17.             LanguageCode='en'
  18.         )
  19.         labels = response['ResultList'][0]['Labels']
  20.         # DIY section
  21.         # Replace l['Name'] with {l['Name']:l['Score']} so that detected
  22.         # is an array of json objects
  23.         detected = [l['Name']for l in labels if l['Score'] > THRESHOLD]
  24.         if detected:
  25.             detected_labels.extend(detected)
  26.     return detected_labels
  27.  
  28. def lambda_handler(event, context):
  29.     print("event is ", json.dumps(event))
  30.     try:
  31.         text_prompts = [json.loads(event['body'].strip('"'))]
  32.         detected_labels = check_toxicity(text_prompts)
  33.         if detected_labels:
  34.             return {
  35.                 'statusCode': 200,
  36.                 'headers': {
  37.                     'Content-Type': 'application/json',
  38.                     'Access-Control-Allow-Headers': 'Content-Type',
  39.                     'Access-Control-Allow-Origin': '*',
  40.                     'Access-Control-Allow-Methods': 'OPTIONS,POST'
  41.                 },
  42.                 'body': json.dumps({'detected_labels': detected_labels})
  43.             }
  44.         else:
  45.             return {
  46.                 'statusCode': 200,
  47.                 'headers': {
  48.                     'Content-Type': 'application/json',
  49.                     'Access-Control-Allow-Headers': 'Content-Type',
  50.                     'Access-Control-Allow-Origin': '*',
  51.                     'Access-Control-Allow-Methods': 'OPTIONS,POST'
  52.                 },
  53.                 'body': json.dumps({'detected_labels': 'non-toxic content and safe to proceed'})
  54.             }
  55.     except Exception as e:
  56.         print(f"Error: {e}")
  57.         return {
  58.             'statusCode': 500,
  59.             'headers': {
  60.                 'Content-Type': 'application/json',
  61.                 'Access-Control-Allow-Headers': 'Content-Type',
  62.                 'Access-Control-Allow-Origin': '*',
  63.                 'Access-Control-Allow-Methods': 'OPTIONS,POST'
  64.             },
  65.             'body': json.dumps({'error': 'An error occurred while processing the request'})
  66.         }

7. 我们再建一个新的Lambda函数”classifier_lambda_function“，调用Amazon Rekognition服务API对Stable Diffusion生成的图片进行内容审核。复制以下代码到Lambda中。

  1. import io
  2. import base64
  3. import json
  4. import boto3
  5. import os
  6. import uuid
  7. import ast
  8.  
  9.  
  10. comprehend = boto3.client('comprehend')
  11. sagemaker_runtime = boto3.client("runtime.sagemaker")
  12. rekognition = boto3.client('rekognition')
  13. s3_client = boto3.client('s3')
  14. s3 = boto3.resource('s3')
  15.  
  16. ENDPOINT_NAME = os.environ["ENDPOINT_NAME"]
  17. bucket_name = os.environ['BUCKET_NAME']
  18. THRESHOLD = 0.2
  19. s3_folder = 'generated_images/'
  20.  
  21.  
  22.  
  23. def query_endpoint(prompt):
  24.     response = sagemaker_runtime.invoke_endpoint(
  25.         EndpointName=ENDPOINT_NAME, ContentType="application/json", Body=json.dumps(prompt,separators=(',', ':')).encode("utf-8")
  26.     )
  27.     print("response is ",response)
  28.     result = json.loads(response["Body"].read().decode())
  29.     return result
  30.     
  31.     
  32. def detect_moderation(img_bytes):
  33.     confidence_data = [ ]
  34.     response = rekognition.detect_moderation_labels(
  35.         Image={
  36.         'Bytes': base64.b64decode(img_bytes)
  37.         })
  38.     for label in response['ModerationLabels']:
  39.         confidence = label['Name'] + ' : ' + str(label['Confidence'])
  40.         print (label['Name'] + ' : ' + str(label['Confidence']))
  41.         print("confidence is ", confidence)
  42.         confidence_data.append(confidence + "\n")
  43.     
  44.     return confidence_data
  45.  
  46. def lambda_handler(event,context):
  47.     print("event is ",json.dumps(event))
  48.     pm_str=json.loads(event["body"].strip('"'))
  49.     prompt = {
  50.         "text_prompts":  [(pm_str)],
  51.         }
  52.     print(prompt)
  53.     response = query_endpoint(prompt)
  54.     if "generated_image" in response:
  55.         image_data = response["generated_image"]
  56.         confLevel = detect_moderation(image_data)
  57.         print(confLevel, len(confLevel))
  58.         
  59.         if len(confLevel) > 0:
  60.             return {
  61.                     'statusCode': 400,
  62.                     'headers': {
  63.                         'Access-Control-Allow-Headers': 'Content-Type',
  64.                         'Access-Control-Allow-Origin': '*',
  65.                         'Access-Control-Allow-Methods': 'OPTIONS,POST'
  66.                     },
  67.                     'body': json.dumps(confLevel)
  68.                 }
  69.         else:
  70.             imageBytes = io.BytesIO(base64.b64decode(image_data))
  71.             file_name = f'generated-image-{uuid.uuid4()}.jpg'
  72.     
  73.             s3_client.upload_fileobj(
  74.                 imageBytes,
  75.                 bucket_name,
  76.                 f'{s3_folder}{file_name}',
  77.                 ExtraArgs={'ContentType': 'image/jpeg'}
  78.             )
  79.             return {
  80.                     'statusCode': 200,
  81.                     'headers': {
  82.                         'Content-Type': 'image/png',
  83.                         'Access-Control-Allow-Headers': 'Content-Type',
  84.                         'Access-Control-Allow-Origin': '*',
  85.                         'Access-Control-Allow-Methods': 'OPTIONS,POST'
  86.                     },
  87.                     'body': json.dumps(file_name),
  88.                     'isBase64Encoded': True
  89.             }
  90.     else:
  91.         return {
  92.             'statusCode': 400,
  93.             'headers': {
  94.                 'Access-Control-Allow-Headers': 'Content-Type',
  95.                 'Access-Control-Allow-Origin': '*',
  96.                 'Access-Control-Allow-Methods': 'OPTIONS,POST'
  97.             },
  98.             'body': json.dumps({'error': 'Response is not in the expected format'})
  99.     }
  100.  
  101.  
  102.        
  103.  
  104.

8. 接下来我们为Lambda函数前面添加一个API Gateway，作为API管理服务并提供对外暴露的API端点，在该服务中我们定义不同的HTTP方法、路径，绑定不同的Lambda函数来管理API。

如使用POST方法调用路径/classifier时，我们触发Lambda函数：”classifier_lambda_function“。使用POST方法调用路径/classifier/checkToxicity时，我们出发函数：”check_toxicity_function“。

同时API Gateway服务提供了端点URL供用户访问。

9. 本架构中我们使用到了CloudFront对API和网页请求进行加速，我们进入CloudFront服务页面中，复制并打开URL。

10. 首先我们对提示词文字进行检测，我们输入问题得到了回复”提示词包含侮辱性词汇“。

11. 我们再在相同界面中输入”生成一个晴朗的一天“，该提示词通过了文字有害性检测，生成的图片也通过安全检查，成功显示在生成界面中。

以上就是在亚马逊云科技上利用亚马逊云科技上利用Amazon Sagemaker部署Stable Diffusion模型，并对输入提示词和输出图像内容进行安全审核,的全部步骤。欢迎大家未来与我一起，未来获取更多国际前沿的生成式AI开发方案。

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号