AI 短剧漫剧生成软件：从分镜到成片的工程化架构

原创

IT大佬 jzit-top

发布于 2026-10-06 13:43:49

370

市面上的"AI 短剧生成软件"大多停留在"输入主题，输出一段拼接视频"。但从工程视角看，一个能真正用于生产短剧/漫剧的系统，需要解决四个核心问题：剧本结构化、角色跨镜头一致、素材生产可恢复、成片质量可评估。本文从软件架构层面拆解这套系统，并给出可运行骨架。

一、系统分层

一个可交付的 AI 短剧生成软件通常分为六层：

接口层    项目创建、任务提交、进度查询、成片下载
编排层    阶段调度、断点续跑、失败重试
生成层    剧本、分镜图、视频片段、配音、字幕
一致性层  角色卡、LoRA、种子管理、参考图
合成层    ffmpeg 拼接、混音、字幕烧录、转场
门禁层    审核、质检、授权留痕、AI 标注

核心原则：每个阶段的产物都落盘并记录元数据，任何一步失败都能从中间恢复。

二、项目状态机：让流程可恢复

短剧生成是长流程任务，动辄几十分钟。用状态机管理阶段，避免"从头再来"。

from dataclasses import dataclass, field, asdict
from enum import Enum
from pathlib import Path
import json, time, uuid

class Stage(str, Enum):
    script = "script"
    assets = "assets"
    assemble = "assemble"
    render = "render"
    gate = "gate"
    done = "done"

