在亚马逊云科技上对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
7. filter_value = And("task == imagegeneration")
8. ss_models = list_jumpstart_models(filter=filter_value)
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
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 "redictor.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
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. }
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()
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
5. comprehend = boto3.client('comprehend')
6. THRESHOLD = float(os.environ['THRESHOLD'])
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
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
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')
16. ENDPOINT_NAME = os.environ["ENDPOINT_NAME"]
17. bucket_name = os.environ['BUCKET_NAME']
18. THRESHOLD = 0.2
19. s3_folder = 'generated_images/'
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
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.     }
102.       

复制代码

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开发方案。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。