电视端AI内容创作流水线：从脚本生成到HLS发布的技术实践

在智能电视平台上，AI 内容创作正在从编辑器里的一个小工具变成整套内容生产链路。Roku 近期在产品方向里引入了 Fairground AI Creator TV，这个方向把生成式 AI 与电视端的内容创建、预览、发布流程放到了一起。对开发者来说，理解这件事不能只停留在“Roku 出了一个 AI 功能”这个层面，更要看清它背后的技术链路：文本怎么变成视频脚本，视频脚本怎么变成镜头列表，生成出来的媒体资源怎么做合规检查、转码、字幕合成，最终又如何以电视端能识别的方式发布成频道或播放列表。

这篇文章会以 Fairground AI Creator TV 为切入点，梳理电视端 AI 内容创作产品的核心模块，再带读者从零搭建一个最小可运行的内容生成流水线。完成之后，你会得到一套本地可跑的 AI 创作服务原型，它能完成“输入主题 -> 生成脚本 -> 生成镜头画面 -> 合成配音和字幕 -> 输出 HLS 播放目录”的完整闭环。这套原型虽然不能直接复刻 Roku 所有的商业能力，但足够帮助你理解电视端 AI 生成产品的工程结构、参数取舍和排错方法。

1. 先理解 Fairground AI Creator TV 这条主线，再动手设计技术方案

1.1 这类产品解决什么问题

电视端的传统内容生产链路非常重：制片方拍好素材，经过剪辑、调色、字幕、配音、审核、编码，再交给平台发布。一个短视频可能在手机上几分钟就能剪完，但同一段内容要进入电视端，至少要满足分辨率、码率、字幕、封面、播放入口和内容审核这些要求。Fairground AI Creator TV 这类产品想解决的问题，就是把这个链条中可以用大模型自动化的部分抽出来，让创作者输入一个主题或一句话，系统自动生成完整的、适合电视端播放的节目内容。

这里要澄清一个容易混淆的点。它不等于“AI 生成一段视频”，更接近“AI 生成一档电视节目”。两者的差别在于：

• 单个生成视频只需要画面和声音。
• 电视节目需要脚本结构、分镜、连贯的画外音、字幕、段落标题、封面图，以及可以顺序播放的内容列表。

从产品角度理解，它是一个面向电视大屏的 AI 内容创作平台。从工程角度理解，它是一条输入主题、输出“可播放节目包”的异步流水线。

1.2 从产品界面到后端能力，拆出技术模块

如果要在工程上复现这类产品，不能只盯住某一个图像生成接口，而要按“创作 -> 生成 -> 合成 -> 发布”四段来拆。

第一段是创作编辑。创作者输入节目主题、风格、时长，这块对应前端交互和提示词管理。

第二段是内容生成。系统调用大语言模型生成脚本，再调用图像或视频生成模型生成镜头画面，同时调用语音合成模型生成画外音。

第三段是合成封装。脚本需要按时间轴把画面、配音、字幕、背景音乐合成到一起，这一步通常需要 FFmpeg 或其他媒体处理工具。

第四段是发布适配。电视端需要识别节目入口、播放列表、海报图、字幕文件和分级信息。Roku 这类平台本身有专门的频道格式和 metadata 规范，自建原型时可以用 HLS 播放目录和 JSON 配置替代。

这四个模块正好对应后端服务里最典型的四个目录：prompt、generator、compositor、publisher。后面搭建原型时会按这个目录组织代码。

1.3 内容安全是电视端 AI 创作的硬门槛

AI 生成内容进入电视端，内容安全审核不是可选项，而是发布之前必须经过的一道工序。电视端的受众范围更广，播放场景是家庭客厅，内容分级、敏感画面、不当语音都需要被拦截。

在工程上，内容安全不能只靠“生成之后再审核”这一道关卡，推荐在流水线的多个节点都做检查：

• 文本脚本生成后，先做一次文本合规判断。
• 图像模型出图后，再对画面做一次识别。
• 配音生成后，对音频转写结果做一次复核。
• 最终成片上线前，根据平台规则做终审。

