Cách thêm tính năng tạo video bằng AI vào ứng dụng SaaS

Việc thêm tính năng tạo video vào ứng dụng của bạn không giống với việc thêm tính năng tạo ảnh. Lời gọi API sẽ trả về ngay lập tức — nhưng video thì chưa sẵn sàng. Bạn nhận được một ID tác vụ và bạn phải liên 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 một API video, chờ phần nội dung phản hồi có URL video, nhưng lại nhận về một ID tác vụ. Hướng dẫn này sẽ đ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 dạng văn bản hoặc hình ảnh, gửi tác vụ tạo video, thăm dò cho đế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 mô hình — Veo 3 Fast, Sora 2, Kling Video và Runway — tất cả thông qua một khóa API duy nhất.

Prerequisites:

• Python 3.8+ hoặc Node.js 18+
• Một 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 ngay trong cùng phản hồi. Tạo video sử dụng hàng đợi tác vụ bất đồng bộ:

1. Gửi yêu cầu tạo → nhận về task_id
2. Thăm dò endpoint trạng thái mỗi vài giây
3. Khi trạng thái đạt tới trạng thái kết thúc, bạn nhận được URL video
4. Tải xuống và lưu trữ video — URL chỉ tạm thời

Nếu bạn xử lý tạo video giống 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 hết thời gian chờ.

Trong dịch vụ web production, vòng lặp thăm dò này nên chạy trong một worker nền (Celery, Bull hoặc tương tự), không phải trong handler của yêu cầu. Các ví dụ bên dưới dùng thăm dò đồng bộ — phù hợp cho script và nguyên mẫu, nhưng không phù hợp khi phục vụ người dùng đồng thời.

Chọn mô hình

Source**: Trang mô hình CometAPI, Tháng 5/2026. Lưu ý: "Sora 2" là mã nhận dạng mô hình của CometAPI — tham khảo trang mô hình của họ để biết chi tiết mô hình nền.

• Veo 3 Fast hỗ trợ cả văn bản thành video và hình ảnh thành video. Rẻ nhất theo giây, điểm khởi đầu tốt.
• Sora 2 tạo âm thanh nguyên bản 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, cài đặt chuyển động camera và chế độ pro. Kiểm soát nhiều nhất trong bốn mô hình.
• Runway qua CometAPI chỉ hỗ trợ image-to-video. Cung cấp một ảnh tĩnh và mô tả chuyển động, hệ thống sẽ tạo chuyển động cho ảnh.

Gửi một tác vụ Veo

Veo dùng multipart/form-data. Dùng files= trong thư viện requests của Python để gửi đúng — data=dict gửi application/x-www-form-urlencoded, không giống nhau:

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

Thăm dò kết quả

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

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 ý chuỗi trạng thái kết thúc của Kling là "succeed" (không phải "succeeded") — khớp 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")

Source**: Tài liệu CometAPI Kling Video

Tạo chuyển động cho ảnh tĩnh với 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")

Source**: Tài liệu CometAPI Runway

Lưu video trước khi URL hết hạn

URL video từ các API tạo nội dung là tạm thời. Hãy tải tệp ngay lập tức và lưu ở nơi bạn kiểm soát:

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

Trong môi trường production, thay việc ghi tệp cục bộ bằng tải lên S3, Cloudflare R2 hoặc bộ nhớ bạn chọn. Mẫu streaming vẫn như cũ — truyền byte trực tiếp thay vì nạp toàn bộ video vào bộ nhớ.

Xử lý lỗi

Source: 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 mô hình:

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

Tiếp theo

Giờ bạn đã có mã hoạt động cho bốn mô hình video, một vòng lặp thăm dò xử lý lỗi và bước tải xuống giúp bạn không bị mất nội dung đã tạo.

Vấn đề tiếp theo mà nhiều nhà phát triển gặp phải: họ hardcode một mô hình, và việc chuyển sang lựa chọn rẻ hơn hoặc nhanh hơn buộc phải chỉnh nhiều tệp. Bài tiếp theo sẽ hướng dẫn cách định tuyến yêu cầu giữa các mô hình mà không phải viết lại mã.

Tiếp theo: How to Switch Between AI Models Without Rewriting Your Code

Câu hỏi thường gặp

Q: Tại sao tôi nhận được ID tác vụ thay vì video trong phản hồi API?

Tạo video là bất đồng bộ — các mô hình như Veo, Sora, Kling và Runway mất 2–5 phút để render. API trả về ID tác vụ ngay để yêu cầu của bạn không bị hết thời gian chờ. Bạn thăm dò một endpoint trạng thái riêng cho đến khi tác vụ đạt trạng thái kết thúc (succeeded, succeed, failed).

Q: 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 là tạm thời. Hãy tải tệp ngay sau khi nhận được URL và lưu vào bộ nhớ của bạn (S3, Cloudflare R2, v.v.). Đừng lưu URL và kỳ vọng nó còn hoạt động sau vài giờ.

Q: Khác biệt giữa Veo 3 Fast và Kling Video là gì?

Veo 3 Fast rẻ hơn ($0.05/giây), nhanh hơn và dễ gọi hơn. Kling Video cho bạn nhiều điều chỉnh hơn: negative_prompt, cfg_scale, cài đặt chuyển động camera và chế độ chất lượng pro. Nếu cần tinh chỉnh đầu ra, dùng Kling. Nếu cần tốc độ và chi phí thấp, dùng Veo 3 Fast.

Q: Tôi có thể tạo video từ hình ả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 nhận prompt chỉ văn bản qua CometAPI.

Q: Tại sao Runway trả về task_not_exist ở lần thăm dò đầu?

Đâ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ò bên trên đã xử lý tự động.

Q: 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 một wrapper thăm dò thống nhất, bạn cần xử lý cả hai chuỗi.

Q: Vòng lặp thăm dò đồng bộ có an toàn để dùng trong web server không?

Không. Vòng lặp trong hướng dẫn này chặn thread 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ả về ID tác vụ cho client và để worker thông báo khi video sẵn sàng.