Midjourney Prompt Message Queue Design

Midjourney Prompt และ Message Queue

Midjourney Prompt สั่ง AI สร้างภาพ ต้องชัดเจน เฉพาะเจาะจง Style Lighting Camera Angle Parameters Message Queue สำหรับ Async Communication ระหว่าง Services

รวมสองหัวข้อ ใช้ Message Queue จัดการ Prompt Queue สำหรับ AI Image Generation ที่ต้อง Process เป็น Batch

Midjourney Prompt Engineering

# === Midjourney Prompt Engineering ===
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class MidjourneyPrompt:
    subject: str
    style: str = ""
    lighting: str = ""
    camera: str = ""
    color_palette: str = ""
    mood: str = ""
    details: List[str] = field(default_factory=list)
    negative: List[str] = field(default_factory=list)
    aspect_ratio: str = "16:9"
    version: str = "6.1"
    quality: float = 1.0
    stylize: int = 100
    chaos: int = 0

    def build(self) -> str:
        """สร้าง Prompt String"""
        parts = [self.subject]

        if self.style:
            parts.append(self.style)
        if self.lighting:
            parts.append(self.lighting)
        if self.camera:
            parts.append(self.camera)
        if self.color_palette:
            parts.append(self.color_palette)
        if self.mood:
            parts.append(self.mood)
        parts.extend(self.details)

        prompt = ", ".join(parts)

        # Parameters
        params = []
        params.append(f"--ar {self.aspect_ratio}")
        params.append(f"--v {self.version}")
        if self.quality != 1.0:
            params.append(f"--q {self.quality}")
        if self.stylize != 100:
            params.append(f"--s {self.stylize}")
        if self.chaos > 0:
            params.append(f"--chaos {self.chaos}")
        if self.negative:
            params.append(f"--no {' '.join(self.negative)}")

        return f"{prompt} {' '.join(params)}"

class PromptLibrary:
    """Prompt Library สำหรับ Midjourney"""

    def __init__(self):
        self.prompts: List[MidjourneyPrompt] = []
        self.categories = {}

    def add(self, category: str, prompt: MidjourneyPrompt):
        self.prompts.append(prompt)
        if category not in self.categories:
            self.categories[category] = []
        self.categories[category].append(prompt)

    def show_all(self):
        for cat, prompts in self.categories.items():
            print(f"\n  [{cat}]")
            for p in prompts:
                print(f"    {p.build()[:80]}...")

# ตัวอย่าง Prompts
library = PromptLibrary()

prompts = [
    ("Product Photography", MidjourneyPrompt(
        "luxury watch on marble surface",
        "product photography", "soft studio lighting",
        "macro lens 100mm", "gold and black tones",
        "elegant minimalist",
        ["sharp focus", "high detail", "reflections"],
        ["text", "watermark"],
        "1:1", stylize=200,
    )),
    ("Web Design", MidjourneyPrompt(
        "modern SaaS dashboard UI design",
        "clean minimal UI/UX", "flat design",
        details=["dark mode", "data visualization", "glassmorphism"],
        negative=["realistic photo"],
        aspect_ratio="16:9", stylize=150,
    )),
    ("Architecture", MidjourneyPrompt(
        "futuristic eco-friendly skyscraper",
        "architectural visualization", "golden hour sunlight",
        "wide angle lens", "green and glass tones",
        "sustainable utopian",
        ["vertical gardens", "solar panels", "sky bridges"],
        ["people", "cars"],
        "9:16", stylize=250,
    )),
    ("Character Design", MidjourneyPrompt(
        "cyberpunk female warrior",
        "anime style illustration", "neon rim lighting",
        "dynamic pose", "purple and cyan neon",
        "futuristic intense",
        ["detailed armor", "holographic HUD", "rain"],
        ["realistic", "photo"],
        "2:3", stylize=300, chaos=20,
    )),
]

for cat, prompt in prompts:
    library.add(cat, prompt)

print("Midjourney Prompt Library:")
library.show_all()