这里不使用任何非合规工具，只提正常工程实践：文本审核可以用关键字加模型分类，图像审核可以用通用视觉理解模型，语音复核可以用 ASR 转写后再走文本策略。多级审核会增加延迟，但在电视端场景里，安全优先级高于生成速度。

2. 电视端 AI 内容创作和普通 AI 生成流程有哪些本质差异

2.1 画面、时长和封装要求完全不同

手机端看 AI 生成视频，通常是一段十几秒到一分钟的竖屏片段，分辨率 720x1280 或 1080x1920，生成完直接播放。电视端内容则是横屏 16:9，单集时长可能是三到五分钟，甚至更长。时长一旦变长，就不能指望“一个视频接口生成完整成片”，必须拆成分镜，再逐段生成并拼接。

电视端对封装格式要求也更高。普通的 MP4 文件可以直接播放，但多集节目、自动续播、字幕切换、清晰度切换这些电视端常见需求，都需要流媒体协议支持。HLS 是最容易自建验证的方案，它能在一份 m3u8 播放列表里组合多个分片，同时携带字幕和音频轨道。

2.2 交互方式从“按钮点击”变成“遥控器确认”

手机应用可以设计复杂表单，用户点击、滑动、拖拽都很自然。电视端只有遥控器，焦点移动、确认、返回是主要操作。这意味着 AI 创作工具在电视端的交互不能设计成一大段文本框，更适合的形态是：用户选择一个主题模板，按确认键触发生成，然后在预览页查看结果。

这块对后端的影响在于：生成任务必须支持异步提交、进度查询和结果回显。用户按下确认后，服务端要立刻返回一个任务 ID，前端轮询任务状态，而不是让用户一直等待模型返回。

2.3 技术架构差异：异步任务、配额、审核和 CDN 缺一不可

电视端 AI 创作的调用链路通常比手机端更长。以一个 3 分钟节目为例，系统可能要先做 10 次以上的模型调用：脚本生成一次、镜头描述生成一次、图像生成 6 到 10 次、配音生成 1 次、字幕生成 1 次。如果每一步都同步等待，任何一个模型超时都会导致请求中断。

所以电视端 AI 创作服务的架构核心是任务队列。任务状态至少包括：等待中、执行中、审核中、已完成、失败。每个任务内部记录当前步骤、重试次数、产物路径。下面的表格展示了手机端生成和电视端生成在架构上的差异。

3. 搭建最小可运行原型：从主题到成片的流水线

3.1 项目结构与技术选型

这一节开始搭建最小原型。技术选型用 Python 加 FastAPI，媒体处理用 FFmpeg，任务队列先用内存队列，不引入 Redis。生成模型相关的接口统一封装成provider，方便切换真实模型和模拟模型。

本地原型目录结构如下：

fairground_ai_creator/ ├── app/ │ ├── main.py │ ├── models/ │ │ ├── task.py │ │ └── script.py │ ├── services/ │ │ ├── script_service.py │ │ ├── image_service.py │ │ ├── audio_service.py │ │ ├── subtitle_service.py │ │ ├── compositor.py │ │ └── task_runner.py │ ├── providers/ │ │ ├── llm_provider.py │ │ ├── image_provider.py │ │ └── tts_provider.py │ └── config.py ├── output/ │ ├── tasks/ │ └── media/ ├── requirements.txt └── README.md

providers层的作用是隔离外部模型厂商。开发调试时可以用本地模拟实现，接入真实服务时只需要替换这一层，不影响上层业务逻辑。

requirements.txt先写成这样：

fastapi==0.110.0 uvicorn[standard]==0.29.0 pydantic==2.6.0 httpx==0.27.0 jinja2==3.1.3 python-multipart==0.0.9

3.2 定义任务模型与状态机

任务是整个流水线的核心数据。一个任务至少需要这些字段：

• task_id：任务唯一 ID。
• status：任务当前状态。
• progress：0 到 100 的进度。
• step：当前执行到哪一步。
• error：失败原因。
• result：产物信息，包括视频路径、播放列表、字幕路径。

用 Pydantic 定义任务模型：

