AI漫剧全流程制作实战：从剧本到成片，手把手搭建自动化管线

用户输入故事主题
    ↓
[LLM] 生成结构化分镜JSON（含场景描述、对话、时长）
    ↓
[分镜→提示词] 用另一个LLM转换为SD正/负向提示词 + ControlNet控制条件
    ↓
[角色一致性] 加载预训练的LoRA（角色A/B） + IP-Adapter（标准照）
    ↓
[关键帧生成] SDXL + ControlNet(OpenPose/Depth) 批量出图，每场景4~6帧
    ↓
[动态化] AnimateDiff v3 生成16帧短视频片段（可调节运动幅度）
    ↓
[音频] Edge TTS 多情感语音生成 + 时间轴对齐
    ↓
[口型] Wav2Lip + GFPGAN 面部增强（仅针对说话人）
    ↓
[合成] MoviePy + FFmpeg 拼接、加字幕、转场

import json
import requests

def generate_script(theme: str) -> list:
    """
    调用本地Qwen-72B服务（OpenAI兼容接口）
    返回 list[dict]，每个dict包含 scene_id, location, characters, 
           visual_desc, dialogue(list of {speaker, text}), duration_sec
    """
    system = """你是一个漫剧编剧。输出严格JSON数组，每个元素结构如下：
    {
      "scene_id": 0,
      "location": "咖啡馆",
      "characters": ["男主", "女主"],
      "visual_desc": "雨天，室内暖光，女主低头端咖啡杯，男主望向窗外",
      "dialogue": [{"speaker": "女主", "text": "我明天就要走了。"}],
      "duration_sec": 6
    }
    对话要简洁，每场景不超过4句。
    """
    user = f"主题：{theme}"
    payload = {
        "model": "Qwen/Qwen-72B-Chat",
        "messages": [{"role": "system", "content": system},
                     {"role": "user", "content": user}],
        "temperature": 0.7
    }
    resp = requests.post("http://localhost:8000/v1/chat/completions", json=payload)
    data = resp.json()["choices"][0]["message"]["content"]
    # 提取JSON（防止LLM输出多余文字）
    import re
    json_str = re.search(r'\[.*\]', data, re.DOTALL).group()
    return json.loads(json_str)

def enhance_prompt(visual_desc: str, style: str = "anime") -> tuple:
    """
    调用Mistral-7B将视觉描述转为SD正负提示词
    返回 (positive, negative)
    """
    prompt = f"""将以下场景描述转化为Stable Diffusion提示词，画风为{style}。
    输出格式：正面提示词：... 负面提示词：...
    描述：{visual_desc}
    """
    # 调用本地Mistral服务（省略）
    # 示例返回：
    pos = "masterpiece, best quality, anime style, rainy day, warm interior lighting, a young woman holding a coffee cup, looking down, sad expression, detailed background"
    neg = "bad anatomy, blurry, ugly, watermark, lowres, extra limbs"
    return pos, neg

# config.json 关键参数
pretrained_model_name_or_path: "stabilityai/stable-diffusion-xl-base-1.0"
v2: false
v_parameterization: false
learning_rate: 1e-4
lr_scheduler: "cosine"
lr_warmup_steps: 100
train_batch_size: 2
epoch: 10
save_every_n_epochs: 2
mixed_precision: "fp16"
network_module: "networks.lora"
network_dim: 32
network_alpha: 16

import torch
from diffusers import StableDiffusionXLControlNetPipeline, ControlNetModel
from diffusers import IPAdapterModel, IPAdapterProcessor
from PIL import Image
import cv2
import numpy as np

def generate_keyframe(
    positive_prompt: str,
    negative_prompt: str,
    pose_image: Image.Image,      # 姿态参考图（由上一帧或预设骨骼生成）
    identity_image: Image.Image,  # 角色标准照
    lora_path: str,
    seed: int = 42
) -> Image.Image:
    # 1. 加载ControlNet
    controlnet = ControlNetModel.from_pretrained(
        "lllyasviel/control_v11p_sd15_openpose",  # 注意SD1.5版
        torch_dtype=torch.float16
    )
    # 2. 加载SDXL pipeline
    pipe = StableDiffusionXLControlNetPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0",
        controlnet=controlnet,
        torch_dtype=torch.float16
    ).to("cuda")
    # 3. 加载LoRA
    pipe.load_lora_weights(lora_path, adapter_name="role")
    pipe.set_adapters(["role"], adapter_weights=[1.0])
    # 4. 加载IP-Adapter（需额外注入）
    ip_model = IPAdapterModel.from_pretrained("h94/IP-Adapter", subfolder="models", torch_dtype=torch.float16)
    ip_processor = IPAdapterProcessor()
    # 将identity_image编码
    ip_image = identity_image.resize((224, 224))
    ip_embeds = ip_processor(ip_image, return_tensors="pt").to("cuda")
    # 手动注入到pipeline（简化版，实际需调用pipe.unet的forward钩子，这里只展示思路）
    # 官方推荐用IPAdapterPipeline，但为保持ControlNet共存，我修改了__call__参数
    # 实际我使用ComfyUI API，此处仅示意
    # 5. 生成
    generator = torch.Generator("cuda").manual_seed(seed)
    image = pipe(
        prompt=positive_prompt,
        negative_prompt=negative_prompt,
        image=pose_image,  # ControlNet输入
        num_inference_steps=30,
        guidance_scale=7.0,
        generator=generator,
        ip_adapter_embeds=ip_embeds,  # 需扩展pipe支持
    ).images[0]
    return image