print("\n\nFull Prompt Examples:")
for cat, prompt in prompts:
    print(f"\n  [{cat}]")
    print(f"  /imagine {prompt.build()}")

Message Queue Architecture

# message_queue.py — Message Queue Design Pattern
import json
import time
import uuid
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Callable, Optional
from datetime import datetime
from collections import deque
from enum import Enum

class MessageStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    DEAD_LETTER = "dead_letter"

@dataclass
class Message:
    id: str
    payload: Dict
    status: MessageStatus = MessageStatus.PENDING
    retries: int = 0
    max_retries: int = 3
    created_at: str = ""
    processed_at: str = ""

    def __post_init__(self):
        if not self.id:
            self.id = str(uuid.uuid4())[:8]
        if not self.created_at:
            self.created_at = datetime.now().isoformat()

class SimpleMessageQueue:
    """Simple Message Queue Implementation"""

    def __init__(self, name: str, max_size: int = 1000):
        self.name = name
        self.max_size = max_size
        self.queue: deque = deque()
        self.dead_letter: List[Message] = []
        self.processed: List[Message] = []
        self.stats = {"enqueued": 0, "processed": 0, "failed": 0}

    def enqueue(self, payload: Dict) -> Message:
        """เพิ่ม Message เข้า Queue"""
        if len(self.queue) >= self.max_size:
            raise Exception(f"Queue full: {self.name}")

        msg = Message(id="", payload=payload)
        self.queue.append(msg)
        self.stats["enqueued"] += 1
        return msg

    def dequeue(self) -> Optional[Message]:
        """ดึง Message ออกจาก Queue"""
        if not self.queue:
            return None
        msg = self.queue.popleft()
        msg.status = MessageStatus.PROCESSING
        return msg

    def complete(self, msg: Message):
        """Mark Message as completed"""
        msg.status = MessageStatus.COMPLETED
        msg.processed_at = datetime.now().isoformat()
        self.processed.append(msg)
        self.stats["processed"] += 1

    def fail(self, msg: Message):
        """Handle failed Message"""
        msg.retries += 1
        if msg.retries >= msg.max_retries:
            msg.status = MessageStatus.DEAD_LETTER
            self.dead_letter.append(msg)
            self.stats["failed"] += 1
        else:
            msg.status = MessageStatus.PENDING
            self.queue.append(msg)  # Re-queue

    def dashboard(self):
        """Queue Dashboard"""
        print(f"\n{'='*50}")
        print(f"Queue: {self.name}")
        print(f"{'='*50}")
        print(f"  Pending: {len(self.queue)}")
        print(f"  Processed: {self.stats['processed']}")
        print(f"  Failed: {self.stats['failed']}")
        print(f"  Dead Letter: {len(self.dead_letter)}")
        print(f"  Total Enqueued: {self.stats['enqueued']}")

# ตัวอย่าง — AI Image Generation Queue
queue = SimpleMessageQueue("ai-image-generation")

# Enqueue Prompts
prompts_to_process = [
    {"prompt": "luxury watch on marble", "style": "product", "priority": "high"},
    {"prompt": "modern dashboard UI", "style": "web", "priority": "medium"},
    {"prompt": "futuristic skyscraper", "style": "architecture", "priority": "low"},
    {"prompt": "cyberpunk warrior", "style": "character", "priority": "high"},
    {"prompt": "cozy coffee shop interior", "style": "interior", "priority": "medium"},
]

for p in prompts_to_process:
    queue.enqueue(p)

# Process Queue
print("Processing Queue:")
while True:
    msg = queue.dequeue()
    if not msg:
        break
    print(f"  Processing: {msg.payload['prompt'][:30]}... [{msg.payload['priority']}]")
    queue.complete(msg)

queue.dashboard()

Production Queue with Redis

# === Production Message Queue with Redis ===
# pip install redis rq celery

