AI漫剧全流程技术解析：从剧本生成到视频成片的生产级实践

from pydantic import BaseModel
from typing import List, Optional

class StoryBeat(BaseModel):
    beat_id: int
    beat_type: str  # "exposition", "rising_action", "climax", "resolution"
    summary: str
    emotional_tone: str
    key_characters: List[str]
    key_props: List[str]

class StoryOutline(BaseModel):
    title: str
    genre: str
    logline: str
    beats: List[StoryBeat]
    total_scenes: int
    estimated_duration: int  # seconds

class CameraAngle:
    """镜头角度枚举"""
    CLOSE_UP = "close_up"
    MEDIUM = "medium"
    WIDE = "wide"
    OVER_SHOULDER = "over_shoulder"

class Shot(BaseModel):
    shot_id: int
    scene_id: int
    camera_angle: CameraAngle
    subject: str  # 画面主体
    background_desc: str
    character_poses: dict  # {char_name: pose_description}
    dialogue: Optional[str]
    sfx: Optional[str]  # 音效
    duration: float  # 秒
    transition: str  # "cut", "fade", "dissolve"

[SYSTEM]
你是一个专业的动画分镜师，擅长将故事转化为视觉化镜头序列。
输出必须符合以下JSON Schema：[schema定义]

[CONTEXT]
故事大纲：{outline}
角色设定：{character_sheets}
风格参考：{style_reference}

[TASK]
将以下场景拆解为3-8个镜头：
场景描述：{scene_description}
情绪基调：{emotion}
关键动作：{key_actions}

[CONSTRAINTS]
- 每个镜头必须包含主体、背景、景别
- 对话场景需明确谁在说话
- 动作场景需拆分连续动作

主体描述模板：
[Character: 林晓] 是一名 [28岁] 的 [女性] 侦探，
[发型: 黑色及肩直发，左侧有一缕挑染银色]，
[面部特征: 丹凤眼，右眼角有一颗小痣]，
[着装: 深灰色风衣，内搭白色衬衫，领口别着银色胸针]，
[标志性元素: 总戴着一副金丝边圆框眼镜]

import cv2
import insightface
from PIL import Image

def align_face_to_reference(generated_img, ref_face_embedding):
    # 1. 检测生成图中的人脸
    face_detector = insightface.model_zoo.get_model('buffalo_l')
    faces = face_detector.get(generated_img)
    
    if len(faces) == 0:
        return generated_img  # 无脸则跳过
    
    # 2. 计算生成脸与参考脸的相似度
    target_face = max(faces, key=lambda f: f['det_score'])
    similarity = cosine_similarity(target_face['embedding'], ref_face_embedding)
    
    if similarity < 0.65:  # 低于阈值触发修复
        # 使用GFPGAN进行面部重建
        repaired = gfpgan_enhancer.enhance(
            generated_img, 
            upscale=1, 
            weight=0.8
        )
        return repaired
    
    return generated_img

┌─────────────────────────────────────────────────────┐
│                   调度层 (Celery)                    │
│  任务队列: shot_generation / bg_generation / fix    │
└─────────────────────────────────────────────────────┘
                           │
┌─────────────────────────────────────────────────────┐
│               ComfyUI Worker Pool                   │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐             │
│  | GPU0 | | GPU1 | | GPU2 | | GPU3 |  (A100 80G) │
│  └──────┘ └──────┘ └──────┘ └──────┘             │
│  每个Worker加载: SDXL + 5个LoRA + ControlNet       │
└─────────────────────────────────────────────────────┘
                           │
┌─────────────────────────────────────────────────────┐
│              共享存储 (NFS / S3)                    │
│  - models/ (基础模型 + LoRA权重)                    │
│  - outputs/ (生成结果 + 元数据)                     │
│  - cache/ (提示词嵌入缓存)                          │
└─────────────────────────────────────────────────────┘

{
  "workflow": {
    "nodes": [
      {
        "id": "load_checkpoint",
        "type": "CheckpointLoaderSimple",
        "inputs": {
          "ckpt_name": "sd_xl_base_1.0.safetensors"
        }
      },
      {
        "id": "lora_loader",
        "type": "LoraLoader",
        "inputs": {
          "lora_name": "character_linxiao_v3.safetensors",
          "strength_model": 0.85,
          "strength_clip": 0.75
        }
      },
      {
        "id": "ip_adapter",
        "type": "IPAdapterAdvanced",
        "inputs": {
          "ipadapter_file": "ip-adapter-faceid-plus_sdxl.bin",
          "image": ["load_ref_image", 0],
          "weight": 0.7,
          "weight_type": "linear"
        }
      },
      {
        "id": "controlnet",
        "type": "ControlNetLoader",
        "inputs": {
          "control_net_name": "controlnet-canny-sdxl-1.0.safetensors"
        }
      },
      {
        "id": "ksampler",
        "type": "KSampler",
        "inputs": {
          "model": ["lora_loader", 0],
          "positive": ["clip_text_encode", 0],
          "negative": ["clip_text_encode", 1],
          "latent_image": ["empty_latent", 0],
          "steps": 30,
          "cfg": 7.5,
          "sampler_name": "dpmpp_2m",
          "scheduler": "karras",
          "denoise": 1.0
        }
      }
    ]
  }
}

