Veo 3 API 教程：用 Google 最新模型生成电影级视频

3 个关键数据：视频生成延迟约 2–4 分钟（异步队列）；veo-3-0 模型目前通过 Gemini API 按使用量计费，每秒视频约 $0.35（Vertex AI 定价）；原生音频生成无需额外后处理，输出分辨率支持 720p / 1080p / 4K。

---

目录

1. 前提条件
2. 认证与环境配置
3. 第一个视频请求（基础实现）
4. 异步轮询：等待生成完成
5. 生产级实现：封装完整 Pipeline
6. API 参数参考表
7. 错误处理
8. 性能与成本对比表
9. 结语

---

1. 前提条件

在开始之前，确认你已具备以下条件：

账户与权限

• Google Cloud 账户，已启用计费
• 项目已启用 Vertex AI API 或已申请 Gemini API 的 veo-3-0 访问权限
• 在 Google AI Studio 申请 Veo 3 访问（目前部分地区仍需 waitlist）

本地环境

• Python 3.10+
• 以下库：

# 安装 Google Gen AI SDK（官方推荐方式）
pip install google-genai>=0.8.0

# 可选：用于视频文件处理
pip install requests tqdm

环境变量

# 将 API Key 写入环境变量，不要硬编码在代码里
export GOOGLE_API_KEY="your_api_key_here"

# 如果使用 Vertex AI 而非 Gemini API，额外设置
export GOOGLE_CLOUD_PROJECT="your_project_id"
export GOOGLE_CLOUD_LOCATION="us-central1"

⚠️ 注意：veo-3-0 和 veo-3-1 目前通过 Gemini API（google-genai SDK）调用。Vertex AI 路径支持企业级 SLA，但配置更复杂。本教程使用 Gemini API 路径。

---

2. 认证与环境配置

# setup_check.py
# 运行这个脚本验证你的配置是否正确，再进入正式开发

import os
import sys
from google import genai
from google.genai import types

def verify_setup():
    """验证 API key 是否存在，以及 SDK 是否能正常初始化"""
    
    api_key = os.environ.get("GOOGLE_API_KEY")
    if not api_key:
        print("[ERROR] 环境变量 GOOGLE_API_KEY 未设置")
        sys.exit(1)
    
    # 初始化客户端 — SDK 会自动从环境变量读取 API key
    client = genai.Client(api_key=api_key)
    
    # 列出可用模型，确认 veo 系列在你的账户里可见
    print("正在获取可用模型列表...")
    try:
        models = client.models.list()
        veo_models = [m.name for m in models if "veo" in m.name.lower()]
        
        if not veo_models:
            print("[WARN] 未找到 Veo 模型 — 可能需要申请访问权限")
            print("申请地址: https://aistudio.google.com/models/veo-3")
        else:
            print(f"[OK] 找到 Veo 模型: {veo_models}")
    except Exception as e:
        print(f"[ERROR] API 调用失败: {e}")
        sys.exit(1)
    
    print("[OK] 配置验证通过")
    return client

if __name__ == "__main__":
    verify_setup()

运行：

python setup_check.py
# 期望输出:
# [OK] 找到 Veo 模型: ['models/veo-3-0', 'models/veo-3-1']
# [OK] 配置验证通过

---

3. 第一个视频请求（基础实现）

Veo 3 的视频生成是异步的，generate_videos() 不会直接返回视频文件，而是返回一个 Operation 对象。你需要轮询这个 operation 直到完成。

下面是最简化的完整流程：

# basic_video_gen.py
# 最小可运行示例 — 生成一个 8 秒视频并下载到本地

import os
import time
import requests
from google import genai
from google.genai import types

# 初始化客户端
client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])

# 定义 prompt — 越具体越好，模型会遵循镜头、光线、风格等描述
PROMPT = (
    "A lone astronaut walks across a rust-red Martian landscape at golden hour. "
    "Low angle shot, dust particles floating in the air, cinematic 35mm lens flare, "
    "ambient wind sounds, no dialogue."
)

