Stable Diffusion ComfyUI DevSecOps Integration —
ComfyUI DevSecOps
Stable Diffusion ComfyUI Node-based Workflow DevSecOps CI/CD Security Model Management SDXL ControlNet LoRA API Automation Container Docker
| Tool | Type | API | Node-based | เหมาะกับ |
|---|---|---|---|---|
| ComfyUI | Node GUI | REST+WS | ใช่ | Advanced/Production |
| Automatic1111 | Web GUI | REST | ไม่ | เริ่มต้น |
| InvokeAI | Web GUI | REST | ไม่ | Creative |
| Fooocus | Simple GUI | ไม่มี | ไม่ | ง่ายสุด |
ComfyUI Setup และ API
=== ComfyUI Setup & API ===
Docker
docker run -d --gpus all \
-p 8188:8188 \
-v ./models:/comfyui/models \
-v ./output:/comfyui/output \
comfyui/comfyui:latest
ComfyUI API — Generate Image
import json
import requests
import websocket
import uuid
COMFYUI_URL = "http://localhost:8188"
client_id = str(uuid.uuid4())
def queue_prompt(workflow):
p = {"prompt": workflow, "client_id": client_id}
response = requests.post(f"{COMFYUI_URL}/prompt", json=p)
return response.json()
def get_image(filename, subfolder, folder_type):
params = {"filename": filename, "subfolder": subfolder, "type": folder_type}
response = requests.get(f"{COMFYUI_URL}/view", params=params)
return response.content
# Load Workflow JSON
with open("workflow_api.json") as f:
workflow = json.load(f)
# Modify Prompt
workflow["6"]["inputs"]["text"] = "a beautiful sunset over mountains"
workflow["3"]["inputs"]["seed"] = 42
# Queue and Wait
result = queue_prompt(workflow)
prompt_id = result["prompt_id"]
# WebSocket for progress
ws = websocket.WebSocket()
ws.connect(f"ws://localhost:8188/ws?clientId={client_id}")
while True:
msg = json.loads(ws.recv())
if msg["type"] == "executing" and msg["data"]["node"] is None:
break # Done
from dataclasses import dataclass
from typing import List
@dataclass
class WorkflowNode:
node_id: str
node_type: str
purpose: str
nodes = [
WorkflowNode("1", "CheckpointLoaderSimple", "Load SD Model"),
WorkflowNode("2", "CLIPTextEncode", "Positive Prompt"),
WorkflowNode("3", "CLIPTextEncode", "Negative Prompt"),
WorkflowNode("4", "KSampler", "Generate Latent"),
WorkflowNode("5", "VAEDecode", "Decode to Image"),
WorkflowNode("6", "SaveImage", "Save Output"),
WorkflowNode("7", "ControlNetApply", "ControlNet Guidance"),
WorkflowNode("8", "LoraLoader", "Load LoRA Weights"),
]
print("=== ComfyUI Workflow Nodes ===")
for n in nodes:
print(f" [Node {n.node_id}] {n.node_type} — {n.purpose}")
DevSecOps Pipeline
=== DevSecOps Pipeline for AI Image Gen ===
GitHub Actions CI/CD
.github/workflows/ai-pipeline.yml
name: AI Image Pipeline
on:
push: { branches: [main] }
jobs:
security-scan:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Scan Docker Image
uses: aquasecurity/trivy-action@master
with:
image-ref: 'comfyui:latest'
severity: 'CRITICAL, HIGH'
- name: Model Checksum Verify
run: |
sha256sum -c models/checksums.sha256
- name: NSFW Filter Test
run: python tests/test_content_safety.py
deploy:
needs: security-scan
runs-on: ubuntu-latest
steps:
- name: Deploy ComfyUI
run: |
docker-compose -f docker-compose.prod.yml up -d
- name: Health Check
run: |
curl -f http://localhost:8188/system_stats
Model Security Scanning
import hashlib
import os
def verify_model(model_path, expected_hash):
sha256 = hashlib.sha256()
with open(model_path, 'rb') as f:
for chunk in iter(lambda: f.read(8192), b''):
sha256.update(chunk)
actual = sha256.hexdigest()
if actual != expected_hash:
raise SecurityError(f"Model hash mismatch: {actual}")
return True
security_checks = {
"Container Scan": {"tool": "Trivy", "frequency": "Every Build", "severity": "Critical+High"},
"Model Checksum": {"tool": "SHA256", "frequency": "Every Load", "severity": "Critical"},
"NSFW Filter": {"tool": "Safety Checker", "frequency": "Every Generation", "severity": "High"},
"Access Control": {"tool": "OAuth2/API Key", "frequency": "Every Request", "severity": "High"},
"Audit Log": {"tool": "ELK Stack", "frequency": "Continuous", "severity": "Medium"},
"License Check": {"tool": "Custom Script", "frequency": "Model Update", "severity": "Medium"},
}
print("\nDevSecOps Security Checks:")
for check, info in security_checks.items():
print(f" [{info['severity']}] {check}")
print(f" Tool: {info['tool']} | Frequency: {info['frequency']}")
Production Architecture
=== Production Architecture ===
docker-compose.prod.yml
version: '3.8'
services:
comfyui:
image: comfyui:latest
deploy:
resources:
reservations:
devices:
- capabilities: [gpu]
ports:
- "8188:8188"
volumes:
- ./models:/comfyui/models:ro
- ./output:/comfyui/output
environment:
- COMFYUI_LISTEN=0.0.0.0
api-gateway:
image: nginx:alpine
ports:
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
redis:
image: redis:7-alpine
ports:
- "6379:6379"
monitoring:
image: grafana/grafana:latest
ports:
- "3000:3000"
architecture = {
"API Gateway": "Nginx + Rate Limiting + Auth",
"Queue": "Redis Queue สำหรับ Job Management",
"ComfyUI Workers": "GPU Nodes รัน Stable Diffusion",
"Model Registry": "S3/HuggingFace เก็บ Models",
"Storage": "S3 เก็บ Output Images",
"CDN": "CloudFront/CloudFlare ส่งภาพ",
"Monitoring": "Grafana + Prometheus Metrics",
"Logging": "ELK Stack Audit Trail",
}
print("Production Architecture:")
for component, desc in architecture.items():
print(f" [{component}]: {desc}")
Cost Estimation
costs = {
"GPU Server (A100)": "$2.50/hr",
"Storage (S3 1TB)": "$23/mo",
"CDN (10TB transfer)": "$85/mo",
"Monitoring": "$50/mo",
"Total (24/7)": "~$1,958/mo",
}
print(f"\n\nCost Estimation:")
for item, cost in costs.items():
print(f" {item}: {cost}")
เคล็ดลับ
- Headless: รัน ComfyUI --listen 0.0.0.0 สำหรับ API Server
- Checksum: ตรวจ SHA256 ทุก Model ก่อนใช้งาน
- Queue: ใช้ Redis Queue จัดการ Job ไม่ให้ GPU Overload
- NSFW: เปิด Safety Checker ทุก Generation ใน Production
- Cache: Cache ผลลัพธ์ที่ Prompt เหมือนกัน ประหยัด GPU
ComfyUI คืออะไร
Node-based GUI Stable Diffusion Workflow SDXL ControlNet LoRA API JSON Automation Headless Production ยืดหยุ่น