D7 一集杀青：把六个环节串成端到端流水线并算清第一笔账

今日目标

能用一条命令从一句话产出一集完整成片，并在任意一步失败后知道卡在哪
能量出每个环节的耗时与花费，指出瓶颈在哪一环
能列出这条线现在的三个脆弱点，并说明第二周分别用什么解决

前六天你把六个环节各自跑通了，但它们之间还隔着你的手：跑完剧本，你手动把 JSON 拷进下一个脚本；跑完配音，你手动确认文件在不在。今天把手拿开。读完回到页面顶部把三条目标勾掉。

小白版讲解

一集杀青那天，问题才开始

剧组有个说法叫「杀青」——最后一个镜头拍完，全组鼓掌。但真正干过后期的人知道，杀青那一刻问题才刚开始：素材散在三张硬盘上，命名各不相同，录音师的时间码和摄影组差了两帧，某一条 NG 的素材和好的那条只差一个字母。

把六个环节串起来，你会一模一样地撞上这些事。单独调试时它们全都不存在，因为你的脑子替程序做了衔接：你知道刚才那个脚本把文件写到哪了，你知道该拿哪一份 JSON 喂给下一步。串起来之后，这些知识必须从你的脑子里搬进代码。

第一个咬人的是产物路径。单独跑图像脚本时，随手写 output/first-frame.png 完全没问题。串起来跑第二遍，第一遍的成片就被覆盖了；跑到一半失败，你分不清磁盘上哪些文件属于这一次；想比较「换一句话之后效果有没有变好」，两次的产物已经混在一起了。

解法只有一条，而且要一开始就立好规矩：每一次运行分配一个运行标识，所有产物都挂在它下面。

TextText

work/
└── runs/run-20260907-001/
    ├── run.json              这次运行的元信息与成本台账
    ├── script/               剧本、人物卡
    ├── assets/               角色定妆图
    ├── shots/<shotId>/       首帧、镜头视频、配音
    ├── timeline/             时间轴表
    └── output/               成片、字幕、封面

这个布局看着朴素，但它一次性解决了三个问题：两次运行不会互相覆盖；失败时残留的产物一目了然；想复现某次结果，把整个目录打包发给同事就行。

run.js

// 一次运行的目录布局。全流水线只通过这个对象拿路径，任何一处直接拼字符串都是隐患。
export class Run {
  constructor(id) {
    this.id = id
    this.dir = `work/runs/${id}`
    this.costs = []
  }
 
  path(...parts) {
    return [this.dir, ...parts].join('/')
  }
 
  // 花费在产生的那一刻就记下来，而不是跑完再回头统计——
  // 跑完再统计的账，失败时一定是错的。
  record(entry) {
    this.costs.push({ ...entry, runId: this.id, at: new Date().toISOString() })
  }
}
 
export function newRunId(now = new Date()) {
  const p = (n, w = 2) => String(n).padStart(w, '0')
  const stamp = `${now.getFullYear()}${p(now.getMonth() + 1)}${p(now.getDate())}`
  return `run-${stamp}-${p(Math.floor((now.getHours() * 60 + now.getMinutes()) % 1000), 3)}`
}

run.py

# 一次运行的目录布局。全流水线只通过这个对象拿路径，任何一处直接拼字符串都是隐患。
from datetime import datetime
 
 
class Run:
    def __init__(self, run_id: str):
        self.id = run_id
        self.dir = f"work/runs/{run_id}"
        self.costs: list[dict] = []
 
    def path(self, *parts: str) -> str:
        return "/".join([self.dir, *parts])
 
    # 花费在产生的那一刻就记下来，而不是跑完再回头统计——
    # 跑完再统计的账，失败时一定是错的。
    def record(self, entry: dict) -> None:
        self.costs.append({**entry, "runId": self.id, "at": datetime.now().isoformat()})
 
 
def new_run_id(now: datetime | None = None) -> str:
    now = now or datetime.now()
    seq = (now.hour * 60 + now.minute) % 1000
    return f"run-{now:%Y%m%d}-{seq:03d}"

第二个问题是日志淹没。六个环节各自打自己的日志，一条命令跑下来几百行滚过去，出了事你根本看不出是哪一环出的。这一天的日志规矩也很简单：每个环节只在开始时打一行、结束时打一行耗时与产物数，剩下的细节压到文件里去。终端上留下的应该是一张能一眼扫完的进度表，不是一部流水账。

那么，串起来之后最该量的是什么？不是「跑通了没有」，而是每一步花了多少时间、多少钱。这两个数字会直接决定你第二周该先修哪里——而它们大概率会推翻你的直觉。

