D1 What an AI Short-Drama Production Pipeline Looks Like: Breaking Down the Stages, a Task-Graph Architecture, and Choosing Among Four Categories of Generation Models

今日目标

能说出一集竖屏短剧从选题到成片要经过哪些工序，以及其中哪几步现在真能交给模型做
能把生产流程画成一张有向无环图，并指出哪些节点可以并行、哪些必须串行
能写出一层统一的生成 provider 接口，让同一段业务代码在离线占位与真实厂商之间切换

这门课默认你会写 TypeScript，不要求你懂影视，也不要求你上过别的 Agent 课。十四天只做一件事：把一句话变成一集能播的竖屏短剧，再把这条脚本扩成一条能并行、可审核、可控成本的生产线。今天不写剧本、不生成画面，先把骨架搭出来——读完回到页面顶部把三条勾掉。

小白版讲解

剧组的账本：这条线上的钱到底花在哪一格

先建立一个类比，这门课每一天都会回到它：你在做的不是一个脚本，而是一个剧组。

真实剧组开工前，制片会先出两样东西：通告单（几点、在哪、拍哪几场、谁到场）和工序表（剧本定稿才画分镜，分镜定了才置景，景搭好了才能开机，拍完才进剪辑）。前者管资源，后者管顺序——它们和后端工程里的调度器、任务图是同一件事，只是早了一百年。

一集竖屏短剧的工序，翻译成我们的语言是六步：剧本 → 角色与场景资产 → 每镜首帧 → 镜头视频 → 配音 → 剪辑合成。这六步就是本课第一周的六天，也是今天实验里那张图的六个节点。

传统剧组每步花多少人多少天，各家差得很远，我们不猜。但 AI 这一侧的账当场能算清楚。按 MiniMax 官方按量付费页的人民币单价：图像每张 ¥0.025；视频按秒计价，档位是 768P 每秒 ¥0.50、2K 每秒 ¥0.80、480P 每秒 ¥0.33；语音官方只公开资源包价（HD 套餐一 ¥630 换 200 万字符），折算约每万字符三元多——这是折算值，不是官方标价。

拿一集四十个分镜、每镜六秒来算：

TextText

视频  40 镜 × 6 秒 × ¥0.50/秒          = ¥120.00
图像  40 张首帧 + 10 张定妆 × ¥0.025    = ¥1.25
语音  约 800 字符，按折算价每万字符 ¥3   = ¥0.24
文本  剧本与评审几万 token，不到一元
                                        合计 约 ¥122

这张账单里有一个数字压倒一切：视频占了 98%，而它同时还是最慢的一步（异步任务，要排队、要轮询）和最容易失败的一步（内容审核、限流、超时）。

所以本课有一个贯穿到底的判断，现在就记住，后面每一天都会落回它：

这条线上最贵、最慢、最容易失败的是视频生成，所以整条工程设计都是围着「怎么少调一次视频接口」转的。

幂等、缓存、参考图复用、草稿档与成片档、预算熔断——第二周那些听起来很工程的东西全是这句话的推论。只想「把 AI 串起来」的人写不出它们，因为他不知道自己在省什么。

把工序画成图：为什么不是一串顺序等待

新手写这类流水线，第一版几乎都是一串 await：写剧本，等；生成图，等；生成视频，等；配音，等。能跑，但三个问题会在第一次真跑时一起冒出来。

第一，它把能同时做的事排成了队：配音只依赖台词、跟画面无关，却要排在四十个视频后面等半小时。第二，它没有断点——第三十七个视频超时，前面三十六个的成果全在内存里，程序一退就得从头重来。第三，它说不清自己卡在哪一步。

剧组早就解决了这个问题：工序表不是一条直线，是一张有依赖关系的图。置景和试妆可以同时进行，因为都只依赖剧本；但开机必须等两者都完成。这种结构叫任务图（directed acyclic graph，有向无环图），三个词各管一件事：有向是「谁在谁前面」，无环是「不许绕回去」，图是「不止一条路」。

今天这张图长这样：

Mermaid source

mermaidmermaid

graph LR
  script[剧本与人物卡] --> assets[角色定妆图]
  assets --> frames[每镜首帧]
  frames --> clips[镜头视频]
  script --> voice[台词配音]
  clips --> timeline[时间轴]
  voice --> timeline

