从剧本到成片：AI 短剧生产平台的工程化架构与落地实践

AI 短剧真正困难的部分，通常不是调用一次大模型，而是把剧本解析、分镜生成、文生图、图生视频、配音、字幕、合成等环节组织成一条稳定、可恢复、可扩展的生产线。本文以 Python、FastAPI、Celery、Redis、MySQL、对象存储与 FFmpeg 为例，完整拆解一个可落地的 AI 短剧生产平台。

一、为什么“接几个模型 API”还不够

一个最小的 AI 短剧流程大致如下：

剧本输入 ↓ 角色/场景/对白解析 ↓ 分镜与提示词生成 ↓ 角色定妆图、场景图生成 ↓ 图生视频或文生视频 ↓ 旁白/角色配音 ↓ 字幕生成与音画对齐 ↓ 转码、拼接、混音、导出

在演示环境里，可以用一个 Python 脚本串行完成这些步骤；进入真实生产后，很快会遇到以下问题：

• 视频模型一次生成可能需要数分钟，HTTP 请求无法一直等待；
• 不同模型的并发限制、计费方式、失败码和返回格式不一致；
• 同一个分镜可能需要多次重试，但不能重复扣费或重复写入数据；
• 单集包含几十个镜头，需要控制并发，否则会触发限流或拖垮 GPU；
• 中间素材体积大，不能塞进 MySQL，也不适合在服务间直接传递；
• 某个镜头失败后，应从失败节点恢复，而不是整集重新生成；
• 用户需要知道当前进度、失败原因以及预计剩余时间；
• 成片还涉及分辨率、帧率、编码格式、响度和字幕安全区等媒体工程细节。

因此，一个可用的平台至少要解决四件事：工作流编排、模型统一接入、媒体资产管理、生产过程可观测。

二、总体架构：控制面与执行面分离

推荐将平台分成控制面和执行面。控制面负责接收请求、保存状态、编排流程；执行面负责模型调用和媒体处理。

┌──────────────── Web / 管理后台 ────────────────┐ │ 剧本编辑、分镜审核、素材替换、任务监控、成片预览 │ └──────────────────────┬─────────────────────────┘ │ REST / WebSocket ┌──────────────────────▼─────────────────────────┐ │ FastAPI 控制面 │ │ 项目管理 | 工作流编排 | 模型路由 | 资产元数据 │ └──────────────┬───────────────┬─────────────────┘ │ │ MySQL（业务状态） Redis（队列、缓存、锁） │ │ ┌──────────────▼───────────────▼─────────────────┐ │ Celery 执行面 │ │ LLM Worker | Image Worker | Video Worker │ │ TTS Worker | FFmpeg Worker │ └──────────────┬─────────────────────────────────┘ │ ┌─────────▼─────────┐ ┌────────────────┐ │ 多模型 API / GPU │ │ S3 / MinIO / OSS│ └───────────────────┘ │ 原始及成品素材 │ └────────────────┘

这套设计有三个关键点：

1. API 服务不执行耗时任务，只创建工作流并快速返回job_id；
2. 队列只传递任务 ID 和对象存储地址，不传递图片、音频或视频二进制；
3. MySQL 保存“事实状态”，Redis 只承担加速、队列和短期协调，避免 Redis 数据丢失后业务状态无法恢复。

三、先设计领域模型，而不是先写模型调用代码

短剧平台最重要的实体并不是“提示词”，而是项目、剧集、镜头、任务和资产。

Project（项目） └─ Episode（剧集） ├─ Character（角色） ├─ Scene（场景） └─ Shot（镜头） ├─ Asset（图片/视频/音频/字幕） └─ TaskRun（每一步执行记录）

建议将“当前业务状态”和“每次执行历史”分开：