def generate_video_basic(prompt: str, output_path: str = "output.mp4"):
    """
    提交视频生成请求并等待完成。
    
    Veo 3 生成 8 秒 720p 视频约需 2–4 分钟。
    这个函数会阻塞直到生成完成或超时（10 分钟）。
    """
    print(f"提交生成请求: {prompt[:60]}...")
    
    # 提交请求 — 注意 model 参数使用完整路径
    operation = client.models.generate_videos(
        model="veo-3-0",          # 最新稳定版本
        prompt=prompt,
        config=types.GenerateVideosConfig(
            aspect_ratio="16:9",   # 支持: "16:9", "9:16", "1:1"
            duration_seconds=8,    # 当前固定为 8 秒
            resolution="720p",     # 支持: "720p", "1080p", "4k"（4k 仅部分区域）
            number_of_videos=1,    # 每次最多生成 4 个变体
            generate_audio=True,   # Veo 3 原生音频，不额外收费
        ),
    )
    
    print(f"Operation ID: {operation.name}")
    print("等待生成完成（预计 2–4 分钟）...")
    
    # 轮询直到完成
    max_wait_seconds = 600  # 10 分钟超时
    poll_interval = 20       # 每 20 秒检查一次，避免 rate limit
    elapsed = 0
    
    while not operation.done:
        if elapsed >= max_wait_seconds:
            raise TimeoutError(f"视频生成超时（已等待 {elapsed}s）")
        
        time.sleep(poll_interval)
        elapsed += poll_interval
        
        # 刷新 operation 状态
        operation = client.operations.get(operation)
        print(f"  [{elapsed}s] 状态: {'完成' if operation.done else '生成中...'}")
    
    # 检查是否有错误
    if operation.error:
        raise RuntimeError(f"生成失败: {operation.error.message}")
    
    # 提取视频 URI 并下载
    video_response = operation.response
    if not video_response.generated_videos:
        raise ValueError("API 返回了空的视频列表")
    
    video = video_response.generated_videos[0]
    
    # download_video 是 SDK 提供的便捷方法
    # 它会处理认证头和分块下载
    client.files.download(file=video.video)
    video.video.save(output_path)
    
    print(f"[OK] 视频已保存: {output_path}")
    return output_path

if __name__ == "__main__":
    generate_video_basic(PROMPT, "mars_astronaut.mp4")

常见误区：不要把 generate_videos() 的返回值当成视频数据处理。它返回的是 Operation，类似于一个异步任务句柄。

---

4. 异步轮询：等待生成完成

上面的 basic_video_gen.py 是同步阻塞的，适合脚本和测试。如果你需要在生成视频的同时做其他事情，或者需要管理多个并发任务，应该分离”提交”和”等待”两个步骤。

# async_polling.py
# 展示如何保存 operation name，稍后再查询结果
# 适用于：Web 服务、批量生成、长时间运行的 pipeline

import os
import json
import time
from google import genai
from google.genai import types

client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])

def submit_video_job(prompt: str, config: dict = None) -> str:
    """
    仅提交任务，立即返回 operation name。
    把 operation name 存到数据库或队列里，稍后查询。
    """
    default_config = types.GenerateVideosConfig(
        aspect_ratio="16:9",
        duration_seconds=8,
        resolution="1080p",
        number_of_videos=1,
        generate_audio=True,
    )
    
    operation = client.models.generate_videos(
        model="veo-3-0",
        prompt=prompt,
        config=default_config,
    )
    
    # operation.name 是类似 "operations/abc123..." 的字符串
    # 这是你唯一需要持久化的东西
    print(f"任务已提交: {operation.name}")
    return operation.name