注意 voice 这一支：它从 script 直接拉出来，绕过了整条画面链路。这就是画图比排队多拿到的第一样东西——并行的可能性由图结构直接表达，不用你在代码里手工判断。

执行这张图只需要一个算法：拓扑排序——把节点排成「每个节点都在它所有依赖之后」的顺序，顺便在发现环时报错。「无环」两个字不是形容词，是这段代码保证出来的：

dag.ts

export type TaskNode = { id: string; deps: string[]; run: () => Promise<string[]> }
 
// 深度优先 + 三态标记：没访问过 / 正在访问 / 访问完。
// 再次走到「正在访问」的节点，就说明沿着依赖绕回了自己，这就是环。
export function topoSort(nodes: TaskNode[]): TaskNode[] {
  const byId = new Map(nodes.map((n) => [n.id, n]))
  const state = new Map<string, 'visiting' | 'done'>()
  const order: TaskNode[] = []
 
  const visit = (id: string, trail: string[]) => {
    if (state.get(id) === 'done') return
    if (state.get(id) === 'visiting') throw new Error(`任务图里有环：${[...trail, id].join(' -> ')}`)
    const node = byId.get(id)
    if (!node) throw new Error(`节点 ${id} 依赖了不存在的节点`) // 静默漏跑一步比报错难查得多
    state.set(id, 'visiting')
    for (const dep of node.deps) visit(dep, [...trail, id])
    state.set(id, 'done')
    order.push(node)
  }
 
  for (const n of nodes) visit(n.id, [])
  return order
}

dag.py

from dataclasses import dataclass
from typing import Awaitable, Callable
 
 
@dataclass
class TaskNode:
    id: str
    deps: list[str]
    run: Callable[[], Awaitable[list[str]]]
 
 
# 深度优先 + 三态标记：没访问过 / 正在访问 / 访问完。
# 再次走到「正在访问」的节点，就说明沿着依赖绕回了自己，这就是环。
def topo_sort(nodes: list[TaskNode]) -> list[TaskNode]:
    by_id = {n.id: n for n in nodes}
    state: dict[str, str] = {}
    order: list[TaskNode] = []
 
    def visit(node_id: str, trail: list[str]) -> None:
        if state.get(node_id) == "done":
            return
        if state.get(node_id) == "visiting":
            raise ValueError("任务图里有环：" + " -> ".join([*trail, node_id]))
        node = by_id.get(node_id)
        if node is None:
            raise ValueError(f"节点 {node_id} 依赖了不存在的节点")  # 静默漏跑一步比报错难查得多
        state[node_id] = "visiting"
        for dep in node.deps:
            visit(dep, [*trail, node_id])
        state[node_id] = "done"
        order.append(node)
 
    for n in nodes:
        visit(n.id, [])
    return order

工程代价也要说清楚：任务图不是免费的。 你得给每个节点定义清楚「输入是哪几个产物、输出是哪几个产物」，否则它只是一张漂亮的依赖声明。今天只做「按图跑」，幂等、断点续跑、并发留给第 8 天和第 9 天——它们全都建立在「今天把产物写进了磁盘固定位置」这个前提上。

四类生成模型的地图：谁站在流水线的哪一格

短剧生产要用到四类生成模型，正好各对应一格工序：

环节	干什么	接口形态
文本	写剧本、评审打分、起标题	`POST /v1/chat/completions`，OpenAI 兼容，一次请求就回
图像	角色定妆图、场景图、每镜首帧	`POST /v1/image_generation`，一次请求就回，返回图片链接
视频	由首帧生成镜头片段	`POST /v1/video_generation` 只返回任务号，要轮询、要取件
语音	台词配音	`POST /v1/t2a_v2`，一次请求就回

本课的第一优先厂商是 MiniMax，理由很实际：它一家覆盖这四类，你配一个 key 就能把整条线真跑通，不用为了学一门课去开四个账号。鉴权只有一个 header：Authorization: Bearer 加上你的 key，再加一个 Content-Type: application/json——老教程里那个 GroupId 查询参数是历史写法，现在的 OpenAPI 里没有它，别照抄。

四类里有三类「一次请求就回」，只有视频是异步任务。视频那一路要走四步：提交拿到 task_id，用 GET /v1/query/video_generation 带着它反复查状态（枚举首字母大写：Preparing、Queueing、Processing、Success、Fail），成功后拿到 file_id，再用 GET /v1/files/retrieve 换 download_url 才能下载。断掉任何一步，你手里就只剩一个任务号。顺带一提，同一家还并存着另一套版本的视频接口，路径、状态枚举、取件方式全不一样——本课代码只用上面这一套，这也预告了下一节要讲的事。