CREATETABLEshots(idBIGINTPRIMARYKEYAUTO_INCREMENT,episode_idBIGINTNOTNULL,shot_noINTNOTNULL,descriptionTEXTNOTNULL,dialogueTEXT,duration_msINTNOTNULL,statusVARCHAR(32)NOTNULL,versionINTNOTNULLDEFAULT0,created_atDATETIMENOTNULL,updated_atDATETIMENOTNULL,UNIQUEKEYuk_episode_shot(episode_id,shot_no));CREATETABLEtask_runs(idBIGINTPRIMARYKEYAUTO_INCREMENT,workflow_idVARCHAR(64)NOTNULL,shot_idBIGINT,stepVARCHAR(32)NOTNULL,idempotency_keyVARCHAR(128)NOTNULL,providerVARCHAR(32),provider_task_idVARCHAR(128),statusVARCHAR(32)NOTNULL,attemptINTNOTNULLDEFAULT0,input_json JSON,output_json JSON,error_codeVARCHAR(64),error_messageTEXT,started_atDATETIME,finished_atDATETIME,UNIQUEKEYuk_idempotency_key(idempotency_key),KEYidx_workflow_status(workflow_id,status));CREATETABLEassets(idBIGINTPRIMARYKEYAUTO_INCREMENT,shot_idBIGINT,asset_typeVARCHAR(32)NOTNULL,object_keyVARCHAR(512)NOTNULL,sha256CHAR(64)NOTNULL,mime_typeVARCHAR(128)NOTNULL,bytesBIGINTNOTNULL,metadata JSON,created_atDATETIMENOTNULL,UNIQUEKEYuk_sha256_type(sha256,asset_type));

TaskRun很关键。它既是重试和断点续跑的依据，也是成本统计、问题排查、供应商对账的数据来源。

四、把生产流程建模为 DAG

短剧生产不是单纯的串行流程。例如，分镜生成后，各镜头的画面和配音可以并行；单个镜头的视频必须等待该镜头的图片完成；整集拼接又必须等待全部镜头完成。这本质上是一个有向无环图（DAG）。

parse_script │ generate_storyboard │ ├─ shot_01: image ─ video ┐ │ └─ tts ─────┤ ├─ shot_02: image ─ video ├─ subtitle ─ compose ─ publish │ └─ tts ─────┤ └─ shot_N: image ─ video ┘ └─ tts ──────┘

Celery 的chain、group和chord可以表达这类关系：

fromceleryimportchord,chain,groupdefbuild_episode_workflow(episode_id:int,shot_ids:list[int]):shot_jobs=group(chain(generate_image.s(shot_id),generate_video.s(),generate_voice.s(),normalize_shot.s(),)forshot_idinshot_ids)returnchain(prepare_episode.s(episode_id),chord(shot_jobs,compose_episode.s(episode_id)),publish_episode.s(episode_id),)

实际项目中，不建议只依赖 Celery 自身保存最终业务状态。每个任务开始、成功和失败时，都应写入task_runs。即使消息代理重启，也可以根据数据库中的状态扫描出“长时间处于 RUNNING”或“应该执行但未执行”的任务并进行补偿。

任务状态机

统一状态比到处写布尔值更容易维护：

PENDING → QUEUED → RUNNING → SUCCEEDED ├──→ RETRYING → RUNNING ├──→ FAILED └──→ CANCELED

状态迁移应使用条件更新，避免多个 Worker 同时处理同一任务：

UPDATEtask_runsSETstatus='RUNNING',started_at=NOW(),attempt=attempt+1WHEREid=:task_idANDstatusIN('PENDING','QUEUED','RETRYING');

只有受影响行数为 1 的 Worker 才获得执行权。这比单独依赖 Redis 锁更稳，因为业务状态与抢占结果在同一个数据库中。

五、用适配器屏蔽不同模型的接口差异

平台通常会接入多个图片或视频供应商。业务代码不应直接依赖某个厂商的字段，而应依赖统一协议。

fromdataclassesimportdataclassfromtypingimportProtocol@dataclass(frozen=True)classVideoRequest:prompt:strimage_url:str|Noneduration_seconds:intaspect_ratio:strseed:int|None=None@dataclass(frozen=True)classSubmitResult:provider_task_id:str@dataclass(frozen=True)classPollResult:status:str# RUNNING / SUCCEEDED / FAILEDoutput_url:str|None=Noneerror_code:str|None=NoneclassVideoProvider(Protocol):defsubmit(self,request:VideoRequest)->SubmitResult:...defpoll(self,provider_task_id:str)->PollResult:...defcancel(self,provider_task_id:str)->None:...

