零基础 AI 漫剧智能量产创作营：从提示词工程到 LoRA 微调的全链路实战

用户输入(故事梗概) 
    → LLM生成分镜JSON 
    → Python调度ComfyUI API 
    → 批量渲染分镜 
    → 自动合成视频

# 创建虚拟环境
conda create -n aimanga python=3.10 -y
conda activate aimanga

# 安装 PyTorch（CUDA 12.1）
pip install torch==2.3.0 torchvision==0.18.0 torchaudio==2.3.0 --index-url https://download.pytorch.org/whl/cu121

# 安装 diffusers + transformers
pip install diffusers==0.30.3 transformers==4.44.2 accelerate==0.34.2

# 安装 ComfyUI（作为后端引擎）
git clone https://github.com/comfyanonymous/ComfyUI.git
cd ComfyUI
pip install -r requirements.txt

# 安装 ControlNet 插件（关键）
cd custom_nodes
git clone https://github.com/Fannovel16/comfyui_controlnet_aux.git

# 安装 AnimateDiff（视频化）
git clone https://github.com/Kosinkadink/ComfyUI-AnimateDiff-Evolved.git

# 回到主环境安装辅助库
pip install opencv-python pillow numpy moviepy edge-tts

import json
import requests

SYSTEM_PROMPT = """你是一位资深漫剧分镜师。请将用户提供的故事情节转化为分镜脚本，输出严格合法的 JSON，格式如下：
{
  "characters": [
    {"id": "C1", "name": "角色名", "gender": "男/女", "age": 20, "style": "日系/美系"},
    ...
  ],
  "scenes": [
    {
      "scene_id": 1,
      "shot_type": "中景/特写/全景",
      "character_ids": ["C1"],
      "action": "角色动作描述",
      "expression": "表情描述",
      "bg": "背景描述",
      "dialogue": "台词（可选）",
      "duration_sec": 3
    }
  ]
}
只输出 JSON，不要任何额外文字。"""

def generate_script(story: str, api_url: str = "http://localhost:8000/v1/chat/completions") -> dict:
    """调用本地部署的 Qwen/DeepSeek 生成分镜脚本"""
    payload = {
        "model": "Qwen2.5-72B-Instruct",
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"故事内容：{story}"}
        ],
        "temperature": 0.3,
        "max_tokens": 4096,
        "response_format": {"type": "json_object"}  # 强制 JSON 输出
    }
    resp = requests.post(api_url, json=payload, timeout=120)
    raw = resp.json()["choices"][0]["message"]["content"]
    # 去除可能存在的 markdown 代码块标记
    raw = raw.strip().strip("```json").strip("```")
    return json.loads(raw)

# 示例输入
story = "一名落魄剑客在沙漠古镇中偶遇被追杀的机械少女，两人结伴寻找传说中的天空之城。"
script = generate_script(story)
print(json.dumps(script, indent=2, ensure_ascii=False))

{
  "characters": [
    {"id": "C1", "name": "剑客·尘", "gender": "男", "age": 28, "style": "水墨武侠"},
    {"id": "C2", "name": "机械少女·翎", "gender": "女", "age": 16, "style": "赛博+古风"}
  ],
  "scenes": [
    {
      "scene_id": 1,
      "shot_type": "全景",
      "character_ids": ["C1"],
      "action": "尘牵马行走在沙尘中",
      "expression": "坚毅而疲惫",
      "bg": "残破的土墙、风沙、血色夕阳",
      "dialogue": "三年了，还是没有找到...",
      "duration_sec": 4
    }
  ]
}

import cv2
import os
from diffusers.utils import load_image
from transformers import pipeline

# 使用 YOLOv8 检测人脸并裁剪
face_detector = pipeline("object-detection", model="hustvl/yolov8x")