class ProgressiveGenerator:
    def __init__(self, target_size=1024, preview_size=512):
        self.target_size = target_size
        self.preview_size = preview_size
    
    def generate_with_preview(self, prompt, **kwargs):
        # 阶段1: 生成预览图
        preview = self.generate(
            prompt, 
            size=self.preview_size, 
            steps=20,
            **kwargs
        )
        
        # 质量评分 (使用CLIP或手工规则)
        score = self.quality_score(preview, prompt)
        if score < 0.7:
            return self.retry_with_adjusted_params(prompt, score)
        
        # 阶段2: 高清重绘
        highres = self.upscale_and_refine(
            preview, 
            target_size=self.target_size,
            denoise_strength=0.35
        )
        
        return highres

class VideoMotionGenerator:
    def __init__(self):
        self.svd_pipeline = StableVideoDiffusionPipeline.from_pretrained(
            "stabilityai/stable-video-diffusion-img2vid",
            torch_dtype=torch.float16,
            variant="fp16"
        )
        self.svd_pipeline.enable_model_cpu_offload()
    
    def generate_motion(self, 
                        image: Image.Image,
                        motion_bucket_id: int = 127,
                        fps: int = 8,
                        frames: int = 25,
                        seed: int = 42):
        """
        生成带运动效果的视频片段
        
        Args:
            motion_bucket_id: 0-255, 控制运动强度
            fps: 帧率, 漫剧通常8-12fps
            frames: 总帧数, 对应2-3秒
        """
        generator = torch.Generator(device="cuda").manual_seed(seed)
        
        # 关键: 使用一致的噪声种子 + 条件帧约束
        video_frames = self.svd_pipeline(
            image,
            decode_chunk_size=8,
            generator=generator,
            motion_bucket_id=motion_bucket_id,
            noise_aug_strength=0.1,
            num_frames=frames,
        ).frames[0]
        
        # 后处理: 平滑插帧
        return self.interpolate_frames(video_frames, target_fps=fps)
    
    def interpolate_frames(self, frames, target_fps):
        """使用RIFE进行帧插值，提升流畅度"""
        from rife.pytorch import RIFE
        interpolator = RIFE()
        return interpolator.interpolate(frames, target_fps)

class MotionIntent(Enum):
    STATIC = "static"              # 静止，仅呼吸
    PANNING = "panning"            # 镜头平移
    ZOOM_IN = "zoom_in"            # 推镜头
    ZOOM_OUT = "zoom_out"          # 拉镜头
    CHARACTER_TURN = "turn"        # 角色转身
    HAND_GESTURE = "gesture"       # 手势动作
    WALK = "walk"                  # 行走
    EMOTION_SHIFT = "emotion"      # 表情变化
    
def select_motion_strategy(shot_type, emotion, action):
    """根据镜头类型和动作选择运动策略"""
    if shot_type == "close_up" and emotion in ["surprise", "anger"]:
        return MotionIntent.EMOTION_SHIFT
    elif shot_type == "wide" and action == "walk":
        return MotionIntent.PANNING
    elif shot_type == "medium" and action == "talk":
        return MotionIntent.STATIC
    # ... 更多规则

class VoiceSynthesizer:
    def __init__(self):
        # 使用XTTS-v2作为基础引擎
        self.tts = TTS(model_name="tts_models/multilingual/multi-dataset/xtts_v2")
        
        # 预加载角色音色
        self.voice_profiles = {
            "林晓": "voice_linxiao.wav",   # 录制的声音样本
            "张队": "voice_zhangdui.wav",
            "旁白": "voice_narrator.wav",
        }
    
    def synthesize_line(self, text, character, emotion="neutral", speed=1.0):
        # 加载对应的音色嵌入
        speaker_embedding = self.load_speaker_embedding(
            self.voice_profiles[character]
        )
        
        # 情感控制: 调整韵律参数
        prosody_params = self.get_emotion_prosody(emotion)
        
        audio = self.tts.tts(
            text,
            speaker_embedding=speaker_embedding,
            speed=speed * prosody_params["speed_factor"],
            pitch=prosody_params["pitch_shift"],
        )
        return audio

def generate_phoneme_animation(audio_path, fps=8, num_frames=25):
    """
    生成口型关键帧序列
    """
    # 1. 提取音素时间轴
    phonemes = extract_phonemes(audio_path)  # 返回 [(start, end, phoneme)]
    
    # 2. 映射到口型姿态 (A, E, I, O, U + 闭合)
    mouth_shapes = {
        'A': 'wide_open',
        'E': 'semi_open',
        'I': 'narrow',
        'O': 'round',
        'U': 'pursed',
        'sil': 'closed'
    }
    
    # 3. 生成逐帧标签 (按时间插值)
    frame_labels = []
    for frame_idx in range(num_frames):
        time = frame_idx / fps
        active_phoneme = find_phoneme_at_time(phonemes, time)
        shape = mouth_shapes.get(active_phoneme, 'closed')
        frame_labels.append(shape)
    
    # 4. 转换为图像生成控制信号 (如Landmark或关键点坐标)
    return frame_labels