模型路由层再根据场景选择供应商：

classVideoRouter:def__init__(self,providers,health_store):self.providers=providers self.health_store=health_storedefselect(self,*,quality:str,duration:int):candidates=[pforpinself.providersifp.supports(duration=duration,quality=quality)andself.health_store.is_available(p.name)]ifnotcandidates:raiseRuntimeError("no available video provider")# 综合成功率、P95 延迟和预估成本计算分数returnmin(candidates,key=lambdap:self.health_store.score(p.name))

这里需要特别注意：不能在一次已经提交成功的生成任务上盲目切换供应商重试。如果请求已被供应商接收，只是客户端超时，再次提交可能产生两份结果和两次费用。正确做法是先用本地幂等键查找provider_task_id，再查询原任务状态。

六、异步模型任务：轮询不等于阻塞

很多视频 API 采用“提交任务—轮询状态—下载结果”的模式。Worker 不应在一个任务里sleep十分钟，这会长期占用执行槽。

更合适的方式是使用 Celery 的倒计时重新投递：

fromceleryimportshared_task@shared_task(bind=True,autoretry_for=(TimeoutError,),retry_backoff=True,retry_jitter=True,max_retries=5)defsubmit_video(self,task_run_id:int):run=task_repo.get(task_run_id)ifrun.provider_task_id:poll_video.apply_async(args=[task_run_id],countdown=5)returnprovider=video_router.select(quality="standard",duration=5)result=provider.submit(build_video_request(run))task_repo.save_provider_task(task_run_id,provider.name,result.provider_task_id)poll_video.apply_async(args=[task_run_id],countdown=5)@shared_taskdefpoll_video(task_run_id:int):run=task_repo.get(task_run_id)result=providers[run.provider].poll(run.provider_task_id)ifresult.status=="RUNNING":delay=min(60,5*2**min(run.poll_count,4))task_repo.increase_poll_count(task_run_id)poll_video.apply_async(args=[task_run_id],countdown=delay)returnifresult.status=="FAILED":task_repo.mark_failed(task_run_id,result.error_code)returnobject_key=asset_service.import_from_url(result.output_url)task_repo.mark_succeeded(task_run_id,{"object_key":object_key})dispatch_next_step(task_run_id)

轮询间隔采用指数退避并加入随机抖动，可以避免大量任务同时访问供应商，形成“惊群”。

七、幂等、重试与补偿：稳定性的核心

分布式任务系统通常只能提供“至少执行一次”，因此业务处理必须幂等。

一个实用的幂等键可以这样构造：

importhashlibimportjsondefmake_idempotency_key(step:str,entity_id:int,entity_version:int,params:dict)->str:payload=json.dumps(params,sort_keys=True,ensure_ascii=False)digest=hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]returnf"{step}:{entity_id}:v{entity_version}:{digest}"

为什么要包含version？用户修改了某个分镜后，新任务不能误用旧结果；但在同一版本内重复点击生成，又应该命中已存在的任务。

错误分类决定重试策略

并非所有错误都值得重试：

重试上限不能只按次数设置，还应有时间预算。例如，镜头生成最多重试 4 次且总耗时不超过 30 分钟。超过预算后进入人工处理队列。

补偿任务

建议增加一个定时扫描器：

• RUNNING超过最大租约时间：查询供应商后恢复状态；
• SUCCEEDED但对象存储不存在：重新拉取或标记资产损坏；
• 所有镜头已完成但整集未合成：补发合成任务；
• 已取消项目仍有远端任务运行：调用供应商取消接口。

这类补偿机制决定了系统能否从“偶尔能跑通”升级为“可以持续生产”。

八、素材管理：数据库存元数据，对象存储放文件

素材路径建议使用稳定、可追踪的命名规则：