def preprocess_images(input_dir: str, output_dir: str, size: int = 1024):
    os.makedirs(output_dir, exist_ok=True)
    for fname in os.listdir(input_dir):
        if not fname.lower().endswith(('.png', '.jpg', '.jpeg')):
            continue
        img_path = os.path.join(input_dir, fname)
        img = cv2.imread(img_path)
        if img is None:
            continue
        # 检测人脸
        results = face_detector(img)
        if not results:
            # 若无清晰人脸，使用中心裁剪
            h, w = img.shape[:2]
            c_x, c_y = w//2, h//2
            crop = img[c_y-size//2:c_y+size//2, c_x-size//2:c_x+size//2]
        else:
            box = results[0]['box']
            x1, y1, x2, y2 = int(box['xmin']), int(box['ymin']), int(box['xmax']), int(box['ymax'])
            # 扩大 1.5 倍包含肩部
            pad = int(max(x2-x1, y2-y1) * 0.25)
            x1, y1 = max(0, x1-pad), max(0, y1-pad)
            x2, y2 = min(w, x2+pad), min(h, y2+pad)
            crop = img[y1:y2, x1:x2]
        crop = cv2.resize(crop, (size, size))
        cv2.imwrite(os.path.join(output_dir, f"processed_{fname}"), crop)
        print(f"✅ 已处理: {fname}")

preprocess_images("./raw_characters/", "./processed_characters/")

import torch
from diffusers import StableDiffusionXLPipeline, AutoencoderKL, DDPMScheduler
from diffusers.optimization import get_scheduler
from peft import LoraConfig, get_peft_model
from torch.utils.data import DataLoader, Dataset
from PIL import Image
import os

class CharacterDataset(Dataset):
    def __init__(self, image_dir, tokenizer, size=1024, center_crop=True):
        self.image_paths = [os.path.join(image_dir, f) for f in os.listdir(image_dir) if f.endswith(('png','jpg'))]
        self.tokenizer = tokenizer
        self.size = size
        self.center_crop = center_crop

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, idx):
        img = Image.open(self.image_paths[idx]).convert("RGB")
        # 统一 Prompt：角色名 + 风格触发词
        prompt = "a man named Chen, wuxia style, ink wash painting, detailed face, high quality"
        return {"pixel_values": img, "prompt": prompt}

def train_lora(
    model_id="stabilityai/stable-diffusion-xl-base-1.0",
    dataset_dir="./processed_characters/",
    output_dir="./lora_output/",
    rank=32,
    epochs=50,
    batch_size=2,
    lr=1e-4,
):
    # 加载 SDXL
    pipe = StableDiffusionXLPipeline.from_pretrained(
        model_id, torch_dtype=torch.float16, variant="fp16"
    )
    pipe = pipe.to("cuda")
    
    # 冻结 VAE 和 text encoder，只微调 UNet
    pipe.vae.requires_grad_(False)
    pipe.text_encoder.requires_grad_(False)
    pipe.text_encoder_2.requires_grad_(False)
    
    # 配置 LoRA
    lora_config = LoraConfig(
        r=rank,
        lora_alpha=rank * 2,
        target_modules=["to_q", "to_v", "to_k", "to_out.0"],
        lora_dropout=0.1,
        bias="none",
        task_type="text_to_image",
    )
    unet = get_peft_model(pipe.unet, lora_config)
    unet.print_trainable_parameters()  # 只训练约 80M 参数
    
    # 优化器 & 调度器
    optimizer = torch.optim.AdamW(unet.parameters(), lr=lr, weight_decay=0.01)
    scheduler = get_scheduler(
        "cosine",
        optimizer=optimizer,
        num_warmup_steps=100,
        num_training_steps=epochs * len(CharacterDataset(...))
    )
    
    # 噪声调度器
    noise_scheduler = DDPMScheduler.from_pretrained(model_id, subfolder="scheduler")
    
    # 训练循环（简化版，仅核心逻辑）
    for epoch in range(epochs):
        for batch in dataloader:
            # 编码 prompt
            text_embeddings = pipe.encode_prompt(batch["prompt"], device="cuda")
            # 加噪
            latents = pipe.vae.encode(batch["pixel_values"].to("cuda")).latent_dist.sample()
            latents = latents * pipe.vae.config.scaling_factor
            noise = torch.randn_like(latents)
            timesteps = torch.randint(0, noise_scheduler.config.num_train_timesteps, (batch_size,), device="cuda")
            noisy_latents = noise_scheduler.add_noise(latents, noise, timesteps)
            # 预测噪声
            noise_pred = unet(noisy_latents, timesteps, text_embeddings).sample
            loss = torch.nn.functional.mse_loss(noise_pred, noise)
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()
        print(f"Epoch {epoch+1}/{epochs}, Loss: {loss.item():.4f}")
    
    # 保存 LoRA 权重
    unet.save_pretrained(output_dir)
    print(f"✅ LoRA 已保存至 {output_dir}")

