วิธีเพิ่มฟีเจอร์สร้างวิดีโอด้วย AI ให้กับแอป SaaS

การเพิ่มความสามารถสร้างวิดีโอให้แอปของคุณไม่เหมือนกับการเพิ่มการสร้างรูปภาพ คำเรียก API จะส่งผลลัพธ์กลับทันที — แต่ตัววิดีโอยังไม่พร้อมใช้งาน คุณจะได้รับ task ID และต้องคอยถามซ้ำว่า "เสร็จหรือยัง?" จนกว่าจะเสร็จ

นักพัฒนาส่วนใหญ่จะเจอเรื่องนี้ตั้งแต่ครั้งแรกที่เรียกใช้วิดีโอ API โดยรอให้มีวิดีโอ URL ใน response body แต่กลับได้ task ID แทน คู่มือนี้จะพาคุณทำครบทุกขั้นตอน: ส่งงาน, โพลผลลัพธ์, จัดการความล้มเหลว และจัดเก็บเอาต์พุตก่อนที่ URL จะหมดอายุ

สิ่งที่คุณจะสร้าง

บริการแบ็กเอนด์ที่รับพรอมป์ข้อความหรือรูปภาพ ส่งงานสร้างวิดีโอ โพลจนกว่างานจะเสร็จ และส่งคืนวิดีโอ URL สุดท้าย คุณจะทำงานกับสี่โมเดล — Veo 3 Fast, Sora 2, Kling Video และ Runway — ทั้งหมดผ่านคีย์ API เดียว

ข้อกำหนดเบื้องต้น:

• Python 3.8+ หรือ Node.js 18+
• คีย์ CometAPI
• พื้นฐานการใช้งาน REST APIs

ทำความเข้าใจว่าทำไมการสร้างวิดีโอจึงต่างออกไป

สำหรับการสร้างรูปภาพ คุณส่งคำขอแล้วได้รูปกลับมาใน response เดียว การสร้างวิดีโอใช้คิวงานแบบอะซิงก์:

1. ส่งคำขอสร้างวิดีโอ → ได้กลับมาเป็น task_id
2. โพล endpoint สถานะทุกไม่กี่วินาที
3. เมื่อสถานะไปถึงสถานะสุดท้าย คุณจะได้วิดีโอ URL
4. ดาวน์โหลดและจัดเก็บวิดีโอ — URL มีอายุชั่วคราว

ถ้าคุณปฏิบัติกับการสร้างวิดีโอเหมือนการสร้างรูปภาพและรอวิดีโอใน response แรก คำขอของคุณจะ time out ทุกครั้ง

ในบริการเว็บระดับโปรดักชัน วงจรการโพลนี้ควรทำงานในแบ็กกราวด์เวิร์กเกอร์ (Celery, Bull หรือคล้ายกัน) ไม่ใช่ใน request handler ตัวอย่างด้านล่างใช้การโพลแบบซิงก์ — ใช้ได้กับสคริปต์และต้นแบบ แต่ไม่เหมาะสำหรับผู้ใช้พร้อมกันจำนวนมาก

เลือกโมเดล

แหล่งที่มา**: หน้าโมเดลของ CometAPI, พฤษภาคม 2026 หมายเหตุ: "Sora 2" เป็น identifier ของโมเดลใน CometAPI — ดูรายละเอียดโมเดลพื้นฐานได้ที่ model page ของพวกเขา

• Veo 3 Fast รองรับทั้ง text-to-video และ image-to-video ถูกที่สุดต่อวินาที เหมาะเริ่มต้น
• Sora 2 สร้างเสียงในตัวพร้อมกับวิดีโอ — บทสนทนา เสียงบรรยากาศ และเอฟเฟ็กต์ โดยไม่ต้องมีขั้นตอน TTS แยกต่างหาก
• Kling Video ให้คุณตั้งค่า negative_prompt, cfg_scale, การเคลื่อนกล้อง และโหมด pro ควบคุมได้มากที่สุดในทั้งสี่
• Runway ผ่าน CometAPI รองรับเฉพาะ image-to-video ป้อนภาพนิ่งและคำอธิบายการเคลื่อนไหว แล้วมันจะทำให้ภาพนั้นขยับ

ส่งงาน Veo

Veo ใช้ multipart/form-data ใช้ files= ใน Python requests เพื่อส่งให้ถูกต้อง — 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 มีโครงสร้าง endpoint ที่ต่างออกไปและใช้ JSON โปรดสังเกตว่าสถานะสุดท้ายของ Kling คือ "succeed" (ไม่ใช่ "succeeded") — ตรงตามรูปแบบ response ของ 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")

แหล่งที่มา**: เอกสาร CometAPI Kling Video

ทำให้ภาพนิ่งเคลื่อนไหวด้วย Runway