projects/{project_id}/episodes/{episode_id}/shots/{shot_id}/ source/reference.png generated/image_v3.png generated/video_v2.mp4 audio/dialogue_v1.wav subtitle/shot_v1.ass output/normalized_v2.mp4

上传后计算 SHA-256，并记录媒体元数据：分辨率、时长、帧率、编码器、采样率、声道数。不要只相信文件扩展名，应使用ffprobe检查真实格式。

ffprobe-verror-show_streams-show_format-ofjson input.mp4

外部模型返回的临时 URL 往往会过期。拿到结果后应立即流式下载到平台自己的对象存储，并限制最大文件大小、校验 MIME 类型和哈希，避免将不受信任的 URL 长期保存在业务数据中。

九、FFmpeg 成片流水线

模型生成的镜头常常具有不同的分辨率、帧率、音频采样率和编码参数，不能直接拼接。第一步应先归一化。

1. 视频归一化

以下命令将素材统一为 1080×1920、25 fps、H.264，并通过补边避免画面变形：

ffmpeg-ishot.mp4\-vf"scale=1080:1920:force_original_aspect_ratio=decrease,pad=1080:1920:(ow-iw)/2:(oh-ih)/2,fps=25,format=yuv420p"\-c:vlibx264-presetmedium-crf20\-an-movflags+faststart normalized.mp4

2. 音频响度统一

不同 TTS 音色的响度可能差异明显，可按短视频场景统一到约 -16 LUFS：

ffmpeg-idialogue.wav\-af"loudnorm=I=-16:TP=-1.5:LRA=11"\-ar48000-ac2normalized.wav

对质量要求较高时，应使用 FFmpeg 的双遍loudnorm：第一遍测量，第二遍带入测量值处理，结果更稳定。

3. 音画合并

ffmpeg-inormalized.mp4-inormalized.wav\-c:vcopy-c:aaac-b:a192k\-map0:v:0-map1:a:0-shortestshot_with_audio.mp4

4. 字幕烧录

ASS 比 SRT 更适合控制字体、描边、位置和安全区：

ffmpeg-ishot_with_audio.mp4\-vf"subtitles=shot.ass:fontsdir=./fonts"\-c:vlibx264-crf20-c:acopy shot_subtitled.mp4

生产环境要显式打包字体，避免服务器缺少中文字体造成方框或排版差异。同时要为字幕留出底部 UI 遮挡区，不要紧贴画面边缘。

5. 镜头拼接

所有片段编码参数一致时，可使用 concat demuxer 快速拼接：

file 'shot_001.mp4' file 'shot_002.mp4' file 'shot_003.mp4'

ffmpeg-fconcat-safe0-ifiles.txt-ccopy episode.mp4

若需要转场，则使用xfade和acrossfade滤镜。应注意转场会改变时间线，字幕时间戳也需要相应修正。

十、音频与字幕对齐不能只靠字符数

根据“字数 ÷ 平均语速”估算字幕时间，只适合原型。实际配音中存在停顿、语气和多音字，误差会逐句累积。

推荐采用以下优先级：

1. TTS 服务直接返回词级或句级时间戳；
2. 若无时间戳，使用强制对齐模型将已知文本与音频对齐；
3. 最后才使用 ASR 回识别，并把结果映射回原始台词。

字幕数据最好保留词级时间信息，最后再按规则合并成显示行：

{"text":"我们必须在天亮之前离开这里","start_ms":1240,"end_ms":3680,"words":[{"text":"我们","start_ms":1240,"end_ms":1580},{"text":"必须","start_ms":1600,"end_ms":1910}]}

合并字幕时可设置：单行不超过 16 个汉字、每条显示 1～6 秒、优先在标点处换行，并避免一句话被切成语义不完整的两段。

十一、角色一致性是产品问题，也是数据问题

AI 短剧常见问题是同一角色在不同镜头中脸型、服装和发色漂移。不能只靠在每个提示词里重复角色名解决。

平台应建立“角色圣经（Character Bible）”：

