MJ + SD 双引擎工程实践：从审美探索到可控交付的完整管线

import os, time, requests, json, hashlib, pathlib

BASE = "https://api.goapi.ai/mj/v2"
H = {"X-API-Secret": os.getenv("MJ_KEY"), "Content-Type": "application/json"}

def mj_run(prompt, aspect="1:1", version="6.1", mode="relax"):
    r = requests.post(f"{BASE}/imagine", headers=H, json={
        "prompt": prompt, "aspect_ratio": aspect,
        "version": version, "process_mode": mode,
    }).json()
    tid = r["task_id"]
    while True:
        s = requests.get(f"{BASE}/fetch", headers=H,
                         params={"task_id": tid}).json()
        if s["status"] == "finished":
            return s["task_result"]["image_urls"]
        if s["status"] == "failed":
            raise RuntimeError(s.get("message"))
        time.sleep(5)

def cache_prompt(prompt, urls, out="mj_cache"):
    pathlib.Path(out).mkdir(exist_ok=True)
    key = hashlib.md5(prompt.encode()).hexdigest()[:10]
    for i, u in enumerate(urls):
        p = f"{out}/{key}_{i}.png"
        if not os.path.exists(p):
            open(p, "wb").write(requests.get(u).content)
    json.dump({"prompt": prompt, "urls": urls},
              open(f"{out}/{key}.json", "w"), ensure_ascii=False)
    return f"{out}/{key}_0.png"

import requests, base64

API = "http://127.0.0.1:7860"

def txt2img(prompt, negative, lora, seed=42, steps=30):
    payload = {
        "prompt": f"{prompt}, <lora:{lora}:0.8>",
        "negative_prompt": negative,
        "seed": seed, "steps": steps,
        "width": 1024, "height": 576,
        "cfg_scale": 6.5,
        "sampler_name": "DPM++ 2M Karras",
    }
    r = requests.post(f"{API}/sdapi/v1/txt2img",
                      json=payload, timeout=300)
    r.raise_for_status()
    img = r.json()["images"][0]
    with open("sd_output.png", "wb") as f:
        f.write(base64.b64decode(img.split(",", 1)[-1]))

import torch
from PIL import Image
from diffusers import StableDiffusionXLPipeline, DDIMScheduler
from transformers import CLIPVisionModelWithProjection

pipe = StableDiffusionXLPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    torch_dtype=torch.float16,
    image_encoder=CLIPVisionModelWithProjection.from_pretrained(
        "h94/IP-Adapter", subfolder="models/image_encoder",
        torch_dtype=torch.float16),
).to("cuda")

pipe.load_ip_adapter("h94/IP-Adapter", subfolder="sdxl_models",
                     weight_name="ip-adapter_sdxl.bin")
pipe.set_ip_adapter_scale(0.6)  # 0.4~0.7 为常用区间

ref = Image.open("mj_cache/style_0.png").convert("RGB")
out = pipe(
    prompt="luxury perfume bottle, marble, studio lighting, 8k",
    negative_prompt="text, watermark, deformed",
    ip_adapter_image=ref,
    num_inference_steps=30, guidance_scale=7.5,
).images[0]

import cv2, numpy as np
from diffusers import StableDiffusionXLControlNetPipeline, ControlNetModel

cn = ControlNetModel.from_pretrained(
    "diffusers/controlnet-depth-sdxl-1.0", torch_dtype=torch.float16)
pipe2 = StableDiffusionXLControlNetPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    controlnet=cn, torch_dtype=torch.float16).to("cuda")
pipe2.load_ip_adapter("h94/IP-Adapter", subfolder="sdxl_models",
                      weight_name="ip-adapter_sdxl.bin")
pipe2.set_ip_adapter_scale(0.5)

edges = cv2.Canny(np.array(ref), 100, 200)
cond = np.stack([edges] * 3, axis=-1)

final = pipe2(
    prompt="premium product shot, brand style, soft light",
    image=cond, ip_adapter_image=Image.fromarray(np.array(ref)),
    controlnet_conditioning_scale=0.7,
    num_inference_steps=30, guidance_scale=7.5,
).images[0]

import hashlib, json
from dataclasses import dataclass, asdict

@dataclass
class Job:
    mj_prompt: str
    sd_prompt: str
    ip_scale: float
    cn_scale: float
    seed: int

def jid(j):
    return hashlib.md5(json.dumps(asdict(j), sort_keys=True)
                       .encode()).hexdigest()[:12]