API 开发 #

┌─────────────────────────────────────────────────────────────┐
│                      API 方式对比                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  方式            特点              适用场景                  │
│  ─────────────────────────────────────────────────────────  │
│  WebUI API      RESTful 接口     简单集成                   │
│  ComfyUI API    工作流驱动       复杂工作流                 │
│  Diffusers      Python 库        深度定制                   │
│  SD.Next API    增强版 API        高级功能                  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

启动 WebUI API：

启动参数：
webui-user.bat
set COMMANDLINE_ARGS=--api --listen

参数说明：
├── --api: 启用 API
├── --listen: 允许外部访问
├── --port 7860: 指定端口
└── --api-auth user:pass: API 认证

验证 API：
访问 http://127.0.0.1:7860/docs
查看 Swagger 文档

主要 API 端点：

文生图：
POST /sdapi/v1/txt2img

图生图：
POST /sdapi/v1/img2img

局部重绘：
POST /sdapi/v1/img2img (带 mask)

模型列表：
GET /sdapi/v1/sd-models

采样器列表：
GET /sdapi/v1/samplers

进度查询：
GET /sdapi/v1/progress

中断生成：
POST /sdapi/v1/interrupt

选项设置：
POST /sdapi/v1/options

Python 请求示例：

import requests
import base64
import json

url = "http://127.0.0.1:7860"

payload = {
    "prompt": "a beautiful girl, anime style, masterpiece",
    "negative_prompt": "ugly, low quality, blurry",
    "steps": 25,
    "cfg_scale": 7,
    "width": 512,
    "height": 768,
    "sampler_name": "DPM++ 2M Karras",
    "seed": -1,
    "batch_size": 1,
    "n_iter": 1,
}

response = requests.post(
    url=f"{url}/sdapi/v1/txt2img",
    json=payload
)

result = response.json()

# 保存图像
for i, img_data in enumerate(result["images"]):
    img_bytes = base64.b64decode(img_data)
    with open(f"output_{i}.png", "wb") as f:
        f.write(img_bytes)

img2img 请求示例：

import requests
import base64

url = "http://127.0.0.1:7860"

# 读取输入图像
with open("input.png", "rb") as f:
    img_base64 = base64.b64encode(f.read()).decode()

payload = {
    "prompt": "anime style, detailed",
    "negative_prompt": "ugly, low quality",
    "init_images": [img_base64],
    "denoising_strength": 0.5,
    "steps": 25,
    "cfg_scale": 7,
    "width": 512,
    "height": 768,
    "sampler_name": "DPM++ 2M Karras",
}

response = requests.post(
    url=f"{url}/sdapi/v1/img2img",
    json=payload
)

result = response.json()

txt2img 完整参数：

{
    "prompt": "提示词",
    "negative_prompt": "负向提示词",
    "steps": 25,                    // 采样步数
    "cfg_scale": 7,                 // CFG Scale
    "width": 512,                   // 宽度
    "height": 512,                  // 高度
    "sampler_name": "Euler a",      // 采样器
    "seed": -1,                     // 种子
    "subseed": -1,                  // 子种子
    "subseed_strength": 0,          // 子种子强度
    "batch_size": 1,                // 批次大小
    "n_iter": 1,                    // 迭代次数
    "restore_faces": false,         // 面部修复
    "tiling": false,                // 平铺
    "eta": 0,                       // Eta
    "s_churn": 0,                   // Sigma churn
    "s_tmax": 0,                    // Sigma tmax
    "s_tmin": 0,                    // Sigma tmin
    "s_noise": 1,                   // Sigma noise
    "override_settings": {},        // 覆盖设置
    "script_args": [],              // 脚本参数
    "alwayson_scripts": {}          // 始终运行的脚本
}

API 中使用 LoRA：

方法一：提示词语法
payload = {
    "prompt": "<lora:style:0.8>, a beautiful girl",
    ...
}

方法二：alwayson_scripts
payload = {
    "prompt": "a beautiful girl",
    "alwayson_scripts": {
        "LoRA": {
            "args": [
                {"lora_name": "style.safetensors", "weight": 0.8}
            ]
        }
    }
}