Runway รองรับเฉพาะ image-to-video นอกจากนี้ยังต้องใช้ header เพิ่มเติม (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")

แหล่งที่มา**: เอกสาร CometAPI Runway

บันทึกวิดีก่อนที่ URL จะหมดอายุ

วิดีโอ 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 หรือสตอเรจที่คุณเลือก รูปแบบการสตรีมยังเหมือนเดิม — pipe ไบต์โดยตรงแทนการโหลดวิดีโอทั้งก้อนเข้าเมมโมรี

จัดการความล้มเหลว

แหล่งที่มา: เอกสาร CometAPI สำหรับการสร้างวิดีโอ

เวอร์ชัน 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);

ถัดไปคืออะไร

ตอนนี้คุณมีโค้ดที่ใช้งานได้สำหรับสี่โมเดล วงจรการโพลที่จัดการความล้มเหลว และขั้นตอนดาวน์โหลดที่ช่วยให้ไม่สูญเสียคอนเทนต์ที่สร้างขึ้น

ปัญหาต่อไปที่นักพัฒนาส่วนใหญ่เจอ: ผูกติดกับโมเดลเดียวแบบฮาร์ดโค้ด และการสลับไปตัวเลือกที่ถูกหรือเร็วกว่า ทำให้ต้องแก้หลายไฟล์ บทความถัดไปจะอธิบายวิธีส่งคำขอข้ามหลายโมเดลโดยไม่ต้องเขียนโค้ดใหม่

ถัดไป: วิธีสลับระหว่างโมเดล AI โดยไม่ต้องเขียนโค้ดใหม่

คำถามที่พบบ่อย

ถาม: ทำไมฉันถึงได้ task ID แทนวิดีโอใน API response?

การสร้างวิดีโอเป็นอะซิงก์ — โมเดลอย่าง Veo, Sora, Kling และ Runway ใช้เวลาประมาณ 2–5 นาทีในการเรนเดอร์ API จึงส่ง task ID ทันทีเพื่อไม่ให้คำขอ time out คุณต้องโพลที่ endpoint สถานะแยกต่างหากจนกว่างานจะเข้าสู่สถานะสุดท้าย (succeeded, succeed, failed)

ถาม: วิดีโอ URL ที่สร้างมาจะใช้งานได้นานแค่ไหน?

วิดีโอ URL จากบริการสร้างวิดีโอมีอายุชั่วคราว ดาวน์โหลดไฟล์ทันทีหลังได้รับ URL แล้วเก็บไว้ในสตอเรจของคุณเอง (S3, Cloudflare R2 ฯลฯ) อย่าเก็บ URL ไว้แล้วคาดว่าจะยังใช้ได้ผ่านไปหลายชั่วโมง

ถาม: ความแตกต่างระหว่าง Veo 3 Fast กับ Kling Video คืออะไร?

Veo 3 Fast ถูกกว่า ($0.05/sec) เร็วกว่า และเรียกใช้ง่ายกว่า Kling Video ให้การควบคุมมากกว่า: negative_prompt, cfg_scale, การเคลื่อนกล้อง และโหมดคุณภาพ pro หากต้องการจูนผลลัพธ์อย่างละเอียด ใช้ Kling หากต้องการความเร็วและต้นทุนต่ำ ใช้ Veo 3 Fast

ถาม: ฉันสามารถสร้างวิดีโอจากรูปภาพแทนพรอมป์ข้อความได้ไหม?

ได้ Veo รองรับ image-to-video โดยส่งไฟล์ input_reference Kling รองรับผ่าน endpoint /kling/v1/videos/image2video ด้วยพารามิเตอร์ image (URL หรือ base64) Runway รองรับเฉพาะ image-to-video — ไม่รับพรอมป์ข้อความล้วนผ่าน CometAPI

ถาม: ทำไม Runway ส่งกลับ task_not_exist ตอนโพลครั้งแรก?

นี่เป็นพฤติกรรมที่ CometAPI ระบุไว้ — งานยังอยู่ระหว่างเริ่มต้นที่แบ็กเอนด์ รอสักไม่กี่วินาทีแล้วลองใหม่ ไม่ใช่ข้อผิดพลาด โค้ดโพลด้านบนจัดการให้อัตโนมัติแล้ว

ถาม: ทำไม Kling ใช้ "succeed" แทน **"succeeded"?

เพราะเป็นรูปแบบ response ของ API ของ Kling จริง ไม่ใช่พิมพ์ผิด Veo และ Runway ใช้ "succeeded" — ส่วน Kling ใช้ "succeed" ถ้าคุณสร้างตัวห่อโพลแบบรวม จะต้องรองรับทั้งสองสตริง

ถาม: วงจรการโพลแบบซิงก์ปลอดภัยสำหรับใช้บนเว็บเซิร์ฟเวอร์ไหม?

ไม่ วงจรการโพลในคู่มือนี้บล็อกเธรดเป็นนาทีๆ ในบริการเว็บจริงที่มีผู้ใช้พร้อมกันจำนวนมาก ให้รันการโพลในแบ็กกราวด์เวิร์กเกอร์ (Celery สำหรับ Python, Bull สำหรับ Node.js) ส่งงานใน request handler ส่งคืน task ID ให้ไคลเอนต์ และให้เวิร์กเกอร์แจ้งไคลเอนต์เมื่อวิดีโอพร้อมใช้งาน