AI视频生成技术完全教程

教程简介

全面讲解AI视频生成的核心技术与实战应用,涵盖Sora/Runway/可灵等主流模型对比、文本生成视频、图像生成视频、视频编辑与风格迁移、提示词工程优化、视频生成API调用实战、长视频分镜编排、音频同步等核心内容,帮助开发者掌握AI视频生成技术。

AI视频生成技术完全教程

从基础原理到实战应用,全面掌握AI视频生成核心技术


教程简介

AI视频生成是当前人工智能领域最令人瞩目的技术方向之一。从2023年Runway Gen-1的惊艳亮相,到2024年OpenAI Sora的震撼发布,再到国内可灵、VEO等模型的快速迭代,AI视频生成技术正以前所未有的速度改变着内容创作的格局。

本教程将从底层技术原理出发,深入剖析视频生成模型的核心架构,系统对比主流视频生成工具,手把手带你完成文本生成视频、图像生成视频、视频编辑等实战任务,并分享商用视频生成工作流中的成本控制与效率优化策略。

你将学到:

  • 视频生成模型的核心架构(DiT、U-Net、Transformer)
  • Sora、Runway Gen-3、可灵、VEO等主流模型的技术特点与选型
  • 文本生成视频(Text-to-Video)与图像生成视频(Image-to-Video)的完整流程
  • 视频编辑、风格迁移与时序一致性控制
  • 提示词工程与视频质量优化技巧
  • 视频生成API调用实战(Replicate、FAL、自建服务)
  • 长视频生成与分镜编排策略
  • 音频同步与配音集成
  • 商用视频生成工作流与成本控制

一、AI视频生成技术概览

1.1 发展历程

AI视频生成技术的发展可以追溯到几个关键里程碑:

2022年:基础探索期

  • CogVideo(清华)首次展示文本到视频的可能性
  • Make-A-Video(Meta)证明了大规模预训练的潜力
  • Imagen Video(Google)将扩散模型引入视频生成

2023年:快速迭代期

  • Runway Gen-1/Gen-2实现商业化落地
  • Stable Video Diffusion开源视频生成模型
  • Pika Labs带来更易用的视频生成体验
  • 国内可灵、通义万相等模型崭露头角

2024-2025年:成熟爆发期

  • Sora展示了超长视频生成的可能性
  • Runway Gen-3 Alpha大幅提升质量
  • 可灵1.5/2.0在中文场景表现优异
  • Google VEO 2带来电影级质量
  • 开源生态(HunyuanVideo、Wan2.1等)逐步完善

1.2 核心技术路线

当前AI视频生成主要遵循以下技术路线:

技术路线 代表模型 核心思想 优势 劣势
扩散模型 Sora, Runway, 可灵 从噪声逐步去噪生成视频 质量高、可控性强 推理速度较慢
自回归模型 CogVideoX 逐帧或逐块生成 长视频生成自然 误差累积
混合架构 VEO 2, Kling 结合扩散与自回归 兼顾质量与效率 架构复杂
GAN系列 早期模型 对抗生成 速度快 质量不稳定

二、核心模型架构深度解析

2.1 扩散模型基础

扩散模型(Diffusion Model)是当前视频生成的主流架构,其核心思想来源于非平衡热力学:

前向过程(加噪):

x_t = √(α_t) * x_{t-1} + √(1-α_t) * ε,  ε ~ N(0, I)

反向过程(去噪):

x_{t-1} = (1/√α_t) * (x_t - (1-α_t)/√(1-ᾱ_t) * ε_θ(x_t, t))

其中 ε_θ 是去噪网络,α_t 是噪声调度参数。

2.2 U-Net架构(Stable Video Diffusion)

U-Net架构在图像扩散模型中取得了巨大成功,扩展到视频领域时引入了时间维度:

import torch
import torch.nn as nn

class TemporalAttention(nn.Module):
    """时间维度注意力层,用于在帧间建立关联"""
    def __init__(self, dim, num_heads=8):
        super().__init__()
        self.norm = nn.LayerNorm(dim)
        self.attention = nn.MultiheadAttention(dim, num_heads, batch_first=True)
        # 可学习的时间位置编码
        self.temporal_pos_embed = nn.Parameter(torch.randn(1, 64, dim) * 0.02)

    def forward(self, x, num_frames):
        """
        x: [B*H*W, T, C] - 将空间维度合并,沿时间维度做注意力
        """
        B_HW, T, C = x.shape
        residual = x

        x = self.norm(x)
        # 添加时间位置编码
        if T <= self.temporal_pos_embed.shape[1]:
            x = x + self.temporal_pos_embed[:, :T, :]

        # 沿时间维度做自注意力
        x, _ = self.attention(x, x, x)
        return x + residual


class SpatioTemporalBlock(nn.Module):
    """时空注意力块:先空间注意力,再时间注意力"""
    def __init__(self, dim, num_heads=8):
        super().__init__()
        self.spatial_attn = nn.MultiheadAttention(dim, num_heads, batch_first=True)
        self.temporal_attn = TemporalAttention(dim, num_heads)
        self.norm1 = nn.LayerNorm(dim)
        self.norm2 = nn.LayerNorm(dim)

    def forward(self, x, H, W, T):
        """
        x: [B, T*H*W, C]
        """
        B, L, C = x.shape

        # 空间注意力:对每一帧独立做空间注意力
        x_spatial = x.reshape(B * T, H * W, C)
        x_spatial = self.norm1(x_spatial)
        x_spatial, _ = self.spatial_attn(x_spatial, x_spatial, x_spatial)
        x = x + x_spatial.reshape(B, L, C)

        # 时间注意力:对每个空间位置独立做时间注意力
        x_temporal = x.reshape(B, T, H * W, C).permute(0, 2, 1, 3)  # [B, H*W, T, C]
        x_temporal = x_temporal.reshape(B * H * W, T, C)
        x_temporal = self.temporal_attn(x_temporal, T)
        x_temporal = x_temporal.reshape(B, H * W, T, C).permute(0, 2, 1, 3).reshape(B, L, C)

        return x_temporal


class VideoUNet(nn.Module):
    """简化的视频U-Net架构示意"""
    def __init__(self, in_channels=4, model_channels=320, num_frames=16):
        super().__init__()
        self.num_frames = num_frames

        # 编码器
        self.down_blocks = nn.ModuleList([
            self._make_block(in_channels, model_channels),
            self._make_block(model_channels, model_channels * 2),
            self._make_block(model_channels * 2, model_channels * 4),
        ])

        # 中间块
        self.mid_block = SpatioTemporalBlock(model_channels * 4)

        # 解码器
        self.up_blocks = nn.ModuleList([
            self._make_block(model_channels * 4, model_channels * 2),
            self._make_block(model_channels * 2, model_channels),
            self._make_block(model_channels, in_channels),
        ])

    def _make_block(self, in_ch, out_ch):
        return nn.Sequential(
            nn.Conv3d(in_ch, out_ch, kernel_size=(1, 3, 3), padding=(0, 1, 1)),
            nn.GroupNorm(8, out_ch),
            nn.SiLU(),
        )

    def forward(self, x, timestep, text_embedding):
        """
        x: [B, C, T, H, W] - 噪声视频张量
        timestep: 时间步
        text_embedding: 文本条件嵌入
        """
        # 编码
        for block in self.down_blocks:
            x = block(x)

        # 中间处理(时空注意力)
        B, C, T, H, W = x.shape
        x_flat = x.permute(0, 2, 3, 4, 1).reshape(B, T * H * W, C)
        x_flat = self.mid_block(x_flat, H, W, T)
        x = x_flat.reshape(B, T, H, W, C).permute(0, 4, 1, 2, 3)

        # 解码
        for block in self.up_blocks:
            x = block(x)

        return x

U-Net视频架构的关键改进:

  1. 3D卷积层:将2D卷积扩展为3D,在时间和空间维度同时提取特征
  2. 时间注意力层:在各分辨率层级加入时间自注意力,保持帧间一致性
  3. 时间位置编码:为每帧添加可学习的位置信息
  4. 因果卷积:确保生成时只能看到前面的帧,模拟自回归特性

2.3 DiT架构(Sora核心)

DiT(Diffusion Transformer)是Sora的核心架构,将Transformer的强大建模能力与扩散模型结合:

import torch
import torch.nn as nn
import math

class PatchEmbedding3D(nn.Module):
    """3D Patch嵌入:将视频分割为时空Patch"""
    def __init__(self, patch_size=(1, 2, 2), in_channels=3, embed_dim=768):
        super().__init__()
        self.proj = nn.Conv3d(
            in_channels, embed_dim,
            kernel_size=patch_size,
            stride=patch_size
        )
        self.patch_size = patch_size

    def forward(self, x):
        """
        x: [B, C, T, H, W]
        return: [B, N, D] 其中N是patch数量
        """
        x = self.proj(x)  # [B, D, T', H', W']
        B, D, T, H, W = x.shape
        x = x.reshape(B, D, T * H * W).transpose(1, 2)
        return x


class DiTBlock(nn.Module):
    """DiT Transformer块:带自适应Layer Norm"""
    def __init__(self, dim, num_heads=12, mlp_ratio=4.0):
        super().__init__()
        self.norm1 = nn.LayerNorm(dim, elementwise_affine=False)
        self.attn = nn.MultiheadAttention(dim, num_heads, batch_first=True)
        self.norm2 = nn.LayerNorm(dim, elementwise_affine=False)
        self.mlp = nn.Sequential(
            nn.Linear(dim, int(dim * mlp_ratio)),
            nn.GELU(),
            nn.Linear(int(dim * mlp_ratio), dim),
        )
        # 自适应Layer Norm的调制参数
        self.adaLN_modulation = nn.Sequential(
            nn.SiLU(),
            nn.Linear(dim, 6 * dim),  # 6个调制参数
        )

    def forward(self, x, c):
        """
        x: [B, N, D] - patch序列
        c: [B, D] - 条件向量(时间步+文本嵌入)
        """
        # 计算调制参数
        shift_msa, scale_msa, gate_msa, shift_mlp, scale_mlp, gate_mlp = \
            self.adaLN_modulation(c).chunk(6, dim=-1)

        # 自注意力 with adaLN
        x_norm = self.norm1(x)
        x_norm = x_norm * (1 + scale_msa.unsqueeze(1)) + shift_msa.unsqueeze(1)
        attn_out, _ = self.attn(x_norm, x_norm, x_norm)
        x = x + gate_msa.unsqueeze(1) * attn_out

        # MLP with adaLN
        x_norm = self.norm2(x)
        x_norm = x_norm * (1 + scale_mlp.unsqueeze(1)) + shift_mlp.unsqueeze(1)
        mlp_out = self.mlp(x_norm)
        x = x + gate_mlp.unsqueeze(1) * mlp_out

        return x