还有三个坑是「不写下来一定会踩」的：

图片链接 24 小时后失效，视频下载链接官方没写有效期但同样是临时的。所以下载落盘不是可选步骤，是流程的一部分。
语音接口返回的 data.audio 默认是 hex 字符串，不是 base64。 Node 里落盘就是 Buffer.from(json.data.audio, "hex")，当成 base64 解会得到一段噪音。
视频接口的 prompt_optimizer 默认是 true：你的提示词会被官方改写一遍再送进模型。做角色一致性时这是个隐形变量，第 3 天专门讲。

其余厂商——阿里云百炼、快手可灵、火山引擎方舟、Google、Runway——本课只做形态对照，不写实现，也不写它们的具体模型版本号（生成模型迭代极快，写死版本号等于给自己埋一个必然过期的点，实测就有一批老版本模型在 2026 年 9 月被下线）。

真正值得记住的是四条不随版本变的形状差异，它们全都实测过：

对照维度	各家的差异长什么样
鉴权 header	形状各不相同。有的只要一个 Bearer，有的除此之外还要一个异步开关头或者 API 版本头
任务状态枚举	大小写与用词都不一样。有的首字母大写，有的全小写，有的干脆返回一个布尔值
结果链接有效期	从一天到一个月不等，各家自己定
角色一致性	有的靠每次请求传一张参考图，有的要先建一个可复用的「主体 id」，再在提示词里用记号引用它

请特别注意最后一列：这四条差异全都不在「能不能做到」上——四类能力大家都有——而在「形状」上。 这一点直接决定了下一节那层接口的必要性。

一层薄薄的 provider 接口：把厂商差异关在门后面

上一节那张表已经把话说了一半：各家能力雷同，形状全不一样。业务代码只要贴着任何一家写，就会被那一家的形状腌入味——鉴权头的个数、状态枚举的大小写、链接能存多久、参考图怎么传，这些一旦渗进业务逻辑，换厂商就不是「多写一个实现类」，而是「重读一遍全部业务代码」。

顺带说一件影响写法的事：MiniMax 没有面向应用代码的官方 Node SDK，所以今天那层 provider 是裸 fetch 写的。这不是偷懒——没有 SDK 反而让边界更清楚：唯一的网络出口就是你自己那几行 fetch。

把厂商调用散进十几个业务文件，你会同时失去三样东西，这三样就是做这层抽象的三个理由。

理由一：离线可跑。 没有余额、没有网络、在 CI 里，流水线也应该能跑完并产出文件。做法是把网络出口收敛到一处，让离线实现与真实实现共用同一个接口。

理由二：换厂商与多厂商并存。 今天用 MiniMax 做视频，明天为某个镜头的运镜想试别家。因为差异只在形状不在能力，只要业务代码里写的是「一次生成动作」而不是「某一家的那四步」，换家这件事就只是多写一个实现类——状态枚举的大小写、多出来的那个请求头、参考图还是主体 id，全部关在实现类里面。

理由三：计量收口。 每次调用花了多少钱必须有一处统一记下来，散在业务代码里的调用永远补不齐这本账。第 12 天整章都建立在这个收口上。

接口该长什么样？关键是按业务动作定义，不按厂商的 HTTP 请求定义。视频那四步对业务代码来说只是一个动作：给我一段片子。所以接口上只有一个 generate：

providers/types.ts

export interface VideoProvider {
  readonly name: string
  readonly model: string
  /** 异步任务：内部完成提交、轮询、取件、下载四步，对调用方表现为一次 await */
  generate(input: {
    prompt: string
    outPath: string
    durationSec: number
    firstFrame?: string
    resolution?: string
    onProgress?: (stage: string) => void // 状态回调，让调用方能打印进度而不必懂协议
  }): Promise<{ path: string; costCny: number }>
}
 
// 唯一的选择点。业务代码永远只 import 这个函数，不 import 任何厂商实现。
export function createProviders(): Providers {
  return process.env.MOCK === '1' ? createMockProviders() : createMiniMaxProviders()
}

providers/types.py

import os
from typing import Callable, Protocol, TypedDict
 
 
class VideoResult(TypedDict):
    path: str
    cost_cny: float
 
 