@dataclass
class Project:
    project_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    title: str = ""
    stage: Stage = Stage.script
    segments: list[dict] = field(default_factory=list)
    meta: dict = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

    def save(self, root: Path) -> Path:
        self.updated_at = time.time()
        path = root / f"{self.project_id}.json"
        path.write_text(
            json.dumps(asdict(self), ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        return path

    @staticmethod
    def load(path: Path) -> "Project":
        data = json.loads(path.read_text(encoding="utf-8"))
        data["stage"] = Stage(data["stage"])
        return Project(**data)

    def advance(self, next_stage: Stage) -> None:
        self.stage = next_stage

项目 JSON 是唯一事实来源。服务重启后，从 JSON 恢复即可继续未完成的阶段。

三、剧本与分镜：结构化输出

自由文本无法编排。用 Pydantic 约束模型输出，把"分镜"变成可校验的数据结构。

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 Script(BaseModel):
    title: str
    characters: list[str] = Field(min_length=1, max_length=6)
    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_script(topic: str) -> Script:
    resp = client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": f"主题：{topic}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.8,
    )
    return Script.model_validate_json(resp.choices[0].message.content)

Schema 里的 max_length、ge/le、min_length 不是装饰，而是防止模型输出超长对白、负时长、零镜头等异常数据的第一道防线。

四、角色一致性：角色卡 + 种子管理

短剧最怕"同一个角色，每个镜头长得不一样"。解决办法是把角色信息结构化，并绑定固定种子。

from dataclasses import dataclass, asdict
import hashlib

@dataclass(frozen=True)
class CharacterCard:
    name: str
    trigger: str
    appearance: str
    style: str
    forbidden: tuple[str, ...] = ()

    def prompt(self) -> str:
        return f"{self.trigger}, {self.appearance}, {self.style}"

    def negative(self) -> str:
        base = ("text", "watermark", "logo", "real person",
                "deformed", "low quality")
        return ", ".join(base + self.forbidden)

    @property
    def fingerprint(self) -> str:
        raw = json.dumps(asdict(self), sort_keys=True, default=str)
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

XIAO_HUI = CharacterCard(
    name="小灰",
    trigger="xiaohui_char",
    appearance="原创灰色小猫，绿色眼睛，红色围巾，二头身",
    style="扁平卡通，粗描边，低饱和，贴纸感",
    forbidden=("复杂背景", "写实"),
)

def build_prompt(card: CharacterCard, shot: Shot, base_seed: int) -> tuple[str, int]:
    prompt = f"{card.prompt()}，画面：{shot.action}，镜头：{shot.camera}"
    seed = base_seed + shot.idx   # 同角色固定基准 seed，镜头间只做微调
    return prompt, seed

同一个 trigger 和固定的 base_seed 是跨镜头一致的关键。角色卡指纹写入每张图的元数据，方便追溯。

五、素材生产：并行与幂等

素材生成（图、视频、配音）是耗时最长的环节，必须并行且可缓存。

import asyncio, hashlib
from pathlib import Path

def asset_key(project_id: str, idx: int, kind: str, prompt: str) -> str:
    raw = f"{project_id}:{idx}:{kind}:{prompt}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

async def gen_visual(prompt: str, negative: str,
                     seed: int, out: Path) -> str:
    # 实际调用即梦、可灵、Stable Diffusion 等官方 API
    await asyncio.sleep(0.05)
    return str(out)

async def gen_audio(text: str, out: Path, voice: str) -> str:
    # 实际调用 TTS；声音克隆需获得本人授权
    await asyncio.sleep(0.03)
    return str(out)

async def build_assets(project: Project, root: Path,
                       card: CharacterCard,
                       concurrency: int = 3) -> None:
    sem = asyncio.Semaphore(concurrency)
    cache: dict[str, str] = {}

    async def one(seg: dict):
        async with sem:
            key = asset_key(project.project_id, seg["idx"], "visual",
                            seg["prompt"])
            if key in cache:
                seg["visual"] = cache[key]
                return
            out = root / f"visual_{seg['idx']:03d}.png"
            seg["visual"] = await gen_visual(
                seg["prompt"], card.negative(), seg["seed"], out)
            cache[key] = seg["visual"]

            audio = root / f"audio_{seg['idx']:03d}.mp3"
            seg["audio"] = await gen_audio(seg["dialogue"], audio, "default")

    await asyncio.gather(*(one(s) for s in project.segments))

幂等键 asset_key 保证重跑时已生成的素材直接复用，不重复扣费。生产环境把 cache 换成 Redis 或数据库。

六、合成与质检

import subprocess
from pathlib import Path

def run_cmd(cmd: list[str], timeout: int = 300) -> dict:
    p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return {"code": p.returncode, "err": p.stderr[-800:]}

def make_srt(segments: list[dict], out: Path) -> Path:
    def fmt(t: float) -> str:
        h, m, s = int(t // 3600), int(t % 3600 // 60), t % 60
        return f"{h:02d}:{m:02d}:{s:06.3f}".replace(".", ",")

    lines, cur = [], 0.0
    for i, seg in enumerate(segments, 1):
        end = cur + seg["duration"]
        lines.append(f"{i}\n{fmt(cur)} --> {fmt(end)}\n{seg['dialogue']}\n")
        cur = end
    out.write_text("\n".join(lines), encoding="utf-8")
    return out

def assemble(parts: list[Path], srt: Path, out: Path) -> dict:
    listing = out.with_suffix(".txt")
    listing.write_text(
        "\n".join(f"file '{p.resolve()}'" for p in parts),
        encoding="utf-8",
    )
    c1 = run_cmd(["ffmpeg", "-y", "-f", "concat", "-safe", "0",
                  "-i", str(listing), "-c", "copy", str(out)])
    if c1["code"] != 0:
        return {"ok": False, "stage": "concat", "err": c1["err"]}
    c2 = run_cmd(["ffmpeg", "-y", "-i", str(out),
                  "-vf", f"subtitles={srt}", str(out.with_name("final.mp4"))])
    return {"ok": c2["code"] == 0, "stage": "subtitle", "err": c2["err"]}

每个 ffmpeg 命令都要检查返回码，失败时抛出异常，由上层决定重试还是告警。

七、门禁与合规

import re

BAD = {"违法", "暴力", "色情", "歧视", "虚假", "最好", "第一"}
PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

def audit_script(script: Script) -> list[str]:
    issues = []
    blob = " ".join(
        s.dialogue + s.action + s.scene for s in script.shots
    )
    for w in BAD:
        if w in blob:
            issues.append(f"含敏感或绝对化用语：{w}")
    if PII.search(blob):
        issues.append("含个人信息，需脱敏")
    return issues

def audit_assets(meta: dict) -> list[str]:
    issues = []
    if not meta.get("model_license"):
        issues.append("缺少模型许可说明")
    if not meta.get("voice_license"):
        issues.append("缺少配音授权")
    if meta.get("has_real_person") and not meta.get("portrait_license"):
        issues.append("含真人肖像但无授权")
    if not meta.get("ai_disclosed"):
        issues.append("未按平台要求标注 AI 生成")
    return issues

门禁不通过就不能进入成片阶段。审核要覆盖剧本和素材两端，留痕要覆盖模型、seed、提示词指纹和授权状态。

八、工程化要点

维度	做法	缺失后果
可恢复	状态机 + 项目 JSON	崩溃后从头重跑
一致性	角色卡 + 触发词 + 固定种子	角色跨镜头漂移
幂等	素材指纹做缓存键	重复生成、重复扣费
质检	自动指标 + 人工终审	废片流入成片
留痕	trace、模型、提示词版本	问题无法追溯
合规	双向审核 + 授权确认	侵权、平台处罚

九、总结

AI 短剧漫剧生成软件的工程核心，不是"接几个模型 API"，而是把剧本、角色、素材、合成、审核串成一条可恢复、可复现、可合规的生产线：

状态机：阶段化、可断点续跑。
结构化剧本：Pydantic 约束，拒绝自由文本。
角色一致性：角色卡 + 触发词 + 固定种子 + LoRA。
并行素材：信号量限流 + 幂等缓存。
合成质检：ffmpeg 检查返回码，自动过滤废片。
合规门禁：剧本与素材双向审核，全程留痕。

代码可以简单，但状态管理、幂等、审核、授权和留痕不能省。先把 15 秒单角色短片跑通，再扩展到多角色、多集、批量生产。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

小程序·短剧