如何在 SaaS 應用程式中加入 AI 影片生成

將影片生成功能加入你的應用程式，與加入圖片生成並不相同。API 呼叫會立即返回——但影片此時尚未準備好。你會拿到一個任務 ID，之後必須持續詢問「完成了嗎？」直到完成為止。

多數開發者第一次呼叫影片 API 時，會等著在回應本文中拿到影片 URL，結果卻只拿到一個任務 ID。本指南將帶你走完完整流程：提交任務、輪詢結果、處理失敗，以及在 URL 失效前儲存輸出。

你將構建的內容

一個後端服務，可接收文字提示或圖片、提交影片生成任務、輪詢直到完成，並返回最終影片 URL。你將使用四個模型——Veo 3 Fast、Sora 2、Kling Video 與 Runway——全部透過單一 API 金鑰完成。

前置需求：

• Python 3.8+ 或 Node.js 18+
• 一把 CometAPI 金鑰
• 基本 REST API 使用經驗

瞭解為什麼影片生成不同

圖片生成是同步的：你送出請求並在同一個回應中拿到圖片。影片生成則使用非同步的任務佇列：

1. 提交生成請求 → 返回一個 task_id
2. 每隔幾秒輪詢狀態端點
3. 當狀態到達終止狀態時，會返回影片 URL
4. 下載並儲存影片——URL 是暫時的

如果你把影片生成當作圖片生成來處理，等待第一個回應就包含影片，你的請求每次都會逾時。

在正式環境的網路服務中，這個輪詢迴圈應在背景工作者（如 Celery、Bull 等）中執行，而不是在請求處理器內。下方範例採用同步輪詢——適合腳本與原型，但不適合處理並發使用者。

選擇模型

Source*: CometAPI model pages, May 2026. Note: "Sora 2" is CometAPI's model* identifier — refer to their model page for the underlying model details.

• Veo 3 Fast 支援文字轉影片與圖像轉影片。每秒成本最低，是很好的起點。
• Sora 2 可原生同時生成影片與音訊——對白、環境音與音效，無需額外的 TTS 步驟。
• Kling Video 提供 negative_prompt、cfg_scale、鏡頭運動設定，以及 pro 模式。在這四者中可控性最高。
• Runway 透過 CometAPI 僅支援圖像轉影片。提供一張靜態圖片與動作描述，它會將其動態化。

提交 Veo 任務

Veo 使用 multipart/form-data。在 Python requests 中請使用 files= 正確傳送——data=dict 會送出 application/x-www-form-urlencoded，這並不相同：