API 中使用 ControlNet：

import base64

# 读取 ControlNet 图像
with open("control_image.png", "rb") as f:
    control_img = base64.b64encode(f.read()).decode()

payload = {
    "prompt": "a beautiful girl",
    "alwayson_scripts": {
        "controlnet": {
            "args": [
                {
                    "enabled": True,
                    "module": "canny",
                    "model": "control_canny",
                    "weight": 1.0,
                    "image": control_img,
                    "guidance_start": 0.0,
                    "guidance_end": 1.0,
                }
            ]
        }
    }
}

启动 ComfyUI API：

python main.py --listen

默认端口: 8188
API 文档: http://127.0.0.1:8188/docs

ComfyUI API 端点：

提交任务：
POST /prompt

查询历史：
GET /history/{prompt_id}

获取图像：
GET /view

WebSocket：
WS /ws

上传图像：
POST /upload/image

获取对象信息：
GET /object_info

ComfyUI API 调用：

import requests
import json

url = "http://127.0.0.1:8188"

# 加载工作流
with open("workflow.json") as f:
    workflow = json.load(f)

# 提交任务
response = requests.post(
    url=f"{url}/prompt",
    json={"prompt": workflow}
)

prompt_id = response.json()["prompt_id"]

# 查询状态
import time
while True:
    history = requests.get(
        url=f"{url}/history/{prompt_id}"
    ).json()
    
    if prompt_id in history:
        print("完成")
        break
    
    time.sleep(1)

WebSocket 实时监听：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    
    if data["type"] == "executing":
        print(f"执行中: {data['data']['node']}")
    
    elif data["type"] == "progress":
        print(f"进度: {data['data']['value']}/{data['data']['max']}")
    
    elif data["type"] == "executed":
        print("完成")
        images = data["data"]["output"]["images"]
        for img in images:
            print(f"图像: {img['filename']}")

ws = websocket.WebSocketApp(
    "ws://127.0.0.1:8188/ws",
    on_message=on_message
)
ws.run_forever()

安装 Diffusers：

pip install diffusers transformers accelerate

基础使用：

from diffusers import StableDiffusionPipeline
import torch

# 加载模型
pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    torch_dtype=torch.float16
)
pipe.to("cuda")

# 生成图像
image = pipe(
    prompt="a beautiful girl, anime style",
    negative_prompt="ugly, low quality",
    num_inference_steps=25,
    guidance_scale=7,
    width=512,
    height=768,
).images[0]

image.save("output.png")

Diffusers 使用 LoRA：

from diffusers import StableDiffusionPipeline
import torch

pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    torch_dtype=torch.float16
)
pipe.to("cuda")

# 加载 LoRA
pipe.load_lora_weights("path/to/lora")

# 生成
image = pipe(
    prompt="a beautiful girl, anime style",
    cross_attention_kwargs={"scale": 0.8},  # LoRA 权重
).images[0]

Diffusers 使用 ControlNet：

from diffusers import StableDiffusionControlNetPipeline, ControlNetModel
from diffusers.utils import load_image
import torch

# 加载 ControlNet
controlnet = ControlNetModel.from_pretrained(
    "lllyasviel/sd-controlnet-canny",
    torch_dtype=torch.float16
)

# 加载管道
pipe = StableDiffusionControlNetPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    controlnet=controlnet,
    torch_dtype=torch.float16
)
pipe.to("cuda")

# 加载控制图像
control_image = load_image("control.png")

# 生成
image = pipe(
    prompt="a beautiful girl",
    image=control_image,
).images[0]

Stable Diffusion API 客户端类：

import requests
import base64
import time
from typing import List, Optional

