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

การเพิ่มการสร้างวิดีโอให้กับแอปของคุณไม่เหมือนกับการเพิ่มการสร้างภาพ การเรียก API จะตอบกลับทันที — แต่วิดีโอยังไม่พร้อม คุณจะได้รับรหัสงาน (task ID) และต้องคอยถามว่า “เสร็จหรือยัง?” ไปเรื่อย ๆ จนกว่าจะเสร็จ

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

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

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

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

• Python 3.8+ หรือ Node.js 18+
• คีย์ CometAPI
• ความคุ้นเคยพื้นฐานกับ REST APIs

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

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

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

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

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

เลือกรุ่นโมเดล

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

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

ส่งงาน 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") — ซึ่งตรงกับรูปแบบการตอบสนองจริงของ 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*:* เอกสาร Kling Video ของ CometAPI

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

Runway รองรับภาพสู่วิดีโอเท่านั้น และต้องมี 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")

Source*:* เอกสาร Runway ของ CometAPI

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

URL วิดีโอจาก API การสร้างเป็นแบบชั่วคราว ดาวน์โหลดไฟล์ทันทีและจัดเก็บไว้ในที่ที่คุณควบคุม:

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

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

Source: เอกสารการสร้างวิดีโอของ 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 ทันทีเพื่อไม่ให้คำขอหมดเวลา คุณต้องโพล endpoint สถานะแยกต่างหากจนกว่างานจะถึงสถานะสิ้นสุด (succeeded, succeed, failed)

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

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

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

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

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

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

ถาม: ทำไม Runway ถึงคืนค่า task_not_exist ในการโพลครั้งแรก?

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

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

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

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

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