def poll_video_job(operation_name: str, timeout: int = 600) -> list[str]:
    """
    根据 operation name 查询任务状态，完成后返回本地文件路径列表。
    
    Args:
        operation_name: submit_video_job() 返回的 operation name
        timeout: 最大等待秒数
    
    Returns:
        下载后的视频文件路径列表
    """
    # 用 operation name 重建 operation 对象
    # SDK 支持通过 name 字符串恢复 operation
    operation = client.operations.get({"name": operation_name})
    
    start_time = time.time()
    
    while not operation.done:
        elapsed = time.time() - start_time
        if elapsed > timeout:
            raise TimeoutError(f"超时: {operation_name}")
        
        # 指数退避：前期频繁检查，后期降低频率
        # 视频生成通常在 90–240 秒内完成
        if elapsed < 60:
            sleep_time = 15
        elif elapsed < 180:
            sleep_time = 20
        else:
            sleep_time = 30
        
        time.sleep(sleep_time)
        operation = client.operations.get(operation)
        print(f"  [{int(elapsed)}s] {operation_name[-12:]}... {'完成' if operation.done else '等待中'}")
    
    if operation.error:
        raise RuntimeError(f"[{operation.error.code}] {operation.error.message}")
    
    # 下载所有生成的视频（number_of_videos 可以 > 1）
    output_paths = []
    for i, gen_video in enumerate(operation.response.generated_videos):
        path = f"video_{operation_name[-8:]}_{i}.mp4"
        client.files.download(file=gen_video.video)
        gen_video.video.save(path)
        output_paths.append(path)
        print(f"[OK] 保存: {path}")
    
    return output_paths


# 演示：先提交多个任务，再并发等待
if __name__ == "__main__":
    prompts = [
        "A time-lapse of cherry blossoms falling in slow motion, macro lens, soft bokeh, ambient forest sounds.",
        "An underwater shot of a coral reef, schools of fish darting through sunbeams, oceanic ambient sound.",
    ]
    
    # 批量提交（几乎同时发出，节省总等待时间）
    op_names = [submit_video_job(p) for p in prompts]
    
    # 保存任务 ID（实际项目中写入数据库）
    with open("pending_jobs.json", "w") as f:
        json.dump(op_names, f)
    print(f"已提交 {len(op_names)} 个任务，ID 已保存到 pending_jobs.json")
    
    # 依次等待完成（也可以用 threading 并发）
    all_videos = []
    for op_name in op_names:
        videos = poll_video_job(op_name)
        all_videos.extend(videos)
    
    print(f"\n全部完成，共 {len(all_videos)} 个视频文件")

---

5. 生产级实现：封装完整 Pipeline

以下是可以直接集成到生产服务的完整封装，包含重试逻辑、日志、以及多分辨率支持：

# veo_pipeline.py
# 生产级 Veo 3 视频生成 Pipeline
# 包含: 重试、日志、配置验证、下载验证

import os
import time
import logging
import hashlib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from google import genai
from google.genai import types
from google.api_core import exceptions as gcp_exceptions

# 配置结构化日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("veo_pipeline")


@dataclass
class VideoGenConfig:
    """Veo 3 生成配置 — 所有参数都有明确的默认值和注释"""
    
    # 模型选择: "veo-3-0" 或 "veo-3-1"（3.1 质量更高但可能更慢）
    model: str = "veo-3-0"
    
    aspect_ratio: str = "16:9"    # "16:9" | "9:16" | "1:1"
    duration_seconds: int = 8     # 目前固定为 8，API 后续可能支持更长
    resolution: str = "1080p"     # "720p" | "1080p" | "4k"
    number_of_videos: int = 1     # 1–4，生成多个变体用于挑选
    generate_audio: bool = True   # 原生音频，Veo 3 独有特性
    
    # 负面提示词 — 用于排除不需要的内容
    negative_prompt: Optional[str] = None
    
    # 输出目录
    output_dir: str = "./veo_outputs"
    
    # 重试配置
    max_retries: int = 3
    retry_delay: float = 5.0
    poll_timeout: int = 600        # 10 分钟轮询超时
    poll_interval_base: int = 20   # 基础轮询间隔（秒）