一张摊开的账单，比十条经验管用

先看一次真实运行打出来的账单，四个镜头、二十一秒的一集：

TextText

账单（耗时占比 / 折算花费）
  环节           耗时      占比                    调用  产物    折算花费
  script           32ms                         1%    1     2  ¥0.0000
  assets           93ms                         2%    2     2  ¥0.0500
  first-frame     120ms                         2%    3     3  ¥0.0750
  clip           2518ms  █████████             47%    3     3  ¥10.4950
  voice           236ms  █                      4%    5     7  ¥0.0193
  compose        2403ms  █████████             44%    0     2  ¥0.0000
  合计        5402ms                                 ¥10.6393

这张表要怎么读？先说一句必须说清的话：上面的花费是折算值。 离线模式下每一笔的真实金额都是 0，所以程序按官方按量付费页上的单价——图像每张两分半、视频按 768P 每秒五毛——把用量折算了一遍；语音那一档官方只公开资源包价格，所以那个数字是由套餐折算出来的近似值，不是官方标价。账单允许是估算，但必须标明它是估算，这是做成本可观测性时最容易被忽略的职业素养。

读出来的第一个结论很刺眼：这一集里超过百分之九十八的钱花在视频那一环。 剧本几乎不要钱，配音不到两分，五张图加起来一毛二五，而三个镜头的视频是十块五。这个比例不是本课的特例，它是整个 AI 视频生产的基本形状。

由此推出的东西非常多，多到值得单列一句：这条线上所有的工程设计，都是围着「怎么少调一次视频接口」转的。 幂等、缓存、参考图复用、先出草稿档再出成片档、预算熔断——第二周你会学到的每一样，追到根上都是这一条。

耗时那一列要小心读。离线模式下 compose 看起来最慢，那是因为视频接口被打了桩，本地 ffmpeg 反而成了大头。真实模式下 clip 一镜要排队好几分钟，耗时那一列会被彻底改写。所以账单里哪些数字在离线模式下有意义、哪些没有，你自己心里要有数，否则你会照着一张假图去做优化。还有一层：上面那些毫秒数是某台机器上跑出来的，你自己跑一定不是这个数。这张表里可复现的是调用次数、产物数与折算花费（同样的输入必然算出同样的结果），不可复现的是耗时——它只能用来比较环节之间的相对关系，不能当基准值抄。

聚合这张表的代码本身很简单，值得注意的只有一件事：折算逻辑要和真实金额分开。

billing.js

// 官方按量付费页上的单价（人民币）。语音那一档是由资源包折算的近似值，不是官方标价。
const PRICE = { imagePerItem: 0.025, videoPerSecond: 0.5, ttsPerCharacter: 0.00035 }
 
// 真实跑过就用真实金额，离线才折算。两者绝不能混在一起算，否则账永远对不上。
export function estimateCny(entry) {
  if (entry.costCny > 0) return entry.costCny
  if (entry.kind === 'image') return entry.units * PRICE.imagePerItem
  if (entry.kind === 'video') return entry.units * PRICE.videoPerSecond
  if (entry.kind === 'tts') return entry.units * PRICE.ttsPerCharacter
  return 0
}
 
export function buildBill(reports, costs) {
  return reports.map((r) => {
    const mine = costs.filter((c) => c.nodeId === r.id)
    return {
      nodeId: r.id,
      ms: r.ms,
      calls: mine.length,
      artifacts: r.artifacts.length,
      estimatedCny: Number(mine.reduce((s, c) => s + estimateCny(c), 0).toFixed(4)),
    }
  })
}

billing.py

# 官方按量付费页上的单价（人民币）。语音那一档是由资源包折算的近似值，不是官方标价。
PRICE = {"image_per_item": 0.025, "video_per_second": 0.5, "tts_per_character": 0.00035}
 
 
# 真实跑过就用真实金额，离线才折算。两者绝不能混在一起算，否则账永远对不上。
def estimate_cny(entry: dict) -> float:
    if entry["costCny"] > 0:
        return entry["costCny"]
    key = {"image": "image_per_item", "video": "video_per_second", "tts": "tts_per_character"}.get(entry["kind"])
    return entry["units"] * PRICE[key] if key else 0.0
 
 
def build_bill(reports: list[dict], costs: list[dict]) -> list[dict]:
    bill = []
    for r in reports:
        mine = [c for c in costs if c["nodeId"] == r["id"]]
        bill.append({
            "nodeId": r["id"],
            "ms": r["ms"],
            "calls": len(mine),
            "artifacts": len(r["artifacts"]),
            "estimatedCny": round(sum(estimate_cny(c) for c in mine), 4),
        })
    return bill

