Việc thêm khả năng tạo video vào ứng dụng không giống với thêm khả năng tạo ảnh. Lời gọi API trả về ngay — nhưng video thì chưa sẵn sàng. Bạn nhận được một task ID và phải tiếp tục hỏi “xong chưa?” cho đến khi hoàn tất.
Hầu hết nhà phát triển gặp tình huống này lần đầu khi gọi API video, chờ phần thân phản hồi chứa URL video, và nhận về task ID thay vào đó. Hướng dẫn này đi qua toàn bộ luồng: gửi tác vụ, thăm dò kết quả, xử lý lỗi, và lưu trữ đầu ra trước khi URL hết hạn.
Bạn sẽ xây dựng gì
Một dịch vụ backend nhận prompt văn bản hoặc ảnh, gửi tác vụ tạo video, thăm dò đến khi hoàn tất, và trả về URL video cuối cùng. Bạn sẽ làm việc với bốn model — Veo 3 Fast, Sora 2, Kling Video và Runway — tất cả dùng chung một API key.
Điều kiện tiên quyết:
- Python 3.8+ hoặc Node.js 18+
- Khóa CometAPI
- Hiểu biết cơ bản về REST API
Hiểu vì sao tạo video khác biệt
Với tạo ảnh, bạn gửi yêu cầu và nhận ảnh trong cùng một phản hồi. Tạo video dùng hàng đợi tác vụ bất đồng bộ:
- Gửi yêu cầu tạo → nhận lại
task_id - Thăm dò endpoint trạng thái mỗi vài giây
- Khi trạng thái đạt trạng thái kết thúc, bạn nhận URL video
- Tải xuống và lưu trữ video — URL chỉ tạm thời
Nếu bạn xử lý tạo video như tạo ảnh và chờ phản hồi đầu tiên chứa video, yêu cầu của bạn sẽ luôn bị timeout.
Trong dịch vụ web production, vòng thăm dò này nên chạy trong worker nền (Celery, Bull, hoặc tương tự), không phải trong request handler. Ví dụ dưới đây dùng thăm dò đồng bộ — phù hợp cho script và nguyên mẫu, nhưng không dành cho xử lý người dùng đồng thời.
Chọn model
| Model | Nhà cung cấp | Thời lượng tối đa | Giá (qua CometAPI) | Phù hợp nhất cho |
|---|---|---|---|---|
| Veo 3 Fast | 8 sec | $0.05/sec | Dựng mẫu nhanh, clip mạng xã hội | |
| Sora 2 | OpenAI (qua ID model CometAPI) | ~10 sec | $0.08/sec | Short sáng tạo chất lượng cao |
| Kling Video | Kuaishou | 10 sec | $0.13–$2.64/task | Nội dung marketing, kiểm soát chi tiết |
| Runway Gen-3A Turbo | Runway | 5 hoặc 10 sec | $0.32/task | Image-to-video, nội dung thương mại |
Nguồn*:* Trang model CometAPI, tháng 5/2026. Lưu ý: “Sora 2” là* định danh model của CometAPI — tham khảo trang model để biết chi tiết về model nền tảng.
- Veo 3 Fast hỗ trợ cả text-to-video và image-to-video. Rẻ nhất theo giây, điểm khởi đầu tốt.
- Sora 2 tạo âm thanh nội tại cùng với video — hội thoại, âm thanh nền và hiệu ứng mà không cần bước TTS riêng.
- Kling Video cung cấp
negative_prompt,cfg_scale, thiết lập chuyển động camera và chế độpro. Kiểm soát nhiều nhất trong bốn model. - Runway chỉ hỗ trợ image-to-video qua CometAPI. Cung cấp một ảnh tĩnh và mô tả chuyển động, nó sẽ tạo hoạt ảnh.
Gửi tác vụ Veo
Veo dùng multipart/form-data. Hãy dùng files= trong Python requests để gửi đúng — data=dict sẽ gửi application/x-www-form-urlencoded, không phải cùng một thứ:
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}")
Thăm dò kết quả
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}")
Dùng Kling Video để có nhiều quyền kiểm soát hơn
Kling có cấu trúc endpoint khác và dùng JSON. Lưu ý rằng trạng thái kết thúc của Kling là "succeed" (không phải "succeeded") — đúng với định dạng phản hồi của 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")
Nguồn*:* Tài liệu Kling Video của CometAPI
Tạo chuyển động cho ảnh tĩnh bằng Runway
Runway chỉ hỗ trợ image-to-video. Nó cũng yêu cầu một header bổ sung (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")
Nguồn*:* Tài liệu Runway của CometAPI
Lưu video trước khi URL hết hạn
URL video từ các API tạo nội dung chỉ tạm thời. Hãy tải tệp về ngay và lưu ở nơi bạn kiểm soát:
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")
Trong môi trường production, thay ghi tệp cục bộ bằng tải lên S3, Cloudflare R2, hoặc dịch vụ lưu trữ bạn chọn. Mẫu streaming giữ nguyên — chuyển bytes trực tiếp thay vì tải toàn bộ video vào bộ nhớ.
Xử lý lỗi
| Triệu chứng | Nguyên nhân có thể | Cách khắc phục |
|---|---|---|
| Tác vụ kẹt ở queued trên 10 phút | Tải máy chủ cao hoặc model không sẵn sàng | Thử lại với model khác |
| task_not_exist khi thăm dò Runway lần đầu | Tác vụ vẫn đang khởi tạo | Chờ 5 giây và thử lại — hành vi được CometAPI ghi nhận |
| failed mà không có thông báo lỗi | Prompt kích hoạt bộ lọc nội dung | Diễn đạt lại prompt |
| URL video trả về 403 | URL hết hạn trước khi tải | Tải ngay sau khi nhận được URL |
| Timeout sau 10 phút | Quá trình tạo mất quá lâu | Tăng max_wait hoặc chuyển sang Veo 3 Fast |
| Kling trả về "succeed" không phải "succeeded" | API của Kling dùng chuỗi trạng thái không chuẩn | Điều này đúng — xem mã thăm dò Kling ở trên |
Nguồn: Tài liệu tạo video CometAPI
Phiên bản Node.js
Node.js 18+ tích hợp sẵn fetch và FormData. Ví dụ này bao phủ cả bốn 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);
Tiếp theo là gì
Giờ bạn đã có mã hoạt động cho bốn model video, một vòng thăm dò xử lý lỗi, và bước tải xuống giúp bạn không mất nội dung đã tạo.
Vấn đề tiếp theo nhiều nhà phát triển gặp phải: họ đã hardcode một model, và việc chuyển sang tùy chọn rẻ hơn hoặc nhanh hơn đòi hỏi sửa nhiều tệp. Bài viết tiếp theo sẽ hướng dẫn cách định tuyến yêu cầu qua nhiều model mà không phải viết lại mã.
Tiếp theo: Cách Chuyển Đổi Giữa Các Model AI Mà Không Cần Viết Lại Mã
Câu hỏi thường gặp
Hỏi: Tại sao tôi nhận được task ID thay vì video trong phản hồi API?
Tạo video là bất đồng bộ — các model như Veo, Sora, Kling, và Runway mất 2–5 phút để render. API trả về task ID ngay để tránh timeout. Bạn thăm dò endpoint trạng thái cho đến khi tác vụ đạt trạng thái kết thúc (succeeded, succeed, failed).
Hỏi: URL video được tạo có hiệu lực trong bao lâu?
URL video từ các API tạo nội dung chỉ tạm thời. Hãy tải tệp về ngay sau khi nhận URL và lưu vào kho của bạn (S3, Cloudflare R2, v.v.). Đừng lưu URL và kỳ vọng nó hoạt động sau vài giờ.
Hỏi: Khác biệt giữa Veo 3 Fast và Kling Video là gì?
Veo 3 Fast rẻ hơn ($0.05/sec), nhanh hơn và gọi đơn giản hơn. Kling Video cho phép bạn kiểm soát nhiều hơn: negative_prompt, cfg_scale, thiết lập chuyển động camera và chế độ chất lượng pro. Nếu bạn cần tinh chỉnh đầu ra, hãy dùng Kling. Nếu bạn cần tốc độ và chi phí thấp, hãy dùng Veo 3 Fast.
Hỏi: Tôi có thể tạo video từ ảnh thay vì prompt văn bản không?
Có. Veo hỗ trợ image-to-video bằng cách truyền một tệp input_reference. Kling hỗ trợ qua endpoint /kling/v1/videos/image2video với tham số image (URL hoặc base64). Runway chỉ hỗ trợ image-to-video — không chấp nhận prompt chỉ văn bản qua CometAPI.
Hỏi: Tại sao Runway trả về task_not_exist ở lần thăm dò đầu tiên?
Đây là hành vi được CometAPI ghi nhận — tác vụ vẫn đang khởi tạo ở backend. Chờ vài giây và thử lại. Đây không phải lỗi. Mã thăm dò ở trên đã xử lý tự động.
Hỏi: Tại sao Kling dùng "succeed" thay vì "succeeded"?
Đó là định dạng phản hồi thực tế của API Kling. Không phải lỗi chính tả. Veo và Runway dùng "succeeded" — Kling dùng "succeed". Nếu bạn xây dựng wrapper thăm dò thống nhất, bạn cần xử lý cả hai chuỗi.
Hỏi: Vòng thăm dò đồng bộ có an toàn để dùng trong web server không?
Không. Vòng thăm dò trong hướng dẫn này chặn luồng trong vài phút. Trong dịch vụ web thực tế với người dùng đồng thời, hãy chạy thăm dò trong worker nền (Celery cho Python, Bull cho Node.js). Gửi tác vụ trong request handler, trả task ID cho client, và để worker thông báo client khi video sẵn sàng.