class VideoProvider(Protocol):
    name: str
    model: str
 
    # 异步任务：内部完成提交、轮询、取件、下载四步，对调用方表现为一次 await
    async def generate(
        self,
        prompt: str,
        out_path: str,
        duration_sec: int,
        first_frame: str | None = None,
        resolution: str | None = None,
        on_progress: Callable[[str], None] | None = None,  # 让调用方能打印进度而不必懂协议
    ) -> VideoResult: ...
 
 
# 唯一的选择点。业务代码永远只 import 这个函数，不 import 任何厂商实现。
def create_providers() -> Providers:
    return create_mock_providers() if os.getenv("MOCK") == "1" else create_minimax_providers()

注意 outPath：接口要求实现方把文件落到指定路径，而不是返回 URL 或者一段 Buffer。这是被上一节那条「链接会失效」逼出来的设计——把落盘写进契约，就没有哪个实现能忘掉它。

这层抽象的代价也要讲清楚，否则就成了「抽象总是好的」这种空话：你会磨掉各家的独有能力。 某家支持首尾帧同时给，某家支持结构化运镜参数，通用接口装不下。办法不是把接口撑大，而是留一个可选的透传字段，让需要它的那一处显式地用——也就显式承认「这一处绑定了某一家」。

离线也要能跑：占位素材不是敷衍

「先接上真接口再说，跑通了再补 mock」——这是本课最想让你改掉的习惯。

原因和调试节奏有关。这条线一次完整运行要调四十多次视频接口，真跑一遍几十分钟、一百多块钱。时间轴写错一个下标，你得等半小时、花一百块才看得到——三次之后你就不想动这段代码了。

离线模式把这个循环压到几秒。今天实验里的离线实现不返回假 JSON，而是用 ffmpeg 真的生成文件：占位图是按镜头描述哈希取色的纯色竖屏图，占位视频是带音轨的测试画面，占位配音是正弦波，时长按台词字数估算（全课统一每汉字 0.22 秒）。

providers/mock.ts

// 桩只打在网络出口上：这里没有 fetch，但产出的是真文件。
class MockImageProvider implements ImageProvider {
  async generate(input: { prompt: string; outPath: string; seed?: number }) {
    // 颜色由输入哈希得来 —— 换一句话，图就换一个颜色，说明业务逻辑真的跑过了
    const key = `${input.prompt}|${input.seed ?? ''}`
    await ffmpeg(['-f', 'lavfi', '-i', `color=c=${colorOf(key)}:s=1080x1920`, '-frames:v', '1', input.outPath])
    return { path: input.outPath, costCny: 0 }
  }
}

providers/mock.py

# 桩只打在网络出口上：这里没有 httpx，但产出的是真文件。
class MockImageProvider:
    async def generate(self, prompt: str, out_path: str, seed: int | None = None) -> ImageResult:
        # 颜色由输入哈希得来 —— 换一句话，图就换一个颜色，说明业务逻辑真的跑过了
        key = f"{prompt}|{seed if seed is not None else ''}"
        await ffmpeg(["-f", "lavfi", "-i", f"color=c={color_of(key)}:s=1080x1920", "-frames:v", "1", out_path])
        return {"path": out_path, "cost_cny": 0.0}

判断一个离线实现合不合格只有一条标准：改一个输入，输出必须跟着变。 换一句话，占位图颜色变了、占位音频时长变了、时间轴总长变了——说明中间那些业务逻辑（状态机、时间轴计算、ffmpeg 参数拼装）真的执行了。如果换什么输入产物都一样，你验的只是「程序没崩」。

所以桩的位置有硬规定：只打在四个 provider 的网络出口上，业务代码里一个 if (process.env.MOCK) 都不该有——业务逻辑一旦分叉，你在离线模式下验的就是另一个程序。

先看一眼终局：第二周这条线会长成什么样

最后花两分钟看这十四天的形状，你会更清楚今天这层地基为什么这么打。

第一周把一集做出来：D2 剧本结构化，D3 角色与场景资产，D4 镜头视频，D5 配音与字幕，D6 剪辑合成，D7 串成一条能一次跑完的线。第二周把这条脚本变成生产线：D8 工作流引擎，D9 并发与配额，D10 审核台，D11 质检与合规，D12 成本与模型路由，D13 多平台发行，D14 一季五集加作品集。