class DiT(nn.Module):
    """Diffusion Transformer 完整架构"""
    def __init__(self, input_size=(16, 64, 64), patch_size=(1, 2, 2),
                 in_channels=4, dim=1152, depth=28, num_heads=16):
        super().__init__()

        # Patch嵌入
        self.patch_embed = PatchEmbedding3D(patch_size, in_channels, dim)

        # 计算patch数量
        T, H, W = input_size
        pt, ph, pw = patch_size
        num_patches = (T // pt) * (H // ph) * (W // pw)

        # 位置编码
        self.pos_embed = nn.Parameter(torch.randn(1, num_patches, dim) * 0.02)

        # Transformer块
        self.blocks = nn.ModuleList([
            DiTBlock(dim, num_heads) for _ in range(depth)
        ])

        # 最终层
        self.final_norm = nn.LayerNorm(dim, elementwise_affine=False)
        self.final_linear = nn.Linear(dim, patch_size[0] * patch_size[1] * patch_size[2] * in_channels)

        # 时间步嵌入
        self.timestep_embedder = TimestepEmbedder(dim)

        # 文本条件嵌入
        self.text_proj = nn.Linear(4096, dim)  # 假设文本编码器输出4096维

    def forward(self, x, timestep, text_embedding):
        """
        x: [B, C, T, H, W] - 噪声视频
        timestep: [B] - 扩散时间步
        text_embedding: [B, L, 4096] - 文本编码
        """
        # Patch嵌入
        x = self.patch_embed(x)
        x = x + self.pos_embed

        # 条件向量:时间步 + 文本池化
        t_emb = self.timestep_embedder(timestep)
        text_pooled = text_embedding.mean(dim=1)
        c = t_emb + self.text_proj(text_pooled)

        # Transformer处理
        for block in self.blocks:
            x = block(x, c)

        # 输出投影
        x = self.final_norm(x)
        x = self.final_linear(x)

        # 重塑为视频张量(需要unpatchify)
        x = self.unpatchify(x)
        return x

    def unpatchify(self, x):
        """将patch序列重塑回视频张量"""
        # 实现省略,核心是将[B, N, D]转回[B, C, T, H, W]
        pass


class TimestepEmbedder(nn.Module):
    """正弦时间步嵌入"""
    def __init__(self, dim):
        super().__init__()
        self.mlp = nn.Sequential(
            nn.Linear(dim, dim * 4),
            nn.SiLU(),
            nn.Linear(dim * 4, dim),
        )

    def forward(self, t):
        half_dim = self.mlp[0].in_features // 2
        emb = math.log(10000) / (half_dim - 1)
        emb = torch.exp(torch.arange(half_dim, device=t.device) * -emb)
        emb = t[:, None].float() * emb[None, :]
        emb = torch.cat([torch.sin(emb), torch.cos(emb)], dim=-1)
        return self.mlp(emb)

DiT相比U-Net的核心优势:

  1. 全局注意力:每个patch可以关注所有其他patch,远距离依赖建模更强
  2. 可扩展性:Transformer架构天然适合大规模扩展(Sora据报道使用了数十亿参数)
  3. 灵活的条件注入:通过adaLN机制优雅地注入时间步和文本条件
  4. 统一架构:图像和视频可以在同一个架构中处理

2.4 VAE与潜空间设计

视频VAE(变分自编码器)将高维视频压缩到低维潜空间,是提升效率的关键:

import torch
import torch.nn as nn

class VideoVAE(nn.Module):
    """视频VAE:时空压缩"""
    def __init__(self, in_channels=3, latent_dim=4, temporal_compress=4, spatial_compress=8):
        super().__init__()
        self.temporal_compress = temporal_compress
        self.spatial_compress = spatial_compress

        # 编码器
        self.encoder = nn.Sequential(
            # 空间压缩层
            nn.Conv3d(in_channels, 64, kernel_size=(1, 3, 3), padding=(0, 1, 1)),
            nn.SiLU(),
            nn.Conv3d(64, 128, kernel_size=(1, 3, 3), stride=(1, 2, 2), padding=(0, 1, 1)),
            nn.SiLU(),
            nn.Conv3d(128, 256, kernel_size=(1, 3, 3), stride=(1, 2, 2), padding=(0, 1, 1)),
            nn.SiLU(),
            # 时间压缩层
            nn.Conv3d(256, 256, kernel_size=(3, 1, 1), stride=(2, 1, 1), padding=(1, 0, 0)),
            nn.SiLU(),
            nn.Conv3d(256, latent_dim * 2, kernel_size=(3, 1, 1), stride=(2, 1, 1), padding=(1, 0, 0)),
        )

        # 解码器
        self.decoder = nn.Sequential(
            # 时间解压
            nn.ConvTranspose3d(latent_dim, 256, kernel_size=(4, 1, 1), stride=(2, 1, 1), padding=(1, 0, 0)),
            nn.SiLU(),
            nn.ConvTranspose3d(256, 256, kernel_size=(4, 1, 1), stride=(2, 1, 1), padding=(1, 0, 0)),
            nn.SiLU(),
            # 空间解压
            nn.ConvTranspose3d(256, 128, kernel_size=(1, 4, 4), stride=(1, 2, 2), padding=(0, 1, 1)),
            nn.SiLU(),
            nn.ConvTranspose3d(128, 64, kernel_size=(1, 4, 4), stride=(1, 2, 2), padding=(0, 1, 1)),
            nn.SiLU(),
            nn.Conv3d(64, in_channels, kernel_size=(1, 3, 3), padding=(0, 1, 1)),
        )

    def encode(self, x):
        """编码视频到潜空间"""
        h = self.encoder(x)
        mean, logvar = h.chunk(2, dim=1)
        std = torch.exp(0.5 * logvar)
        z = mean + std * torch.randn_like(std)
        return z, mean, logvar

    def decode(self, z):
        """从潜空间解码视频"""
        return self.decoder(z)

    def forward(self, x):
        z, mean, logvar = self.encode(x)
        recon = self.decode(z)
        return recon, mean, logvar

视频VAE设计要点:

  • 时间压缩比通常为4-8倍(如16帧压缩为2-4帧)
  • 空间压缩比通常为8倍(如512x512压缩为64x64)
  • 分离时空压缩可以更好地保持时间一致性
  • 使用KL散度正则化潜空间分布

三、主流视频生成模型对比

3.1 Sora(OpenAI)

技术特点:

  • 基于DiT(Diffusion Transformer)架构
  • 将视频表示为"时空Patch"序列
  • 支持最长60秒、最高1080p的视频生成
  • 具备强大的物理世界模拟能力
  • 支持文本到视频、图像到视频、视频编辑

核心创新:

  • 统一的视频表示:不同分辨率、时长、宽高比的视频都可以处理
  • 大规模训练:使用了海量视频数据训练
  • 涌现能力:自发学会了3D一致性、物体持久性等物理特性

使用方式:

  • ChatGPT Plus/Pro用户可直接使用
  • 通过API调用(需申请)

3.2 Runway Gen-3 Alpha

技术特点:

  • 在电影级质感和运动流畅度上表现出色
  • 支持文本到视频、图像到视频
  • 最长10秒、最高1280x768分辨率
  • 提供运动画笔(Motion Brush)精细控制

API调用示例:

import requests
import time

class RunwayClient:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.runwayml.com/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def generate_video(self, prompt, duration=5, resolution="1280x768"):
        """文本生成视频"""
        payload = {
            "model": "gen3a_turbo",
            "promptText": prompt,
            "duration": duration,
            "resolution": resolution,
        }
        response = requests.post(
            f"{self.base_url}/image_to_video",
            headers=self.headers,
            json=payload
        )
        return response.json()

    def generate_from_image(self, image_url, prompt, duration=5):
        """图像生成视频"""
        payload = {
            "model": "gen3a_turbo",
            "promptImage": image_url,
            "promptText": prompt,
            "duration": duration,
        }
        response = requests.post(
            f"{self.base_url}/image_to_video",
            headers=self.headers,
            json=payload
        )
        task_id = response.json().get("id")

        # 轮询等待结果
        while True:
            status = self.check_status(task_id)
            if status["status"] == "SUCCEEDED":
                return status["output"]
            elif status["status"] == "FAILED":
                raise Exception(f"Generation failed: {status.get('failure')}")
            time.sleep(5)

    def check_status(self, task_id):
        response = requests.get(
            f"{self.base_url}/tasks/{task_id}",
            headers=self.headers
        )
        return response.json()


# 使用示例
client = RunwayClient("your-api-key")
result = client.generate_video(
    prompt="A cinematic shot of a golden retriever running through autumn leaves, "
           "warm lighting, shallow depth of field, 4K quality",
    duration=5
)
print(f"Video URL: {result}")

3.3 可灵(Kling)

技术特点:

  • 快手推出的AI视频生成模型
  • 对中文语义理解出色
  • 支持长达2分钟的视频生成
  • 运动幅度大、物理效果好
  • 提供专业模式和标准模式

API调用示例:

import requests
import time
import json

class KlingClient:
    def __init__(self, access_key, secret_key):
        self.access_key = access_key
        self.secret_key = secret_key
        self.base_url = "https://api.klingai.com/v1"
        self.token = self._get_token()

    def _get_token(self):
        """获取访问令牌"""
        import jwt
        headers = {"alg": "HS256", "typ": "JWT"}
        payload = {
            "iss": self.access_key,
            "exp": int(time.time()) + 1800,
            "nbf": int(time.time()) - 5,
        }
        return jwt.encode(payload, self.secret_key, headers=headers)

    def text_to_video(self, prompt, mode="std", duration="5",
                      aspect_ratio="16:9", model_name="kling-v1"):
        """文本生成视频"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.token}",
        }
        payload = {
            "model_name": model_name,
            "prompt": prompt,
            "mode": mode,  # std 或 pro
            "duration": duration,  # 5 或 10
            "aspect_ratio": aspect_ratio,
        }
        response = requests.post(
            f"{self.base_url}/videos/text2video",
            headers=headers,
            json=payload
        )
        task_id = response.json()["data"]["task_id"]
        return self._poll_result(task_id, headers)

    def image_to_video(self, image_url, prompt="", mode="std", duration="5"):
        """图像生成视频"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.token}",
        }
        payload = {
            "model_name": "kling-v1",
            "image": image_url,
            "prompt": prompt,
            "mode": mode,
            "duration": duration,
        }
        response = requests.post(
            f"{self.base_url}/videos/image2video",
            headers=headers,
            json=payload
        )
        task_id = response.json()["data"]["task_id"]
        return self._poll_result(task_id, headers)

    def _poll_result(self, task_id, headers, max_wait=600):
        """轮询任务结果"""
        start = time.time()
        while time.time() - start < max_wait:
            response = requests.get(
                f"{self.base_url}/videos/text2video/{task_id}",
                headers=headers
            )
            data = response.json()["data"]
            if data["task_status"] == "succeed":
                return data["task_result"]["videos"]
            elif data["task_status"] == "failed":
                raise Exception(f"Task failed: {data.get('task_status_msg')}")
            time.sleep(10)
        raise TimeoutError("Video generation timed out")