{"character_id":17,"name":"林夏","appearance":{"age":26,"hair":"齐肩黑发","costume":"米白色风衣","distinctive_features":"左眼下方有一颗小痣"},"reference_asset_ids":[301,302,303],"prompt_fragment":"26-year-old Chinese woman, shoulder-length black hair...","negative_prompt":"different clothes, different hairstyle, face distortion"}

生成镜头时，把角色版本、参考图版本、模型版本、LoRA 或身份适配器参数全部写入任务输入。这样某次生成效果异常时，才能准确复现。

一个实际有效的流程是：先生成并人工确认角色三视图，再批量生成分镜；重要角色启用参考图、固定种子或身份保持能力；角色设定修改后，只让受影响的镜头失效，而不是全项目重做。

十二、API 设计：长任务必须异步化

创建生产任务时立即返回202 Accepted：

fromfastapiimportFastAPI,Header,statusfrompydanticimportBaseModel app=FastAPI()classGenerateEpisodeRequest(BaseModel):quality:str="standard"regenerate_failed_only:bool=False@app.post("/episodes/{episode_id}/generate",status_code=status.HTTP_202_ACCEPTED)defgenerate_episode(episode_id:int,body:GenerateEpisodeRequest,idempotency_key:str=Header(alias="Idempotency-Key"),):workflow=workflow_service.create_or_get(episode_id=episode_id,idempotency_key=idempotency_key,options=body.model_dump(),)ifworkflow.is_new:start_workflow.delay(workflow.id)return{"job_id":workflow.id,"status":workflow.status,"status_url":f"/jobs/{workflow.id}",}

查询接口可以返回聚合进度：

{"job_id":"wf_01J...","status":"RUNNING","progress":63,"current_stage":"GENERATE_VIDEO","shots":{"total":24,"succeeded":14,"running":4,"failed":1},"estimated_remaining_seconds":420}

进度不能简单用“完成步骤数 ÷ 总步骤数”计算，因为视频生成与写数据库耗时相差巨大。更合理的方法是根据历史数据为各阶段设置权重，并根据模型、时长和队列等待时间估算剩余时间。

十三、资源隔离与并发控制

不同任务对资源的需求差别很大，应拆分队列：

task_routes={"tasks.llm.*":{"queue":"llm"},"tasks.image.*":{"queue":"image"},"tasks.video.*":{"queue":"video"},"tasks.tts.*":{"queue":"tts"},"tasks.ffmpeg.*":{"queue":"media"},}

这样可以分别扩容，也能避免大量轻量 LLM 任务阻塞耗 CPU 的 FFmpeg 任务。

并发限制至少分三层：

• 平台级：保护整体服务和数据库；
• 供应商级：遵守 RPM、并发数和账户配额；
• 租户级：防止一个大客户占满全部资源。

Redis 令牌桶适合做分布式限流。对于本地 GPU Worker，还应根据显存而不是只按进程数调度。例如，720P 图生视频和 1080P 图生视频可以消耗不同数量的资源令牌。

十四、可观测性：不仅看接口 QPS

平台至少要采集以下指标：

系统指标

• 各队列长度与最老任务等待时间；
• Worker 在线数、CPU、内存、GPU 利用率和显存；
• MySQL 连接池、慢查询和 Redis 内存；
• 对象存储上传、下载失败率。

业务指标

• 每个供应商的成功率、P50/P95/P99 延迟；
• 每个生成步骤的重试率和人工介入率；
• 单镜头、单集、单租户的平均成本；
• 首次成片通过率和平均返工次数；
• 从提交剧本到可预览成片的总时长。

日志应统一包含：

trace_id, workflow_id, task_run_id, episode_id, shot_id, provider, provider_task_id, attempt, elapsed_ms, error_code

不要在日志中记录完整 API Key、签名 URL、用户隐私数据或未经脱敏的剧本全文。

十五、成本控制必须进入架构设计

生成式模型的成本远高于普通 Web API，工程上应主动减少无效调用：

