基于DeepSeek提示词工程与Stable Diffusion的AI绘画自动化管线：从文本到图像的高效生成实践

┌─────────────┐     ┌─────────────────┐     ┌─────────────────────┐
│  用户输入   │ ──> │ DeepSeek提示词   │ ──> │  提示词解析器       │
│ (中文/口语) │     │ 重构引擎         │     │ (提取正/负提示词+    │
└─────────────┘     └─────────────────┘     │  参数建议)           │
                                             └──────────┬──────────┘
                                                        ▼
┌─────────────┐     ┌─────────────────┐     ┌─────────────────────┐
│  COS存储    │ <── │ 后处理/水印     │ <── │ Stable Diffusion   │
│  /结果图    │     │ 裁剪/超分       │     │ 生成引擎 (v1.5/XL)  │
└─────────────┘     └─────────────────┘     └──────────┬──────────┘
                                                        │
                                              ┌─────────┴─────────┐
                                              │ ControlNet + LoRA │
                                              │ 条件注入模块       │
                                              └───────────────────┘

# config/prompt_template.py

SYSTEM_PROMPT = """你是一位顶级AI绘画提示词工程师，精通Stable Diffusion、Midjourney提示词语法。
你的任务是将用户输入的自然语言描述，重构为结构化的英文prompt，并附加负面提示词与生成参数建议。

输出必须严格遵循JSON格式：
{
    "positive_prompt": "主提示词，包含主体、环境、光线、风格、画质词，以逗号分隔",
    "negative_prompt": "负面提示词，防止畸形、低画质等",
    "cfg_scale": 浮点数,
    "steps": 整数,
    "seed": 整数或null,
    "width": 整数,
    "height": 整数,
    "style_hint": "写实/二次元/油画/赛博朋克等"
}

注意：positive_prompt必须包含至少5个专业美术词汇，且长度控制在75-150 token之间。
"""

USER_TEMPLATE = """用户原始需求：{user_input}
当前风格偏好：{style}（若无则留空）
参考艺术家：{artist}（若无则留空）
请输出JSON。"""

# core/deepseek_client.py

import os
import json
import asyncio
from typing import Dict, Any, Optional
from openai import AsyncOpenAI  # deepseek兼容openai sdk
from tenacity import retry, stop_after_attempt, wait_exponential

class DeepSeekRefiner:
    def __init__(self, api_key: str = None, base_url: str = "https://api.deepseek.com"):
        self.client = AsyncOpenAI(
            api_key=api_key or os.getenv("DEEPSEEK_API_KEY"),
            base_url=base_url
        )
        self.model = "deepseek-chat"
        self.temperature = 0.7  # 保持一定创造性
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def refine(self, user_input: str, style: str = "", artist: str = "") -> Dict[str, Any]:
        from config.prompt_template import SYSTEM_PROMPT, USER_TEMPLATE
        
        user_msg = USER_TEMPLATE.format(user_input=user_input, style=style, artist=artist)
        
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_msg}
            ],
            response_format={"type": "json_object"},  # 强制JSON
            temperature=self.temperature,
            max_tokens=512
        )
        
        raw = response.choices[0].message.content
        try:
            result = json.loads(raw)
            # 校验必要字段
            required = ["positive_prompt", "negative_prompt"]
            for key in required:
                if key not in result:
                    raise ValueError(f"Missing key: {key}")
            return result
        except json.JSONDecodeError as e:
            # 降级：尝试从文本中抽取JSON
            import re
            match = re.search(r'\{.*\}', raw, re.DOTALL)
            if match:
                return json.loads(match.group())
            raise RuntimeError(f"DeepSeek返回非法JSON: {raw}") from e

# core/sd_generator.py

import torch
from diffusers import (
    StableDiffusionXLPipeline, 
    StableDiffusionPipeline,
    ControlNetModel,
    AutoencoderKL,
    DPMSolverMultistepScheduler
)
from diffusers.utils import load_image
from PIL import Image
from typing import Optional, List, Tuple
import numpy as np