# 使用示例
client = KlingClient("your-access-key", "your-secret-key")
videos = client.text_to_video(
    prompt="一只金色的猫咪在阳光下的草地上奔跑,微风吹动毛发,电影级画质",
    mode="pro",
    duration="10",
    aspect_ratio="16:9"
)
print(f"Generated video: {videos[0]['url']}")

3.4 Google VEO 2

技术特点:

  • Google DeepMind推出的视频生成模型
  • 支持最高4K分辨率、最长2分钟
  • 物理世界理解能力突出
  • 支持电影级镜头控制
  • 通过Vertex AI和Google AI Studio使用

3.5 开源模型生态

HunyuanVideo(腾讯)

# HunyuanVideo 使用示例
from diffusers import HunyuanVideoPipeline
import torch

pipe = HunyuanVideoPipeline.from_pretrained(
    "tencent/HunyuanVideo",
    torch_dtype=torch.bfloat16,
)
pipe.to("cuda")
pipe.enable_model_cpu_offload()  # 节省显存

video = pipe(
    prompt="A beautiful sunset over the ocean, waves gently lapping at the shore",
    num_frames=61,
    height=720,
    width=1280,
    num_inference_steps=30,
).frames[0]

# 保存视频
from diffusers.utils import export_to_video
export_to_video(video, "output.mp4", fps=24)

Wan2.1(阿里万相)

# Wan2.1 使用示例
from diffusers import WanPipeline
import torch

pipe = WanPipeline.from_pretrained(
    "Wan-AI/Wan2.1-T2V-14B",
    torch_dtype=torch.bfloat16,
)
pipe.to("cuda")
pipe.enable_model_cpu_offload()

video = pipe(
    prompt="A cat wearing sunglasses driving a car through a futuristic city",
    num_frames=81,
    guidance_scale=5.0,
).frames[0]

from diffusers.utils import export_to_video
export_to_video(video, "wan_output.mp4", fps=16)

3.6 模型选型指南

需求场景 推荐模型 理由
中文场景、短视频 可灵 中文理解强、运动幅度大
电影级质感 Runway Gen-3 画面质感最佳
超长视频 Sora 支持60秒连贯生成
4K高分辨率 VEO 2 分辨率最高
开源自建 HunyuanVideo 完全开源、社区活跃
低成本批量 Stable Video Diffusion 开源免费、可本地部署

四、文本生成视频(Text-to-Video)实战

4.1 基础流程

文本生成视频的完整流程包括:

  1. 文本编码:将自然语言提示词编码为语义向量
  2. 条件生成:以语义向量为条件,指导扩散模型生成
  3. 时空去噪:在潜空间逐步去噪,生成视频潜表示
  4. 视频解码:VAE解码器将潜表示还原为像素级视频
  5. 后处理:帧插值、超分辨率、色彩校正

4.2 提示词工程

提示词质量直接决定视频质量,以下是系统化的提示词工程方法:

class VideoPromptEngineer:
    """视频生成提示词工程工具"""

    # 提示词模板结构
    TEMPLATE = {
        "subject": "",      # 主体
        "action": "",       # 动作
        "scene": "",        # 场景
        "style": "",        # 风格
        "camera": "",       # 镜头
        "lighting": "",     # 光线
        "mood": "",         # 情绪
        "quality": "",      # 质量修饰词
    }

    # 镜头运动词汇表
    CAMERA_MOVEMENTS = {
        "pan": "镜头平移",
        "tilt": "镜头俯仰",
        "dolly": "推拉镜头",
        "tracking": "跟踪镜头",
        "crane": "摇臂镜头",
        "zoom": "变焦",
        "steadicam": "斯坦尼康稳定镜头",
        "handheld": "手持镜头",
        "aerial": "航拍",
        "static": "固定机位",
    }

    # 风格修饰词
    STYLE_MODIFIERS = [
        "cinematic", "photorealistic", "anime", "watercolor",
        "oil painting", "3D render", "Studio Ghibli style",
        "film noir", "documentary", "music video",
    ]

    @classmethod
    def build_prompt(cls, subject, action, scene="", style="cinematic",
                     camera="tracking", lighting="natural", mood="neutral"):
        """构建结构化提示词"""
        parts = []

        # 质量前缀
        quality_prefix = "High quality, detailed, "

        # 主体和动作
        parts.append(f"{subject} {action}")

        # 场景
        if scene:
            parts.append(f"in {scene}")

        # 风格
        parts.append(f"{style} style")

        # 镜头
        if camera in cls.CAMERA_MOVEMENTS:
            parts.append(f"{camera} shot")

        # 光线
        lighting_map = {
            "natural": "natural lighting",
            "golden": "golden hour lighting",
            "dramatic": "dramatic lighting with strong shadows",
            "soft": "soft diffused lighting",
            "neon": "neon lighting",
            "studio": "professional studio lighting",
        }
        if lighting in lighting_map:
            parts.append(lighting_map[lighting])

        # 情绪/氛围
        mood_map = {
            "neutral": "",
            "warm": "warm and inviting atmosphere",
            "cold": "cold and mysterious atmosphere",
            "vibrant": "vibrant and energetic atmosphere",
            "serene": "serene and peaceful atmosphere",
        }
        if mood in mood_map and mood_map[mood]:
            parts.append(mood_map[mood])

        prompt = quality_prefix + ", ".join(parts)
        return prompt

    @classmethod
    def enhance_prompt(cls, base_prompt, enhancements=None):
        """增强提示词"""
        default_enhancements = [
            "masterpiece quality",
            "professional cinematography",
            "smooth motion",
            "highly detailed",
        ]
        if enhancements is None:
            enhancements = default_enhancements

        enhanced = base_prompt + ", " + ", ".join(enhancements)
        return enhanced

    @classmethod
    def negative_prompt(cls):
        """通用负面提示词"""
        return (
            "blurry, low quality, distorted, deformed, disfigured, "
            "bad anatomy, watermark, text overlay, static, frozen, "
            "flickering, jittery, morphing artifacts, extra limbs, "
            "mutation, ugly, duplicate, error"
        )


# 使用示例
prompt = VideoPromptEngineer.build_prompt(
    subject="a golden retriever puppy",
    action="playfully chasing butterflies",
    scene="a sunlit meadow with wildflowers",
    style="cinematic",
    camera="tracking",
    lighting="golden",
    mood="warm"
)
print(f"Generated prompt: {prompt}")
# 输出: High quality, detailed, a golden retriever puppy playfully chasing butterflies
# in a sunlit meadow with wildflowers, cinematic style, tracking shot,
# golden hour lighting, warm and inviting atmosphere

enhanced = VideoPromptEngineer.enhance_prompt(prompt)
print(f"Enhanced: {enhanced}")

4.3 使用Diffusers生成视频

import torch
from diffusers import StableVideoDiffusionPipeline
from diffusers.utils import load_image, export_to_video

# 加载模型
pipe = StableVideoDiffusionPipeline.from_pretrained(
    "stabilityai/stable-video-diffusion-img2vid-xt",
    torch_dtype=torch.float16,
    variant="fp16",
)
pipe.to("cuda")

# 加载起始帧
image = load_image("input_image.png")
image = image.resize((1024, 576))

# 生成视频
generator = torch.manual_seed(42)
frames = pipe(
    image,
    decode_chunk_size=8,
    generator=generator,
    motion_bucket_id=127,      # 运动幅度 (1-255)
    noise_aug_strength=0.1,    # 噪声增强强度
    num_frames=25,             # 生成帧数
    height=576,
    width=1024,
).frames[0]

# 保存视频
export_to_video(frames, "output.mp4", fps=7)

五、图像生成视频(Image-to-Video)实战

5.1 技术原理

图像生成视频以一张静态图片为起始帧,通过运动预测和帧生成扩展为动态视频:

import torch
import torch.nn as nn