if __name__ == "__main__":
    train_lora()

import json
import requests
import base64
from io import BytesIO
import time

class ComfyUIClient:
    def __init__(self, server="127.0.0.1:8188"):
        self.server = server
        self.base_url = f"http://{server}"
        self.client_id = "manga_producer_001"

    def queue_prompt(self, workflow_json: dict) -> str:
        """提交工作流并返回 prompt_id"""
        payload = {"prompt": workflow_json, "client_id": self.client_id}
        resp = requests.post(f"{self.base_url}/prompt", json=payload)
        return resp.json()["prompt_id"]

    def wait_for_result(self, prompt_id: str, timeout=300):
        """轮询结果直到完成"""
        url = f"{self.base_url}/history/{prompt_id}"
        start = time.time()
        while time.time() - start < timeout:
            resp = requests.get(url)
            if resp.status_code == 200 and prompt_id in resp.json():
                history = resp.json()[prompt_id]
                if "outputs" in history:
                    return history["outputs"]
            time.sleep(1)
        raise TimeoutError(f"Prompt {prompt_id} 超时未完成")

    def generate_scene(self, prompt: str, negative_prompt: str = "", 
                        controlnet_image: str = None) -> Image.Image:
        """渲染单个分镜（动态替换 prompt）"""
        # 加载工作流模板
        with open("workflow_api.json", "r") as f:
            workflow = json.load(f)

        # 查找并替换 prompt 节点 (节点 id 需根据实际导出调整)
        for node_id, node in workflow.items():
            if node["class_type"] == "CLIPTextEncode":
                if "positive" in node["_meta"]["title"].lower():
                    node["inputs"]["text"] = prompt
                elif "negative" in node["_meta"]["title"].lower():
                    node["inputs"]["text"] = negative_prompt
            if node["class_type"] == "LoadImage" and controlnet_image:
                # 上传 ControlNet 参考图（Base64）
                img_b64 = base64.b64encode(controlnet_image).decode('utf-8')
                node["inputs"]["image"] = img_b64
                node["inputs"]["upload"] = "image.png"

        prompt_id = self.queue_prompt(workflow)
        outputs = self.wait_for_result(prompt_id)
        # 提取生成的图片（根据实际节点调整）
        for node_id, node_output in outputs.items():
            if "images" in node_output:
                img_data = node_output["images"][0]
                img_bytes = base64.b64decode(img_data["image"])
                return Image.open(BytesIO(img_bytes))
        return None

# 批量生成
client = ComfyUIClient()
scenes = script["scenes"]
for idx, scene in enumerate(scenes):
    # 构建 prompt: 角色 + 动作 + 背景 + 风格词
    char_names = [c["name"] for c in script["characters"] if c["id"] in scene["character_ids"]]
    prompt = f"{', '.join(char_names)}, {scene['action']}, {scene['bg']}, ink wash style, masterpiece, best quality"
    negative = "bad anatomy, distorted face, extra limbs, lowres, watermark, text"
    
    # 生成 ControlNet 姿态图（使用 OpenPose）
    pose_image = generate_openpose(scene["action"])  # 自定义函数，简化
    
    img = client.generate_scene(prompt, negative, pose_image)
    img.save(f"./outputs/scene_{idx+1:03d}.png")
    print(f"✅ 场景 {idx+1}/{len(scenes)} 渲染完成")

import torch
from diffusers import AnimateDiffPipeline, MotionAdapter
from diffusers.utils import export_to_gif

def animate_scene(image_pil, prompt, motion_scale=1.0, num_frames=16):
    """将单张分镜转换为动态短片"""
    adapter = MotionAdapter.from_pretrained("guoyww/animatediff-motion-adapter-v1-5-2")
    pipeline = AnimateDiffPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0",
        motion_adapter=adapter,
        torch_dtype=torch.float16
    ).to("cuda")
    pipeline.enable_model_cpu_offload()
    
    # 使用图像引导 (Image-to-Video)
    output = pipeline(
        prompt=prompt,
        image=image_pil,
        num_frames=num_frames,
        motion_scale=motion_scale,
        guidance_scale=7.5,
        num_inference_steps=25
    ).frames[0]
    
    # 导出为 GIF 或 MP4
    export_to_gif(output, "./outputs/animated_scene.gif")
    return output