class VeoPipeline:
    """
    封装 Veo 3 API 的完整工作流。
    
    设计原则:
    - 每个方法只做一件事
    - 所有错误都包含可操作的信息
    - 不吞掉异常
    """
    
    # 已知可重试的错误码
    RETRYABLE_ERRORS = {429, 503, 504}
    
    def __init__(self, config: VideoGenConfig):
        self.config = config
        self.client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
        
        # 确保输出目录存在
        Path(config.output_dir).mkdir(parents=True, exist_ok=True)
        logger.info(f"VeoPipeline 初始化完成，输出目录: {config.output_dir}")
    
    def _build_gen_config(self) -> types.GenerateVideosConfig:
        """从 VideoGenConfig 构建 SDK 所需的配置对象"""
        kwargs = {
            "aspect_ratio": self.config.aspect_ratio,
            "duration_seconds": self.config.duration_seconds,
            "resolution": self.config.resolution,
            "number_of_videos": self.config.number_of_videos,
            "generate_audio": self.config.generate_audio,
        }
        if self.config.negative_prompt:
            kwargs["negative_prompt"] = self.config.negative_prompt
        
        return types.GenerateVideosConfig(**kwargs)
    
    def _submit_with_retry(self, prompt: str):
        """提交请求，失败时按配置重试"""
        last_error = None
        
        for attempt in range(self.config.max_retries):
            try:
                logger.info(f"提交请求 (尝试 {attempt + 1}/{self.config.max_retries})")
                operation = self.client.models.generate_videos(
                    model=self.config.model,
                    prompt=prompt,
                    config=self._build_gen_config(),
                )
                logger.info(f"Operation 创建成功: {operation.name}")
                return operation
            
            except gcp_exceptions.ResourceExhausted as e:
                # 429 Rate limit — 需要等待更长时间
                wait = self.config.retry_delay * (2 ** attempt)  # 指数退避
                logger.warning(f"Rate limited，等待 {wait:.1f}s 后重试")
                time.sleep(wait)
                last_error = e
            
            except gcp_exceptions.ServiceUnavailable as e:
                # 503 服务暂时不可用
                wait = self.config.retry_delay * (attempt + 1)
                logger.warning(f"服务不可用，等待 {wait:.1f}s 后重试")
                time.sleep(wait)
                last_error = e
            
            except gcp_exceptions.InvalidArgument as e:
                # 400 参数错误 — 不重试，直接抛出
                logger.error(f"参数无效，请检查 prompt 或 config: {e}")
                raise
        
        raise RuntimeError(f"达到最大重试次数 ({self.config.max_retries})") from last_error
    
    def _wait_for_completion(self, operation):
        """轮询 operation 直到完成或超时"""
        start = time.time()
        poll_count = 0
        
        while not operation.done:
            elapsed = time.time() - start
            
            if elapsed > self.config.poll_timeout:
                raise TimeoutError(
                    f"生成超时 ({self.config.poll_timeout}s)，"
                    f"Operation: {operation.name}"
                )
            
            # 自适应轮询间隔：早期快，后期慢，减少 API 调用
            interval = min(
                self.config.poll_interval_base + poll_count * 5,
                60  # 最大 60 秒一次
            )
            time.sleep(interval)
            
            operation = self.client.operations.get(operation)
            poll_count += 1
            logger.info(f"  [{int(elapsed)}s] 轮询 #{poll_count}: {'完成' if operation.done else '生成中'}")
        
        if operation.error:
            error_code = operation.error.code
            error_msg = operation.error.message
            raise RuntimeError(f"生成失败 [{error_code}]: {error_msg}")
        
        return operation
    
    def _download_videos(self, operation, prompt: str) -> list[Path]:
        """下载所有生成的视频文件，文件名包含 prompt hash 便于追踪"""
        
        # 用 prompt 的 hash 前 8 位作为文件名的一部分
        prompt_hash = hashlib.md5(prompt.encode()).hexdigest()[:8]
        timestamp = int(time.time())
        
        output_paths = []
        for i, gen_video in enumerate(operation.response.generated_videos):
            filename = f"{timestamp}_{prompt_hash}_{i}.mp4"
            output_path = Path(self.config.output_dir) / filename
            
            # SDK 的 download + save 处理分块传输和认证
            self.client.files.download(file=gen_video.video)
            gen_video.video.save(str(output_path))
            
            # 验证文件确实被写入且大小合理（> 10KB）
            if not output_path.exists() or output_path.stat().st_size < 10_000:
                logger.warning(f"文件 {output_path} 可能不完整，大小: {output_path.stat().st_size if output_path.exists() else 0} bytes")
            else:
                logger.info(f"[OK] 保存: {output_path} ({output_path.stat().st_size / 1024:.1f} KB)")
            
            output_paths.append(output_path)
        
        return output_paths
    
    def generate(self, prompt: str) -> list[Path]:
        """
        主入口：提交 → 等待 → 下载
        
        Returns:
            本地视频文件路径列表
        
        Raises:
            RuntimeError: 生成失败（包含错误码）
            TimeoutError: 超过 poll_timeout 未完成
            ValueError: prompt 为空
        """
        if not prompt or not prompt.strip():
            raise ValueError("prompt 不能为空")
        
        if len(prompt) > 4096:
            # API 限制，超出会返回 400
            raise ValueError(f"prompt 超过 4096 字符限制（当前 {len(prompt)} 字）")
        
        logger.info(f"开始生成视频: '{prompt[:80]}...'")
        
        operation = self._submit_with_retry(prompt)
        operation = self._wait_for_completion(operation)
        paths = self._download_videos(operation, prompt)
        
        logger.info(f"生成完成，共 {len(paths)} 个视频")
        return paths


