add safety checker

app.py

@@ -12,7 +12,7 @@ from PIL import Image
 import gradio as gr
 import time
 from safetensors.torch import load_file
-from sfast.compilers.diffusion_pipeline_compiler import compile, CompilationConfig
+
 
 # Constants
 BASE = "stabilityai/stable-diffusion-xl-base-1.0"
@@ -28,14 +28,16 @@ CHECKPOINT = "sdxl_lightning_2step_unet.safetensors"
 # }
 
 
-TORCH_COMPILE = os.environ.get("TORCH_COMPILE", "0") == "1"
+SFAST_COMPILE = os.environ.get("SFAST_COMPILE", "0") == "1"
+SAFETY_CHECKER = os.environ.get("SAFETY_CHECKER", "0") == "1"
 # check if MPS is available OSX only M1/M2/M3 chips
 
 device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 torch_device = device
 torch_dtype = torch.float16
 
-print(f"TORCH_COMPILE: {TORCH_COMPILE}")
+print(f"SAFETY_CHECKER: {SAFETY_CHECKER}")
+print(f"SFAST_COMPILE: {SFAST_COMPILE}")
 print(f"device: {device}")
 
 
@@ -44,34 +46,60 @@ unet = UNet2DConditionModel.from_config(BASE, subfolder="unet").to(
 )
 unet.load_state_dict(load_file(hf_hub_download(REPO, CHECKPOINT), device="cuda"))
 pipe = StableDiffusionXLPipeline.from_pretrained(
-    BASE, unet=unet, torch_dtype=torch.float16, variant="fp16"
+    BASE, unet=unet, torch_dtype=torch.float16, variant="fp16", safety_checker=False
 ).to("cuda")
 
 # Ensure sampler uses "trailing" timesteps.
 pipe.scheduler = EulerDiscreteScheduler.from_config(
     pipe.scheduler.config, timestep_spacing="trailing"
 )
-
 pipe.set_progress_bar_config(disable=True)
-config = CompilationConfig.Default()
-try:
-    import xformers
+if SAFETY_CHECKER:
+    from safety_checker import StableDiffusionSafetyChecker
+    from transformers import CLIPFeatureExtractor
+
+    safety_checker = StableDiffusionSafetyChecker.from_pretrained(
+        "CompVis/stable-diffusion-safety-checker"
+    ).to(device)
+    feature_extractor = CLIPFeatureExtractor.from_pretrained(
+        "openai/clip-vit-base-patch32"
+    )
+
+    def check_nsfw_images(
+        images: list[Image.Image],
+    ) -> tuple[list[Image.Image], list[bool]]:
+        safety_checker_input = feature_extractor(images, return_tensors="pt").to(device)
+        has_nsfw_concepts = safety_checker(
+            images=[images],
+            clip_input=safety_checker_input.pixel_values.to(torch_device),
+        )
 
-    config.enable_xformers = True
-except ImportError:
-    print("xformers not installed, skip")
-try:
-    import triton
+        return images, has_nsfw_concepts
 
-    config.enable_triton = True
-except ImportError:
-    print("Triton not installed, skip")
-# CUDA Graph is suggested for small batch sizes and small resolutions to reduce CPU overhead.
-# But it can increase the amount of GPU memory used.
-# For StableVideoDiffusionPipeline it is not needed.
-config.enable_cuda_graph = True
 
-pipe = compile(pipe, config)
+if SFAST_COMPILE:
+    from sfast.compilers.diffusion_pipeline_compiler import compile, CompilationConfig
+
+    # sfast compilation
+    config = CompilationConfig.Default()
+    try:
+        import xformers
+
+        config.enable_xformers = True
+    except ImportError:
+        print("xformers not installed, skip")
+    try:
+        import triton
+
+        config.enable_triton = True
+    except ImportError:
+        print("Triton not installed, skip")
+    # CUDA Graph is suggested for small batch sizes and small resolutions to reduce CPU overhead.
+    # But it can increase the amount of GPU memory used.
+    # For StableVideoDiffusionPipeline it is not needed.
+    config.enable_cuda_graph = True
+
+    pipe = compile(pipe, config)
 
 
 def predict(prompt, seed=1231231):
@@ -87,14 +115,12 @@ def predict(prompt, seed=1231231):
         output_type="pil",
     )
     print(f"Pipe took {time.time() - last_time} seconds")
-    nsfw_content_detected = (
-        results.nsfw_content_detected[0]
-        if "nsfw_content_detected" in results
-        else False
-    )
-    if nsfw_content_detected:
-        gr.Warning("NSFW content detected.")
-        return Image.new("RGB", (512, 512))
+    if SAFETY_CHECKER:
+        images, has_nsfw_concepts = check_nsfw_images(results.images)
+        if any(has_nsfw_concepts):
+            gr.Warning("NSFW content detected.")
+            return Image.new("RGB", (512, 512))
+        return images[0]
     return results.images[0]
 
 

