AI视频自动化流水线：用Python整合脚本生成、素材检索、TTS配音与字幕合成

最近做短视频和知识类内容的同学应该都有一个体会剪片子本身不难真正烦的是写脚本、找素材、配音、加字幕这些重复劳动。如果你同时维护好几个账号每天光“找素材”就能耗掉俩小时。这次我们来看一条可以串起来的 AI 自动化流水线写脚本、找素材、配音、字幕全部用脚本编排完成。这条流水线的思路不复杂把 LLM 的文案生成、素材站点的检索下载、TTS 语音合成、ASR 语音识别和 FFmpeg 视频拼接串成一个 Python Pipeline。你只需要输入一个选题系统自动输出一段带配音和字幕的成片素材。核心价值不是单点工具而是把四五个零散能力组合成一条可重复执行、可批量跑的生产线。先说门槛。这条流水线不是典型的“显存敏感型 AI 应用”它更偏 API 编排。如果所有 AI 能力都走云端接口CPU 电脑也能跑如果想把 TTS 和字幕识别换成本地模型才需要关注显存。因此本文会同时覆盖云端接口版和本地模型版的部署思路并给出通用代码模板。想要完整跑通建议准备 Python 3.10 以上环境、一个 LLM API Key、一个 TTS 服务或本地模型、一个素材源以及 FFmpeg。1. 核心能力速览能力项说明项目类型AI 内容生产自动化流水线脚本生成 素材采集 TTS 配音 字幕生成 视频封装核心模块AI 写脚本、自动找素材、自动配音、字幕生成、FFmpeg 合成硬件门槛纯云端 API 方案可 CPU 运行本地 TTS/ASR 方案建议 NVIDIA GPU显存需按具体模型测试启停方式Python 命令行执行支持任务配置驱动可封装为 Web API 服务接口能力LLM API、TTS API、ASR API、素材站公开接口可统一封装为 pipeline 服务批量任务支持读取任务清单批量生成关键步骤支持断点重试输出格式脚本 JSON、配音音频、SRT 字幕、合成 MP4适合场景短视频批量制作、知识科普视频、图文转视频、内部培训素材生成部署难度中低主要是多个 API 的鉴权和数据格式适配这条流水线的本质是“AI 能力 自动化编排”。单看写脚本ChatGPT、DeepSeek、本地 Qwen 都可以做单看配音各种 TTS 工具已经成熟。把模块用代码连起来才形成“全自动化”。2. 流水线的整体架构2.1 主流程流水线的完整处理链路如下输入一个选题例如“介绍 Python 的列表推导式”。脚本模块调用 LLM 生成口播文案并把文案拆成若干场景段落每个段落包含“口播词”和“画面描述”。素材模块根据“画面描述”生成搜索关键词到本地素材库或素材站点搜索下载素材。配音模块将每个场景的口播词合成为语音文件。字幕模块根据配音生成带时间轴的字幕可以用语音识别也可以直接利用 TTS 的时间戳。封装模块用 FFmpeg 把素材、配音、字幕合成为最终视频。2.2 模块职责表模块输入输出关键依赖script_agent选题、风格、时长场景化 JSON 文案LLM API / 本地模型material_finder画面描述关键词素材文件列表素材站 API / 本地目录tts_engine每段口播词每段音频文件TTS API / 本地 TTSsubtitle_engine音频文件、口播词SRT 字幕ASR API / TTS 时间戳merge_pipeline素材、音频、字幕MP4 成片FFmpeg2.3 数据流设计流水线各模块之间通过临时文件和 JSON 交换数据。这样做的好处是方便断点重跑脚本生成完写入output/scripts/配音生成完写入output/audio/某一步失败后不需要重新执行前面的模块。工作目录建议这样规划pipeline/ ├── config/ │ ├── settings.yaml │ └── tasks.json ├── input/ │ └── topics.txt ├── output/ │ ├── scripts/ │ ├── materials/ │ ├── audio/ │ ├── subtitle/ │ └── videos/ ├── pipeline/ │ ├── __init__.py │ ├── script_agent.py │ ├── material_finder.py │ ├── tts_engine.py │ ├── subtitle_engine.py │ └── merge_pipeline.py └── main.py3. 环境准备与前置条件3.1 基础运行环境建议按以下环境准备版本号以实际安装为准Python 3.10 或更高版本。FFmpeg用于音频格式转换、字幕烧录和视频合成。pip 包管理工具建议使用 venv 或 conda 隔离环境。如果使用本地 TTS 或 ASR 模型按模型要求安装 PyTorch 和 CUDA 驱动。创建并激活虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install --upgrade pip3.2 Python 依赖根据模块不同需要安装以下依赖未用到本地模型的可按需裁剪pip install requests pyyaml openai-whisper edge-tts faster-whisper pip install openai1.0.0说明openai用于调用兼容 OpenAI 协议的 LLM APIDeepSeek、本地 vLLM、Ollama 等都兼容。edge-tts是微软 Edge 的免费 TTS 接口适合快速验证配音流程。faster-whisper用于本地语音转写适合生成字幕时间轴。requests用于调用素材站和通用 HTTP API。pyyaml用于读取配置文件。3.3 环境变量配置把各类 API Key 统一放到.env文件中避免写死在代码里LLM_API_KEYsk-xxxx LLM_BASE_URLhttps://api.deepseek.com/v1 LLM_MODELdeepseek-chat TTS_VOICEzh-CN-XiaoxiaoNeuralPython 端读取环境变量推荐使用python-dotenvpip install python-dotenvfrom dotenv import load_dotenv import os load_dotenv() api_key os.getenv(LLM_API_KEY)3.4 端口与进程检查如果后续把流水线封装成 Web API 服务注意端口冲突。启动前检查端口占用# Linux / macOS lsof -i :8080 # Windows netstat -ano | findstr :80804. 模块一AI 写脚本4.1 功能说明这个模块负责把“一个选题”扩展成“一条带画面设计的视频脚本”。相比直接让 LLM 写一段文字流水线模式要求输出结构化 JSON方便后续素材模块和配音模块直接消费。4.2 提示词设计提示词要明确要求模型输出 JSON并指定字段结构和约束你是一个短视频脚本策划。请根据选题“{topic}”生成一段约 60 秒的口播脚本。 要求 1. 输出 JSON不要输出其他文字。 2. JSON 字段为 scenesscenes 是数组。 3. 每个 scene 包含 title、narration、visual、duration。 4. narration 是口播词不超过 80 字。 5. visual 是画面描述用于视频素材检索要求具体、可翻译成搜索关键词。 6. duration 是场景时长单位为秒所有场景时长之和约等于 60。 输出示例 { scenes: [ { title: 开场, narration: 今天讲一个 Python 小技巧列表推导式。, visual: 程序员在电脑前写代码屏幕上出现列表推导式代码, duration: 8 } ] }4.3 调用示例下面是用 OpenAI 兼容协议调用 LLM 的通用示例接口地址和模型名需要按你自己的服务商调整import json import os from openai import OpenAI client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), ) def generate_script(topic: str) - dict: prompt f你是一个短视频脚本策划。请根据选题“{topic}”生成一段约 60 秒的口播脚本。 要求 1. 输出 JSON不要输出其他文字。 2. JSON 字段为 scenesscenes 是数组。 3. 每个 scene 包含 title、narration、visual、duration。 4. narration 是口播词不超过 80 字。 5. visual 是画面描述具体、可翻译成搜索关键词。 6. duration 是场景时长单位秒所有场景时长之和约等于 60。 response client.chat.completions.create( modelos.getenv(LLM_MODEL, deepseek-chat), messages[ {role: system, content: 你只输出合法 JSON。}, {role: user, content: prompt}, ], temperature0.7, ) content response.choices[0].message.content # 有些模型会在 JSON 外包裹 json需要清理 content content.strip().removeprefix(json).removesuffix().strip() return json.loads(content) if __name__ __main__: data generate_script(Python 列表推导式) print(json.dumps(data, ensure_asciiFalse, indent2))4.4 预期输出与失败排查正常返回时scenes会包含多个场景。容易出的问题有三个模型返回的不是纯 JSON包含大量解释文本。处理方式是清理首尾的 json 标记再尝试解析。duration之和与目标时长不一致。可以在后处理中按比例缩放或者让模型先列大纲再逐段生成。API 超时。建议设置请求超时时间并在失败后重试 2 到 3 次。5. 模块二自动找素材5.1 功能说明素材模块的核心是把visual画面描述变成视频素材列表。根据资源来源分为两种模式本地素材库模式适合有积累的创作者从本地目录按关键词匹配已有视频或图片。在线素材源模式调用免费版权素材站的 API 或下载页面按关键词下载素材。无论哪种模式都要注意素材版权。优先选择 CC0、CC BY 或明确允许商用的素材源不要直接抓取无授权网站。5.2 本地素材库匹配本地素材模式的思路是维护一个素材清单文件每行记录素材路径和标签然后用关键词匹配import os import json import shutil def search_local_material(tag: str, material_root: str, output_dir: str, topk: int 1): 按标签在本地素材库中检索并复制到当前任务目录。 tag tag.lower() matched [] for root, dirs, files in os.walk(material_root): for name in files: if not name.lower().endswith((.mp4, .mov, .jpg, .png, .webp)): continue # 用文件名或同目录的 tags.txt 做简单匹配 if tag in name.lower(): matched.append(os.path.join(root, name)) else: tag_file os.path.join(root, tags.txt) if os.path.exists(tag_file): with open(tag_file, r, encodingutf-8) as f: if tag in f.read().lower(): matched.append(os.path.join(root, name)) matched matched[:topk] copied [] for src in matched: dst os.path.join(output_dir, os.path.basename(src)) shutil.copy2(src, dst) copied.append(dst) return copied5.3 在线素材源下载在线素材源一般需要对接具体平台的 API这里给出通用下载模板。实际使用时以素材平台提供的接口文档为准import requests import os def download_material(url: str, save_dir: str, filename: str) - str: 下载素材文件到指定目录返回本地路径。 os.makedirs(save_dir, exist_okTrue) save_path os.path.join(save_dir, filename) headers {User-Agent: Mozilla/5.0} with requests.get(url, headersheaders, streamTrue, timeout60) as r: r.raise_for_status() with open(save_path, wb) as f: for chunk in r.iter_content(chunk_size8192): f.write(chunk) return save_path5.4 素材与场景的映射关系素材模块输出的本地文件需要按场景编号保存在output/materials/scene_001/、output/materials/scene_002/这样的目录下方便封装阶段对齐。如果某个场景没有匹配到素材可以先用文字底图兜底由 LLM 生成一张背景图或者直接使用模板背景。6. 模块三自动配音6.1 功能说明配音模块把每个场景的narration文本转换成语音文件。推荐优先使用 Edge TTS 做快速验证优点是免费、无需 GPU、音色中文支持好生产环境可以换成商业 TTS API 或本地模型。6.2 Edge TTS 快速配音Edge TTS 通过edge-tts命令行和 Python API 使用示例命令如下edge-tts --voice zh-CN-YunxiNeural --text 今天讲一个 Python 小技巧 --write-media output.mp3在流水线中推荐用异步方式逐句生成import asyncio import edge_tts VOICE zh-CN-YunxiNeural async def text_to_audio(text: str, output_path: str): communicate edge_tts.Communicate(text, VOICE) await communicate.save(output_path) def generate_voice_for_scene(scene: dict, scene_index: int, audio_dir: str) - str: os.makedirs(audio_dir, exist_okTrue) audio_path os.path.join(audio_dir, fscene_{scene_index:03d}.mp3) asyncio.run(text_to_audio(scene[narration], audio_path)) return audio_path6.3 本地 TTS 模型方案如果对音色有强需求比如需要固定某位配音老师的音色建议使用本地或云端商业 TTS。本地 TTS 模型通常提供 Python SDK 或 HTTP 接口。显存占用与具体模型有关6GB 显存可以跑轻量模型更大模型需要 12GB 以上实际情况以模型文档为准。本地 TTS 启动后通常可以通过 HTTP 接口调用调用模板如下import requests def tts_local(text: str, server_url: str, output_path: str): resp requests.post( server_url, json{text: text, voice: default, format: mp3}, timeout120, ) resp.raise_for_status() with open(output_path, wb) as f: f.write(resp.content)实际字段名以你部署的 TTS 服务文档为准。生产环境建议直接读取 TTS 返回的音频二进制并落盘。6.4 多音字和语气词处理TTS 合成时经常出现多音字念错的问题。治本的方法是用支持 SSML 标记的 TTS 服务通过phoneme或alias字段指定读音。如果 TTS 不支持 SSML就维护一个“多音字替换表”在配音前对文本做替换。PRONUNCIATION_ALIAS { 重载: 重(zhong4)载(zai4), 单行: 单(dan1)行(xing2), } def apply_pronunciation_alias(text: str) - str: for word, replaced in PRONUNCIATION_ALIAS.items(): text text.replace(word, replaced) return text这种做法不是最优解但在不支持 SSML 的免费 TTS 上非常有效。7. 模块四字幕生成与对齐7.1 两种方案对比方案原理优点缺点ASR 转写对配音音频做语音识别输出带时间戳的文本通用任何音频都能处理识别准确率影响字幕正确性TTS 时间戳从 TTS 服务直接获取每句话的时间戳准确率高无识别误差依赖 TTS 服务是否输出时间戳7.2 基于 ASR 的 SRT 生成使用 faster-whisper 生成带时间段的文本from faster_whisper import WhisperModel model WhisperModel(small, devicecpu, compute_typeint8) def generate_srt(audio_path: str, srt_path: str): segments, info model.transcribe(audio_path, languagezh, vad_filterTrue) entries [] for i, seg in enumerate(segments, start1): start_ms int(seg.start * 1000) end_ms int(seg.end * 1000) text seg.text.strip() entries.append((i, start_ms, end_ms, text)) with open(srt_path, w, encodingutf-8) as f: for i, start_ms, end_ms, text in entries: f.write(f{i}\n) f.write(f{format_srt_time(start_ms)} -- {format_srt_time(end_ms)}\n) f.write(f{text}\n\n) def format_srt_time(ms: int) - str: hours ms // 3600000 minutes (ms % 3600000) // 60000 seconds (ms % 60000) // 1000 millis ms % 1000 return f{hours:02}:{minutes:02}:{seconds:02},{millis:03}这里用的是 CPU 推理small模型在 CPU 上也能处理短视频音频。如果批量任务多建议换 GPU 和更大的模型识别准确率更高。7.3 字幕渲染生成 SRT 后用 FFmpeg 把字幕烧录到视频中。常见有两种方式硬字幕把字幕烧录到画面里所有播放器都能显示。软字幕把 SRT 封装进 MKV 或 MP4播放器可开关。流水线场景通常选择硬字幕ffmpeg -i input.mp4 -vf subtitlessubtitle.srt:force_styleFontSize16,PrimaryColourH00FFFFFF,OutlineColourH00000000 -c:a copy output.mp4注意 Windows 下字幕路径中的反斜杠和冒号需要转义。如果 ffmpeg 找不到 SRT建议先把字幕复制到工作目录并使用相对路径。8. 模块五流水线串联与批量任务8.1 串行执行主脚本主脚本负责调用四个模块并记录执行日志import json import os import traceback from datetime import datetime def run_pipeline(topic: str, task_id: str): base_dir foutput/{task_id} dirs { scripts: os.path.join(base_dir, scripts), materials: os.path.join(base_dir, materials), audio: os.path.join(base_dir, audio), subtitle: os.path.join(base_dir, subtitle), videos: os.path.join(base_dir, videos), } for d in dirs.values(): os.makedirs(d, exist_okTrue) log_path os.path.join(base_dir, pipeline.log) def log(msg): line f[{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}] {msg} print(line) with open(log_path, a, encodingutf-8) as f: f.write(line \n) try: log(f开始处理选题: {topic}) script generate_script(topic) script_path os.path.join(dirs[scripts], script.json) with open(script_path, w, encodingutf-8) as f: json.dump(script, f, ensure_asciiFalse, indent2) for idx, scene in enumerate(script[scenes], start1): log(f场景 {idx}: 查找素材) material_list search_local_material(scene[visual], materials_lib, dirs[materials]) log(f场景 {idx}: 生成配音) audio_path generate_voice_for_scene(scene, idx, dirs[audio]) log(f场景 {idx}: 生成字幕) srt_path os.path.join(dirs[subtitle], fscene_{idx:03d}.srt) generate_srt(audio_path, srt_path) log(所有场景处理完成进入合成阶段) merge_all(dirs) log(合成完成) except Exception as e: log(流水线执行失败: str(e)) log(traceback.format_exc()) return False return True8.2 任务清单批量处理批量任务通过config/tasks.json驱动每一行是一个选题{ tasks: [ { id: task_001, topic: Python 列表推导式, style: 知识科普, duration: 60 }, { id: task_002, topic: Docker 容器基础入门, style: 教程讲解, duration: 90 } ] }批量执行脚本import json def run_batch(config_path: str): with open(config_path, r, encodingutf-8) as f: config json.load(f) success_count 0 for task in config[tasks]: ok run_pipeline(task[topic], task[id]) if ok: success_count 1 else: print(f任务 {task[id]} 失败请查看日志) print(f批量执行完成成功 {success_count}/{len(config[tasks])}) if __name__ __main__: run_batch(config/tasks.json)批量任务的关键是每个任务使用独立task_id目录避免输出互相覆盖。如果需要并发执行可以用concurrent.futures.ThreadPoolExecutor但要注意素材下载、TTS 服务是否有并发限制。8.3 失败断点与重试流水线最怕跑一半挂了所有场景都要重新生成。建议加“结果文件检查”逻辑如果某个场景的音频已存在且文件大小不为 0就直接跳过 TTS。import os def scene_audio_exists(audio_dir: str, scene_index: int) - bool: path os.path.join(audio_dir, fscene_{scene_index:03d}.mp3) return os.path.exists(path) and os.path.getsize(path) 100这一步能从很大程度上提升批量任务的稳定性。9. 资源占用与性能观察9.1 不同环节的消耗对比环节CPU 占用GPU 或云端依赖主要耗时因素LLM 写脚本低云端 API网络延迟、模型响应速度素材下载低无网络带宽、素材数量Edge TTS 配音低云端服务音频长度、并发数本地 TTS 配音中GPU 建议模型大小、合成时长faster-whisper 字幕中GPU 可加速CPU 核数、音频时长、模型大小FFmpeg 合成中高无视频分辨率、素材长度9.2 显存占用观察方法如果用本地 TTS 或 faster-whisper可以通过命令行持续观察显存nvidia-smi -l 2更建议在执行脚本中记录推理前和推理后的显存占用import subprocess def get_gpu_memory_mb() - int: result subprocess.run( [nvidia-smi, --query-gpumemory.used, --formatcsv,noheader,nounits], capture_outputTrue, textTrue, ) return int(result.stdout.strip().split(\n)[0])显存占用与模型版本、推理分辨率、batch size 有关。实际部署时先用最小的模型参数跑一遍再逐步增大找到一个稳定不爆显存的配置。9.3 性能优化思路LLM 写脚本时限制输出 token 数避免超长响应拖慢整体流程。素材下载采用并发下载但控制并发数防止被素材站点限流。TTS 按场景并行合成注意 TTS 服务的 QPS 限制超出后做退避重试。faster-whisper 在 CPU 上处理长音频较慢建议开启 VAD 过滤静音段。FFmpeg 合成阶段转码是耗时大户尽量让素材与输出目标分辨率一致避免缩放。9.4 端口和进程残留如果把流水线封装成 Web API要特别注意进程残留问题。启动服务后记录 PID停止时主动清理# 查看监听 8080 端口的进程 lsof -i :8080 # 按 PID 结束进程 kill -9 PID10. 常见问题与排查方法问题现象可能原因排查方式解决方案LLM 返回的内容解析失败模型输出非 JSON打印原始 content清理 json 包裹增加 json 修复逻辑LLM API 超时网络波动或模型响应慢看请求日志设置 timeout失败后指数退避重试素材匹配为空关键词与素材标签不匹配打印 visual 和检索 tag让 LLM 输出多个备选关键词素材下载失败网络限制或防盗链测试单个 URL 是否可访问调整 User-Agent换成素材站官方 SDK配音音频为空TTS 服务限流检查返回状态码增加重试降低并发多音字读错TTS 无 SSML 支持试听定位错误词使用多音字替换表字幕时间轴偏移ASR 识别产生偏差播放验证使用 TTS 时间戳方案或对 ASR 分段做偏移校正ffmpeg 找不到字幕文件路径未转义查看 ffmpeg 报错使用相对路径Windows 路径做转义批量任务中途卡住某步骤等待网络响应查看 pipeline.log增加单步超时和看门狗重试10.1 日志排查方法建议所有模块都通过统一的log()函数写日志并记录每个场景开始和结束时间。排查问题时先看日志中最后一个成功的步骤再检查对应输出目录下是否生成了文件。10.2 API 调用失败排查顺序遇到 API 调用失败按以下顺序排查环境变量是否读取成功打印 Key 是否存在但不打印完整 Key。API 服务地址是否可访问用 curl 测试基础连通性。请求参数是否符合文档要求重点检查模型名和鉴权头。是否触发频率限制查看响应头中的限流字段。11. 最佳实践与合规边界11.1 工程化建议第一次执行先跑单条任务不要直接批量。单条任务定位问题最快。保留一套最小可运行配置。比如只跑“写脚本 Edge TTS 配音”确认稳定后再逐步添加素材和字幕模块。模型文件、输入素材、输出结果分目录管理。不要把第三方素材和生成结果混在一起。批量任务一定要有日志和失败重试。建议记录每个任务每个场景的执行状态而不是只记录最后是否成功。接口服务要限制访问范围。如果封装成 Web API默认绑定 127.0.0.1不要暴露到公网。配置文件和环境变量分离。API Key 只放.env不要提交到代码仓库。11.2 版权与隐私合规这条流水线本质上涉及素材、声音、内容生成三类风险需要特别注意素材版权。找素材只能使用明确授权的免费版权素材例如 CC0、CC BY 或平台允许商用的素材库。不要批量抓取未经授权的图片、视频和音频。影视片段、音乐、他人摄影作品都不应该直接进入流水线。声音授权。如果要把某位真人或特定配音演员的声音训练成 TTS 音色必须获得本人明确授权。用于商用内容时授权范围更严格。内容合规。AI 生成的文案、视频内容发布前必须人工复核确认不包含侵权、不实信息或违规内容。肖像权。任何情况下不要用生成式 AI 处理真实人物肖像做虚假代言或误导性内容。11.3 发布前的复核流程自动生成的视频不应直接发布。建议增加一个“人工复核”环节检查脚本是否有事实错误。检查素材是否匹配口播内容。检查配音音色是否自然。检查字幕是否有错别字和时间轴偏差。确认素材授权信息和出处可追溯。12. 总结这条“写脚本、找素材、配音、字幕全自动化”的 AI 流水线最值得尝试的点不是某一个模块而是把多个 AI 能力串联成可复用的生产线。你可以在一小时内部署一个最小版本LLM 生成脚本、Edge TTS 配音、faster-whisper 出字幕、FFmpeg 合成全部跑在 CPU 环境里。最先应该验证的功能是“AI 写脚本 配音”。这两个模块链路短、反馈直接能快速确认 API 配置和格式处理是否正确。最容易踩的坑是两个一是 LLM 输出的 JSON 不稳定需要增加清理和解析逻辑二是素材版权批量下载前务必确认授权范围。建议收藏备用后续可以继续扩展的方向包括接入更自然的本地 TTS 音色、增加视频画面智能匹配、把流水线封装成 Web 服务、接入任务队列实现真正的无人值守批量生产。