Extreme simplification for stability

app.py

@@ -5,7 +5,6 @@ import logging
 import sys
 import os
 from PIL import Image as PILImage
-import io
 
 # 设置日志记录
 logging.basicConfig(level=logging.INFO, 
@@ -13,88 +12,61 @@ logging.basicConfig(level=logging.INFO,
                     stream=sys.stdout)
 logger = logging.getLogger(__name__)
 
-# 确保缓存目录存在
-os.makedirs(".gradio/cached_examples", exist_ok=True)
-
-# 修复 Gradio JSON Schema 错误
-try:
-    import gradio_client.utils
-    
-    # 保存原始函数
-    original_get_type = gradio_client.utils.get_type
-    original_json_schema_to_python_type = gradio_client.utils._json_schema_to_python_type
-    
-    # 修复 get_type 函数
-    def patched_get_type(schema):
-        if isinstance(schema, bool):
-            return "bool"
-        if not isinstance(schema, dict):
-            return "any"
-        return original_get_type(schema)
-    
-    # 修复 _json_schema_to_python_type 函数
-    def patched_json_schema_to_python_type(schema, defs=None):
-        # 处理基本类型
-        if schema is True or schema is False:
-            return "bool"
-        if schema is None:
-            return "None"
-        if not isinstance(schema, dict):
-            return "any"
-            
-        try:
-            return original_json_schema_to_python_type(schema, defs)
-        except Exception as e:
-            logger.warning(f"Error in JSON schema parsing: {str(e)}")
-            return "any"  # 作为备用类型返回
-    
-    # 应用补丁
-    gradio_client.utils.get_type = patched_get_type
-    gradio_client.utils._json_schema_to_python_type = patched_json_schema_to_python_type
-    
-    logger.info("Successfully patched Gradio JSON schema processing")
-except Exception as e:
-    logger.error(f"Failed to patch Gradio: {str(e)}")
-
 # 创建一个简单的示例图像，在模型加载失败或生成失败时使用
 def create_dummy_image():
-    # 创建一个256x256的红色图像
-    img = PILImage.new('RGB', (256, 256), color = (255, 0, 0))
+    # 创建一个简单的彩色图像
+    img = PILImage.new('RGB', (256, 256), color = (255, 100, 100))
     return img
 
-# 加载模型
-try:
-    from diffusers import DiffusionPipeline
-    import torch
-    
-    device = "cuda" if torch.cuda.is_available() else "cpu"
-    model_repo_id = "stabilityai/sdxl-turbo"
-    
-    logger.info(f"Using device: {device}")
-    logger.info(f"Loading model: {model_repo_id}")
-    
-    if torch.cuda.is_available():
-        torch_dtype = torch.float16
-    else:
-        torch_dtype = torch.float32
-    
-    pipe = DiffusionPipeline.from_pretrained(model_repo_id, torch_dtype=torch_dtype)
-    pipe = pipe.to(device)
-    logger.info("Model loaded successfully")
-    
-    MAX_SEED = np.iinfo(np.int32).max
-except Exception as e:
-    logger.error(f"Error loading model: {str(e)}")
-    # 创建一个空的函数以避免错误
-    pipe = None
+# 全局变量，管理模型状态
+pipe = None
+MAX_SEED = np.iinfo(np.int32).max
 
-# 简单的推理函数
+# 简化的推理函数
 def generate_image(prompt):
+    global pipe
+    
     try:
+        # 懒加载模型 - 仅在第一次调用时加载
         if pipe is None:
-            logger.error("Model not loaded, cannot generate image")
-            return create_dummy_image()
+            try:
+                logger.info("First request - loading model...")
+                from diffusers import DiffusionPipeline
+                import torch
+                
+                device = "cuda" if torch.cuda.is_available() else "cpu"
+                model_repo_id = "stabilityai/sdxl-turbo"
+                
+                logger.info(f"Using device: {device}")
+                logger.info(f"Loading model: {model_repo_id}")
+                
+                if torch.cuda.is_available():
+                    torch_dtype = torch.float16
+                else:
+                    torch_dtype = torch.float32
+                
+                # 优化内存使用
+                pipe = DiffusionPipeline.from_pretrained(
+                    model_repo_id, 
+                    torch_dtype=torch_dtype,
+                    variant="fp16" if torch.cuda.is_available() else None,
+                    use_safetensors=True
+                )
+                pipe = pipe.to(device)
+                
+                # 优化内存
+                if torch.cuda.is_available():
+                    pipe.enable_attention_slicing()
+                    # 释放不必要的内存
+                    torch.cuda.empty_cache()
+                
+                logger.info("Model loaded successfully")
+                
+            except Exception as e:
+                logger.error(f"Error loading model: {str(e)}")
+                return create_dummy_image()
             
+        # 处理空提示
         if not prompt or prompt.strip() == "":
             prompt = "A beautiful landscape"
             logger.info(f"Empty prompt, using default: {prompt}")
@@ -103,79 +75,63 @@ def generate_image(prompt):
         seed = random.randint(0, MAX_SEED)
         generator = torch.Generator().manual_seed(seed)
         
-        # 添加异常处理，防止数组索引越界
+        # 简化参数和异常处理
         try:
+            # 使用最小的推理步骤以减轻资源压力
             image = pipe(
                 prompt=prompt,
-                guidance_scale=0.0,
-                num_inference_steps=2,
-                generator=generator
+                guidance_scale=0.0,  # 设为0以获得最快的推理时间
+                num_inference_steps=1,  # 减少步骤
+                generator=generator,
+                height=256,  # 减小图像尺寸
+                width=256    # 减小图像尺寸
             ).images[0]
             
-            # 确保图像是 PIL.Image 类型
+            # 确保图像是有效的PIL图像
             if not isinstance(image, PILImage.Image):
-                logger.warning(f"Converting image from {type(image)} to PIL.Image")
-                if hasattr(image, 'numpy'):
-                    image = PILImage.fromarray(image.numpy())
-                else:
-                    image = PILImage.fromarray(np.array(image))
-                    
-            # 转换为 RGB 模式，确保兼容性
+                logger.warning("Converting image to PIL format")
+                image = PILImage.fromarray(np.array(image))
+                
+            # 转换图像模式
             if image.mode != 'RGB':
                 image = image.convert('RGB')
                 
             logger.info("Image generation successful")
-            # 保存图像以供调试
-            debug_path = "debug_image.jpg"
-            image.save(debug_path)
-            logger.info(f"Debug image saved to {debug_path}")
             
+            # 释放内存
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+                
             return image
-        except IndexError as e:
-            logger.error(f"Index error in pipe: {str(e)}")
+            
+        except Exception as e:
+            logger.error(f"Error in image generation: {str(e)}")
             return create_dummy_image()
         
     except Exception as e:
-        logger.error(f"Error generating image: {str(e)}")
+        logger.error(f"Unexpected error: {str(e)}")
         return create_dummy_image()
 
-# 创建简单的 Gradio 界面，禁用示例缓存
-with gr.Blocks(title="SDXL Turbo Text-to-Image") as demo:
-    gr.Markdown("# SDXL Turbo Text-to-Image Generator")
-    
-    with gr.Row():
-        with gr.Column():
-            prompt_input = gr.Textbox(label="Enter your prompt", placeholder="Type your text prompt here...")
-            generate_button = gr.Button("Generate Image")
-        
-        with gr.Column():
-            image_output = gr.Image(label="Generated Image", type="pil")
-    
-    # 示例
-    gr.Examples(
-        examples=["A cute cat", "Sunset over mountains"],
-        inputs=prompt_input,
-        outputs=image_output,
-        fn=generate_image,
-        cache_examples=False
-    )
-    
-    # 绑定生成按钮
-    generate_button.click(fn=generate_image, inputs=prompt_input, outputs=image_output)
-    
-    # 直接绑定文本框的提交事件
-    prompt_input.submit(fn=generate_image, inputs=prompt_input, outputs=image_output)
+# 使用最简单的界面
+demo = gr.Interface(
+    fn=generate_image,
+    inputs=gr.Textbox(label="Enter your prompt"),
+    outputs=gr.Image(label="Generated Image", type="pil"),
+    title="SDXL Turbo Text-to-Image",
+    description="Enter a text prompt to generate an image.",
+    examples=["A cute cat"],  # 只保留一个简单示例
+    cache_examples=False
+)
 
 # 启动应用
 if __name__ == "__main__":
     try:
         logger.info("Starting Gradio app")
-        # 添加更多启动选项，以提高稳定性
         demo.launch(
-            debug=True,
-            show_error=True,
-            server_name="0.0.0.0",  # 确保可以从外部访问
-            max_threads=1  # 降低并发以避免竞态条件
+            server_name="0.0.0.0",
+            server_port=7860,
+            show_api=False,  # 禁用API
+            share=False
         )
     except Exception as e:
         logger.error(f"Error launching app: {str(e)}")

requirements.txt

@@ -1,7 +1,6 @@
-accelerate
-diffusers
-invisible_watermark
-torch
-transformers
-xformers
-gradio==3.34.0
+accelerate==0.21.0
+diffusers==0.20.0
+torch==2.0.1
+transformers==4.34.0
+gradio==3.32.0
+Pillow==10.0.0