# 批量生成视频片段
for i in range(len(scenes)):
    static_img = Image.open(f"./outputs/scene_{i+1:03d}.png")
    prompt = scenes[i]["action"] + ", " + scenes[i]["bg"]
    animate_scene(static_img, prompt, motion_scale=0.8, num_frames=24)

import asyncio
import edge_tts

async def generate_voice(text: str, output_file: str, voice: str = "zh-CN-XiaoxiaoNeural"):
    """使用 Edge TTS 生成配音"""
    communicate = edge_tts.Communicate(text, voice)
    await communicate.save(output_file)

# 为每个分镜生成配音
for idx, scene in enumerate(scenes):
    if scene.get("dialogue"):
        asyncio.run(generate_voice(scene["dialogue"], f"./outputs/audio_{idx+1:03d}.mp3"))

# 使用 moviepy 合成完整视频
from moviepy.editor import VideoFileClip, AudioFileClip, concatenate_videoclips

clips = []
for i in range(len(scenes)):
    video = VideoFileClip(f"./outputs/scene_{i+1:03d}.mp4")  # 假设已导出 MP4
    audio = AudioFileClip(f"./outputs/audio_{i+1:03d}.mp3")
    video = video.set_audio(audio)
    clips.append(video)

final = concatenate_videoclips(clips)
final.write_videofile("./outputs/final_manga_episode.mp4", fps=24, codec="libx264")
print("🎬 完整漫剧已合成！")

# tasks.py
from celery import Celery
from celery.result import AsyncResult
import json

app = Celery('manga_producer', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def produce_episode(self, story_text: str, episode_id: str):
    try:
        # 1. 生成脚本
        script = generate_script(story_text)
        # 2. 训练 LoRA（如果有新角色）
        for char in script["characters"]:
            if not lora_exists(char["id"]):
                train_lora(char["id"], collect_images(char["id"]))
        # 3. 批量渲染
        for scene in script["scenes"]:
            render_scene(scene)
        # 4. 合成视频
        final_video = compose_video(script)
        # 5. 上传至 COS
        upload_to_tencent_cos(final_video, f"episodes/{episode_id}.mp4")
        return {"episode_id": episode_id, "status": "success"}
    except Exception as e:
        self.retry(exc=e, countdown=60)
        return {"episode_id": episode_id, "status": "failed", "error": str(e)}

# 调用
result = produce_episode.delay("一个关于星际探险家的故事...", "EP001")
print(f"任务 ID: {result.id}")

from qcloud_cos import CosConfig, CosS3Client
import os

def upload_to_tencent_cos(local_path: str, cos_key: str):
    config = CosConfig(
        Region=os.getenv("COS_REGION", "ap-guangzhou"),
        SecretId=os.getenv("COS_SECRET_ID"),
        SecretKey=os.getenv("COS_SECRET_KEY"),
        Scheme="https"
    )
    client = CosS3Client(config)
    response = client.upload_file(
        Bucket=os.getenv("COS_BUCKET"),
        LocalFilePath=local_path,
        Key=cos_key,
        PartSize=10,
        MAXThread=5
    )
    print(f"✅ 已上传至 COS: {cos_key}")
    return response

# 集成到 Pipeline
upload_to_tencent_cos("./outputs/final_manga_episode.mp4", "episodes/EP001.mp4")

ai_manga_production/
├── comfyui/                     # ComfyUI 引擎
├── models/                      # 所有模型权重
│   ├── sdxl/
│   ├── lora/
│   └── controlnet/
├── scripts/
│   ├── generate_script.py       # LLM 剧本生成
│   ├── train_lora.py           # LoRA 训练
│   ├── comfyui_client.py       # API 调度
│   ├── animate_engine.py       # 视频化
│   └── pipeline.py             # 主流程编排
├── config.yaml                  # 全局配置（API key，路径等）
├── requirements.txt
└── README.md