第四步炸了：这一天故意用最笨的办法

现在做一件反直觉的事：主动让流水线失败一次。把视频那一环——也就是第四个环节——人为打断，看看这条线现在是什么表现。

TextText

  ▸ script 剧本与人物卡 ... 32ms，产物 2 个
  ▸ assets 角色定妆图 ... 80ms，产物 2 个
  ▸ frames 每镜首帧 ... 158ms，产物 4 个
  ▸ clips 镜头视频 ... 625ms ✗
 
✗ clip 失败：（注入 failnode4）clip 环节失败：厂商返回了一个不可重试的错误
 
痛点报告
  失败环节：clip
  已经产生的产物：7 个，它们还好好地躺在 work/runs/run-20260907-741 里
  已经花掉的折算金额：¥3.1250
  现在唯一的办法：把整条命令从头再跑一遍。
  于是上面那 ¥3.1250 会被原封不动再花一次，而且是必然的，不是偶然的。

三个细节值得盯着看。

第一，八个产物还在磁盘上，但它们没有任何用。 剧本、人物卡、两张定妆图、四张首帧，全都是完好的、可以直接复用的文件。可是这条线不知道它们可以复用，所以重跑时会把它们原样再生成一遍。

第二，失败的那一环自己也花了钱。 视频是逐镜调用的，第一镜已经出去了才炸的，那三块钱是真的付了。如果账单只统计「成功的环节」，这笔钱在复盘时就凭空消失了——所以本课的账单会把失败节点也记进去，产物数是 0，花费不是 0。这一条在真实的事故复盘里非常关键：你必须能回答「这次故障烧了多少钱」。

第三，程序对此毫无办法。 它能做的只有：把已经跑完的记下来，打印一份报告，退出。没有重试、没有缓存、没有断点续跑。

这是故意的。今天如果顺手加上重试和缓存，你就永远感受不到「没有它们有多痛」，明天的工作流引擎也就变成了一个不知道为什么要学的东西。所以这一天的失败处理策略只有三个动作：记账、报告、退出。

不过有一处是必须现在就做对的：失败时不能把已经跑完的环节的耗时与花费一起丢掉。这一点决定了你的痛点报告有没有数字。

sequential.js

// 最笨的顺序执行器：边跑边记，出错就停。
// 关键不在「怎么跑」，而在「出错时手里还剩什么」——一个 reject 到底的实现，
// 会把前面几个环节的耗时与花费一起扔掉，而那正是失败时最该看的东西。
export async function runSequential(stages) {
  const reports = []
  for (const node of topoSort(stages)) {
    const started = Date.now()
    try {
      const result = await node.run()
      reports.push({ id: node.id, ms: Date.now() - started, artifacts: result.artifacts })
    } catch (err) {
      // 失败的这一环也要进账单：它往往已经调过一两次付费接口了。
      reports.push({ id: node.id, ms: Date.now() - started, artifacts: [], note: '失败，产物作废' })
      return { reports, failedAt: node.id, error: err }
    }
  }
  return { reports }
}

sequential.py

# 最笨的顺序执行器：边跑边记，出错就停。
# 关键不在「怎么跑」，而在「出错时手里还剩什么」——一个直接向上抛的实现，
# 会把前面几个环节的耗时与花费一起扔掉，而那正是失败时最该看的东西。
import time
 
 
def run_sequential(stages: list) -> dict:
    reports = []
    for node in topo_sort(stages):
        started = time.time()
        try:
            result = node.run()
            reports.append({"id": node.id, "ms": int((time.time() - started) * 1000), "artifacts": result["artifacts"]})
        except Exception as err:
            # 失败的这一环也要进账单：它往往已经调过一两次付费接口了。
            reports.append({"id": node.id, "ms": int((time.time() - started) * 1000), "artifacts": [], "note": "失败"})
            return {"reports": reports, "failedAt": node.id, "error": err}
    return {"reports": reports}

第一次看片：眼睛要往哪儿放

跑通了，账也算了，接下来是这一天最没有技术含量、却最容易被跳过的一步：把成片从头到尾看一遍。

自动化流水线最危险的状态不是报错，是一路绿灯地产出垃圾。所有环节都返回成功，文件也都在，但片子根本没法看。所以这一步不能省，而且要带着清单看，不能凭感觉。

第一遍看连贯性：同一个角色在四镜里是不是同一张脸？这是 AI 短剧最容易崩的地方，也是 D3 那些参考图手段真正要解决的问题。第二遍听音画：每句台词说完之前画面有没有提前切走？字幕出现和声音起来是不是同一时刻？第三遍看画面质量：有没有哪一镜明显比别的镜暗、糊、色调不对？第四遍看封面：那一帧看不看得出这一集在讲什么？