class MotionPredictor(nn.Module):
    """运动预测网络:从单帧预测运动场"""
    def __init__(self, in_channels=3, hidden_dim=256):
        super().__init__()
        # 光流估计网络
        self.encoder = nn.Sequential(
            nn.Conv2d(in_channels, 64, 7, stride=2, padding=3),
            nn.ReLU(),
            nn.Conv2d(64, 128, 5, stride=2, padding=2),
            nn.ReLU(),
            nn.Conv2d(128, hidden_dim, 3, stride=2, padding=1),
            nn.ReLU(),
        )

        # 运动解码器
        self.motion_decoder = nn.Sequential(
            nn.ConvTranspose2d(hidden_dim, 128, 4, stride=2, padding=1),
            nn.ReLU(),
            nn.ConvTranspose2d(128, 64, 4, stride=2, padding=1),
            nn.ReLU(),
            nn.ConvTranspose2d(64, 32, 4, stride=2, padding=1),
            nn.ReLU(),
            nn.Conv2d(32, 2, 3, padding=1),  # 2通道光流 (dx, dy)
            nn.Tanh(),
        )

        # 时间运动预测
        self.temporal_predictor = nn.LSTM(
            input_size=hidden_dim * 8 * 8,  # 假设特征图大小
            hidden_size=512,
            num_layers=2,
            batch_first=True,
        )

    def forward(self, first_frame, num_frames=16):
        """
        first_frame: [B, C, H, W] - 起始帧
        return: [B, T, 2, H, W] - 每帧的运动场
        """
        B, C, H, W = first_frame.shape

        # 提取特征
        features = self.encoder(first_frame)  # [B, D, H', W']

        # 预测第一帧运动
        motion = self.motion_decoder(features)  # [B, 2, H, W]

        # 扩展到多帧
        motions = [motion]
        hidden = None

        feat_flat = features.reshape(B, -1).unsqueeze(1)  # [B, 1, D*H'*W']

        for t in range(1, num_frames):
            # 使用LSTM预测后续帧的运动变化
            out, hidden = self.temporal_predictor(feat_flat, hidden)
            motion_delta = out.reshape(B, 1, -1)

            # 这里简化处理,实际需要更复杂的运动场更新
            motion = motion + 0.1 * motion.mean()  # 简化运动累积
            motions.append(motion)

        return torch.stack(motions, dim=1)


def apply_optical_flow(image, flow):
    """使用光流对图像进行变形"""
    B, C, H, W = image.shape

    # 生成网格
    grid_y, grid_x = torch.meshgrid(
        torch.arange(H, device=image.device, dtype=torch.float32),
        torch.arange(W, device=image.device, dtype=torch.float32),
        indexing='ij'
    )
    grid = torch.stack([grid_x, grid_y], dim=0)  # [2, H, W]

    # 应用光流
    new_grid = grid + flow  # [B, 2, H, W]
    new_grid = new_grid.permute(0, 2, 3, 1)  # [B, H, W, 2]

    # 归一化到[-1, 1]
    new_grid[..., 0] = 2.0 * new_grid[..., 0] / (W - 1) - 1.0
    new_grid[..., 1] = 2.0 * new_grid[..., 1] / (H - 1) - 1.0

    # 双线性插值采样
    warped = torch.nn.functional.grid_sample(
        image, new_grid, mode='bilinear', padding_mode='border', align_corners=True
    )
    return warped

5.2 图像到视频实战流程

import torch
from PIL import Image
import numpy as np

class ImageToVideoGenerator:
    """图像生成视频的完整工作流"""

    def __init__(self, model_path="stabilityai/stable-video-diffusion-img2vid-xt"):
        from diffusers import StableVideoDiffusionPipeline
        self.pipe = StableVideoDiffusionPipeline.from_pretrained(
            model_path,
            torch_dtype=torch.float16,
            variant="fp16",
        )
        self.pipe.to("cuda")

    def preprocess_image(self, image_path, target_size=(1024, 576)):
        """图像预处理"""
        image = Image.open(image_path).convert("RGB")

        # 保持宽高比缩放
        w, h = image.size
        target_w, target_h = target_size
        ratio = min(target_w / w, target_h / h)
        new_w, new_h = int(w * ratio), int(h * ratio)
        image = image.resize((new_w, new_h), Image.LANCZOS)

        # 居中填充
        padded = Image.new("RGB", target_size, (0, 0, 0))
        offset_x = (target_w - new_w) // 2
        offset_y = (target_h - new_h) // 2
        padded.paste(image, (offset_x, offset_y))

        return padded

    def generate(self, image_path, num_frames=25, fps=7,
                 motion_bucket_id=127, seed=42):
        """
        从图像生成视频

        参数:
            image_path: 输入图像路径
            num_frames: 生成帧数
            fps: 帧率
            motion_bucket_id: 运动强度 (1-255, 越大运动越剧烈)
            seed: 随机种子
        """
        # 预处理
        image = self.preprocess_image(image_path)

        # 生成
        generator = torch.manual_seed(seed)
        frames = self.pipe(
            image,
            num_frames=num_frames,
            decode_chunk_size=8,
            generator=generator,
            motion_bucket_id=motion_bucket_id,
            noise_aug_strength=0.1,
        ).frames[0]

        return frames

    def postprocess(self, frames, output_path, fps=7, enhance=True):
        """后处理:超分、色彩校正"""
        if enhance:
            frames = self._color_correct(frames)

        from diffusers.utils import export_to_video
        export_to_video(frames, output_path, fps=fps)
        return output_path

    def _color_correct(self, frames):
        """简单色彩校正"""
        enhanced = []
        for frame in frames:
            img = np.array(frame)
            # 对比度增强
            img = np.clip(img * 1.1, 0, 255).astype(np.uint8)
            enhanced.append(Image.fromarray(img))
        return enhanced


# 使用示例
generator = ImageToVideoGenerator()
frames = generator.generate(
    "product_photo.jpg",
    num_frames=25,
    fps=7,
    motion_bucket_id=150,
)
generator.postprocess(frames, "product_video.mp4", fps=7)

六、视频编辑与风格迁移

6.1 视频风格迁移

import torch
import torch.nn as nn
from torchvision import transforms

class VideoStyleTransfer:
    """视频风格迁移:保持时序一致性"""

    def __init__(self, style_model_path=None):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.transform = transforms.Compose([
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                               std=[0.229, 0.224, 0.225]),
        ])

    def transfer_style(self, video_frames, style_image, strength=0.7):
        """
        对视频进行风格迁移,保持时序一致性

        参数:
            video_frames: 视频帧列表 [PIL.Image, ...]
            style_image: 风格参考图 PIL.Image
            strength: 风格强度 0-1
        """
        import cv2
        import numpy as np

        styled_frames = []
        prev_flow = None

        for i, frame in enumerate(video_frames):
            frame_np = np.array(frame)

            if i == 0:
                # 第一帧直接风格化
                styled = self._stylize_single(frame, style_image, strength)
            else:
                # 后续帧使用光流保持一致性
                prev_frame = np.array(video_frames[i - 1])
                flow = cv2.calcOpticalFlowFarneback(
                    cv2.cvtColor(prev_frame, cv2.COLOR_RGB2GRAY),
                    cv2.cvtColor(frame_np, cv2.COLOR_RGB2GRAY),
                    None, 0.5, 3, 15, 3, 5, 1.2, 0
                )

                # 用光流传播上一帧的风格结果
                warped_style = self._warp_frame(
                    np.array(styled_frames[-1]), flow
                )

                # 混合当前帧风格化和传播结果
                current_style = self._stylize_single(frame, style_image, strength)
                styled = cv2.addWeighted(
                    current_style, 0.6,
                    warped_style, 0.4, 0
                )

            styled_frames.append(Image.fromarray(styled))

        return styled_frames

    def _stylize_single(self, content, style, strength):
        """单帧风格化(简化版,实际应使用预训练模型)"""
        import numpy as np
        content_np = np.array(content).astype(np.float32)
        style_np = np.array(style.resize(content.size)).astype(np.float32)
        styled = content_np * (1 - strength) + style_np * strength
        return np.clip(styled, 0, 255).astype(np.uint8)

    def _warp_frame(self, frame, flow):
        """使用光流变形帧"""
        import cv2
        h, w = flow.shape[:2]
        flow_map = np.column_stack((
            np.repeat(np.arange(w), h),
            np.tile(np.arange(h), w)
        )).reshape(h, w, 2).astype(np.float32)
        flow_map += flow
        warped = cv2.remap(frame, flow_map[..., 0], flow_map[..., 1], cv2.INTER_LINEAR)
        return warped

6.2 视频编辑(InstructPix2Pix风格)

import torch
from diffusers import StableDiffusionInstructPix2PixPipeline
from diffusers.utils import load_video

class VideoEditor:
    """基于指令的视频编辑"""

    def __init__(self):
        self.pipe = StableDiffusionInstructPix2PixPipeline.from_pretrained(
            "timbrooks/instruct-pix2pix",
            torch_dtype=torch.float16,
        )
        self.pipe.to("cuda")

    def edit_frame(self, frame, instruction, guidance_scale=7.5,
                   image_guidance_scale=1.5):
        """编辑单帧"""
        result = self.pipe(
            prompt=instruction,
            image=frame,
            guidance_scale=guidance_scale,
            image_guidance_scale=image_guidance_scale,
        ).images[0]
        return result

    def edit_video(self, video_path, instruction, output_path):
        """编辑整个视频"""
        from diffusers.utils import export_to_video
        from PIL import Image

        # 加载视频帧
        frames = load_video(video_path)

        # 编辑每帧
        edited_frames = []
        for i, frame in enumerate(frames):
            print(f"Editing frame {i+1}/{len(frames)}")
            edited = self.edit_frame(frame, instruction)
            edited_frames.append(edited)

        # 保存
        export_to_video(edited_frames, output_path, fps=24)
        return output_path

    def selective_edit(self, video_path, mask_region, edit_instruction,
                       output_path):
        """选择性区域编辑"""
        from diffusers.utils import load_video, export_to_video
        import numpy as np

        frames = load_video(video_path)
        edited_frames = []

        for frame in frames:
            frame_np = np.array(frame)
            # 应用mask区域编辑
            edited = self.edit_frame(frame, edit_instruction)
            edited_np = np.array(edited)

            # 仅替换mask区域
            x1, y1, x2, y2 = mask_region
            result_np = frame_np.copy()
            result_np[y1:y2, x1:x2] = edited_np[y1:y2, x1:x2]
            edited_frames.append(Image.fromarray(result_np))

        export_to_video(edited_frames, output_path, fps=24)
        return output_path