class SDGenerator:
    def __init__(
        self,
        model_id: str = "stabilityai/stable-diffusion-xl-base-1.0",
        device: str = "cuda",
        torch_dtype: torch.dtype = torch.float16,
        use_compile: bool = True,
        lora_path: Optional[str] = None
    ):
        self.device = device
        self.dtype = torch_dtype
        
        # 加载VAE（使用fp16加速）
        vae = AutoencoderKL.from_pretrained(
            "madebyollin/sdxl-vae-fp16-fix", 
            torch_dtype=torch_dtype
        )
        
        # 主管线
        self.pipe = StableDiffusionXLPipeline.from_pretrained(
            model_id,
            vae=vae,
            torch_dtype=torch_dtype,
            variant="fp16",
            use_safetensors=True
        )
        
        # 调度器（DPMSolver++ 减少步数）
        self.pipe.scheduler = DPMSolverMultistepScheduler.from_config(
            self.pipe.scheduler.config,
            algorithm_type="dpmsolver++",
            use_karras_sigma=True
        )
        
        self.pipe = self.pipe.to(device)
        
        # 可选LoRA
        if lora_path:
            self.pipe.load_lora_weights(lora_path)
            self.pipe.fuse_lora()
        
        # 编译UNet（首次运行会耗时，但后续加速）
        if use_compile and device == "cuda":
            self.pipe.unet = torch.compile(
                self.pipe.unet, 
                mode="reduce-overhead", 
                fullgraph=True
            )
        
        # 启用VAE切片和注意力切片减少显存
        self.pipe.enable_vae_slicing()
        self.pipe.enable_attention_slicing()
        
    @torch.no_grad()
    def generate(
        self,
        positive_prompt: str,
        negative_prompt: str = "",
        cfg_scale: float = 7.5,
        steps: int = 30,
        width: int = 1024,
        height: int = 1024,
        seed: Optional[int] = None,
        control_image: Optional[Image.Image] = None,
        controlnet_conditioning_scale: float = 0.8
    ) -> Image.Image:
        generator = torch.Generator(device=self.device)
        if seed is not None:
            generator.manual_seed(seed)
        else:
            generator.seed()
        
        # ControlNet注入（如果提供）
        if control_image is not None:
            # 此处需单独加载ControlNet模型，简化示例：使用canny预处理器
            from diffusers import StableDiffusionXLControlNetPipeline
            controlnet = ControlNetModel.from_pretrained(
                "diffusers/controlnet-canny-sdxl-1.0",
                torch_dtype=self.dtype
            ).to(self.device)
            # 重新构建管线（实际生产可缓存）
            pipe = StableDiffusionXLControlNetPipeline(
                vae=self.pipe.vae,
                unet=self.pipe.unet,
                controlnet=controlnet,
                scheduler=self.pipe.scheduler,
                tokenizer=self.pipe.tokenizer,
                tokenizer_2=self.pipe.tokenizer_2,
                text_encoder=self.pipe.text_encoder,
                text_encoder_2=self.pipe.text_encoder_2,
            ).to(self.device)
            result = pipe(
                prompt=positive_prompt,
                negative_prompt=negative_prompt,
                image=control_image,
                controlnet_conditioning_scale=controlnet_conditioning_scale,
                num_inference_steps=steps,
                guidance_scale=cfg_scale,
                width=width,
                height=height,
                generator=generator
            ).images[0]
        else:
            result = self.pipe(
                prompt=positive_prompt,
                negative_prompt=negative_prompt,
                num_inference_steps=steps,
                guidance_scale=cfg_scale,
                width=width,
                height=height,
                generator=generator
            ).images[0]
        
        return result

# core/async_pipeline.py

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict
import aiofiles
from PIL import Image
import io

class AsyncGenerationPipeline:
    def __init__(self, generator: SDGenerator, refiner: DeepSeekRefiner):
        self.generator = generator
        self.refiner = refiner
        self.executor = ThreadPoolExecutor(max_workers=2)  # 根据GPU显存调整
        
    async def generate_single(self, user_input: str, **kwargs) -> Dict:
        # 1. DeepSeek优化提示词
        refined = await self.refiner.refine(user_input, **kwargs)
        
        # 2. 同步生成（但放到线程池避免阻塞事件循环）
        loop = asyncio.get_event_loop()
        image = await loop.run_in_executor(
            self.executor,
            self.generator.generate,
            refined["positive_prompt"],
            refined["negative_prompt"],
            refined.get("cfg_scale", 7.5),
            refined.get("steps", 30),
            refined.get("width", 1024),
            refined.get("height", 1024),
            refined.get("seed", None)
        )
        return {"image": image, "metadata": refined}
    
    async def batch_generate(self, inputs: List[str], concurrency: int = 4) -> List[Dict]:
        semaphore = asyncio.Semaphore(concurrency)
        async def limited_task(input_text):
            async with semaphore:
                return await self.generate_single(input_text)
        tasks = [limited_task(inp) for inp in inputs]
        return await asyncio.gather(*tasks)

# storage/cos_uploader.py

from qcloud_cos import CosConfig, CosS3Client
import os
import hashlib
from datetime import datetime