# 使用示例
if __name__ == "__main__":
    config = VideoGenConfig(
        model="veo-3-0",
        resolution="1080p",
        number_of_videos=2,        # 生成 2 个变体
        generate_audio=True,
        negative_prompt="blurry, low quality, watermark, text overlay",
        output_dir="./my_videos",
    )
    
    pipeline = VeoPipeline(config)
    
    prompt = (
        "A vintage steam locomotive crossing a stone viaduct in the Scottish Highlands, "
        "mist rolling through valleys, late afternoon golden light, "
        "steam engine sounds and distant bird calls, wide establishing shot."
    )
    
    try:
        video_paths = pipeline.generate(prompt)
        print(f"\n成功生成 {len(video_paths)} 个视频:")
        for p in video_paths:
            print(f"  {p}")
    except TimeoutError as e:
        print(f"[TIMEOUT] {e}")
    except RuntimeError as e:
        print(f"[ERROR] {e}")

---

6. API 参数参考表

参数名	类型	默认值	有效范围	影响
model	str	—	"veo-3-0", "veo-3-1"	3.1 质量更高，延迟略长
prompt	str	—	1–4096 字符	视频内容、风格、音频氛围
negative_prompt	str	None	最大 1024 字符	排除不需要的元素
aspect_ratio	str	"16:9"	"16:9", "9:16", "1:1"	输出画幅比例
duration_seconds	int	8	目前仅 8	视频时长（API 后续扩展中）
resolution	str	"720p"	"720p", "1080p", "4k"	输出分辨率，影响成本和生成时间
number_of_videos	int	1	1–4	一次生成多少个变体（按个数计费）
generate_audio	bool	True	True, False	是否生成原生音频（Veo 3 特有）
seed	int	None	任意整数	固定随机种子，提升可复现性
enhance_prompt	bool	True	True, False	是否让模型自动扩写 prompt 细节