# 使用示例
editor = VideoEditor()
editor.edit_video(
    "input.mp4",
    "Change the sky to a beautiful sunset with orange and pink clouds",
    "edited_output.mp4"
)

七、时序一致性与运动控制

7.1 时序一致性问题

视频生成中的时序一致性是最核心的挑战之一。主要问题包括:

  • 闪烁(Flickering):相邻帧之间的像素级不一致
  • 形变(Morphing):物体形状在帧间不自然地变化
  • 消失(Disappearance):物体突然消失又出现
  • 风格漂移(Style Drift):画面风格随时间逐渐偏离

7.2 帧间一致性优化

import torch
import torch.nn.functional as F
import numpy as np

class TemporalConsistencyLoss:
    """时序一致性损失函数"""

    @staticmethod
    def optical_flow_loss(frames):
        """基于光流的一致性损失"""
        import cv2
        total_loss = 0.0

        for i in range(1, len(frames)):
            prev = np.array(frames[i - 1]).astype(np.float32)
            curr = np.array(frames[i]).astype(np.float32)

            # 计算光流
            flow = cv2.calcOpticalFlowFarneback(
                cv2.cvtColor(prev.astype(np.uint8), cv2.COLOR_RGB2GRAY),
                cv2.cvtColor(curr.astype(np.uint8), cv2.COLOR_RGB2GRAY),
                None, 0.5, 3, 15, 3, 5, 1.2, 0
            )

            # 变形前一帧
            h, w = flow.shape[:2]
            flow_map = np.column_stack((
                np.repeat(np.arange(w), h),
                np.tile(np.arange(h), w)
            )).reshape(h, w, 2).astype(np.float32)
            flow_map += flow
            warped = cv2.remap(prev, flow_map[..., 0], flow_map[..., 1],
                              cv2.INTER_LINEAR)

            # 计算与当前帧的差异
            loss = np.mean((warped - curr) ** 2)
            total_loss += loss

        return total_loss / (len(frames) - 1)

    @staticmethod
    def temporal_smoothness_loss(video_tensor):
        """时间平滑性损失"""
        # video_tensor: [B, T, C, H, W]
        diff = video_tensor[:, 1:] - video_tensor[:, :-1]
        loss = torch.mean(diff ** 2)
        return loss

    @staticmethod
    def lpips_temporal_loss(frames, lpips_model):
        """基于感知相似度的时序损失"""
        total_loss = 0.0
        for i in range(1, len(frames)):
            loss = lpips_model(frames[i-1], frames[i])
            total_loss += loss.mean()
        return total_loss / (len(frames) - 1)


class MotionController:
    """运动控制器:精细控制视频中的运动"""

    def __init__(self):
        self.motion_presets = {
            "zoom_in": {"scale": [1.0, 1.3], "center": [0.5, 0.5]},
            "zoom_out": {"scale": [1.3, 1.0], "center": [0.5, 0.5]},
            "pan_left": {"offset_x": [-0.2, 0.0], "offset_y": [0.0, 0.0]},
            "pan_right": {"offset_x": [0.0, 0.2], "offset_y": [0.0, 0.0]},
            "tilt_up": {"offset_x": [0.0, 0.0], "offset_y": [-0.2, 0.0]},
            "tilt_down": {"offset_x": [0.0, 0.0], "offset_y": [0.0, 0.2]},
            "rotate_cw": {"rotation": [0, 15]},
            "rotate_ccw": {"rotation": [0, -15]},
        }

    def apply_motion(self, frame, motion_type, t, total_frames):
        """
        对单帧应用运动变换

        参数:
            frame: PIL.Image
            motion_type: 运动类型
            t: 当前帧序号
            total_frames: 总帧数
        """
        from PIL import Image
        import numpy as np

        if motion_type not in self.motion_presets:
            return frame

        preset = self.motion_presets[motion_type]
        progress = t / max(total_frames - 1, 1)  # 0 -> 1

        w, h = frame.size
        frame_np = np.array(frame)

        if "scale" in preset:
            # 缩放
            s0, s1 = preset["scale"]
            scale = s0 + (s1 - s0) * progress
            new_w, new_h = int(w * scale), int(h * scale)
            resized = frame.resize((new_w, new_h), Image.LANCZOS)
            # 裁剪回原始大小
            left = (new_w - w) // 2
            top = (new_h - h) // 2
            frame = resized.crop((left, top, left + w, top + h))

        elif "offset_x" in preset:
            # 平移
            ox0, ox1 = preset["offset_x"]
            oy0, oy1 = preset["offset_y"]
            dx = int((ox0 + (ox1 - ox0) * progress) * w)
            dy = int((oy0 + (oy1 - oy0) * progress) * h)
            shifted = np.zeros_like(frame_np)
            # 计算有效区域
            src_x1 = max(0, -dx)
            src_y1 = max(0, -dy)
            src_x2 = min(w, w - dx)
            src_y2 = min(h, h - dy)
            dst_x1 = max(0, dx)
            dst_y1 = max(0, dy)
            dst_x2 = dst_x1 + (src_x2 - src_x1)
            dst_y2 = dst_y1 + (src_y2 - src_y1)
            shifted[dst_y1:dst_y2, dst_x1:dst_x2] = frame_np[src_y1:src_y2, src_x1:src_x2]
            frame = Image.fromarray(shifted)

        return frame

    def create_motion_sequence(self, frames, motion_type):
        """对整个视频序列应用运动"""
        total = len(frames)
        return [
            self.apply_motion(f, motion_type, i, total)
            for i, f in enumerate(frames)
        ]

八、视频生成API调用实战

8.1 Replicate平台

import replicate
import time
import requests

class ReplicateVideoGenerator:
    """通过Replicate API生成视频"""

    def __init__(self, api_token):
        self.client = replicate.Client(api_token=api_token)

    def generate_svd(self, image_url, num_frames=25, fps=7,
                     motion_bucket_id=127):
        """使用Stable Video Diffusion生成"""
        output = self.client.run(
            "stability-ai/stable-video-diffusion:3f0457e4619daac51203dedb472816fd4af51f3149fa7a9e0b5ffcf1b8172438",
            input={
                "input_image": image_url,
                "num_frames": num_frames,
                "fps": fps,
                "motion_bucket_id": motion_bucket_id,
                "cond_aug": 0.02,
                "decoding_t": 7,
            }
        )
        return output

    def generate_animate_diff(self, prompt, negative_prompt="",
                              num_frames=16, width=512, height=512):
        """使用AnimateDiff生成"""
        output = self.client.run(
            "lucataco/animate-diff:beecf59c4aee8d81bf04f0381033dfa10dc16e845b4ae00d281e2fa377e48a9f",
            input={
                "path": "mm_sd_v15_v2.ckpt",
                "prompt": prompt,
                "n_prompt": negative_prompt,
                "steps": 25,
                "cfg_scale": 7.5,
                "seed": 42,
                "width": width,
                "height": height,
                "num_frames": num_frames,
            }
        )
        return output

    def generate_with_polling(self, model, input_params, poll_interval=5):
        """带轮询的异步生成"""
        # 创建预测
        prediction = self.client.predictions.create(
            version=model,
            input=input_params,
        )

        # 轮询等待
        while prediction.status not in ["succeeded", "failed"]:
            time.sleep(poll_interval)
            prediction = self.client.predictions.get(prediction.id)
            print(f"Status: {prediction.status}")

        if prediction.status == "succeeded":
            return prediction.output
        else:
            raise Exception(f"Generation failed: {prediction.error}")


# 使用示例
gen = ReplicateVideoGenerator("r8_your_api_token")
result = gen.generate_animate_diff(
    prompt="A cute cat dancing in the rain, anime style, vibrant colors",
    num_frames=16,
)
print(f"Video URL: {result}")

8.2 FAL.ai平台

import fal_client
import requests
import time

class FALVideoGenerator:
    """通过FAL.ai API生成视频"""

    def __init__(self, api_key):
        self.api_key = api_key
        fal_client.api_key = api_key

    def generate_text_to_video(self, prompt, model="fal-ai/fast-svd",
                               num_frames=25):
        """文本生成视频"""
        handler = fal_client.submit(
            model,
            arguments={
                "prompt": prompt,
                "num_frames": num_frames,
            },
        )
        result = handler.get()
        return result

    def generate_image_to_video(self, image_url, model="fal-ai/fast-svd",
                                num_frames=25, fps=7):
        """图像生成视频"""
        result = fal_client.run(
            model,
            arguments={
                "image_url": image_url,
                "num_frames": num_frames,
                "fps": fps,
            },
        )
        return result

    def generate_kling(self, prompt, duration=5, aspect_ratio="16:9",
                       model="fal-ai/kling-video"):
        """使用可灵模型生成"""
        result = fal_client.run(
            model,
            arguments={
                "prompt": prompt,
                "duration": duration,
                "aspect_ratio": aspect_ratio,
                "model": "kling-v1",
                "mode": "pro",
            },
        )
        return result

    def generate_with_callback(self, model, params, callback_url=None):
        """带回调的异步生成"""
        if callback_url:
            result = fal_client.subscribe(
                model,
                arguments=params,
                webhook_url=callback_url,
            )
        else:
            # 同步等待
            result = fal_client.run(model, arguments=params)
        return result


