AI工作流生成短视频：从零搭建自动化内容生产系统

1次阅读

共计 2012 个字符，预计需要花费 6 分钟才能阅读完成。

短视频内容生产面临三大效率瓶颈：

脚本创作耗时：人工撰写需要创意构思和反复修改，单条脚本平均消耗 1 - 2 小时
素材匹配困难：寻找符合主题的图片 / 视频素材占整体制作时间的 40%
后期制作复杂：剪辑、配音、字幕同步等工序需要专业软件操作经验

技术组件	适用场景	优势	局限性
GPT-4	脚本生成	语义理解强，支持多轮迭代	API 调用成本较高
Stable Diffusion	素材生成	可定制化生成图片	需要 GPU 资源
CLIP	图文匹配	跨模态检索精度高	需预构建素材库
Edge-TTS	语音合成	免费、支持多语种	音色选择较少
FFmpeg	视频合成	跨平台、支持硬件加速	命令行参数复杂

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("作为短视频编剧，请创作关于 {theme} 的 30 秒脚本，包含 5 个分镜描述"
)
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
chain = prompt | llm
response = chain.invoke({"theme": "科技改变生活"})

import clip
import torch
from PIL import Image

# 加载预训练模型
device = "cuda" if torch.cuda.is_available() else "cpu"
model, preprocess = clip.load("ViT-B/32", device=device)

# 计算文本特征
text_input = clip.tokenize(["科技感实验室"]).to(device)
with torch.no_grad():
    text_features = model.encode_text(text_input)

# 在素材库中搜索最匹配图片
best_match = None
max_sim = -1
for img_path in material_library:
    image = preprocess(Image.open(img_path)).unsqueeze(0).to(device)
    image_features = model.encode_image(image)
    sim = torch.cosine_similarity(text_features, image_features)
    if sim > max_sim:
        max_sim = sim
        best_match = img_path

import edge_tts
import asyncio

async def generate_voice(text, output_file):
    voice = edge_tts.Communicate(
        text=text,
        voice="zh-CN-YunxiNeural",
        rate="+10%"
    )
    await voice.save(output_file)

asyncio.run(generate_voice(response.content, "output.mp3"))

ffmpeg \
  -loop 1 -i bg.png -i audio.mp3 \
  -vf "drawtext=text=' 科技改变生活 ':fontcolor=white:fontsize=48:box=1:boxcolor=black@0.5:boxborderw=5:x=(w-text_w)/2:y=h-th-60" \
  -c:v libx264 -preset fast -crf 22 \
  -c:a aac -b:a 192k \
  -shortest output.mp4

flowchart TD
    A[输入主题] --> B(GPT- 4 生成脚本)
    B --> C{脚本解析}
    C --> D[CLIP 匹配素材]
    C --> E[TTS 生成配音]
    D --> F[FFmpeg 合成视频]
    E --> F
    F --> G[输出成品]

API 调用优化
使用异步请求并发处理多个生成任务
实现请求结果缓存机制
设置 rate limit 重试策略

本地缓存设计

from diskcache import Cache

cache = Cache("./clip_cache")

@cache.memoize()
def get_clip_features(text):
    return model.encode_text(clip.tokenize(text))