enhance_prompt 说明：默认开启时，模型会在你的 prompt 基础上自动补充光线、摄影角度等细节，通常提升质量。如果你需要严格控制输出（如品牌内容），建议设为 False。

---

7. 错误处理

错误码	错误类型	触发原因	处理方式
400	InvalidArgument	prompt 超长、参数值无效、resolution 拼写错误	检查参数，不要重试
403	PermissionDenied	API key 无效，或未申请 Veo 3 访问权限	检查 key，前往 AI Studio 申请访问
404	NotFound	model name 错误（如写成 "veo3-0"）	确认使用 "veo-3-0" 格式
429	ResourceExhausted	超出 QPM（每分钟请求数）限制	指数退避重试，免费层 QPM 极低
500	InternalServerError	服务器内部错误，生成失败（内容过滤等）	等待后重试；如持续失败换 prompt
503	ServiceUnavailable	服务暂时过载	退避重试，通常 30–60s 后恢复

内容安全过滤（HTTP 200 但视频列表为空）：

# 当 prompt 触发内容安全过滤时，API 返回 200
# 但 generated_videos 为空列表，不会抛出异常
# 必须主动检查

if not operation.response.generated_videos:
    # 不是 API 错误，是内容被过滤
    # 常见触发: 真实人名、暴力、版权角色
    raise ValueError(
        "生成结果为空，prompt 可能触发了内容安全过滤。"
        "建议: 删除真实人名、品牌名、版权角色名称。"
    )

Prompt 最佳实践（减少过滤触发）：

• ❌ "Tom Hanks walking on a beach" → 使用真实人名会被过滤
• ✅ "A middle-aged man with kind eyes walking on a beach"
• ❌ "Spiderman swinging through New York" → 版权角色
• ✅ "A figure in a red and blue suit swinging between skyscrapers"

---

8. 性能与成本对比表

以下数据基于 Vertex AI 公开定价和社区实测数据（2025 年 6 月）：

配置	分辨率	音频	预计生成时间	估算成本/视频	适用场景
veo-3-0 + 720p + 无音频	720p	❌	~90–150s	~$2.80	快速原型、批量草稿
veo-3-0 + 720p + 有音频	720p	✅	~120–180s	~$2.80	标准内容生产
veo-3-0 + 1080p + 有音频	1080p	✅	~150–240s	~$4.20	商业发布内容
veo-3-1 + 1080p + 有音频	1080p	✅	~180–300s	~$4.20	高保真需求
veo-3-0 + 4K + 有音频	4K	✅	~300–480s	~$8.40+	专业影视制作
number_of_videos=4 × 1080p	1080p	✅	~240–360s	~$16.80	A/B 测试多变体

成本说明：以上估算基于 $0.35/秒的参考定价 × 8 秒。实际费用会因区域、套餐和 API 更新而变化，以 Google Cloud 定价页面 为准。免费层（Free Tier）提供有限配额，不适合批量生产。

何时不应该用 Veo 3 API：

• 需要生成 超过 8 秒 的连续视频（当前 API 限制）
• 需要 实时或低延迟（< 30 秒）的应用场景
• 预算 < $5/视频 的大规模批量任务（考虑开源模型）
• 需要 精确时序控制（如配合固定旁白的广告）

---

9. 结语

本教程涵盖了从环境配置到生产级封装的完整 Veo 3 API 工作流，所有代码块均可直接复制使用。关键点：异步轮询是必须的，不是可选的；generate_audio=True 是 Veo 3 区别于其他模型的核心特性，默认开启即可；内容安全过滤以静默方式返回空列表，必须在代码里主动检查。后续关注 duration_seconds 参数的扩展，Google 已在文档中标注该参数将支持更长时长。

提示： 如果你需要在同一个项目中使用多个 AI 模型，AtlasCloud 提供统一 API 接入 300+ 模型（Kling、Flux、Seedance、Claude、GPT 等），一个 key 全部搞定。新用户首次充值享 25% 赠送（最高 $100）。