from enum import Enum from typing import Optional from pydantic import BaseModel, Field class TaskStatus(str, Enum): PENDING = "pending" RUNNING = "running" REVIEWING = "reviewing" COMPLETED = "completed" FAILED = "failed" class Task(BaseModel): task_id: str = Field(default_factory=lambda: uuid4().hex) status: TaskStatus = TaskStatus.PENDING progress: int = 0 step: str = "queued" error: Optional[str] = None title: str = "" style: str = "documentary" duration_seconds: int = 180 output_dir: Optional[str] = None video_url: Optional[str] = None subtitle_url: Optional[str] = None playlist_url: Optional[str] = None

状态流转规则是：pending -> running -> reviewing -> completed，任意running阶段出错则进入failed。reviewing状态单独拿出来，是为了让内容审核在正式发布前有独立停留点。

3.3 生成脚本并把内容拆成镜头列表

脚本生成这一步调用大语言模型。输入是主题和风格，输出是一份结构化的节目脚本。为了让后续流程稳定，要求模型输出 JSON，而不是自由文本。

import json from app.providers.llm_provider import LLMProvider class ScriptService: def __init__(self, llm: LLMProvider): self.llm = llm def generate_script(self, title: str, style: str, duration_seconds: int) -> dict: estimated_shots = max(4, duration_seconds // 20) prompt = f""" 你是一个电视节目编剧。请根据以下要求生成一个适合电视端播放的节目脚本： 标题：{title} 风格：{style} 目标时长：{duration_seconds}秒 分镜数量：{estimated_shots} 输出要求：只输出 JSON，不要输出其他内容。JSON 结构如下： {{ "title": "节目标题", "summary": "一句话摘要", "shots": [ {{ "index": 1, "narration": "画外音文本", "image_prompt": "画面描述，用于图像生成", "duration": 20 }} ] }} """ text = self.llm.chat(prompt) return json.loads(text)

这里有一个容易出错的点：模型输出的 JSON 可能带有 Markdown 代码块标记，比如json ...。解析之前要先清理。更稳妥的方法是在提示词里要求“只输出 JSON”，并在解析失败时做一次正则清理。

import re def parse_json_response(text: str) -> dict: text = text.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text) return json.loads(text)

3.4 调用图像生成服务并做合规检查

拿到分镜列表后，每个image_prompt对应一张画面。图像生成模型通常按张数计费，所以最小原型里要控制并发，避免一瞬间把配额用完。

import asyncio from pathlib import Path class ImageService: def __init__(self, image_provider, output_root: Path, review_func=None): self.image_provider = image_provider self.output_root = output_root self.review_func = review_func async def generate_shot_images(self, shots: list[dict], task_id: str) -> list[Path]: image_paths = [] for shot in shots: prompt = shot.get("image_prompt", "") # 生成前先做一次文本合规判断 if self.review_func and not self.review_func(prompt): raise RuntimeError(f"prompt denied: {shot.get('index')}") image_path = await self.image_provider.generate( prompt=prompt, output_path=self.output_root / task_id / f"shot_{shot['index']:02d}.png", ) image_paths.append(image_path) # 生成后再对图片内容做一次视觉审核 if self.review_func: self.review_func(image_path) return image_paths

在实际项目中，图像生成调用要加超时和重试。默认超时可以设置为 60 秒，重试次数设置为 2 次。如果提供商返回内容违规，立刻中断任务而不是继续生成后面的分镜。

3.5 自动配音并生成字幕

配音生成要按分镜逐段完成，每段配音时长尽量与分镜时长对齐。语音合成接口通常返回音频字节或临时文件，这里把每段配音保存为独立文件。

class AudioService: def __init__(self, tts_provider, output_root: Path): self.tts_provider = tts_provider self.output_root = output_root async def generate_narrations(self, shots: list[dict], task_id: str) -> list[dict]: audio_segments = [] for shot in shots: narration = shot["narration"] audio_path = await self.tts_provider.synthesize( text=narration, output_path=self.output_root / task_id / f"audio_{shot['index']:02d}.mp3", ) audio_segments.append({ "index": shot["index"], "path": audio_path, "duration": shot.get("duration", 20), }) return audio_segments

