如何將 AI 影片生成功能加入 SaaS 應用程式

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

多數開發者第一次呼叫影片 API 時，會等待回應內直接附上影片 URL，結果卻拿到任務 ID。本文將帶你完整走一遍流程：提交任務、輪詢結果、處理失敗情況，以及在 URL 過期前儲存輸出。

你將建置的內容

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

先決條件：

• 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);

接下來做什麼

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

多數開發者接下來遇到的問題是：他們把模型硬編碼了，切換到更便宜或更快的選項需要改動多個檔案。下一篇將介紹如何在不重寫程式碼的情況下，跨模型路由請求。

下一篇：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/sec）、更快、呼叫更簡單。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 回傳給客戶端，並讓背景工作在影片就緒時通知客戶端。