class SDWebUIClient:
    def __init__(self, url: str = "http://127.0.0.1:7860"):
        self.url = url
    
    def txt2img(
        self,
        prompt: str,
        negative_prompt: str = "",
        steps: int = 25,
        cfg_scale: float = 7,
        width: int = 512,
        height: int = 512,
        sampler: str = "DPM++ 2M Karras",
        seed: int = -1,
        batch_size: int = 1,
    ) -> List[bytes]:
        
        payload = {
            "prompt": prompt,
            "negative_prompt": negative_prompt,
            "steps": steps,
            "cfg_scale": cfg_scale,
            "width": width,
            "height": height,
            "sampler_name": sampler,
            "seed": seed,
            "batch_size": batch_size,
        }
        
        response = requests.post(
            f"{self.url}/sdapi/v1/txt2img",
            json=payload
        )
        
        images = []
        for img_data in response.json()["images"]:
            images.append(base64.b64decode(img_data))
        
        return images
    
    def img2img(
        self,
        prompt: str,
        init_image: bytes,
        denoising_strength: float = 0.5,
        **kwargs
    ) -> List[bytes]:
        
        img_base64 = base64.b64encode(init_image).decode()
        
        payload = {
            "prompt": prompt,
            "init_images": [img_base64],
            "denoising_strength": denoising_strength,
            **kwargs
        }
        
        response = requests.post(
            f"{self.url}/sdapi/v1/img2img",
            json=payload
        )
        
        images = []
        for img_data in response.json()["images"]:
            images.append(base64.b64decode(img_data))
        
        return images
    
    def get_progress(self) -> dict:
        response = requests.get(f"{self.url}/sdapi/v1/progress")
        return response.json()
    
    def interrupt(self):
        requests.post(f"{self.url}/sdapi/v1/interrupt")

# 使用示例
client = SDWebUIClient()

images = client.txt2img(
    prompt="a beautiful girl, anime style",
    width=512,
    height=768,
)

for i, img in enumerate(images):
    with open(f"output_{i}.png", "wb") as f:
        f.write(img)

批量生成脚本：

import os
from typing import List
import time

def batch_generate(
    client: SDWebUIClient,
    prompts: List[str],
    output_dir: str = "outputs",
    **kwargs
):
    os.makedirs(output_dir, exist_ok=True)
    
    for i, prompt in enumerate(prompts):
        print(f"生成 {i+1}/{len(prompts)}: {prompt[:50]}...")
        
        images = client.txt2img(prompt=prompt, **kwargs)
        
        for j, img in enumerate(images):
            filename = f"{output_dir}/img_{i}_{j}.png"
            with open(filename, "wb") as f:
                f.write(img)
        
        print(f"保存到 {filename}")

# 使用示例
prompts = [
    "a beautiful girl with long hair",
    "a handsome man in suit",
    "a cute cat sitting on a chair",
]

batch_generate(client, prompts, output_dir="batch_outputs")

完善的错误处理：

import requests
from requests.exceptions import RequestException
import time

def generate_with_retry(
    client: SDWebUIClient,
    prompt: str,
    max_retries: int = 3,
    retry_delay: float = 5.0,
    **kwargs
):
    for attempt in range(max_retries):
        try:
            return client.txt2img(prompt=prompt, **kwargs)
        except RequestException as e:
            print(f"请求失败 (尝试 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                time.sleep(retry_delay)
            else:
                raise

并发生成：

import concurrent.futures
from typing import List

def concurrent_generate(
    client: SDWebUIClient,
    prompts: List[str],
    max_workers: int = 2,
    **kwargs
):
    results = []
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(client.txt2img, prompt=prompt, **kwargs): prompt
            for prompt in prompts
        }
        
        for future in concurrent.futures.as_completed(futures):
            prompt = futures[future]
            try:
                images = future.result()
                results.append((prompt, images))
            except Exception as e:
                print(f"生成失败: {prompt}, 错误: {e}")
    
    return results

进度监控：

import time

def generate_with_progress(client: SDWebUIClient, prompt: str, **kwargs):
    import threading
    
    result = None
    exception = None
    
    def generate():
        nonlocal result, exception
        try:
            result = client.txt2img(prompt=prompt, **kwargs)
        except Exception as e:
            exception = e
    
    thread = threading.Thread(target=generate)
    thread.start()
    
    while thread.is_alive():
        progress = client.get_progress()
        print(f"\r进度: {progress['progress']:.1%} | ETA: {progress['eta_relative']:.1f}s", end="")
        time.sleep(0.5)
    
    thread.join()
    
    if exception:
        raise exception
    
    return result