import requestsimport osfrom dotenv import load_dotenv​load_dotenv()​def submit_veo_task(prompt: str, size: str = "16x9") -> str:    """Submit a Veo 3 Fast text-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    response = requests.post(        "https://api.cometapi.com/v1/videos",        headers={"Authorization": f"Bearer {api_key}"},        files={            "prompt": (None, prompt),            "model": (None, "veo3-fast"),            "size": (None, size)        },        timeout=30    )    response.raise_for_status()    return response.json()["id"]​​task_id = submit_veo_task("A paper kite drifting above a wheat field on a windy afternoon")print(f"Task submitted: {task_id}")

輪詢結果

import time​def poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Poll until Veo task completes. Returns video URL."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    headers = {"Authorization": f"Bearer {api_key}"}    url = f"https://api.cometapi.com/v1/videos/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result.get("status")​        if status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(                f"Task {task_id} failed with status '{status}': "                f"{result.get('error', 'no error detail returned')}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Task {task_id} did not complete within {max_wait} seconds")​​video_url = poll_veo_task(task_id)print(f"Video ready: {video_url}")

需要更多控制時使用 Kling Video

Kling 的端點結構不同，並使用 JSON。注意 Kling 的終止狀態字串是「succeed」（不是「succeeded」）——這與 API 的實際回應格式相符：

def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str:    """Submit a Kling text-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    response = requests.post(        "https://api.cometapi.com/kling/v1/videos/text2video",        headers={            "Authorization": f"Bearer {api_key}",            "Content-Type": "application/json"        },        json={            "model_name": "kling-v1-6",            "prompt": prompt,            "negative_prompt": "blurry, low quality, watermark",            "cfg_scale": 0.5,            "mode": mode,         # "std" or "pro"            "aspect_ratio": "16:9",            "duration": duration  # "5" or "10"        },        timeout=30    )    response.raise_for_status()    return response.json()["data"]["task_id"]​​def poll_kling_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Poll Kling task until complete. Returns video URL."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    headers = {"Authorization": f"Bearer {api_key}"}    url = f"https://api.cometapi.com/kling/v1/videos/text2video/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result["data"]["task_status"]​        if status == "succeed":  # Kling uses "succeed", not "succeeded"            return result["data"]["task_result"]["videos"][0]["url"]        elif status == "failed":            error_detail = result.get("data", {}).get("task_result", "no detail")            raise RuntimeError(                f"Kling task {task_id} failed: {error_detail}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Kling task {task_id} timed out after {max_wait}s")

Source*:* CometAPI Kling Video docs

用 Runway 讓靜態圖片動起來

Runway 僅支援圖像轉影片，且需要額外的標頭（X-Runway-Version）：

def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str:    """Submit a Runway image-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    response = requests.post(        "https://api.cometapi.com/runwayml/v1/image_to_video",        headers={            "Authorization": f"Bearer {api_key}",            "X-Runway-Version": "2024-11-06",            "Content-Type": "application/json"        },        json={            "model": "gen3a_turbo",            "promptImage": image_url,  # must be a stable HTTPS URL            "promptText": motion_prompt,            "duration": duration,            "ratio": "1280:720",            "watermark": False        },        timeout=30    )    response.raise_for_status()    return response.json()["id"]​​def poll_runway_task(task_id: str, interval: int = 5, max_wait: int = 600) -> str:    """Poll Runway task. Returns video URL when done."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    headers = {        "Authorization": f"Bearer {api_key}",        "X-Runway-Version": "2024-11-06"    }    url = f"https://api.cometapi.com/runwayml/v1/tasks/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result.get("status")​        if status == "task_not_exist":            # CometAPI-specific: task is still initializing, retry after a few seconds            time.sleep(interval)            elapsed += interval            continue        elif status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(f"Runway task {task_id} failed: {result.get('error', 'no detail')}")​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Runway task {task_id} timed out after {max_wait}s")

Source*:* CometAPI Runway docs

在 URL 失效前儲存影片

生成 API 提供的影片 URL 為暫時性。立即下載檔案並儲存到你可控的位置：

import requestsimport pathlib​def download_video(url: str, output_path: str) -> None:    """Download video from URL to local file using streaming."""    out = pathlib.Path(output_path)    if out.parent != pathlib.Path("."):        out.parent.mkdir(parents=True, exist_ok=True)​    with requests.get(url, stream=True, timeout=60) as r:        r.raise_for_status()        with open(out, "wb") as f:            for chunk in r.iter_content(chunk_size=8192):                f.write(chunk)    print(f"Saved to {output_path}")​​# Full flowtask_id = submit_veo_task("A timelapse of clouds moving over a city skyline")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")

在正式環境中，將本機檔案寫入替換為上傳至 S3、Cloudflare R2 或你選擇的儲存服務。串流模式保持不變——直接管線傳輸位元組，而非將整個影片載入記憶體。

處理失敗情況

Source: CometAPI video generation docs

Node.js 版本

Node.js 18+ 原生包含 fetch 與 FormData。此範例涵蓋四個模型：

// Node.js 18+ — no extra packages needed​const API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY is not set');​// --- Veo 3 Fast ---async function submitVeoTask(prompt, size = '16x9') {  const form = new FormData();  form.append('prompt', prompt);  form.append('model', 'veo3-fast');  form.append('size', size);​  const res = await fetch('https://api.cometapi.com/v1/videos', {    method: 'POST',    headers: { 'Authorization': `Bearer ${API_KEY}` },    body: form  });  if (!res.ok) throw new Error(`Veo submit failed: ${res.status}`);  return (await res.json()).id;}​async function pollVeoTask(taskId, intervalMs = 10000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(`https://api.cometapi.com/v1/videos/${taskId}`, {      headers: { 'Authorization': `Bearer ${API_KEY}` }    });    if (!res.ok) throw new Error(`Poll failed: ${res.status}`);    const result = await res.json();​    if (result.status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(result.status)) {      throw new Error(`Task ${taskId} failed: ${result.error ?? 'no detail'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Task ${taskId} timed out`);}​// --- Kling Video ---async function submitKlingTask(prompt, duration = '5', mode = 'std') {  const res = await fetch('https://api.cometapi.com/kling/v1/videos/text2video', {    method: 'POST',    headers: {      'Authorization': `Bearer ${API_KEY}`,      'Content-Type': 'application/json'    },    body: JSON.stringify({      model_name: 'kling-v1-6',      prompt,      negative_prompt: 'blurry, low quality, watermark',      cfg_scale: 0.5,      mode,      aspect_ratio: '16:9',      duration    })  });  if (!res.ok) throw new Error(`Kling submit failed: ${res.status}`);  return (await res.json()).data.task_id;}​async function pollKlingTask(taskId, intervalMs = 10000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(      `https://api.cometapi.com/kling/v1/videos/text2video/${taskId}`,      { headers: { 'Authorization': `Bearer ${API_KEY}` } }    );    if (!res.ok) throw new Error(`Kling poll failed: ${res.status}`);    const result = await res.json();    const status = result.data.task_status;​    if (status === 'succeed') return result.data.task_result.videos[0].url;    if (status === 'failed') {      throw new Error(`Kling task ${taskId} failed: ${JSON.stringify(result.data.task_result ?? 'no detail')}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Kling task ${taskId} timed out`);}​// --- Runway (image-to-video) ---async function submitRunwayTask(imageUrl, motionPrompt, duration = 5) {  const res = await fetch('https://api.cometapi.com/runwayml/v1/image_to_video', {    method: 'POST',    headers: {      'Authorization': `Bearer ${API_KEY}`,      'X-Runway-Version': '2024-11-06',      'Content-Type': 'application/json'    },    body: JSON.stringify({      model: 'gen3a_turbo',      promptImage: imageUrl,      promptText: motionPrompt,      duration,      ratio: '1280:720',      watermark: false    })  });  if (!res.ok) throw new Error(`Runway submit failed: ${res.status}`);  return (await res.json()).id;}​async function pollRunwayTask(taskId, intervalMs = 5000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(      `https://api.cometapi.com/runwayml/v1/tasks/${taskId}`,      { headers: { 'Authorization': `Bearer ${API_KEY}`, 'X-Runway-Version': '2024-11-06' } }    );    if (!res.ok) throw new Error(`Runway poll failed: ${res.status}`);    const result = await res.json();    const status = result.status;​    if (status === 'task_not_exist') {      // CometAPI-specific: task still initializing      await new Promise(r => setTimeout(r, intervalMs));      elapsed += intervalMs;      continue;    }    if (status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(status)) {      throw new Error(`Runway task ${taskId} failed: ${result.error ?? 'no detail'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Runway task ${taskId} timed out`);}​// Usage exampleconst taskId = await submitVeoTask('A paper kite drifting above a wheat field');const videoUrl = await pollVeoTask(taskId);console.log('Video ready:', videoUrl);

下一步

你現在已擁有四個影片模型的可用程式碼、一個可處理失敗的輪詢迴圈，以及在失效前下載輸出的步驟，避免遺失生成內容。

多數開發者接下來會遇到的問題是：他們把模型寫死在程式裡，切換到更便宜或更快的選項時需要動到多個檔案。下一篇將介紹如何在不重寫程式碼的情況下在模型間導流。

Next: How to Switch Between AI Models Without Rewriting Your Code

常見問題

Q: 為什麼我在 API 回應中拿到的是任務 ID，而不是影片？

影片生成是非同步的——像 Veo、Sora、Kling 與 Runway 等模型需要 2–5 分鐘渲染。API 會立即返回任務 ID，避免你的請求逾時。你需要輪詢另一個狀態端點，直到任務達到終止狀態（succeeded、succeed、failed）。

Q: 生成的影片 URL 可以維持多久有效？

影片生成 API 的 URL 是暫時性的。取得 URL 後請立即下載並儲存到自己的儲存空間（S3、Cloudflare R2 等）。不要只存 URL 並期待數小時後仍可使用。

Q: Veo 3 Fast 與 Kling Video 有何不同？

Veo 3 Fast 更便宜（$0.05/秒）、更快、呼叫更簡單。Kling Video 給你更多控制：negative_prompt、cfg_scale、鏡頭運動設定與 pro 品質模式。如果你需要精細調整輸出，使用 Kling；如果你需要速度與低成本，使用 Veo 3 Fast。

Q: 我可以用影像而不是文字提示來生成影片嗎？

可以。Veo 支援透過提供 input_reference 檔案進行圖像轉影片。Kling 可透過 /kling/v1/videos/image2video 端點與 image 參數（URL 或 base64）達成。Runway 只支援圖像轉影片——透過 CometAPI 不接受純文字提示。

Q: 為什麼 Runway 在第一次輪詢時會返回 task_not_exist？

這是 CometAPI 文件記載的行為——後端任務仍在初始化。等待幾秒後重試。這不是錯誤。上面的輪詢程式碼已自動處理此情況。

Q: 為什麼 Kling 使用 "succeed" 而不是 "succeeded"？

這是 Kling 的實際 API 回應格式，並非拼寫錯誤。Veo 與 Runway 使用 "succeeded"——Kling 使用 "succeed"。如果你要做統一的輪詢包裝，需要同時處理這兩個字串。

Q: 同步輪詢迴圈在網路伺服器中使用是否安全？

不安全。本指南的輪詢迴圈會阻塞執行緒數分鐘。在有並發使用者的實際服務中，請在背景工作者中執行輪詢（Python 用 Celery、Node.js 用 Bull）。在請求處理器中提交任務、將任務 ID 返回給客戶端，並由工作者在影片就緒時通知客戶端。