从剧本到成片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│ └───────────────────┘ │ 原始及成品素材 │ └────────────────┘这套设计有三个关键点API 服务不执行耗时任务只创建工作流并快速返回job_id队列只传递任务 ID 和对象存储地址不传递图片、音频或视频二进制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_jobsgroup(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_runsSETstatusRUNNING,started_atNOW(),attemptattempt1WHEREid:task_idANDstatusIN(PENDING,QUEUED,RETRYING);只有受影响行数为 1 的 Worker 才获得执行权。这比单独依赖 Redis 锁更稳因为业务状态与抢占结果在同一个数据库中。五、用适配器屏蔽不同模型的接口差异平台通常会接入多个图片或视频供应商。业务代码不应直接依赖某个厂商的字段而应依赖统一协议。fromdataclassesimportdataclassfromtypingimportProtocoldataclass(frozenTrue)classVideoRequest:prompt:strimage_url:str|Noneduration_seconds:intaspect_ratio:strseed:int|NoneNonedataclass(frozenTrue)classSubmitResult:provider_task_id:strdataclass(frozenTrue)classPollResult:status:str# RUNNING / SUCCEEDED / FAILEDoutput_url:str|NoneNoneerror_code:str|NoneNoneclassVideoProvider(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.providersproviders self.health_storehealth_storedefselect(self,*,quality:str,duration:int):candidates[pforpinself.providersifp.supports(durationduration,qualityquality)andself.health_store.is_available(p.name)]ifnotcandidates:raiseRuntimeError(no available video provider)# 综合成功率、P95 延迟和预估成本计算分数returnmin(candidates,keylambdap:self.health_store.score(p.name))这里需要特别注意不能在一次已经提交成功的生成任务上盲目切换供应商重试。如果请求已被供应商接收只是客户端超时再次提交可能产生两份结果和两次费用。正确做法是先用本地幂等键查找provider_task_id再查询原任务状态。六、异步模型任务轮询不等于阻塞很多视频 API 采用“提交任务—轮询状态—下载结果”的模式。Worker 不应在一个任务里sleep十分钟这会长期占用执行槽。更合适的方式是使用 Celery 的倒计时重新投递fromceleryimportshared_taskshared_task(bindTrue,autoretry_for(TimeoutError,),retry_backoffTrue,retry_jitterTrue,max_retries5)defsubmit_video(self,task_run_id:int):runtask_repo.get(task_run_id)ifrun.provider_task_id:poll_video.apply_async(args[task_run_id],countdown5)returnprovidervideo_router.select(qualitystandard,duration5)resultprovider.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],countdown5)shared_taskdefpoll_video(task_run_id:int):runtask_repo.get(task_run_id)resultproviders[run.provider].poll(run.provider_task_id)ifresult.statusRUNNING:delaymin(60,5*2**min(run.poll_count,4))task_repo.increase_poll_count(task_run_id)poll_video.apply_async(args[task_run_id],countdowndelay)returnifresult.statusFAILED:task_repo.mark_failed(task_run_id,result.error_code)returnobject_keyasset_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:payloadjson.dumps(params,sort_keysTrue,ensure_asciiFalse)digesthashlib.sha256(payload.encode(utf-8)).hexdigest()[:16]returnf{step}:{entity_id}:v{entity_version}:{digest}为什么要包含version用户修改了某个分镜后新任务不能误用旧结果但在同一版本内重复点击生成又应该命中已存在的任务。错误分类决定重试策略并非所有错误都值得重试错误类型示例策略瞬时错误连接超时、HTTP 502指数退避重试限流错误HTTP 429读取Retry-After延迟重试内容错误提示词违规、图片格式非法不自动重试返回用户修改资源错误GPU 显存不足降低并发或路由到其他节点永久错误API Key 无效、账户欠费熔断供应商并告警重试上限不能只按次数设置还应有时间预算。例如镜头生成最多重试 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\-vfscale1080:1920:force_original_aspect_ratiodecrease,pad1080:1920:(ow-iw)/2:(oh-ih)/2,fps25,formatyuv420p\-c:vlibx264-presetmedium-crf20\-an-movflagsfaststart normalized.mp42. 音频响度统一不同 TTS 音色的响度可能差异明显可按短视频场景统一到约 -16 LUFSffmpeg-idialogue.wav\-afloudnormI-16:TP-1.5:LRA11\-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.mp44. 字幕烧录ASS 比 SRT 更适合控制字体、描边、位置和安全区ffmpeg-ishot_with_audio.mp4\-vfsubtitlesshot.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.mp4ffmpeg-fconcat-safe0-ifiles.txt-ccopy episode.mp4若需要转场则使用xfade和acrossfade滤镜。应注意转场会改变时间线字幕时间戳也需要相应修正。十、音频与字幕对齐不能只靠字符数根据“字数 ÷ 平均语速”估算字幕时间只适合原型。实际配音中存在停顿、语气和多音字误差会逐句累积。推荐采用以下优先级TTS 服务直接返回词级或句级时间戳若无时间戳使用强制对齐模型将已知文本与音频对齐最后才使用 ASR 回识别并把结果映射回原始台词。字幕数据最好保留词级时间信息最后再按规则合并成显示行{text:我们必须在天亮之前离开这里,start_ms:1240,end_ms:3680,words:[{text:我们,start_ms:1240,end_ms:1580},{text:必须,start_ms:1600,end_ms:1910}]}合并字幕时可设置单行不超过 16 个汉字、每条显示 16 秒、优先在标点处换行并避免一句话被切成语义不完整的两段。十一、角色一致性是产品问题也是数据问题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 AcceptedfromfastapiimportFastAPI,Header,statusfrompydanticimportBaseModel appFastAPI()classGenerateEpisodeRequest(BaseModel):quality:strstandardregenerate_failed_only:boolFalseapp.post(/episodes/{episode_id}/generate,status_codestatus.HTTP_202_ACCEPTED)defgenerate_episode(episode_id:int,body:GenerateEpisodeRequest,idempotency_key:strHeader(aliasIdempotency-Key),):workflowworkflow_service.create_or_get(episode_idepisode_id,idempotency_keyidempotency_key,optionsbody.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工程上应主动减少无效调用内容寻址缓存模型版本、提示词、参数和输入素材哈希完全一致时复用结果分级预览分镜审核阶段使用低分辨率图片和低成本模型确认后再生成高清视频局部失效修改一句台词只重做对应配音、字幕与后续合成预算预检任务启动前估算费用超出项目预算则要求确认失败熔断某供应商连续失败时暂停新提交防止错误请求持续计费资产去重相同哈希的素材只保存一份物理文件。可以为每次模型调用记录成本快照estimated_cost → reserved_cost → actual_cost任务提交前预占预算结束后按实际费用结算失败或取消则释放剩余额度。这能避免高并发时多个任务同时通过预算检查导致超支。十六、安全与内容合规AI 内容平台不能把安全当作上线前的附加功能。至少应覆盖对剧本、提示词和生成结果进行分阶段内容审核限制外部下载地址防止 SSRF 访问内网对上传文件执行类型、大小和解码校验对对象存储使用短期签名 URL 和最小权限凭证API Key 存入密钥管理服务不写进代码、数据库明文或日志记录素材来源、模型版本、生成时间与编辑历史满足内容溯源为删除项目设计异步清理和审计流程覆盖数据库、缓存、对象存储及备份策略。尤其要注意ffmpeg的输入文件和滤镜参数不能直接拼接用户输入后交给 Shell。应使用参数数组启动子进程并将用户素材限制在受控目录中。十七、一个可执行的 MVP 迭代路线不要第一版就接入十种模型。更合理的迭代方式是第一阶段跑通闭环单一 LLM、图片、视频和 TTS 供应商支持剧本解析、分镜人工确认、单集生成Celery 异步任务、MySQL 状态、MinIO 素材FFmpeg 归一化、拼接、字幕烧录失败任务手工重试。第二阶段提升稳定性引入幂等键、任务租约、自动重试和补偿扫描拆分队列并设置租户级限流增加指标、链路追踪和成本统计支持从失败镜头断点续跑。第三阶段提升生产效率多供应商模型适配与动态路由角色圣经、参考图与一致性控制批量素材调度、版本管理和局部失效在线分镜编辑、低清预览与高清终稿自动质量检测与人工审核工作台。十八、工程落地时最容易踩的坑最后总结几个高频问题把耗时生成放在 Web 请求中会造成超时、重复提交和连接资源耗尽只用 Celery 状态当业务状态队列数据无法替代可审计的业务数据库任务失败就整集重跑成本高且会覆盖已经通过审核的镜头直接拼接模型视频分辨率、时基和编码参数不一致时容易音画不同步外部结果 URL 永久入库临时链接过期后资产不可用所有错误统一重试内容违规、欠费等永久错误只会放大故障和费用只记录最终文件缺少输入参数、模型版本和中间资产时无法复现按平均耗时估算进度长尾模型任务会让进度长时间卡在 99%忽略取消语义本地任务取消但远端仍在生成费用仍会发生过早追求完全自动化短剧审美具有主观性在角色定妆、分镜和终稿阶段保留人工确认往往更省成本。结语AI 短剧生产平台并不是一个“大模型套壳”而是一个融合了分布式任务、媒体工程、对象存储、模型网关、成本治理和内容审核的复杂生产系统。真正有价值的工程能力是让任意一个步骤失败后都能定位、重试和恢复让每一份素材都可追踪、可复现让模型供应商可以替换而上层业务无需重写让创作者能在低成本预览和高质量终稿之间顺畅迭代。当系统具备这些能力后AI 才不只是一次生成而会成为一条稳定、可规模化的内容生产线。