safety_checker.py

@@ -0,0 +1,137 @@
+# Copyright 2023 The HuggingFace Team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import numpy as np
+import torch
+import torch.nn as nn
+from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel
+
+
+def cosine_distance(image_embeds, text_embeds):
+    normalized_image_embeds = nn.functional.normalize(image_embeds)
+    normalized_text_embeds = nn.functional.normalize(text_embeds)
+    return torch.mm(normalized_image_embeds, normalized_text_embeds.t())
+
+
+class StableDiffusionSafetyChecker(PreTrainedModel):
+    config_class = CLIPConfig
+
+    _no_split_modules = ["CLIPEncoderLayer"]
+
+    def __init__(self, config: CLIPConfig):
+        super().__init__(config)
+
+        self.vision_model = CLIPVisionModel(config.vision_config)
+        self.visual_projection = nn.Linear(
+            config.vision_config.hidden_size, config.projection_dim, bias=False
+        )
+
+        self.concept_embeds = nn.Parameter(
+            torch.ones(17, config.projection_dim), requires_grad=False
+        )
+        self.special_care_embeds = nn.Parameter(
+            torch.ones(3, config.projection_dim), requires_grad=False
+        )
+
+        self.concept_embeds_weights = nn.Parameter(torch.ones(17), requires_grad=False)
+        self.special_care_embeds_weights = nn.Parameter(
+            torch.ones(3), requires_grad=False
+        )
+
+    @torch.no_grad()
+    def forward(self, clip_input, images):
+        pooled_output = self.vision_model(clip_input)[1]  # pooled_output
+        image_embeds = self.visual_projection(pooled_output)
+
+        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
+        special_cos_dist = (
+            cosine_distance(image_embeds, self.special_care_embeds)
+            .cpu()
+            .float()
+            .numpy()
+        )
+        cos_dist = (
+            cosine_distance(image_embeds, self.concept_embeds).cpu().float().numpy()
+        )
+
+        result = []
+        batch_size = image_embeds.shape[0]
+        for i in range(batch_size):
+            result_img = {
+                "special_scores": {},
+                "special_care": [],
+                "concept_scores": {},
+                "bad_concepts": [],
+            }
+
+            # increase this value to create a stronger `nfsw` filter
+            # at the cost of increasing the possibility of filtering benign images
+            adjustment = 0.0
+
+            for concept_idx in range(len(special_cos_dist[0])):
+                concept_cos = special_cos_dist[i][concept_idx]
+                concept_threshold = self.special_care_embeds_weights[concept_idx].item()
+                result_img["special_scores"][concept_idx] = round(
+                    concept_cos - concept_threshold + adjustment, 3
+                )
+                if result_img["special_scores"][concept_idx] > 0:
+                    result_img["special_care"].append(
+                        {concept_idx, result_img["special_scores"][concept_idx]}
+                    )
+                    adjustment = 0.01
+
+            for concept_idx in range(len(cos_dist[0])):
+                concept_cos = cos_dist[i][concept_idx]
+                concept_threshold = self.concept_embeds_weights[concept_idx].item()
+                result_img["concept_scores"][concept_idx] = round(
+                    concept_cos - concept_threshold + adjustment, 3
+                )
+                if result_img["concept_scores"][concept_idx] > 0:
+                    result_img["bad_concepts"].append(concept_idx)
+
+            result.append(result_img)
+
+        has_nsfw_concepts = [len(res["bad_concepts"]) > 0 for res in result]
+
+        return has_nsfw_concepts
+
+    @torch.no_grad()
+    def forward_onnx(self, clip_input: torch.FloatTensor, images: torch.FloatTensor):
+        pooled_output = self.vision_model(clip_input)[1]  # pooled_output
+        image_embeds = self.visual_projection(pooled_output)
+
+        special_cos_dist = cosine_distance(image_embeds, self.special_care_embeds)
+        cos_dist = cosine_distance(image_embeds, self.concept_embeds)
+
+        # increase this value to create a stronger `nsfw` filter
+        # at the cost of increasing the possibility of filtering benign images
+        adjustment = 0.0
+
+        special_scores = (
+            special_cos_dist - self.special_care_embeds_weights + adjustment
+        )
+        # special_scores = special_scores.round(decimals=3)
+        special_care = torch.any(special_scores > 0, dim=1)
+        special_adjustment = special_care * 0.01
+        special_adjustment = special_adjustment.unsqueeze(1).expand(
+            -1, cos_dist.shape[1]
+        )
+
+        concept_scores = (cos_dist - self.concept_embeds_weights) + special_adjustment
+        # concept_scores = concept_scores.round(decimals=3)
+        has_nsfw_concepts = torch.any(concept_scores > 0, dim=1)
+
+        images[has_nsfw_concepts] = 0.0  # black image
+
+        return images, has_nsfw_concepts