AI 漫剧短剧训练：角色一致性、分镜生成与效果评估的工程化实践

dataset/
  character_xiaohui/
    01_front.png
    02_side.png
    03_back.png
    04_smile.png
    05_angry.png
    06_cry.png
    ...
    metadata.json

import json
from pathlib import Path

def build_metadata(char_dir: Path, trigger: str,
                   base_tags: list[str]) -> Path:
    items = []
    for img in sorted(char_dir.glob("*.png")):
        items.append({
            "file_name": img.name,
            "text": f"{trigger}, " + ", ".join(base_tags),
        })
    meta = {"trigger_word": trigger, "items": items}
    path = char_dir / "metadata.json"
    path.write_text(json.dumps(meta, ensure_ascii=False, indent=2),
                    encoding="utf-8")
    return path

build_metadata(
    Path("dataset/character_xiaohui"),
    trigger="xiaohui_char",
    base_tags=["flat cartoon", "thick outline", "low saturation",
               "sticker style", "clean edges"],
)

from dataclasses import dataclass

@dataclass
class LoRAConfig:
    base_model: str = "runwayml/stable-diffusion-v1-5"
    resolution: int = 512
    train_batch_size: int = 1
    gradient_accumulation: int = 4
    learning_rate: float = 1e-4
    lr_scheduler: str = "cosine"
    max_train_steps: int = 1500
    rank: int = 32
    network_alpha: int = 16
    mixed_precision: str = "fp16"
    save_every: int = 250
    seed: int = 42

    @property
    def effective_batch(self) -> int:
        return self.train_batch_size * self.gradient_accumulation

def build_prompt(trigger: str, action: str, style: str) -> str:
    return (f"{trigger}, {style}, action: {action}, "
            "clean edges, no text, no watermark")

def negative_prompt() -> str:
    return ("text, watermark, logo, real person, complex background, "
            "deformed, low quality, extra fingers")

import os, json
from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

class Shot(BaseModel):
    idx: int
    scene: str = Field(max_length=120)
    camera: str = Field(max_length=60)
    action: str = Field(max_length=120)
    dialogue: str = Field(default="", max_length=80)
    duration: float = Field(ge=1, le=8)

class Episode(BaseModel):
    title: str
    characters: list[str]
    shots: list[Shot] = Field(min_length=4, max_length=12)

SYSTEM = """你是漫剧分镜师。根据主题生成 6 个镜头，每镜 3-5 秒。
只输出 JSON：{title, characters, shots:[{idx,scene,camera,action,dialogue,duration}]}。
禁止侵权、违法、暴力、色情内容。角色必须来自给定角色表。"""

def gen_episode(topic: str, character_names: list[str]) -> Episode:
    resp = client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user",
             "content": f"主题：{topic}\n角色表：{', '.join(character_names)}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.8,
    )
    return Episode.model_validate_json(resp.choices[0].message.content)

import numpy as np
from PIL import Image
from pathlib import Path

def image_embedding(path: Path, size: int = 64) -> np.ndarray:
    img = Image.open(path).convert("RGB").resize((size, size))
    arr = np.asarray(img, dtype=np.float32) / 255.0
    return arr.flatten()

def cosine(a: np.ndarray, b: np.ndarray) -> float:
    denom = np.linalg.norm(a) * np.linalg.norm(b)
    return float(a @ b / denom) if denom else 0.0

def consistency_score(ref: Path, samples: list[Path]) -> dict:
    ref_vec = image_embedding(ref)
    scores = [cosine(ref_vec, image_embedding(s)) for s in samples]
    return {
        "mean": round(float(np.mean(scores)), 4),
        "min": round(float(np.min(scores)), 4),
        "pass": float(np.min(scores)) >= 0.75,
    }

def validate_episode(ep: Episode, target_duration: tuple[float, float],
                     banned: set[str]) -> list[str]:
    issues = []
    total = sum(s.duration for s in ep.shots)
    if not (target_duration[0] <= total <= target_duration[1]):
        issues.append(f"总时长 {total}s 不在目标区间")
    for s in ep.shots:
        if any(w in s.dialogue for w in banned):
            issues.append(f"镜头 {s.idx} 对白含禁用词")
        if any(w in s.action for w in banned):
            issues.append(f"镜头 {s.idx} 动作含禁用词")
    return issues