class COSUploader:
    def __init__(
        self,
        secret_id: str,
        secret_key: str,
        region: str = "ap-guangzhou",
        bucket: str = "ai-art-1234567890"
    ):
        config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)
        self.client = CosS3Client(config)
        self.bucket = bucket
    
    def upload_image(self, image: Image.Image, prefix: str = "generated") -> str:
        # 生成文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        hash_id = hashlib.md5(str(time.time()).encode()).hexdigest()[:8]
        key = f"{prefix}/{timestamp}_{hash_id}.png"
        
        # 转JPEG压缩
        buffer = io.BytesIO()
        image.convert("RGB").save(buffer, format="JPEG", quality=92)
        buffer.seek(0)
        
        response = self.client.put_object(
            Bucket=self.bucket,
            Body=buffer,
            Key=key,
            ContentType="image/jpeg"
        )
        # 返回访问URL（需配置公有读或预签名）
        return f"https://{self.bucket}.cos.{self.region}.myqcloud.com/{key}"

# deploy/values.yaml
replicaCount: 1
image:
  repository: ccr.ccs.tencentyun.com/ai-pipeline/sd-deepseek
  tag: latest
  pullPolicy: Always

resources:
  limits:
    nvidia.com/gpu: 1
    memory: 32Gi
    cpu: 8
  requests:
    nvidia.com/gpu: 1
    memory: 24Gi
    cpu: 4

env:
  - name: DEEPSEEK_API_KEY
    valueFrom:
      secretKeyRef:
        name: deepseek-secret
        key: api-key
  - name: COS_SECRET_ID
    valueFrom:
      secretKeyRef:
        name: cos-secret
        key: secret-id
  - name: COS_SECRET_KEY
    valueFrom:
      secretKeyRef:
        name: cos-secret
        key: secret-key

service:
  type: ClusterIP
  port: 8000

# 水平自动伸缩基于GPU利用率
autoscaling:
  enabled: true
  minReplicas: 1
  maxReplicas: 4
  targetGPUUtilization: 70

# api/server.py

from fastapi import FastAPI, UploadFile, File, Form, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import uuid
from core.async_pipeline import AsyncGenerationPipeline
from storage.cos_uploader import COSUploader
from core.sd_generator import SDGenerator
from core.deepseek_client import DeepSeekRefiner
import os

app = FastAPI(title="AI绘画生成服务")

# 全局初始化（生产环境建议懒加载）
refiner = DeepSeekRefiner()
generator = SDGenerator()
pipeline = AsyncGenerationPipeline(generator, refiner)
uploader = COSUploader(
    secret_id=os.getenv("COS_SECRET_ID"),
    secret_key=os.getenv("COS_SECRET_KEY")
)

class GenerateRequest(BaseModel):
    text: str
    style: Optional[str] = None
    artist: Optional[str] = None
    control_image_url: Optional[str] = None  # 暂不实现远程下载

@app.post("/generate")
async def generate_image(request: GenerateRequest, background_tasks: BackgroundTasks):
    # 异步生成
    result = await pipeline.generate_single(
        request.text, 
        style=request.style or "",
        artist=request.artist or ""
    )
    image = result["image"]
    metadata = result["metadata"]
    
    # 上传COS（可后台执行）
    url = uploader.upload_image(image, prefix="user_generated")
    return {
        "code": 0,
        "data": {
            "url": url,
            "metadata": metadata,
            "seed": metadata.get("seed")
        }
    }

@app.post("/batch")
async def batch_generate(texts: list[str], concurrency: int = 4):
    results = await pipeline.batch_generate(texts, concurrency)
    urls = [uploader.upload_image(res["image"]) for res in results]
    return {"code": 0, "data": [{"url": u, "meta": res["metadata"]} for u, res in zip(urls, results)]}

# 安装依赖
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118
pip install diffusers transformers accelerate openai qcloud-cos fastapi uvicorn aiofiles tenacity

# 设置环境变量
export DEEPSEEK_API_KEY="sk-xxx"
export COS_SECRET_ID="xxx"
export COS_SECRET_KEY="xxx"

# 启动服务
uvicorn api.server:app --host 0.0.0.0 --port 8000 --workers 1  # 单worker因GPU独占

curl -X POST http://localhost:8000/generate -H "Content-Type: application/json" -d '{"text":"一只赛博朋克风格的猫，霓虹灯光，雨夜"}'

{
  "code": 0,
  "data": {
    "url": "https://ai-art-1234567890.cos.ap-guangzhou.myqcloud.com/user_generated/20260812_143022_a1b2c3d4.jpg",
    "metadata": {
      "positive_prompt": "cyberpunk cat, neon lights, rainy night, volumetric lighting, intricate details, 8k, sharp focus, by Ross Tran",
      "negative_prompt": "blurry, deformed, low quality, bad anatomy",
      "cfg_scale": 8.5,
      "steps": 25,
      "seed": 420,
      "style_hint": "赛博朋克"
    }
  }
}