Cara Menambah Penjanaan Video AI ke dalam Aplikasi SaaS

Menambah penjanaan video pada aplikasi anda tidak sama dengan menambah penjanaan imej. Panggilan API memulangkan respons serta-merta — tetapi video belum siap. Anda akan mendapat ID tugasan, dan anda perlu terus bertanya “sudah siap belum?” sehingga ia siap.

Kebanyakan pembangun mengalami perkara ini kali pertama mereka memanggil API video, menunggu badan respons dengan URL video, tetapi menerima ID tugasan sebagai gantinya. Panduan ini menerangkan keseluruhan aliran: menghantar tugasan, meninjau (poll) hasil, mengendalikan kegagalan, dan menyimpan output sebelum URL tamat tempoh.

Apa yang anda akan bina

Satu perkhidmatan backend yang menerima gesaan teks atau imej, menghantar tugasan penjanaan video, meninjau sehingga siap, dan memulangkan URL video akhir. Anda akan bekerja dengan empat model — Veo 3 Fast, Sora 2, Kling Video, dan Runway — semuanya menggunakan satu kunci API.

Prasyarat:

• Python 3.8+ atau Node.js 18+
• Kunci CometAPI
• Kefahaman asas tentang REST API

Fahami mengapa penjanaan video berbeza

Dengan penjanaan imej, anda menghantar permintaan dan menerima imej dalam respons yang sama. Penjanaan video menggunakan baris gilir tugasan async:

1. Hantar permintaan penjanaan → dapatkan task_id
2. Tinjau titik akhir status setiap beberapa saat
3. Apabila status mencapai keadaan terminal, anda akan mendapat URL video
4. Muat turun dan simpan video — URL adalah sementara

Jika anda melayan penjanaan video seperti penjanaan imej dan menunggu respons pertama mengandungi video anda, permintaan anda akan tamat masa setiap kali.

Dalam perkhidmatan web produksi, gelung tinjauan ini patut berjalan dalam pekerja latar (Celery, Bull, atau seumpamanya), bukan dalam pengendali permintaan anda. Contoh di bawah menggunakan tinjauan segerak — sesuai untuk skrip dan prototaip, tetapi tidak untuk mengendalikan pengguna serentak.

Pilih model

Sumber**: Halaman model CometAPI, Mei 2026. Nota: "Sora 2" ialah pengenal pasti model CometAPI — rujuk halaman model mereka untuk butiran model asas.

• Veo 3 Fast menyokong kedua-dua teks-ke-video dan imej-ke-video. Paling murah per saat, titik permulaan yang baik.
• Sora 2 menjana audio secara asli bersama video — dialog, bunyi ambien, dan efek tanpa langkah TTS berasingan.
• Kling Video memberikan anda negative_prompt, cfg_scale, tetapan pergerakan kamera, dan mod pro. Kawalan paling banyak antara empat.
• Runway hanya imej-ke-video melalui CometAPI. Berikan imej statik dan keterangan gerakan, dan ia akan menganimasikannya.

Hantar tugasan Veo

Veo menggunakan multipart/form-data. Gunakan files= dalam requests Python untuk menghantarnya dengan betul — data=dict menghantar application/x-www-form-urlencoded, yang tidak sama:

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}")

Tinjau keputusan

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}")

Guna Kling Video untuk lebih kawalan

Kling mempunyai struktur titik akhir berbeza dan menggunakan JSON. Ambil perhatian bahawa rentetan status terminal Kling ialah "succeed" (bukan "succeeded") — ini sepadan dengan format respons API sebenar:

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")

Sumber**: Dokumen Kling Video CometAPI

Animasi imej statik dengan Runway

