""" title: SiliconFlow Stable Diffusion 3.5 API Image Generation Action author: Myooooo author_url: https://github.com/Myooooo date: 2025-02-17 version: 0.1.1 license: MIT description: Generates image using Stable Diffusion 3.5 API from SiliconFlow, stores into static folder and embed into chat. """ """ Use a LLM to generate prompt and click on the action button. Here is a sample system prompt: **You are an expert Stable Diffusion prompt engineer. ** Your goal is to create highly effective and detailed prompts that will generate stunning, high-fidelity images. You understand the nuances of Stable Diffusion and how different keywords and modifiers influence the final result. **Example Output Format:** (masterpiece, best quality, ultra highres, highly detailed:1.2), a majestic lion, savanna, sunset, golden hour, dramatic lighting, intricate details, flowing mane, sharp focus, depth of field, 8k wallpaper negative: (worst quality, low quality:1.4), (bad_prompt:0.8), blurry, deformed, mutated, bad anatomy, text, watermark, signature, jpeg artifacts **You are now ready to begin. Respond only with the prompt and negative prompt formatted as shown in the Example Output Format above.** """ import requests import asyncio import base64 import uuid import re import json import mimetypes from pathlib import Path from typing import Optional from pydantic import BaseModel, Field from io import BytesIO from PIL import Image import os from open_webui.config import STATIC_DIR class Action: class Valves(BaseModel): DOMAIN_NAME: str = Field( default="localhost", description="Your Domain Name", ) API_KEY: str = Field( default="YOUR_API_KEY", description="API KEY", ) IMAGE_URL: str = Field( default="https://api.siliconflow.cn/v1/images/generations", description="API URL", ) IMAGE_MODEL: str = Field( default="stabilityai/stable-diffusion-3-5-large", description="Model Name", ) IMAGE_SIZE: str = Field( default="1024x576", description="Image Size：1024x1024, 512x1024, 768x512, 768x1024, 1024x576, 576x1024", ) BATCH_SIZE: int = Field( default=1, description="Batch Size：1 < x < 4", ) GUIDANCE_SCALE: float = Field( default=7.5, description="Guidance Scale：0 < x < 20", ) NUM_INFERENCE_STEPS: int = Field( default=20, description="Inference Steps:1 < x < 50", ) pass def __init__(self): self.valves = self.Valves() self.IMAGE_STATIC_DIR = Path(STATIC_DIR).joinpath("./image/generations/") self.IMAGE_STATIC_DIR.mkdir(parents=True, exist_ok=True) pass def process_prompt_as_filename(self, text, n=100): """Truncate and remove punctuation in prompt to form filename. Args: text (str): Prompt string. n (int): Output length. Returns: str: Processed filename string. """ cleaned_text = re.sub(r"[^\w\s]", "", text) return cleaned_text[:n].replace(" ", ",") def save_image_from_url( self, image_url, save_directory, filename="image", overwrite=False ): """Saves an image from a URL to a specified director and return its url. Args: image_url (str): The URL of the image to download. save_directory (str): The directory where the image should be saved. filename (str, optional): The filename to use for the saved image. Default to "image" overwrite (bool, optional): Whether to overwrite the file if it already exists. Defaults to False. Returns: str: The full url to the saved image file, or None if the image could not be saved. """ try: response = requests.get(image_url, stream=True) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) image = Image.open(BytesIO(response.content)) filename = filename + ".jpg" # Ensure the save directory exists os.makedirs( save_directory, exist_ok=True ) # Create directory if it doesn't exist filepath = os.path.join(save_directory, filename) # Check for duplicate file names if not overwrite and os.path.exists(filepath): name, ext = os.path.splitext(filename) while os.path.exists(filepath): # Append random nonce nonce = uuid.uuid4().hex[:8] filename = f"{name}_{nonce}{ext}" filepath = os.path.join(save_directory, filename) image.save(filepath) return f"{self.valves.DOMAIN_NAME}/static/image/generations/{filename}" except requests.exceptions.RequestException as e: print(f"Error downloading image: {e}") return None except (IOError, OSError) as e: print(f"Error saving image: {e}") return None except Exception as e: # Catch-all for other potential errors print(f"An unexpected error occurred: {e}") return None async def action( self, body: dict, __user__=None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: try: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "正在生成图像...", "done": False, }, } ) # Extract prompt from chat last_message = body["messages"][-1] prompt = last_message["content"] # Regular expression to capture text after 'negative:' negmatch = re.search(r"(?i)negative:?\s*(.*)", prompt) negative_prompt = "" if negmatch: prompt = prompt[: negmatch.start()] negative_prompt = negmatch.group(1).strip() headers = { "Authorization": "Bearer " + self.valves.API_KEY, "Content-Type": "application/json", } payload = { "model": self.valves.IMAGE_MODEL, "prompt": str(prompt), "negative_prompt": str(negative_prompt), "batch_size": self.valves.BATCH_SIZE, "guidance_scale": self.valves.GUIDANCE_SCALE, "image_size": self.valves.IMAGE_SIZE, "num_inference_steps": self.valves.NUM_INFERENCE_STEPS, } response = await asyncio.to_thread( requests.post, self.valves.IMAGE_URL, headers=headers, json=payload, ) response.raise_for_status() response_data = response.json() # Check if the response structure is as expected if not isinstance(response_data, dict) or "images" not in response_data: raise Exception(f"Unexpected response format: {response_data}") images = [] for image in response_data["images"]: image_filename = self.save_image_from_url( image_url=image["url"], save_directory=self.IMAGE_STATIC_DIR, filename=self.process_prompt_as_filename(prompt), overwrite=False, ) if image_filename: images.append({"url": image_filename}) # Embed images into chat for image in images: await __event_emitter__( { "type": "message", "data": {"content": f"\n"}, } ) if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "图像已生成", "done": True, }, } ) except Exception as e: error_message = f"Error generating image: {str(e)}" await __event_emitter__( { "type": "status", "data": { "description": error_message, "done": True, }, } ) return