字幕文件用 SRT 格式生成。每段字幕的开始时间和结束时间可以通过前序分镜的时长累加得到。

def build_srt(shots: list[dict], audio_segments: list[dict], output_path: Path): lines = [] cursor = 0 for shot, audio in zip(shots, audio_segments): start_ms = cursor * 1000 duration_ms = audio.get("duration", 20) * 1000 end_ms = start_ms + duration_ms lines.append(f"{shot['index']}") lines.append(f"{format_srt_time(start_ms)} --> {format_srt_time(end_ms)}") lines.append(shot["narration"]) lines.append("") cursor += audio.get("duration", 20) output_path.write_text("\n".join(lines), encoding="utf-8")

4. 关键参数与配置说明：分辨率、异步任务、审核阈值怎么定

4.1 生成模型的画面参数

电视端画面默认使用 1920x1080。如果内容要在性能较低的电视设备上播放，建议输出 1920x1080 的 H.264 视频，不要直接输出 4K。4K 生成成本高，兼容性差，对 3 到 5 分钟的节目来说不是首选。

图像模型参数需要关注分辨率、画面比例和质量。电视端固定 16:9，所以图像生成时宽高比要明确传参，避免模型按默认的 1:1 输出。

4.2 异步任务的超时与重试参数

任务队列是电视端 AI 创作服务的核心。每个子步骤都不能无限等待，必须设置超时和重试。

4.3 内容审核阈值

内容审核采用模型分类时，通常会输出每个类别的置信度分数。推荐设置一个严格的拦截阈值和一个转人工阈值。

不要把所有风险都交给图像审核模型。第一道防线是文本提示词审核，第二道防线是生成后的图像审核，第三道防线是配音 ASR 复核。

4.4 输出媒体参数

合成视频时，FFmpeg 的编码参数直接影响文件大小和播放兼容性。

ffmpeg -i shot_01.png -i audio_01.mp3 -c:v libx264 -preset veryfast -crf 23 -c:a aac -b:a 128k -pix_fmt yuv420p segment_01.mp4

-pix_fmt yuv420p很重要。许多电视播放器不支持 yuv444 或 yuv422，输出 yuv420p 能保证兼容性。使用-crf 23可以在画质和体积之间取得平衡，如果想要更高画质，可以调整到 18 到 20，但文件体积会上升。

5. 把生成结果组装成电视端可识别的播放结构

5.1 定义输出目录约定

每一个生成任务对应一个独立目录，目录名使用任务 ID。目录内部按角色拆分文件，方便排查问题时定位。

output/tasks/<task_id>/ ├── script.json ├── images/ │ ├── shot_01.png │ └── shot_02.png ├── audios/ │ ├── audio_01.mp3 │ └── audio_02.mp3 ├── subtitles/ │ └── subtitles.srt ├── segments/ │ ├── segment_01.mp4 │ └── segment_02.mp4 ├── playlist.m3u8 ├── cover.png └── metadata.json

这种约定的好处是：脚本、素材、中间产物和最终产物都保留在任务目录里，调试时可以直接查看任意一步的结果。

5.2 生成 HLS 播放列表

电视端播放器通常支持 HLS，不需要一次性把所有分镜拼接成一个巨型 MP4。把每个分镜生成独立切片，再生成m3u8播放列表，既方便单段重试，也方便按需加载。

def build_playlist(segment_paths: list[Path], output_playlist: Path): lines = ["#EXTM3U", "#EXT-X-VERSION:3", "#EXT-X-TARGETDURATION:30"] for segment in segment_paths: duration = probe_duration(segment) lines.append(f"#EXTINF:{duration:.2f},") lines.append(segment.name) lines.append("#EXT-X-ENDLIST") output_playlist.write_text("\n".join(lines), encoding="utf-8")

实际生产环境中，HLS 分段需要经过转码工具生成真正的.ts切片，并设置#EXT-X-MEDIA-SEQUENCE等字段。最小原型里可以先构建索引文件，验证播放器是否能正确识别。

5.3 遥控器场景下的展示与控制接口

