AI视频生成工程：Veo 3与Kling 2.0指南【2026】

interface VideoJobRequest {
  provider: "veo3" | "kling2" | "runway" | "pika";
  prompt: string;
  durationSec: number;
  aspectRatio: "16:9" | "9:16" | "1:1";
  quality: "draft" | "standard" | "premium";
  webhookUrl: string;
  idempotencyKey: string;
}

async function submitVeoJob(job: VideoJobRequest) {
  const response = await fetch("https://api.example-veo.com/v1/videos", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${process.env.VEO_API_KEY}`,
      "Idempotency-Key": job.idempotencyKey,
    },
    body: JSON.stringify({
      prompt: job.prompt,
      duration_seconds: job.durationSec,
      aspect_ratio: job.aspectRatio,
      quality: job.quality,
      webhook_url: job.webhookUrl,
    }),
  });

  if (!response.ok) {
    throw new Error(`Veo submission failed: ${response.status}`);
  }

  return response.json();
}

import os
import requests

def submit_kling_job(prompt: str, callback_url: str) -> dict:
    payload = {
        "prompt": prompt,
        "duration": 5,
        "aspect_ratio": "9:16",
        "mode": "standard",
        "callback_url": callback_url,
    }
    response = requests.post(
        "https://api.example-kling.com/v2/video/generations",
        headers={"Authorization": f"Bearer {os.environ['KLING_API_KEY']}"},
        json=payload,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

job = submit_kling_job(
    "A clean product demo shot, slow dolly-in, soft studio light, 5 seconds",
    "https://example.com/webhooks/video",
)
print(job["id"])

主体：一只陶瓷咖啡杯放在胡桃木桌面上
场景：清晨工作室，柔和侧光，极简背景
镜头：从中景缓慢推进到特写
运动：蒸汽上升，杯面反光轻微移动
时长：5 秒
风格：真实产品广告，浅景深
负向约束：不要文字、不要手部畸变、不要跳切、不要闪烁

from dataclasses import dataclass

@dataclass
class VideoScore:
    alignment: float
    motion: float
    aesthetic: float
    safety: float

def aggregate_score(score: VideoScore) -> float:
    if score.safety < 0.95:
        return 0.0
    return (
        0.35 * score.alignment +
        0.25 * score.motion +
        0.25 * score.aesthetic +
        0.15 * score.safety
    )

sample = VideoScore(alignment=0.83, motion=0.78, aesthetic=0.81, safety=0.99)
print(round(aggregate_score(sample), 3))

import asyncio
from dataclasses import dataclass
from typing import Literal

Provider = Literal["kling2", "veo3", "runway"]

@dataclass
class Job:
    id: str
    prompt: str
    quality: Literal["draft", "premium"]
    budget_cents: int

async def submit(provider: Provider, job: Job) -> str:
    await asyncio.sleep(0.1)
    return f"{provider}-remote-{job.id}"

def route(job: Job) -> list[Provider]:
    if job.quality == "draft":
        return ["kling2", "runway", "veo3"]
    return ["veo3", "runway", "kling2"]

async def process(job: Job) -> dict:
    errors = []
    for provider in route(job):
        try:
            remote_id = await submit(provider, job)
            return {"job_id": job.id, "provider": provider, "remote_id": remote_id}
        except Exception as error:
            errors.append(str(error))
    raise RuntimeError(f"All providers failed: {errors}")

result = asyncio.run(process(Job("job-123", "cinematic product shot", "draft", 500)))
print(result)