回头看第一节那句判断——最贵最慢最容易失败的是视频——第二周每一天都是它的推论：D8 的幂等是为了失败后不重新生成视频，D9 的配额闸门是为了不撞限流白白浪费额度，D12 的草稿档是为了调参阶段用便宜档位跑，D3 的资产复用是为了让同一张定妆图被四十个镜头共用。

课程顺序也是这么定的。 先写 provider 接口而不是先写剧本，是因为接口一旦定稿，后面十三天都在往里填内容；先做通再做好，再做多做稳做省——反过来你会去优化一个还不存在的东西。

源码导读

动手实验

🧪 D1 实验：一张生产线任务图与一个能在离线模式下跑通全流程的四类 provider 骨架

Code location: labs/ai-drama-pipeline/day-01-pipeline-skeleton

验收标准：

MOCK=1 pnpm start 跑完，终端按 script、assets、frames、clips、voice、timeline 六个节点逐个打印耗时与产物数，最后一行账单显示产物 14 个。
产物目录下真的有文件：两张角色定妆图、三张首帧、三段片子、三段配音、两份 JSON 与一份运行记录。
用 ffprobe 看第一镜的片子，能看到 1080x1920、h264 与 aac 两条流、时长 6 秒。
三张首帧的颜色互不相同（占位色由镜头描述哈希得来，不是同一块灰）。
时间轴 JSON 里三镜首尾相接：0 到 6000、6000 到 12000、12000 到 18000，不重叠也不留缝。
给剧本节点加一条指向时间轴节点的依赖造出一个环，再跑会报「任务图里有环」并直接退出。

动手前确认 ffmpeg 装好了（ffmpeg -version 有输出），并且全程用 MOCK=1——今天不需要密钥。starter/ 原样能跑完，但第 3 到 6 条验收都不达标，四个编号练习各对应一条，卡住了对照 solution/ 同名文件看。

先跑一遍 solution 的 MOCK=1 pnpm start，把终端输出和产物目录对着看一遍，心里有个「做完是什么样」的样板。
回到 starter，打开 src/index.ts 看那六个节点的 deps 是怎么连的，在纸上把这张图画出来，标出每个节点的输入产物与输出产物。
做练习 1 和练习 2：把离线图像与离线视频的 ffmpeg 参数补完整，跑完用 ffprobe 确认片子是 1080x1920、6 秒、带音轨，三张首帧颜色不同。
做练习 3：实现拓扑排序与环检测，跑起来那行「topoSort 还没实现」的警告应该消失；再故意造一个环，确认它会报错退出。
做练习 4：算出时间轴的起止毫秒，打开时间轴 JSON 确认三镜首尾相接；最后换一句自己的一句话再跑一遍，确认占位素材真的跟着变了。

面试题

今天 3 道题在下方题库区，侧重生产线的分层与解耦、provider 抽象的边界、以及离线可跑为什么是工程要求而不是玩具。展开后先看分析过程再看要点——照着推导练，比背要点管用。标注「国内高频 / 海外高频」方便按目标市场取舍。

检查清单与明日预告

能说出一集竖屏短剧从选题到成片要经过哪些工序，以及其中哪几步现在真能交给模型做
能把生产流程画成一张有向无环图，并指出哪些节点可以并行、哪些必须串行
能写出一层统一的生成 provider 接口，让同一段业务代码在离线占位与真实厂商之间切换
能不看讲义说出这条线上钱主要花在哪一格，以及这个事实推出了哪几条工程设计
能说清视频接口比另外三类多了哪四步，以及为什么落盘要写进接口契约
实验的 6 条验收标准全部通过
3 道面试题不看要点也能答出至少 2 道

明天（D2）我们进编剧室：让模型把一句话变成人物卡、场景与分镜的结构化数据，而不是一段散文。为什么是这个顺序？因为今天这张图上，除剧本节点外的五个节点，输入全都是分镜数据里的字段——visual 进图像提示词、camera 进视频提示词、dialogue 进配音、durationSec 进时间轴。剧本不结构化，后面五步就没有东西可吃。D2 还会给剧本 Agent 加一个评审角色让稿子在循环里变好，并把跨集共用的设定沉淀成一份世界观档案——那份档案会一直用到第 14 天的一季五集。

What an AI Short-Drama Production Pipeline Looks Like: Breaking Down the Stages, a Task-Graph Architecture, and Choosing Among Four Categories of Generation Models