电视端展示 AI 创作内容时，需要同时提供元数据接口和播放接口。元数据接口返回标题、摘要、封面、分级、时长。播放接口返回播放列表地址。

from fastapi import FastAPI, HTTPException app = FastAPI() @app.get("/tasks/{task_id}") def get_task(task_id: str): task = task_store.get(task_id) if not task: raise HTTPException(status_code=404, detail="task not found") return task @app.get("/tasks/{task_id}/playlist") def get_playlist(task_id: str): task = task_store.get(task_id) if task.status != "completed": raise HTTPException(status_code=400, detail="task not ready") return {"playlist_url": task.playlist_url}

在电视端产品中，这里的接口要考虑两点：一是返回字段尽量少，方便前端快速渲染；二是图片和视频的 URL 要能跨域访问，不能在服务端把二进制数据直接返回，否则电视播放器难以缓存和快进。

6. 运行验证：从提交任务到检查成片

6.1 启动本地服务

在项目根目录安装依赖并启动 FastAPI 服务：

pip install -r requirements.txt uvicorn app.main:app --reload --port 8000

启动成功后访问http://127.0.0.1:8000/docs，可以看到 Swagger 接口文档。这是检查接口是否注册成功最直接的方式。

6.2 提交一个生成任务

向任务创建接口发送一个主题，例如“城市夜景介绍”，风格选择“纪录片”。原型里可以先使用模拟 Provider，不调用真实付费模型。

curl -X POST "http://127.0.0.1:8000/tasks" \ -H "Content-Type: application/json" \ -d '{"title": "城市夜景介绍", "style": "documentary", "duration_seconds": 60}'

预期返回：

{ "task_id": "a1b2c3d4e5f6", "status": "pending", "progress": 0, "step": "queued" }

6.3 查询任务状态

任务执行需要时间，电视端场景下使用轮询方式获取状态。

curl -X GET "http://127.0.0.1:8000/tasks/a1b2c3d4e5f6"

正常情况下，状态会经历pending -> running -> reviewing -> completed。如果中间某一步失败，状态会变成failed，同时error字段会记录失败原因。

6.4 检查产物文件

任务完成后，进入任务目录检查文件是否齐全：

find output/tasks/a1b2c3d4e5f6 -type f | sort

至少应该看到script.json、metadata.json、playlist.m3u8、cover.png和多个segment_*.mp4文件。如果缺少任何关键文件，说明对应生成步骤没有执行成功。

6.5 用 ffprobe 验证媒体参数是否符合电视端要求

检查视频编码、分辨率和像素格式：

ffprobe -v error -show_entries stream=codec_name,width,height,pix_fmt -of json output/tasks/a1b2c3d4e5f6/segments/segment_01.mp4

预期输出中codec_name是h264，width是 1920，height是 1080，pix_fmt是yuv420p。如果 pix_fmt 不是 yuv420p，电视播放器可能无法正常解码，需要在合成命令里加-pix_fmt yuv420p重新生成。

7. 常见问题排查：AI 创作流水线为什么卡住或产出不符合预期

7.1 任务长时间停留在 pending

现象：任务提交后一直没有进入 running。

可能原因有三个：

1. 任务队列没有消费者线程，代码里创建任务后没有调用task_runner.start()。
2. 队列并发数被设置为 0，导致没有 worker 可以执行。
3. 前置审核接口一直未返回，任务在进入队列前就卡住。

建议按这个顺序检查：先看日志是否输出“task worker started”，再看任务队列长度，最后检查审核服务是否有响应。最小原型中可以在创建任务后显式asyncio.create_task(runner.execute(task_id))，避免依赖未启动的后台线程。

7.2 图像内容生成后出现拉伸变形

现象：生成画面是 16:9 容器，但内容被明显拉宽。

原因通常是图像模型按 1:1 生成图片，合成阶段直接强制缩放成 1920x1080。图像内容本身没有按 16:9 构图，强制拉伸就会变形。

推荐做法是：在图像生成请求里明确传aspect_ratio=16:9。如果模型不支持比例参数，先生成更大画幅，再用 FFmpeg 的 pad 或 crop 方式适配 1920x1080，不要直接拉伸。

7.3 配音与画面不同步