# 使用示例
gen = FALVideoGenerator("your-fal-api-key")
result = gen.generate_kling(
    prompt="A majestic dragon flying over a medieval castle at sunset, "
           "cinematic wide shot, epic fantasy atmosphere",
    duration=10,
    aspect_ratio="16:9",
)
print(f"Generated video: {result}")

8.3 自建推理服务

# 自建视频生成推理服务
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import torch
import uuid
import asyncio

app = FastAPI(title="Video Generation API")

class GenerationRequest(BaseModel):
    prompt: str
    negative_prompt: str = ""
    num_frames: int = 16
    width: int = 512
    height: int = 512
    guidance_scale: float = 7.5
    num_inference_steps: int = 25
    seed: int = -1

class GenerationTask(BaseModel):
    task_id: str
    status: str  # pending, processing, completed, failed
    result_url: str = None
    error: str = None

# 任务存储
tasks = {}

# 模型加载
model = None

def load_model():
    global model
    from diffusers import AnimateDiffPipeline, MotionAdapter
    adapter = MotionAdapter.from_pretrained(
        "guoyww/animatediff-motion-adapter-v1-5-3",
        torch_dtype=torch.float16,
    )
    pipe = AnimateDiffPipeline.from_pretrained(
        "emilianJR/epiCRealism",
        motion_adapter=adapter,
        torch_dtype=torch.float16,
    )
    pipe.to("cuda")
    pipe.enable_vae_slicing()
    model = pipe

@app.on_event("startup")
async def startup():
    load_model()

@app.post("/generate", response_model=GenerationTask)
async def generate_video(request: GenerationRequest,
                         background_tasks: BackgroundTasks):
    """提交视频生成任务"""
    task_id = str(uuid.uuid4())
    tasks[task_id] = GenerationTask(
        task_id=task_id,
        status="pending",
    )
    background_tasks.add_task(process_generation, task_id, request)
    return tasks[task_id]

@app.get("/tasks/{task_id}", response_model=GenerationTask)
async def get_task(task_id: str):
    """查询任务状态"""
    if task_id not in tasks:
        return {"error": "Task not found"}
    return tasks[task_id]

async def process_generation(task_id: str, request: GenerationRequest):
    """后台处理视频生成"""
    tasks[task_id].status = "processing"

    try:
        seed = request.seed if request.seed >= 0 else torch.randint(0, 2**32, (1,)).item()
        generator = torch.manual_seed(seed)

        output = model(
            prompt=request.prompt,
            negative_prompt=request.negative_prompt,
            num_frames=request.num_frames,
            width=request.width,
            height=request.height,
            guidance_scale=request.guidance_scale,
            num_inference_steps=request.num_inference_steps,
            generator=generator,
        )

        # 保存视频
        video_path = f"outputs/{task_id}.mp4"
        from diffusers.utils import export_to_video
        export_to_video(output.frames[0], video_path, fps=8)

        tasks[task_id].status = "completed"
        tasks[task_id].result_url = f"/videos/{task_id}.mp4"

    except Exception as e:
        tasks[task_id].status = "failed"
        tasks[task_id].error = str(e)

九、长视频生成与分镜编排

9.1 分镜脚本设计

from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum

class ShotType(Enum):
    WIDE = "wide shot"
    MEDIUM = "medium shot"
    CLOSE_UP = "close-up"
    EXTREME_CLOSE_UP = "extreme close-up"
    OVER_SHOULDER = "over the shoulder"
    BIRD_EYE = "bird's eye view"
    LOW_ANGLE = "low angle"
    HIGH_ANGLE = "high angle"

class CameraMovement(Enum):
    STATIC = "static"
    PAN_LEFT = "panning left"
    PAN_RIGHT = "panning right"
    TILT_UP = "tilting up"
    TILT_DOWN = "tilting down"
    DOLLY_IN = "dolly in"
    DOLLY_OUT = "dolly out"
    TRACKING = "tracking shot"
    CRANE = "crane shot"
    AERIAL = "aerial shot"

class Transition(Enum):
    CUT = "cut"
    FADE = "fade"
    DISSOLVE = "dissolve"
    WIPE = "wipe"

@dataclass
class Shot:
    """单个镜头"""
    scene_number: int
    shot_number: int
    description: str
    shot_type: ShotType
    camera_movement: CameraMovement
    duration: float  # 秒
    prompt: str
    negative_prompt: str = ""
    transition: Transition = Transition.CUT
    audio_description: str = ""
    notes: str = ""

@dataclass
class Storyboard:
    """分镜脚本"""
    title: str
    shots: List[Shot] = field(default_factory=list)
    target_fps: int = 24
    resolution: tuple = (1280, 720)

    def add_shot(self, **kwargs):
        shot = Shot(**kwargs)
        self.shots.append(shot)
        return shot

    def total_duration(self):
        return sum(shot.duration for shot in self.shots)

    def to_prompt_sequence(self):
        """转换为可执行的提示词序列"""
        return [
            {
                "scene": f"Scene {s.scene_number} - Shot {s.shot_number}",
                "prompt": s.prompt,
                "negative": s.negative_prompt,
                "duration": s.duration,
                "type": s.shot_type.value,
                "movement": s.camera_movement.value,
                "transition": s.transition.value,
            }
            for s in self.shots
        ]


class StoryboardGenerator:
    """AI辅助分镜生成"""

    @staticmethod
    def from_script(script_text: str) -> Storyboard:
        """从剧本文字生成分镜"""
        # 这里可以调用LLM来解析剧本并生成分镜
        # 示例实现
        storyboard = Storyboard(title="Generated Storyboard")

        # 简单的段落分割
        paragraphs = [p.strip() for p in script_text.split("\n\n") if p.strip()]

        for i, para in enumerate(paragraphs):
            storyboard.add_shot(
                scene_number=i // 3 + 1,
                shot_number=i % 3 + 1,
                description=para[:100],
                shot_type=ShotType.MEDIUM,
                camera_movement=CameraMovement.STATIC,
                duration=5.0,
                prompt=para,
                negative_prompt="blurry, low quality, distorted",
            )

        return storyboard

    @staticmethod
    def from_outline(outline: dict) -> Storyboard:
        """从大纲字典生成分镜"""
        storyboard = Storyboard(title=outline.get("title", "Untitled"))

        for scene_idx, scene in enumerate(outline.get("scenes", [])):
            for shot_idx, shot_info in enumerate(scene.get("shots", [])):
                storyboard.add_shot(
                    scene_number=scene_idx + 1,
                    shot_number=shot_idx + 1,
                    description=shot_info.get("description", ""),
                    shot_type=ShotType(shot_info.get("shot_type", "medium shot")),
                    camera_movement=CameraMovement(
                        shot_info.get("camera_movement", "static")
                    ),
                    duration=shot_info.get("duration", 5.0),
                    prompt=shot_info.get("prompt", ""),
                    negative_prompt=shot_info.get("negative_prompt", ""),
                    transition=Transition(shot_info.get("transition", "cut")),
                )

        return storyboard


# 使用示例
storyboard = Storyboard(title="产品宣传视频")

storyboard.add_shot(
    scene_number=1, shot_number=1,
    description="开场:城市全景",
    shot_type=ShotType.WIDE,
    camera_movement=CameraMovement.AERIAL,
    duration=4.0,
    prompt="Aerial view of a modern city at golden hour, skyscrapers "
           "reflecting sunset light, cinematic drone shot, 4K quality",
    negative_prompt="blurry, low quality, cartoon",
    transition=Transition.DISSOLVE,
)

storyboard.add_shot(
    scene_number=1, shot_number=2,
    description="推近到产品展示",
    shot_type=ShotType.MEDIUM,
    camera_movement=CameraMovement.DOLLY_IN,
    duration=3.0,
    prompt="Smooth dolly in to a sleek smartphone on a white pedestal, "
           "professional studio lighting, product photography style",
    negative_prompt="blurry, distorted",
    transition=Transition.CUT,
)

print(f"Total duration: {storyboard.total_duration()}s")
for shot in storyboard.to_prompt_sequence():
    print(f"  [{shot['scene']}] {shot['prompt'][:60]}...")

9.2 长视频拼接与过渡

import cv2
import numpy as np
from PIL import Image

class VideoStitcher:
    """视频片段拼接器"""

    @staticmethod
    def crossfade(frame_a, frame_b, progress):
        """交叉溶解过渡"""
        return cv2.addWeighted(frame_a, 1 - progress, frame_b, progress, 0)

    @staticmethod
    def fade_to_black(frame, progress):
        """淡入淡出到黑"""
        return (frame * (1 - progress)).astype(np.uint8)

    @staticmethod
    def stitch_clips(clips, transitions=None, crossfade_frames=12):
        """
        拼接多个视频片段

        参数:
            clips: 视频片段路径列表
            transitions: 过渡类型列表
            crossfade_frames: 交叉溶解帧数
        """
        all_frames = []

        for i, clip_path in enumerate(clips):
            cap = cv2.VideoCapture(clip_path)
            frames = []
            while True:
                ret, frame = cap.read()
                if not ret:
                    break
                frames.append(frame)
            cap.release()

            if i == 0:
                all_frames.extend(frames)
            else:
                # 应用过渡效果
                transition = transitions[i - 1] if transitions else "cut"

                if transition == "crossfade":
                    # 交叉溶解
                    overlap = min(crossfade_frames, len(all_frames), len(frames))
                    for j in range(overlap):
                        progress = j / overlap
                        idx_a = len(all_frames) - overlap + j
                        blended = VideoStitcher.crossfade(
                            all_frames[idx_a], frames[j], progress
                        )
                        all_frames[idx_a] = blended
                    all_frames.extend(frames[overlap:])
                elif transition == "fade":
                    # 淡入淡出
                    fade_out = min(12, len(all_frames))
                    fade_in = min(12, len(frames))
                    for j in range(fade_out):
                        progress = j / fade_out
                        idx = len(all_frames) - fade_out + j
                        all_frames[idx] = VideoStitcher.fade_to_black(
                            all_frames[idx], progress
                        )
                    for j in range(fade_in):
                        progress = 1 - j / fade_in
                        frames[j] = VideoStitcher.fade_to_black(
                            frames[j], progress
                        )
                    all_frames.extend(frames)
                else:  # cut
                    all_frames.extend(frames)

        return all_frames

    @staticmethod
    def save_video(frames, output_path, fps=24):
        """保存帧列表为视频"""
        if not frames:
            return
        h, w = frames[0].shape[:2]
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        writer = cv2.VideoWriter(output_path, fourcc, fps, (w, h))
        for frame in frames:
            writer.write(frame)
        writer.release()