# 1. Celery Task Queue (Production)
# tasks.py
# from celery import Celery
# import time
#
# app = Celery('tasks',
#     broker='redis://localhost:6379/0',
#     backend='redis://localhost:6379/1',
# )
#
# app.conf.update(
#     task_serializer='json',
#     accept_content=['json'],
#     result_serializer='json',
#     timezone='Asia/Bangkok',
#     task_acks_late=True,          # Acknowledge after processing
#     worker_prefetch_multiplier=1, # Process one at a time
#     task_reject_on_worker_lost=True,
#     task_default_retry_delay=60,  # Retry after 60 seconds
#     task_max_retries=3,
# )
#
# @app.task(bind=True, max_retries=3)
# def generate_image(self, prompt_data):
#     """Generate AI Image from Prompt"""
#     try:
#         prompt = prompt_data['prompt']
#         style = prompt_data.get('style', 'default')
#         # Call Midjourney API or Stable Diffusion
#         time.sleep(30)  # Simulate generation
#         return {"status": "completed", "image_url": f"/images/{self.request.id}.png"}
#     except Exception as exc:
#         self.retry(exc=exc, countdown=60 * (self.request.retries + 1))
#
# # Send task
# result = generate_image.delay({"prompt": "luxury watch", "style": "product"})
# print(f"Task ID: {result.id}")
# print(f"Status: {result.status}")
# print(f"Result: {result.get(timeout=120)}")

# 2. Docker Compose — Redis + Celery
# version: '3.8'
# services:
#   redis:
#     image: redis:7-alpine
#     ports: ["6379:6379"]
#
#   celery-worker:
#     build: .
#     command: celery -A tasks worker --loglevel=info --concurrency=4
#     depends_on: [redis]
#     environment:
#       CELERY_BROKER_URL: redis://redis:6379/0
#
#   celery-beat:
#     build: .
#     command: celery -A tasks beat --loglevel=info
#     depends_on: [redis]
#
#   flower:
#     build: .
#     command: celery -A tasks flower --port=5555
#     ports: ["5555:5555"]
#     depends_on: [redis]

# 3. Monitoring Commands
# celery -A tasks inspect active    # Active tasks
# celery -A tasks inspect reserved  # Reserved tasks
# celery -A tasks inspect stats     # Worker stats
# celery -A tasks purge             # Purge all tasks

queue_architecture = {
    "Broker": "Redis (Message Storage)",
    "Backend": "Redis (Result Storage)",
    "Workers": "Celery Workers (Concurrent Processing)",
    "Scheduler": "Celery Beat (Periodic Tasks)",
    "Monitoring": "Flower (Web Dashboard port 5555)",
    "Retry": "Exponential Backoff (60s, 120s, 180s)",
    "DLQ": "Dead Letter Queue for failed tasks",
}

print("Production Queue Architecture:")
for component, desc in queue_architecture.items():
    print(f"  {component}: {desc}")

Best Practices

• Prompt Structure: Subject + Style + Lighting + Camera + Details + Parameters
• Negative Prompts: ใช้ --no ลบสิ่งไม่ต้องการออกจากภาพ
• Idempotent: ทำ Consumer ให้ Idempotent ประมวลผลซ้ำได้ผลเหมือนเดิม
• Dead Letter Queue: ตั้ง DLQ สำหรับ Messages ที่ fail หลาย retry
• Monitoring: ติดตาม Queue Depth, Consumer Lag, Processing Time
• Retry Policy: ใช้ Exponential Backoff สำหรับ Retry

Midjourney Prompt คืออะไร

ข้อความสั่ง Midjourney AI สร้างภาพ Subject Style Parameters --ar --v --q Prompt ดีต้องชัดเจน เฉพาะเจาะจง Style Lighting Camera Angle Color Palette

สรุป

Midjourney Prompt ต้อง Subject Style Lighting Camera Parameters Message Queue จัดการ Async Processing Redis Celery Production Queue Retry Exponential Backoff Dead Letter Queue Monitor ด้วย Flower Idempotent Consumers ป้องกัน Duplicate