class AudioPostProcessor:
    def __init__(self):
        self.sfx_library = load_sfx_library()  # 音效素材库
        self.bgm_generator = MusicGen.from_pretrained("facebook/musicgen-medium")
    
    def compose_audio_track(self, shots, dialogues, emotion_curve):
        """
        合成完整音频轨道
        """
        timeline = AudioTimeline()
        
        for shot in shots:
            # 1. 对话轨道
            if shot.dialogue:
                audio = self.synthesize_dialogue(shot.dialogue, shot.character)
                timeline.add_track(audio, shot.start_time, shot.duration)
            
            # 2. 音效轨道
            if shot.sfx:
                sfx = self.match_sfx(shot.sfx, shot.emotion)
                timeline.add_track(sfx, shot.start_time, shot.duration)
        
        # 3. BGM生成（基于整体情感曲线）
        bgm = self.bgm_generator.generate(
            description=self.emotion_to_music_prompt(emotion_curve),
            duration=timeline.total_duration,
        )
        timeline.add_bgm(bgm, volume=-12)  # -12dB 作为背景
        
        return timeline.render()

┌─────────────────────────────────────────────────────────────────┐
│                       用户输入层                               │
│   (故事梗概 / 角色设定 / 风格参考图 / 控制参数)                │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Orchestrator (Apache Airflow)               │
│   DAG: ai_comic_production_v2                                  │
│   ┌─────────┐   ┌─────────┐   ┌─────────┐   ┌─────────┐     │
│   │ 剧本生成 │ → │ 分镜拆解 │ → │ 角色设计 │ → │ 场景生成 │     │
│   └─────────┘   └─────────┘   └─────────┘   └─────────┘     │
│        │             │             │             │            │
│        ▼             ▼             ▼             ▼            │
│   ┌─────────┐   ┌─────────┐   ┌─────────┐   ┌─────────┐     │
│   │ 视频生成 │ ← │ 图像生成 │ ← │ 批量渲染 │ ← │ Prompt  │     │
│   └─────────┘   └─────────┘   └─────────┘   └─────────┘     │
│        │             │                                        │
│        ▼             ▼                                        │
│   ┌─────────┐   ┌─────────┐   ┌─────────┐                   │
│   │ 音频合成 │ → │ 后期剪辑 │ → │ 编码输出 │                   │
│   └─────────┘   └─────────┘   └─────────┘                   │
└─────────────────────────────────────────────────────────────────┘

# 将生成任务拆分为可并行执行的子任务
def build_task_graph(scene_count, shot_per_scene):
    tasks = []
    
    for scene_idx in range(scene_count):
        # 场景级依赖: 背景先于角色
        bg_task = Task(
            type='background',
            scene=scene_idx,
            depends_on=[]  # 无依赖
        )
        
        for shot_idx in range(shot_per_scene):
            char_task = Task(
                type='character',
                scene=scene_idx,
                shot=shot_idx,
                depends_on=[bg_task.id]  # 依赖背景
            )
            tasks.extend([bg_task, char_task])
    
    # 按GPU数量分组调度
    return schedule_by_gpu(tasks, num_gpus=4)

def auto_quality_check(image, shot_spec):
    """自动化质检"""
    checks = []
    
    # 检查1: 主体是否存在 (使用GroundingDINO)
    subjects = grounding_dino.detect(image, shot_spec.subject)
    checks.append(('subject_present', len(subjects) > 0))
    
    # 检查2: 构图是否符合景别要求
    composition = analyze_composition(image)
    checks.append(('composition_match', 
                   composition['shot_type'] == shot_spec.camera_angle))
    
    # 检查3: 角色一致性 (使用面部嵌入相似度)
    if shot_spec.character:
        similarity = compute_face_similarity(image, reference_embedding)
        checks.append(('face_consistency', similarity > 0.7))
    
    # 综合评分
    score = sum(1 for _, passed in checks if passed) / len(checks)
    return score >= 0.7, {k: v for k, v in checks}

def smooth_video_frames(frames, window_size=5):
    """
    使用滑动窗口平滑帧间色彩
    """
    smoothed = []
    for i in range(len(frames)):
        start = max(0, i - window_size // 2)
        end = min(len(frames), i + window_size // 2 + 1)
        window = frames[start:end]
        
        # 在LAB色彩空间进行平滑
        lab_frames = [cv2.cvtColor(f, cv2.COLOR_RGB2LAB) for f in window]
        mean_lab = np.mean(lab_frames, axis=0)
        smoothed.append(cv2.cvtColor(mean_lab.astype(np.uint8), cv2.COLOR_LAB2RGB))
    
    return smoothed

构图规范：
- 双人对话使用"过肩镜头"或"正反打"
- 左侧人物面向右，右侧人物面向左
- 使用 [人物A: x=0.3, y=0.5] 格式指定位置