# 关键参数（存于config.yaml）
motion_adapter: "guoyww/animatediff-motion-adapter-v1-5-2"
num_frames: 16   # 每次生成16帧
context_length: 16
context_overlap: 4  # 重叠4帧用于平滑
fps: 8            # 生成后插值到24fps
motion_scale: 1.0  # 基础幅度

# 在pipeline中注入平移向量
pipe.motion_adapter.set_motion_scale(motion_scale)
pipe.motion_adapter.set_translation(dx=0.0, dy=-0.02)  # 缓慢上移

import asyncio
import edge_tts

async def tts_generate(text, voice, output_path, rate='+0%', pitch='+0Hz'):
    communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch)
    await communicate.save(output_path)
    # 返回时长（秒）
    import wave
    with wave.open(output_path, 'rb') as wf:
        frames = wf.getnframes()
        rate = wf.getframerate()
        duration = frames / float(rate)
    return duration

def wav2lip_sync(video_path, audio_path, face_box=None):
    # 1. 提取所有帧
    # 2. 对每一帧裁剪人脸，使用FaceXLib的detection
    # 3. 调用wav2lip的inference，批量生成人脸帧
    # 4. 将生成的人脸贴回原图（使用mask融合）
    # 5. 保存新视频
    # 具体代码较长，可参考开源项目Wav2Lip的demo

from moviepy.editor import VideoFileClip, AudioFileClip, CompositeVideoClip, TextClip, concatenate_videoclips
import os

class ComicCompositor:
    def __init__(self, scenes, output_dir):
        self.scenes = scenes  # list of dict with keys: video_path, audio_path, subtitle_text, start_time, end_time
        self.output_dir = output_dir
        self.temp_clips = []
    
    def add_scene(self, scene):
        # video clip without audio
        vid = VideoFileClip(scene['video_path'])
        # trim to exact duration
        dur = scene.get('duration', vid.duration)
        vid = vid.subclip(0, min(dur, vid.duration))
        # audio
        if os.path.exists(scene['audio_path']):
            aud = AudioFileClip(scene['audio_path'])
            # if audio shorter, loop or pad
            if aud.duration < dur:
                aud = aud.loop(duration=dur)
            else:
                aud = aud.subclip(0, dur)
            vid = vid.set_audio(aud)
        # subtitle
        txt = TextClip(scene['subtitle_text'], fontsize=28, color='white', stroke_color='black', stroke_width=1, font='SimHei')
        txt = txt.set_position(('center', 0.85), relative=True).set_duration(dur)
        final = CompositeVideoClip([vid, txt])
        self.temp_clips.append(final)
    
    def export(self, output_path, transition_duration=0.5):
        # 添加淡入淡出转场
        final_clips = []
        for i, clip in enumerate(self.temp_clips):
            if i > 0:
                clip = clip.crossfadein(transition_duration)
            final_clips.append(clip)
        final = concatenate_videoclips(final_clips, method="compose")
        final.write_videofile(output_path, fps=24, codec='libx264', audio_codec='aac')

comic_pipeline/
├── config.yaml            # 所有模型路径、参数
├── scripts/
│   ├── generate_script.py
│   ├── generate_frames.py
│   ├── animate_diff.py
│   ├── tts.py
│   ├── wav2lip_sync.py
│   └── compositor.py
├── models/                # 存放所有预训练权重
├── outputs/               # 每部作品输出目录
├── temp/                  # 中间文件
└── main.py                # 总入口

from scripts.pipeline import run_pipeline
run_pipeline(theme="雨天咖啡馆告别", output_dir="outputs/my_comic")