现象：画面切换速度明显快于配音进度，或者配音已经结束但画面还在播放。

原因通常在于每段分镜的duration是脚本生成阶段估计出来的，而实际配音时长更长或更短。合成视频时以画面时长为基准，导致配音被截断或留白。

建议以实际音频时长为准，动态更新每个分镜的展示时长。代码中可以用ffprobe读取音频文件时长，再回写任务数据。如果音频过长，可以在不改变语义的前提下用语音合成服务的speed参数适当加速。

7.4 电视端播放黑屏但声音正常

现象：播放器有声音，但画面黑屏。

可能原因是视频像素格式不支持，或者帧率过高。电视端播放器对 H.264 的兼容性较好，但对 HEVC 和 yuv444 的支持参差不齐。检查ffprobe输出，如果pix_fmt是yuv444p，重新转码为yuv420p。如果codec_name是hevc，改用libx264。

7.5 内容审核误伤过多，导致生成任务频繁失败

现象：正常内容被审核接口拦截，任务大量失败。

原因通常是审核阈值设置过严，或者提示词里包含容易触发模型误判的词汇。比如在纪录片里出现“危险动作”“吸烟画面”等描述，可能被模型判定为不适合播放。

建议把审核结果分为三级：直接拒绝、转人工、通过。对于转人工内容，不要立刻终止任务，先让成片进入reviewing状态，由人工确认后再发布。这样既能保证安全，也能减少误伤。

7.6 模型返回 JSON 解析失败

现象：脚本生成步骤报JSONDecodeError。

原因是大语言模型可能返回 Markdown 代码块，或者在 JSON 前后附加了说明文字。虽然提示词里写了“只输出 JSON”，但模型仍有概率不遵守。

解决方案是在parse_json_response中增加清理逻辑，去掉 ```json 标记和首尾空白。如果清理后仍解析失败，可以把原始文本写入output/tasks/<task_id>/error_response.txt，方便人工检查模型输出。

8. 生产环境要补的工程能力和可复用清单

8.1 学习原型与生产系统的差距

最小原型追求的是跑通链路，生产环境则需要补充以下能力：

1. 持久化任务存储。内存队列在服务重启后会丢失所有任务，生产环境至少使用 Redis 加数据库的组合。
2. 可追踪的日志链路。每次模型调用、每段合成命令都要记录任务 ID、耗时、状态码和失败原因。
3. 配额与计费。模型调用不是免费的，需要按用户、按任务记录 credits 消耗，防止异常任务打爆账单。
4. 分布式锁。如果同一个任务被重复提交，消费端需要保证幂等，避免重复生成扣费。
5. 内容版本管理。AI 生成内容可能有多次重试，要保留历史版本，方便回滚。
6. 文件存储改造。本地目录要替换为对象存储，并配置 CDN 加速，保证电视端播放不卡顿。

8.2 可复用的发布前检查清单

以下清单适用于把 AI 创作服务发布到测试或生产环境之前：

8.3 扩展方向：从单集生成到频道化运营

原型实现的是单集节目生成，Fairground AI Creator TV 这类产品更有价值的方向是频道化运营。频道不是单条视频，而是可以按主题、风格、播放顺序组织起来的内容集合。扩展思路有三个：

1. 模板化生成系列节目。比如“科技简史”“城市漫游”这些固定栏目，只替换脚本背景和画面风格，形成批量生产能力。
2. 根据观看数据动态生成后续内容。电视端播放数据回传后，可以调整下一集的选题方向和画面风格。
3. 在生成流水线中加入人工编辑节点。AI 生成初稿，人工在电视端或 Web 端确认、修改提示词、替换镜头，再发布。这个模式比全自动生成更适合高品质内容。

从工程角度看，Fairground AI Creator TV 这类产品最值得借鉴的不是某一个生成模型的效果，而是把“生成能力”和“电视端发布能力”组合成一条可控流水线的方法。新手可以先从最小原型里把 LLM 生成脚本、图像生成、配音、字幕合成这些环节分别摸清楚，再逐步加上审核、任务持久化和流媒体封装，最终才能形成一个真正能上线服务的系统。