# 使用示例
stitcher = VideoStitcher()
clips = ["scene1.mp4", "scene2.mp4", "scene3.mp4"]
transitions = ["crossfade", "cut"]
frames = stitcher.stitch_clips(clips, transitions)
stitcher.save_video(frames, "final_output.mp4", fps=24)

十、音频同步与配音

10.1 AI配音集成

import subprocess
import json
from pathlib import Path

class AudioVideoSync:
    """音视频同步工具"""

    @staticmethod
    def add_audio_to_video(video_path, audio_path, output_path,
                           audio_start=0, volume=1.0):
        """将音频添加到视频"""
        cmd = [
            "ffmpeg", "-y",
            "-i", video_path,
            "-i", audio_path,
            "-filter_complex",
            f"[1:a]atrim=start={audio_start},volume={volume}[a]",
            "-map", "0:v", "-map", "[a]",
            "-c:v", "copy", "-c:a", "aac",
            "-shortest",
            output_path,
        ]
        subprocess.run(cmd, check=True)
        return output_path

    @staticmethod
    def generate_tts_audio(text, output_path, voice="zh-CN-YunxiNeural",
                           rate="+0%", pitch="+0Hz"):
        """使用Edge TTS生成配音"""
        import edge_tts
        import asyncio

        async def _generate():
            communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch)
            await communicate.save(output_path)

        asyncio.run(_generate())
        return output_path

    @staticmethod
    def add_background_music(video_path, music_path, output_path,
                            music_volume=0.3, voice_volume=1.0):
        """添加背景音乐"""
        cmd = [
            "ffmpeg", "-y",
            "-i", video_path,
            "-i", music_path,
            "-filter_complex",
            f"[0:a]volume={voice_volume}[voice];"
            f"[1:a]volume={music_volume}[music];"
            f"[voice][music]amix=inputs=2:duration=first[a]",
            "-map", "0:v", "-map", "[a]",
            "-c:v", "copy",
            "-shortest",
            output_path,
        ]
        subprocess.run(cmd, check=True)
        return output_path

    @staticmethod
    def sync_audio_to_scenes(scenes, tts_voice="zh-CN-YunxiNeural"):
        """
        为每个场景生成配音并同步

        参数:
            scenes: [{"video": path, "narration": text}, ...]
        """
        audio_files = []
        for i, scene in enumerate(scenes):
            # 生成配音
            audio_path = f"temp_audio_{i}.mp3"
            AudioVideoSync.generate_tts_audio(
                scene["narration"], audio_path, voice=tts_voice
            )
            audio_files.append(audio_path)

        # 拼接音频和视频
        final_clips = []
        for scene, audio in zip(scenes, audio_files):
            output = f"synced_{Path(scene['video']).stem}.mp4"
            AudioVideoSync.add_audio_to_video(scene["video"], audio, output)
            final_clips.append(output)

        return final_clips


# 使用示例
sync = AudioVideoSync()

# 生成配音
sync.generate_tts_audio(
    "在这个美丽的秋天,让我们一起走进大自然,感受金色的阳光和温暖的微风。",
    "narration.mp3",
    voice="zh-CN-YunxiNeural",
    rate="-10%",
)

# 添加到视频
sync.add_audio_to_video(
    "generated_video.mp4",
    "narration.mp3",
    "final_with_audio.mp4",
)

# 添加背景音乐
sync.add_background_music(
    "final_with_audio.mp4",
    "background_music.mp3",
    "final_complete.mp4",
    music_volume=0.2,
)

十一、商用视频生成工作流

11.1 完整工作流设计

from dataclasses import dataclass
from typing import List, Dict, Optional
from pathlib import Path
import json
import time

@dataclass
class VideoProject:
    """视频项目"""
    project_id: str
    title: str
    description: str
    scenes: List[Dict]
    style_guide: Dict
    target_duration: float
    resolution: tuple = (1920, 1080)
    fps: int = 24

class CommercialVideoPipeline:
    """商用视频生成流水线"""

    def __init__(self, output_dir="./projects"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def create_project(self, title, description, brief) -> VideoProject:
        """创建视频项目"""
        project_id = f"proj_{int(time.time())}"
        project_dir = self.output_dir / project_id
        project_dir.mkdir(exist_ok=True)

        # 从简报生成分镜
        storyboard = self._generate_storyboard(brief)

        project = VideoProject(
            project_id=project_id,
            title=title,
            description=description,
            scenes=storyboard,
            style_guide=self._extract_style(brief),
            target_duration=brief.get("duration", 30),
        )

        # 保存项目配置
        with open(project_dir / "project.json", "w") as f:
            json.dump(vars(project), f, indent=2, ensure_ascii=False)

        return project

    def _generate_storyboard(self, brief):
        """从简报生成分镜脚本"""
        # 这里可以集成LLM来智能生成分镜
        scenes = []
        for i, scene_desc in enumerate(brief.get("scenes", [])):
            scenes.append({
                "scene_number": i + 1,
                "description": scene_desc,
                "prompt": self._scene_to_prompt(scene_desc),
                "duration": brief.get("scene_duration", 5),
                "shot_type": "medium",
                "movement": "static",
            })
        return scenes

    def _scene_to_prompt(self, description):
        """将场景描述转换为生成提示词"""
        return (
            f"High quality cinematic video: {description}. "
            f"Professional cinematography, smooth motion, "
            f"detailed textures, natural lighting, 4K resolution"
        )

    def _extract_style(self, brief):
        """提取风格指南"""
        return {
            "color_palette": brief.get("colors", ["warm", "natural"]),
            "mood": brief.get("mood", "professional"),
            "camera_style": brief.get("camera_style", "cinematic"),
            "lighting": brief.get("lighting", "natural"),
        }

    def generate_scene(self, project, scene, generator):
        """生成单个场景"""
        project_dir = self.output_dir / project.project_id
        scene_dir = project_dir / f"scene_{scene['scene_number']:03d}"
        scene_dir.mkdir(exist_ok=True)

        # 生成视频
        print(f"Generating scene {scene['scene_number']}: {scene['description'][:50]}...")

        # 提示词优化
        full_prompt = self._optimize_prompt(
            scene["prompt"],
            project.style_guide
        )

        # 生成
        result = generator.generate(
            prompt=full_prompt,
            negative_prompt="blurry, low quality, distorted, watermark",
            num_frames=int(scene["duration"] * project.fps),
            width=project.resolution[0],
            height=project.resolution[1],
        )

        # 保存
        scene_video_path = scene_dir / "generated.mp4"
        self._save_video(result, scene_video_path)

        return {
            "scene_number": scene["scene_number"],
            "video_path": str(scene_video_path),
            "prompt_used": full_prompt,
            "duration": scene["duration"],
        }

    def _optimize_prompt(self, base_prompt, style_guide):
        """根据风格指南优化提示词"""
        style_parts = []

        if "cinematic" in style_guide.get("camera_style", ""):
            style_parts.append("cinematic camera movement")

        mood = style_guide.get("mood", "")
        if mood:
            style_parts.append(f"{mood} mood")

        lighting = style_guide.get("lighting", "")
        if lighting:
            style_parts.append(f"{lighting} lighting")

        if style_parts:
            return f"{base_prompt}, {', '.join(style_parts)}"
        return base_prompt

    def assemble_final_video(self, project, scene_results,
                            add_audio=True, audio_script=None):
        """组装最终视频"""
        project_dir = self.output_dir / project.project_id

        # 拼接场景
        stitcher = VideoStitcher()
        video_paths = [r["video_path"] for r in scene_results]
        transitions = ["crossfade"] * (len(video_paths) - 1)

        frames = stitcher.stitch_clips(video_paths, transitions)
        stitched_path = project_dir / "stitched.mp4"
        stitcher.save_video(frames, str(stitched_path), fps=project.fps)

        final_path = project_dir / "final.mp4"

        if add_audio and audio_script:
            # 生成配音
            audio_path = project_dir / "narration.mp3"
            AudioVideoSync.generate_tts_audio(
                audio_script, str(audio_path)
            )
            AudioVideoSync.add_audio_to_video(
                str(stitched_path), str(audio_path), str(final_path)
            )
        else:
            final_path = stitched_path

        print(f"Final video saved to: {final_path}")
        return str(final_path)

    def _save_video(self, frames, path):
        """保存视频帧"""
        from diffusers.utils import export_to_video
        export_to_video(frames, str(path), fps=24)


# 完整使用示例
pipeline = CommercialVideoPipeline(output_dir="./video_projects")

# 创建项目
project = pipeline.create_project(
    title="品牌宣传视频",
    description="高端电子产品品牌宣传片",
    brief={
        "duration": 30,
        "scenes": [
            "A sleek smartphone floating in a void with soft blue light",
            "The smartphone transforming into particles of light",
            "Light particles forming a beautiful abstract pattern",
        ],
        "scene_duration": 10,
        "mood": "premium and futuristic",
        "camera_style": "cinematic",
        "lighting": "studio",
        "colors": ["blue", "white", "silver"],
    },
)

print(f"Project created: {project.project_id}")
print(f"Total scenes: {len(project.scenes)}")

11.2 质量控制与审核

import torch
from torchvision import models, transforms
from PIL import Image
import numpy as np

class VideoQualityChecker:
    """视频质量自动检测"""

    def __init__(self):
        self.transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                               std=[0.229, 0.224, 0.225]),
        ])

    def check_video(self, frames):
        """综合质量检测"""
        results = {
            "sharpness": self.check_sharpness(frames),
            "brightness": self.check_brightness(frames),
            "consistency": self.check_consistency(frames),
            "artifacts": self.check_artifacts(frames),
            "overall_score": 0,
        }

        # 计算综合分数
        scores = [v for k, v in results.items() if k != "overall_score" and isinstance(v, (int, float))]
        results["overall_score"] = sum(scores) / len(scores) if scores else 0

        return results

    def check_sharpness(self, frames):
        """检测清晰度"""
        import cv2
        sharpness_scores = []
        for frame in frames:
            gray = cv2.cvtColor(np.array(frame), cv2.COLOR_RGB2GRAY)
            laplacian = cv2.Laplacian(gray, cv2.CV_64F)
            score = laplacian.var()
            sharpness_scores.append(score)

        avg_sharpness = np.mean(sharpness_scores)
        # 归一化到0-100
        return min(100, avg_sharpness / 500 * 100)

    def check_brightness(self, frames):
        """检测亮度均匀性"""
        brightness_values = []
        for frame in frames:
            gray = np.array(frame.convert("L"))
            brightness_values.append(np.mean(gray))

        # 亮度稳定性
        std = np.std(brightness_values)
        score = max(0, 100 - std * 2)
        return score

    def check_consistency(self, frames):
        """检测帧间一致性"""
        if len(frames) < 2:
            return 100

        consistency_scores = []
        for i in range(1, len(frames)):
            prev = np.array(frames[i-1]).astype(float)
            curr = np.array(frames[i]).astype(float)
            diff = np.mean(np.abs(prev - curr))
            # 帧间差异应该适中(太大会闪烁,太小会静止)
            if 1 < diff < 30:
                consistency_scores.append(100)
            elif diff <= 1:
                consistency_scores.append(80)  # 可能太静止
            else:
                consistency_scores.append(max(0, 100 - diff))

        return np.mean(consistency_scores)

    def check_artifacts(self, frames):
        """检测视觉伪影"""
        artifact_scores = []
        for frame in frames:
            img = np.array(frame)
            # 检测异常色彩块
            r, g, b = img[:,:,0], img[:,:,1], img[:,:,2]
            # 检测纯色区域(可能是生成伪影)
            uniformity = np.std([np.std(r), np.std(g), np.std(b)])
            score = min(100, uniformity * 2)
            artifact_scores.append(score)

        return np.mean(artifact_scores)