1. 内容寻址缓存：模型版本、提示词、参数和输入素材哈希完全一致时复用结果；
2. 分级预览：分镜审核阶段使用低分辨率图片和低成本模型，确认后再生成高清视频；
3. 局部失效：修改一句台词，只重做对应配音、字幕与后续合成；
4. 预算预检：任务启动前估算费用，超出项目预算则要求确认；
5. 失败熔断：某供应商连续失败时暂停新提交，防止错误请求持续计费；
6. 资产去重：相同哈希的素材只保存一份物理文件。

可以为每次模型调用记录成本快照：

estimated_cost → reserved_cost → actual_cost

任务提交前预占预算，结束后按实际费用结算，失败或取消则释放剩余额度。这能避免高并发时多个任务同时通过预算检查导致超支。

十六、安全与内容合规

AI 内容平台不能把安全当作上线前的附加功能。至少应覆盖：

• 对剧本、提示词和生成结果进行分阶段内容审核；
• 限制外部下载地址，防止 SSRF 访问内网；
• 对上传文件执行类型、大小和解码校验；
• 对对象存储使用短期签名 URL 和最小权限凭证；
• API Key 存入密钥管理服务，不写进代码、数据库明文或日志；
• 记录素材来源、模型版本、生成时间与编辑历史，满足内容溯源；
• 为删除项目设计异步清理和审计流程，覆盖数据库、缓存、对象存储及备份策略。

尤其要注意：ffmpeg的输入文件和滤镜参数不能直接拼接用户输入后交给 Shell。应使用参数数组启动子进程，并将用户素材限制在受控目录中。

十七、一个可执行的 MVP 迭代路线

不要第一版就接入十种模型。更合理的迭代方式是：

第一阶段：跑通闭环

• 单一 LLM、图片、视频和 TTS 供应商；
• 支持剧本解析、分镜人工确认、单集生成；
• Celery 异步任务、MySQL 状态、MinIO 素材；
• FFmpeg 归一化、拼接、字幕烧录；
• 失败任务手工重试。

第二阶段：提升稳定性

• 引入幂等键、任务租约、自动重试和补偿扫描；
• 拆分队列并设置租户级限流；
• 增加指标、链路追踪和成本统计；
• 支持从失败镜头断点续跑。

第三阶段：提升生产效率

• 多供应商模型适配与动态路由；
• 角色圣经、参考图与一致性控制；
• 批量素材调度、版本管理和局部失效；
• 在线分镜编辑、低清预览与高清终稿；
• 自动质量检测与人工审核工作台。

十八、工程落地时最容易踩的坑

最后总结几个高频问题：

1. 把耗时生成放在 Web 请求中：会造成超时、重复提交和连接资源耗尽；
2. 只用 Celery 状态当业务状态：队列数据无法替代可审计的业务数据库；
3. 任务失败就整集重跑：成本高，且会覆盖已经通过审核的镜头；
4. 直接拼接模型视频：分辨率、时基和编码参数不一致时容易音画不同步；
5. 外部结果 URL 永久入库：临时链接过期后资产不可用；
6. 所有错误统一重试：内容违规、欠费等永久错误只会放大故障和费用；
7. 只记录最终文件：缺少输入参数、模型版本和中间资产时无法复现；
8. 按平均耗时估算进度：长尾模型任务会让进度长时间卡在 99%；
9. 忽略取消语义：本地任务取消但远端仍在生成，费用仍会发生；
10. 过早追求完全自动化：短剧审美具有主观性，在角色定妆、分镜和终稿阶段保留人工确认，往往更省成本。

结语

AI 短剧生产平台并不是一个“大模型套壳”，而是一个融合了分布式任务、媒体工程、对象存储、模型网关、成本治理和内容审核的复杂生产系统。

真正有价值的工程能力，是让任意一个步骤失败后都能定位、重试和恢复；让每一份素材都可追踪、可复现；让模型供应商可以替换，而上层业务无需重写；让创作者能在低成本预览和高质量终稿之间顺畅迭代。

当系统具备这些能力后，AI 才不只是一次生成，而会成为一条稳定、可规模化的内容生产线。