Runway hanya menyokong imej-ke-video. Ia juga memerlukan pengepala tambahan (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")

Sumber**: Dokumen Runway CometAPI

Simpan video sebelum URL tamat tempoh

URL video daripada API penjanaan adalah sementara. Muat turun fail serta-merta dan simpan di tempat yang anda kawal:

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")

Dalam produksi, gantikan penulisan fail tempatan dengan muat naik ke S3, Cloudflare R2, atau storan pilihan anda. Corak penstriman kekal sama — alirkan bait terus tanpa memuatkan keseluruhan video ke dalam memori.

Tangani kegagalan

Sumber: Dokumen penjanaan video CometAPI

Versi Node.js

Node.js 18+ menyertakan fetch dan FormData secara asli. Contoh ini merangkumi keempat-empat model:

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

Langkah seterusnya

Anda kini mempunyai kod berfungsi untuk empat model video, gelung tinjauan yang mengendalikan kegagalan, dan langkah muat turun yang mengelakkan anda kehilangan kandungan terjana.

Masalah seterusnya yang sering dihadapi pembangun: mereka telah mengkodkan tetap satu model, dan beralih kepada pilihan lebih murah atau lebih pantas memerlukan pengubahsuaian pada berbilang fail. Artikel seterusnya menerangkan cara menghala tuju permintaan merentas model tanpa menulis semula kod anda.

Seterusnya: Cara Bertukar Antara Model AI Tanpa Menulis Semula Kod Anda

Soalan Lazim

S: Mengapa saya menerima ID tugasan bukannya video dalam respons API?

Penjanaan video adalah async — model seperti Veo, Sora, Kling, dan Runway mengambil masa 2–5 minit untuk merender. API memulangkan ID tugasan serta-merta supaya permintaan anda tidak tamat masa. Anda meninjau titik akhir status berasingan sehingga tugasan mencapai keadaan terminal (succeeded, succeed, failed).

S: Berapa lama URL video terjana kekal sah?

URL video daripada API penjanaan adalah sementara. Muat turun fail serta-merta selepas mendapat URL dan simpan dalam storan anda sendiri (S3, Cloudflare R2, dsb.). Jangan simpan URL dan mengharapkan ia berfungsi beberapa jam kemudian.

S: Apakah perbezaan antara Veo 3 Fast dan Kling Video?

Veo 3 Fast lebih murah ($0.05/saat), lebih pantas, dan lebih ringkas untuk dipanggil. Kling Video memberikan lebih kawalan: negative_prompt, cfg_scale, tetapan pergerakan kamera, dan mod pro berkualiti. Jika anda perlu melaras hasil dengan halus, gunakan Kling. Jika anda perlukan kelajuan dan kos rendah, gunakan Veo 3 Fast.

S: Bolehkah saya menjana video daripada imej dan bukannya gesaan teks?

Ya. Veo menyokong imej-ke-video dengan menghantar fail input_reference. Kling menyokongnya melalui titik akhir /kling/v1/videos/image2video dengan parameter image (URL atau base64). Runway hanya imej-ke-video — ia tidak menerima gesaan teks sahaja melalui CometAPI.

S: Mengapa Runway memulangkan task_not_exist pada tinjauan pertama?

Ini tingkah laku CometAPI yang didokumenkan — tugasan masih memulakan proses di backend. Tunggu beberapa saat dan cuba semula. Ini bukan ralat. Kod tinjauan di atas mengendalikannya secara automatik.

S: Mengapa Kling menggunakan "succeed" dan bukan "succeeded"?

Itulah format respons API sebenar Kling. Ia bukan salah taip. Veo dan Runway menggunakan "succeeded" — Kling menggunakan "succeed". Jika anda membina pembungkus tinjauan bersatu, anda perlu mengendalikan kedua-dua rentetan ini.

S: Adakah gelung tinjauan segerak selamat digunakan dalam pelayan web?

Tidak. Gelung tinjauan dalam panduan ini menyekat thread selama beberapa minit. Dalam perkhidmatan web sebenar dengan pengguna serentak, jalankan tinjauan dalam pekerja latar (Celery untuk Python, Bull untuk Node.js). Hantar tugasan dalam pengendali permintaan, pulangkan ID tugasan kepada klien, dan biarkan pekerja memaklumkan klien apabila video sedia.