# 使用示例
checker = VideoQualityChecker()
# results = checker.check_video(generated_frames)
# print(f"Quality Score: {results['overall_score']:.1f}/100")
# print(f"Sharpness: {results['sharpness']:.1f}")
# print(f"Consistency: {results['consistency']:.1f}")

十二、成本控制与效率优化

12.1 成本分析

@dataclass
class CostEstimator:
    """视频生成成本估算器"""

    # 各平台定价(参考价,实际以官方为准)
    PRICING = {
        "runway_gen3": {"per_second": 0.05, "currency": "USD"},
        "kling_std": {"per_second": 0.02, "currency": "USD"},
        "kling_pro": {"per_second": 0.06, "currency": "USD"},
        "replicate_svd": {"per_second": 0.01, "currency": "USD"},
        "sora": {"per_second": 0.04, "currency": "USD"},
        "local_gpu": {"per_hour": 2.0, "currency": "USD"},  # A100
    }

    @classmethod
    def estimate_project_cost(cls, scenes, platform="kling_std"):
        """估算项目总成本"""
        total_duration = sum(s.get("duration", 5) for s in scenes)
        # 假设平均需要2次重试
        effective_duration = total_duration * 2.5

        pricing = cls.PRICING.get(platform, {})
        if "per_second" in pricing:
            cost = effective_duration * pricing["per_second"]
        elif "per_hour" in pricing:
            # 假设每秒视频需要30秒GPU时间
            gpu_hours = effective_duration * 30 / 3600
            cost = gpu_hours * pricing["per_hour"]
        else:
            cost = 0

        return {
            "platform": platform,
            "total_duration": total_duration,
            "effective_duration": effective_duration,
            "estimated_cost": round(cost, 2),
            "currency": pricing.get("currency", "USD"),
        }

    @classmethod
    def compare_platforms(cls, scenes):
        """跨平台成本对比"""
        results = {}
        for platform in cls.PRICING:
            results[platform] = cls.estimate_project_cost(scenes, platform)
        return results


# 使用示例
scenes = [
    {"duration": 5, "description": "Opening scene"},
    {"duration": 8, "description": "Product showcase"},
    {"duration": 5, "description": "Closing scene"},
]

comparison = CostEstimator.compare_platforms(scenes)
for platform, info in comparison.items():
    print(f"{platform}: ${info['estimated_cost']:.2f} "
          f"({info['total_duration']}s video)")

12.2 效率优化策略

class EfficiencyOptimizer:
    """视频生成效率优化器"""

    @staticmethod
    def batch_generate(prompts, generator, batch_size=4):
        """批量生成优化"""
        results = []
        for i in range(0, len(prompts), batch_size):
            batch = prompts[i:i+batch_size]
            print(f"Processing batch {i//batch_size + 1}/"
                  f"{(len(prompts)-1)//batch_size + 1}")

            # 并行生成(如果GPU显存允许)
            batch_results = []
            for prompt in batch:
                result = generator.generate(prompt=prompt)
                batch_results.append(result)

            results.extend(batch_results)

        return results

    @staticmethod
    def smart_retry(generate_fn, prompt, max_retries=3,
                    quality_threshold=0.7):
        """智能重试:质量不达标时自动重试"""
        for attempt in range(max_retries):
            result = generate_fn(prompt)
            quality = VideoQualityChecker().check_video(result)

            if quality["overall_score"] >= quality_threshold * 100:
                print(f"Quality passed on attempt {attempt + 1}: "
                      f"{quality['overall_score']:.1f}")
                return result

            print(f"Quality below threshold on attempt {attempt + 1}: "
                  f"{quality['overall_score']:.1f}, retrying...")

        print(f"Returning best result after {max_retries} attempts")
        return result

    @staticmethod
    def optimize_resolution(target_resolution, quality_level="standard"):
        """根据需求选择最优分辨率"""
        resolution_presets = {
            "draft": (512, 288),
            "preview": (768, 432),
            "standard": (1280, 720),
            "high": (1920, 1080),
            "ultra": (3840, 2160),
        }
        return resolution_presets.get(quality_level, target_resolution)

    @staticmethod
    def estimate_gpu_time(num_frames, resolution, model="svd"):
        """估算GPU时间"""
        # 基于经验的估算公式
        pixels = resolution[0] * resolution[1]
        base_time_per_frame = {
            "svd": 2.5,
            "animate_diff": 3.0,
            "kling": 5.0,
        }
        time_per_frame = base_time_per_frame.get(model, 3.0)

        # 分辨率缩放因子
        scale_factor = pixels / (1280 * 720)
        total_time = num_frames * time_per_frame * scale_factor

        return {
            "estimated_seconds": round(total_time, 1),
            "estimated_minutes": round(total_time / 60, 1),
            "num_frames": num_frames,
            "resolution": resolution,
        }


# 使用示例
optimizer = EfficiencyOptimizer()

# 估算GPU时间
estimate = optimizer.estimate_gpu_time(
    num_frames=72,  # 3秒@24fps
    resolution=(1920, 1080),
    model="svd"
)
print(f"Estimated GPU time: {estimate['estimated_minutes']} minutes")

十三、最佳实践总结

13.1 提示词最佳实践

  1. 结构化描述:主体 → 动作 → 场景 → 风格 → 镜头 → 光线
  2. 具体优于抽象:"阳光透过树叶洒在地面"优于"美丽的光线"
  3. 避免否定:正面描述期望效果,而非说"不要xxx"
  4. 镜头语言:使用专业术语如"tracking shot"、"dolly in"
  5. 一致性:同一项目的提示词风格保持一致

13.2 技术选型建议

  1. 原型验证:使用开源模型(HunyuanVideo、Wan2.1)快速验证
  2. 质量优先:Sora > VEO 2 > Runway Gen-3 > 可灵 Pro
  3. 成本敏感:可灵标准模式 > Replicate SVD > 自建服务
  4. 中文场景:可灵 > 通义万相 > 其他模型
  5. 大批量生产:自建推理服务 + 批量队列

13.3 常见问题与解决方案

问题 原因 解决方案
画面闪烁 帧间一致性差 使用temporal attention、降低运动强度
运动不自然 物理规律违反 增加运动描述、使用物理先验
人物变形 解剖学错误 使用负面提示词、选择更好的模型
生成速度慢 推理计算量大 使用Turbo模型、降低分辨率、分块解码
风格不一致 缺乏风格锚点 使用参考图、统一提示词模板

总结

AI视频生成技术正处于快速发展期,从底层架构(DiT、U-Net、Transformer)到上层应用(分镜编排、音视频同步),整个技术栈日趋成熟。

关键要点回顾:

  1. 架构演进:从U-Net到DiT,Transformer架构正在成为主流
  2. 工具生态:商业模型(Sora、Runway、可灵)和开源模型(HunyuanVideo、Wan2.1)并存
  3. 实战能力:掌握提示词工程、API调用、工作流编排是核心技能
  4. 商业化路径:成本控制、质量保证、效率优化缺一不可
  5. 未来趋势:更长视频、更高分辨率、更强可控性、更低门槛

AI视频生成不再只是"把文字变成视频"的玩具,它正在成为专业内容创作者的生产力工具。掌握这些技术,你就能在这个新赛道上占据先机。


本教程内容基于截至2025年6月的公开技术资料整理,技术发展迅速,建议持续关注最新进展。

内容声明

本文内容为AI技术学习教程,仅供学习参考。如涉及技术问题,欢迎通过 xurj005@163.com 与我们交流。

目录