Menambah penjanaan video pada aplikasi anda tidak sama seperti menambah penjanaan imej. Panggilan API memulangkan respons serta-merta — tetapi videonya belum siap. Anda akan menerima ID tugas, dan anda perlu terus bertanya "sudah siap?" sehingga siap.
Kebanyakan pembangun mengalami perkara ini pada kali pertama mereka memanggil API video, menunggu badan respons dengan URL video, tetapi menerima ID tugas sebaliknya. Panduan ini menerangkan keseluruhan aliran: menghantar tugas, polling untuk keputusan, mengendalikan kegagalan, dan menyimpan output sebelum URL tamat tempoh.
Apa yang akan anda bina
Satu perkhidmatan backend yang menerima prompt teks atau imej, menghantar tugas penjanaan video, memantau sehingga selesai, dan memulangkan URL video akhir. Anda akan menggunakan 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 barisan tugas asinkron:
- Hantar permintaan penjanaan → menerima
task_id - Poll endpoint status setiap beberapa saat
- Apabila status mencapai keadaan terminal, anda akan menerima URL video
- Muat turun dan simpan video — URL adalah sementara
Jika anda memperlakukan penjanaan video seperti penjanaan imej dan menunggu respons pertama mengandungi video, permintaan anda akan tamat masa setiap kali.
Dalam perkhidmatan web produksi, gelung polling ini patut berjalan dalam pekerja latar (Celery, Bull, atau seumpamanya), bukan dalam pengendali permintaan anda. Contoh di bawah menggunakan polling segerak — sesuai untuk skrip dan prototaip, tetapi bukan untuk menangani pengguna serentak.
Pilih model
| Model | Penyedia | Tempoh maksimum | Harga (melalui CometAPI) | Terbaik untuk |
|---|---|---|---|---|
| Veo 3 Fast | 8 saat | $0.05/saat | Pembuatan prototaip pantas, klip sosial | |
| Sora 2 | OpenAI (melalui ID model CometAPI) | ~10 saat | $0.08/saat | Video pendek kreatif berkualiti tinggi |
| Kling Video | Kuaishou | 10 saat | $0.13–$2.64/tugas | Kandungan pemasaran, kawalan terperinci |
| Runway Gen-3A Turbo | Runway | 5 atau 10 saat | $0.32/tugas | Imej-ke-video, kandungan komersial |
Source**: Halaman model CometAPI, Mei 2026. Nota: "Sora 2" ialah pengenal model CometAPI — rujuk halaman model mereka untuk butiran model asas.
- Veo 3 Fast menyokong teks-ke-video dan imej-ke-video. Paling murah per saat, titik mula yang baik.
- Sora 2 menjana audio secara natif bersama video — dialog, bunyi ambien, dan kesan tanpa langkah TTS berasingan.
- Kling Video menyediakan
negative_prompt,cfg_scale, tetapan pergerakan kamera, dan modpro. Paling banyak kawalan antara empat ini. - Runway melalui CometAPI adalah imej-ke-video sahaja. Beri imej statik dan penerangan pergerakan, dan ia akan menganimaskannya.
Hantar tugas Veo
Veo menggunakan multipart/form-data. Gunakan files= dalam Python requests untuk menghantarnya dengan betul — data=dict akan menghantar application/x-www-form-urlencoded yang tidak sama:
import requestsimport osfrom dotenv import load_dotenvload_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}")
Polling untuk keputusan
import timedef 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 kawalan lebih terperinci
Kling mempunyai struktur endpoint yang berbeza dan menggunakan JSON. Perhatikan 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")
Source**: Dokumen Kling Video CometAPI
Animasi imej statik dengan Runway
Runway adalah imej-ke-video sahaja. Ia juga memerlukan header 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")
Source**: 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 pathlibdef 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 setempat dengan muat naik ke S3, Cloudflare R2, atau storan pilihan anda. Corak penstriman kekal sama — salurkan bait secara terus dan bukannya memuatkan keseluruhan video ke dalam memori.
Tangani kegagalan
| Simptom | Punca berkemungkinan | Penyelesaian |
|---|---|---|
| Tugas tersekat dalam queued selama 10+ min | Beban pelayan atau model tidak tersedia | Cuba semula dengan model berbeza |
| task_not_exist pada polling Runway pertama | Tugas masih dalam proses permulaan | Tunggu 5 saat dan cuba semula — tingkah laku CometAPI yang didokumenkan |
| gagal tanpa mesej ralat | Prompt mencetuskan penapis kandungan | Ubah cara penulisan prompt |
| URL video memulangkan 403 | URL tamat tempoh sebelum muat turun | Muat turun serta-merta selepas memperoleh URL |
| Tamat masa selepas 10 min | Penjanaan mengambil masa terlalu lama | Tingkatkan max_wait atau beralih ke Veo 3 Fast |
| Kling mengembalikan "succeed" bukan "succeeded" | API Kling menggunakan rentetan status tidak piawai | Ini betul — lihat kod polling Kling di atas |
Sumber: Dokumen penjanaan video CometAPI
Versi Node.js
Node.js 18+ menyertakan fetch dan FormData secara natif. Contoh ini merangkumi keempat-empat model:
// Node.js 18+ — no extra packages neededconst 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 polling yang mengendalikan kegagalan, dan langkah muat turun yang mengelakkan anda kehilangan kandungan yang dijana.
Masalah seterusnya yang sering dihadapi pembangun: mereka mengeraskan satu model, dan menukar kepada pilihan yang lebih murah atau pantas memerlukan pengubahsuaian berbilang fail. Artikel seterusnya membincangkan cara menghala permintaan merentasi model tanpa menulis semula kod anda.
Seterusnya: Cara Bertukar Antara Model AI Tanpa Menulis Semula Kod Anda
Soalan Lazim
S: Mengapa saya menerima ID tugas dan bukannya video dalam respons API?
J: Penjanaan video adalah asinkron — model seperti Veo, Sora, Kling, dan Runway mengambil masa 2–5 minit untuk render. API memulangkan ID tugas serta-merta supaya permintaan anda tidak tamat masa. Anda akan melakukan polling pada endpoint status berasingan sehingga tugas mencapai keadaan terminal (succeeded, succeed, failed).
S: Berapa lama URL video yang dijana kekal sah?
J: URL video daripada API penjanaan adalah sementara. Muat turun fail serta-merta selepas memperoleh 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?
J: Veo 3 Fast lebih murah ($0.05/saat), lebih pantas, dan lebih ringkas untuk dipanggil. Kling Video memberikan lebih banyak kawalan: negative_prompt, cfg_scale, tetapan pergerakan kamera, dan mod kualiti pro. Jika anda perlu menala hasil dengan lebih halus, gunakan Kling. Jika anda perlukan kelajuan dan kos rendah, gunakan Veo 3 Fast.
S: Bolehkah saya menjana video daripada imej dan bukannya prompt teks?
J: Ya. Veo menyokong imej-ke-video dengan menghantar fail input_reference. Kling menyokongnya melalui endpoint /kling/v1/videos/image2video dengan parameter image (URL atau base64). Runway adalah imej-ke-video sahaja — ia tidak menerima prompt teks sahaja melalui CometAPI.
S: Mengapa Runway memulangkan task_not_exist pada polling pertama?
J: Ini adalah tingkah laku CometAPI yang didokumenkan — tugas masih dalam proses permulaan di backend. Tunggu beberapa saat dan cuba semula. Ini bukan ralat. Kod polling di atas mengendalikannya secara automatik.
S: Mengapa Kling menggunakan "succeed" dan bukan "succeeded"?
J: Itulah format respons API Kling sebenar. Ia bukan kesilapan taip. Veo dan Runway menggunakan "succeeded" — Kling menggunakan "succeed". Jika anda membina pembungkus polling bersatu, anda perlu mengendalikan kedua-dua rentetan.
S: Adakah gelung polling segerak selamat digunakan dalam pelayan web?
J: Tidak. Gelung polling dalam panduan ini menyekat thread selama beberapa minit. Dalam perkhidmatan web sebenar dengan pengguna serentak, jalankan polling dalam pekerja latar (Celery untuk Python, Bull untuk Node.js). Hantar tugas dalam pengendali permintaan, pulangkan ID tugas kepada klien, dan biarkan pekerja memaklumkan klien apabila video sedia.