从脚本到系统：下周补的三块

这条线现在能跑，但它是一个脚本，不是一个系统。差在哪儿？把今天遇到的问题按性质归一归，正好三块。

第一块是可靠性。 任意一步失败就要从头再来，前面付过钱的产物白扔。今天那份痛点报告上的三块一毛二五就是代价，而一集完整跑下来是十块六——真正做起来，你一天会撞上好几次这种事。补它的是 D8 的工作流引擎：给每个节点算一个幂等键，产物按内容寻址落盘，重跑变成一次差集计算，第二次运行的付费调用次数应该是 0。

第二块是吞吐。 六个环节是一条直线，一集的耗时就是六段相加，多集只能排队。但你不能简单地把并发开到最大，因为厂商那边有额度——语音接口对充值用户是每分钟二十次请求，一集四十个分镜就是四十次合成，不做闸门必然撞上限流。补它的是 D9：按厂商分桶的并发闸门、令牌桶、优先级队列。

第三块是可控。 现在的成片好不好全靠人眼，看完也没地方记，成本只能事后算不能事前拦。补它的是 D10 的审核台、D11 的质检与合规、D12 的成本路由与预算熔断。

源码导读

动手实验

🧪 D7 实验：一条命令从一句话产出一集成片的端到端脚本与一份运行账单

代码位置：labs/ai-drama-pipeline/day-07-end-to-end

验收标准：

一条命令跑完六个环节，本次运行目录下出现成片、字幕与封面。
账单按环节列出耗时、耗时占比、调用次数、产物数与折算花费，并指出最慢和最贵分别是哪一环。
换一句话再跑，六个环节的输入都跟着变，说明业务逻辑是真的在跑而不是在放录像。
人为让第四个环节失败之后：进程非 0 退出，账单里那一行有花费但产物为 0，痛点报告打印出已经花掉的金额与「只能从头再跑一遍」。
终端最后打印了看片检查清单与三个脆弱点，每个脆弱点都点名了第二周由哪一天解决。

动手之前先确认前六天的实验都还能单独跑通，尤其是 ffmpeg 那一步——串起来之后报错会被淹没在几百行日志里，单独确认一遍能省很多时间。今天的实验不需要密钥，离线模式下六个环节全部真实执行，只有四个网络出口被打了桩。

跑一次完整流程，确认六个环节按顺序跑完，本次运行目录下的成片能打开。
读一遍账单，说出最贵的是哪一环、它占了总花费的百分之多少，再说出耗时那一列在离线模式下为什么不可信。
换一句话再跑一次，比较两次运行目录里的产物，确认它们互不覆盖也互不污染。
人为让第四个环节失败，读痛点报告，把「已经花掉多少」这个数字记下来——明天要把它变成 0。
照着看片清单把成片完整看一遍，写下你发现的每一个质量问题，并给每个问题标上「这是哪一环的责任」。

面试题

今天 3 道题在下方题库区，侧重串联之后才出现的工程问题、按环节度量耗时与成本、以及脆弱点的识别与排序。展开后先看分析过程再看要点——照着推导练，比背要点管用。标注国内高频与海外高频，方便按目标市场取舍。

检查清单与明日预告

能用一条命令从一句话产出一集完整成片，并在任意一步失败后知道卡在哪
能量出每个环节的耗时与花费，指出瓶颈在哪一环
能列出这条线现在的三个脆弱点，并说明第二周分别用什么解决
能说清为什么所有产物都要挂在运行标识下面
能说出账单里哪些数字在离线模式下不可信，以及为什么
实验的 5 条验收标准全部通过
3 道面试题不看要点也能答出至少 2 道

明天（D8）我们把这条顺序脚本换成一个有状态的工作流引擎：每个节点算一个幂等键，产物按内容寻址落盘，失败之后再跑只补做没做完的那部分。为什么是它排在第二周第一位？因为今天那份痛点报告上的数字是可以直接量化的浪费，而且视频占了九成八的成本——先修能被数字证明的那一块，这是工程排序里最不会出错的一条原则。

一集杀青：把六个环节串成端到端流水线并算清第一笔账

今日目标

小白版讲解

一集杀青那天，问题才开始

一张摊开的账单，比十条经验管用

第四步炸了：这一天故意用最笨的办法

第一次看片：眼睛要往哪儿放

从脚本到系统：下周补的三块

源码导读

动手实验

面试题

检查清单与明日预告

面试题库

评论