OpenMontage 中 HeyGen 視頻狀態(tài)輪詢實戰(zhàn)從 poll 模式到斷點續(xù)查的完整實現(xiàn)【免費下載鏈接】OpenMontageWorlds first open-source, agentic video production system. 12 production pipelines, 100 tools, 700 agent skill and production-knowledge files. Turn your AI coding assistant into a full video production studio.項目地址: https://gitcode.com/GitHub_Trending/op/OpenMontage本文圍繞 OpenMontage 倉庫中 HeyGencreate-video技能的核心參考文檔 video-status.md 展開系統(tǒng)講解 HeyGen 異步視頻生成的狀態(tài)輪詢機(jī)制狀態(tài)類型語義、生成耗時預(yù)估、completed/failed兩種響應(yīng)結(jié)構(gòu)、帶進(jìn)度回調(diào)的輪詢實現(xiàn)、帶指數(shù)退避的下載重試以及適合長任務(wù)的斷點續(xù)查模式。讀完本篇你可以為任何接入 HeyGen API或其他異步媒體生成服務(wù)的流程編寫生產(chǎn)級輪詢邏輯并理解 OpenMontage 工具層heygen_video是如何在源碼中落地同一套輪詢策略的。1. 背景為什么狀態(tài)輪詢是異步視頻生成的必經(jīng)環(huán)節(jié)HeyGen 的視頻生成是完全異步的提交請求后服務(wù)端只返回一個video_id真正的渲染在后臺排隊執(zhí)行?？蛻舳吮仨毞磸?fù)查詢狀態(tài)接口直到拿到video_url或確認(rèn)失敗。這正是 SKILL.md 中Default Workflow的第 3 步——Callmcp__heygen__get_videowith the returned video_id to poll status and get the download URL。文檔給出了兩條查詢路徑MCP 工具首選若 HeyGen MCP 服務(wù)器已連接直接使用mcp__heygen__get_video并傳入videoId參數(shù)。它一次性返回 status、video_url、thumbnail_url、duration、title、gif_url、captioned_video_url等全部元數(shù)據(jù)直接調(diào)用 REST APIGET /v2/videos/{video_id}需要自行處理狀態(tài)機(jī)與重試。SKILL.md的Tool Selection表格進(jìn)一步明確了這一優(yōu)先級有mcp__heygen__*工具時優(yōu)先用它自動處理鑒權(quán)與請求格式?jīng)]有時才回退到裸 HTTP 調(diào)用。1.1 查詢接口的最小實現(xiàn)文檔給出三種語言的直接調(diào)用示例。curl 版本如下注意鑒權(quán)走X-Api-Key請求頭key 來自環(huán)境變量HEYGEN_API_KEYcurl -X GET https://api.heygen.com/v2/videos/YOUR_VIDEO_ID \ -H X-Api-Key: $HEYGEN_API_KEYTypeScript 版本定義了完整的響應(yīng)結(jié)構(gòu)VideoStatusResponse其中data字段包含id、status四態(tài)之一、video_url、thumbnail_url、duration、title、created_at、completed_at、gif_url、captioned_video_url、subtitle_url、folder_id、output_language以及失敗專用的failure_code與failure_message。錯誤處理約定是頂層error字段非空即代表調(diào)用失敗如 404、鑒權(quán)失敗應(yīng)直接拋錯業(yè)務(wù)狀態(tài)含失敗態(tài)則放在data.status中表達(dá)。async function getVideoStatus(videoId: string): PromiseVideoStatusResponse[data] { const response await fetch( https://api.heygen.com/v2/videos/${videoId}, { headers: { X-Api-Key: process.env.HEYGEN_API_KEY! } } ); const json: VideoStatusResponse await response.json(); if (json.error) { throw new Error(json.error); } return json.data; }Python 版本邏輯等價import requests import os def get_video_status(video_id: str) - dict: response requests.get( fhttps://api.heygen.com/v2/videos/{video_id}, headers{X-Api-Key: os.environ[HEYGEN_API_KEY]} ) data response.json() if data.get(error): raise Exception(data[error]) return data[data]2. 狀態(tài)類型與耗時預(yù)估2.1 四種狀態(tài)及其語義Status含義客戶端行為pending視頻已入隊等待處理繼續(xù)輪詢processing視頻正在生成繼續(xù)輪詢completed視頻可下載讀取video_url下載failed生成失敗讀取failure_message定位原因并終止輪詢邏輯的本質(zhì)就是一個針對這四個狀態(tài)的狀態(tài)機(jī)completed返回 URL、failed拋錯、其余狀態(tài) sleep 后重試。2.2 生成耗時與影響因素文檔給出的經(jīng)驗值是視頻生成通常需要5–15 分鐘高峰負(fù)載或長腳本場景可能超過 20 分鐘。影響耗時的主要因素因素影響腳本長度腳本越長處理時間顯著增加分辨率1080p 比 720p 慢Avatar 復(fù)雜度部分 avatar 渲染更快隊列負(fù)載高峰時段可能等待 15–20 分鐘以上多場景每個場景都增加處理時間基于此文檔給出的工程建議是超時設(shè)置為 15–20 分鐘900,000–1,200,000 ms語音腳本超過 2 分鐘時應(yīng)預(yù)期 15 分鐘以上的等待長視頻建議改用異步模式保存video_id稍后再查見第 5 節(jié)。這一經(jīng)驗值與 OpenMontage 源碼中的實際實現(xiàn)一致heygen_video工具的輪詢默認(rèn)超時為 600 秒見 poll_heygen 的timeout: int 600參數(shù)——它對應(yīng)約 10 分鐘的基礎(chǔ)預(yù)算而文檔建議對長內(nèi)容上調(diào)到 15–20 分鐘兩者并不矛盾前者是工具鏈的保守默認(rèn)后者是面向長腳本的上限建議。3. 響應(yīng)格式詳解理解響應(yīng) JSON 的兩類形態(tài)是編寫正確狀態(tài)處理代碼的前提。3.1 completed 響應(yīng)成功時data中攜帶全部交付元數(shù)據(jù)。文檔示例{ error: null, data: { id: abc123, status: completed, video_url: https://files.heygen.ai/video/abc123.mp4, thumbnail_url: https://files.heygen.ai/thumbnail/abc123.jpg, duration: 45.2, title: My Video, created_at: 2024-01-15T10:30:00Z, completed_at: 2024-01-15T10:38:00Z, gif_url: https://files.heygen.ai/gif/abc123.gif, captioned_video_url: null, subtitle_url: null, folder_id: null, output_language: en } }注意兩個易踩的坑其一captioned_video_url、subtitle_url等字段可能為null取用前必須判空其二從示例中的created_at/completed_at時間差8 分鐘可以看到實際渲染時長可用這兩個字段做生成耗時統(tǒng)計。3.2 failed 響應(yīng)失敗時響應(yīng)結(jié)構(gòu)不變但只有failure_code和failure_message提供診斷信息{ error: null, data: { id: abc123, status: failed, failure_code: script_too_long, failure_message: Script too long for selected avatar } }這里的failure_code如script_too_long是機(jī)器可讀的錯誤分類failure_message是可直接展示給用戶的文本。輪詢代碼在failed分支應(yīng)優(yōu)先把兩者都記錄下來——OpenMontage 源碼中的poll_heygen同樣遵循失敗即拋異常并攜帶錯誤詳情的原則raise RuntimeError(fHeyGen generation failed: {data.get(error, Unknown)})見 tools/video/_shared.py。4. 輪詢實現(xiàn)從基礎(chǔ)循環(huán)到進(jìn)度回調(diào)4.1 基礎(chǔ)輪詢核心是一個截止時刻 固定間隔循環(huán)記錄startTime每輪查一次狀態(tài)completed返回video_urlfailed拋錯pending/processing則 sleep 后繼續(xù)超出maxWaitMs后拋超時錯誤。async function waitForVideo( videoId: string, maxWaitMs 600000, // 10 minutes pollIntervalMs 5000 // 5 seconds ): Promisestring { const startTime Date.now(); while (Date.now() - startTime maxWaitMs) { const status await getVideoStatus(videoId); switch (status.status) { case completed: return status.video_url!; case failed: throw new Error(status.failure_message || Video generation failed); case pending: case processing: await new Promise((resolve) setTimeout(resolve, pollIntervalMs)); break; } } throw new Error(Video generation timed out); }默認(rèn)參數(shù)為 10 分鐘超時、5 秒輪詢間隔——注意這里的maxWaitMs默認(rèn)值偏保守對長視頻應(yīng)按第 2 節(jié)的建議顯式傳入更大的值。4.2 帶進(jìn)度回調(diào)的輪詢在 Agent 或 CLI 場景中用戶需要看到還在跑已等待 X 秒這類反饋。做法是引入ProgressCallback (status, elapsed) void在每次輪詢后把當(dāng)前狀態(tài)與已耗時傳給回調(diào)type ProgressCallback (status: string, elapsed: number) void; async function waitForVideoWithProgress( videoId: string, onProgress?: ProgressCallback, maxWaitMs 600000, pollIntervalMs 5000 ): Promisestring { const startTime Date.now(); while (Date.now() - startTime maxWaitMs) { const elapsed Date.now() - startTime; const status await getVideoStatus(videoId); onProgress?.(status.status, elapsed); switch (status.status) { case completed: return status.video_url!; case failed: throw new Error(status.failure_message || Video generation failed); default: await new Promise((resolve) setTimeout(resolve, pollIntervalMs)); } } throw new Error(Video generation timed out); } // Usage const videoUrl await waitForVideoWithProgress( videoId, (status, elapsed) { console.log(Status: ${status}, Elapsed: ${Math.round(elapsed / 1000)}s); } );Python 版本提供同樣能力on_progress是可選的Callable[[str, int], None]參數(shù)為當(dāng)前狀態(tài)與已等待秒數(shù)import time from typing import Optional, Callable def wait_for_video( video_id: str, max_wait_seconds: int 600, poll_interval: int 5, on_progress: Optional[Callable[[str, int], None]] None ) - str: start_time time.time() while time.time() - start_time max_wait_seconds: elapsed int(time.time() - start_time) status_data get_video_status(video_id) status status_data[status] if on_progress: on_progress(status, elapsed) if status completed: return status_data[video_url] elif status failed: raise Exception(status_data.get(failure_message, Video generation failed)) time.sleep(poll_interval) raise Exception(Video generation timed out) # Usage def progress_callback(status: str, elapsed: int): print(fStatus: {status}, Elapsed: {elapsed}s) video_url wait_for_video(video_id, on_progressprogress_callback)4.3 OpenMontage 源碼中的真實輪詢實現(xiàn)上述示例之外倉庫的工具層給出了一個可直接借鑒的生產(chǎn)實現(xiàn)。tools/video/_shared.py 中的poll_heygen有兩個值得注意的設(shè)計漸進(jìn)式退避。間隔不是固定的 5 秒而是從 5.0 秒開始每輪乘以 1.2上限 30 秒interval min(interval * 1.2, 30.0)。這正好實踐了文檔Best Practices第 1 條——對長任務(wù)增大輪詢間隔——避免在 15 分鐘級的任務(wù)里做無謂的高頻請求。def poll_heygen(execution_id: str, api_key: str, timeout: int 600) - str: ... interval 5.0 while time.time() deadline: response requests.get(url, headersheaders, timeout30) response.raise_for_status() data response.json().get(data, {}) status data.get(status, ) if status completed: video_url ( data.get(output, {}).get(video, {}).get(video_url) or data.get(output, {}).get(video_url) ) ... if status in {failed, error}: raise RuntimeError(fHeyGen generation failed: {data.get(error, Unknown)}) time.sleep(min(interval, max(0.0, deadline - time.time()))) interval min(interval * 1.2, 30.0) raise TimeoutError(fHeyGen execution {execution_id} timed out after {timeout}s)響應(yīng)結(jié)構(gòu)兼容。completed分支同時嘗試output.video.video_url與output.video_url兩條路徑取值并在都取不到時拋出帶完整響應(yīng)體的錯誤Completed but no video_url in output——這是一種防御性寫法應(yīng)對服務(wù)端響應(yīng)結(jié)構(gòu)在不同端點版本間的差異。從源碼結(jié)構(gòu)看poll_heygen服務(wù)于 Workflow 端點/v1/workflows/executions/{id}而文檔示例針對的是標(biāo)準(zhǔn)video_id狀態(tài)端點/v2/videos/{id}兩者狀態(tài)機(jī)語義completed / failed / 輪詢 / 超時完全一致可視為同一套模式在不同端點上的落地。5. 下載階段completed 不等于立即可下載文檔特別強調(diào)了一個容易被忽略的事實狀態(tài)顯示completed之后video_url可能仍短暫不可用文件還在向 CDN 分發(fā)因此下載必須帶重試與指數(shù)退避。5.1 帶重試的下載TypeScriptasync function downloadVideoWithRetry( videoUrl: string, outputPath ./output/video.mp4, maxRetries 5, initialDelayMs 2000 ): Promisevoid { let lastError: Error | null null; for (let attempt 0; attempt maxRetries; attempt) { try { const response await fetch(videoUrl); if (!response.ok) { throw new Error(HTTP ${response.status}: ${response.statusText}); } const arrayBuffer await response.arrayBuffer(); fs.writeFileSync(path.resolve(outputPath), Buffer.from(arrayBuffer)); console.log(Video downloaded to ${outputPath}); return; } catch (error) { lastError error as Error; const delay initialDelayMs * Math.pow(2, attempt); // Exponential backoff console.log(Download attempt ${attempt 1} failed, retrying in ${delay}ms...); await new Promise((resolve) setTimeout(resolve, delay)); } } throw new Error(Failed to download after ${maxRetries} attempts: ${lastError?.message}); }退避序列為 2s → 4s → 8s → 16s → 32sinitialDelayMs * 2^attempt最多 5 次。5.2 帶重試的下載PythonPython 版本使用streamTrue分塊寫入chunk_size8192避免大文件一次性占用內(nèi)存重試邏輯與 TypeScript 版一一對應(yīng)def download_video_with_retry( video_url: str, output_path: str, max_retries: int 5, initial_delay: float 2.0 ) - None: last_error None for attempt in range(max_retries): try: response requests.get(video_url, streamTrue, timeout60) response.raise_for_status() with open(output_path, wb) as f: for chunk in response.iter_content(chunk_size8192): f.write(chunk) print(fVideo downloaded to {output_path}) return except Exception as e: last_error e delay initial_delay * (2 ** attempt) # Exponential backoff print(fDownload attempt {attempt 1} failed, retrying in {delay}s...) time.sleep(delay) raise Exception(fFailed to download after {max_retries} attempts: {last_error})若只是快速腳本、失敗后可手動重跑文檔也提供了無重試的簡版downloadVideo單次 fetch 寫入!response.ok時拋錯。對比倉庫實現(xiàn)generate_heygen_video在 tools/video/_shared.py 中拿到video_url后直接requests.get(video_url, timeout120)一次性下載未做應(yīng)用層重試——但這并不沖突因為 tools/video/heygen_video.py 聲明了retry_policy RetryPolicy(max_retries2, backoff_seconds10.0, retryable_errors[rate_limit, timeout, server_error])由工具框架層對execute整體重試兜底。這說明同一份文檔知識在 OpenMontage 中有兩種落地方式輪詢工具自行實現(xiàn)退避poll_heygen下載重試則委托給工具框架的 RetryPolicy。6. 完整工作流生成 → 輪詢 → 下載把前述環(huán)節(jié)串起來就是一個端到端的一次性流程。以下示例以 Video Agent 生成端點為起點該端點返回data.video_id與 video-agent.md 中的響應(yīng)示例一致再進(jìn)入輪詢與下載async function generateAndDownloadVideo(config: VideoConfig): Promisestring { // 1. Generate video const generateResponse await fetch( https://api.heygen.com/v2/video/generate, { method: POST, headers: { X-Api-Key: process.env.HEYGEN_API_KEY!, Content-Type: application/json, }, body: JSON.stringify(config), } ); const { data: generateData } await generateResponse.json(); const videoId generateData.video_id; console.log(Video ID: ${videoId}); // 2. Poll for completion const videoUrl await waitForVideoWithProgress( videoId, (status, elapsed) { console.log([${Math.round(elapsed / 1000)}s] Status: ${status}); } ); // 3. Download const outputPath ./output/${videoId}.mp4; await downloadVideo(videoUrl, outputPath); return outputPath; }OpenMontage 的generate_heygen_video演示了同樣的三步骨架只是第 1 步換成了 Workflow 端點POST /v1/workflows/executionsworkflow_type: GenerateVideoNode返回execution_id第 2 步調(diào)用內(nèi)部poll_heygen(execution_id, api_key, timeout600)第 3 步把結(jié)果寫入output_path并以ToolResult返回execution_id、provider_variant、aspect_ratio等元數(shù)據(jù)見 tools/video/_shared.py。另外image_to_video場景下本地參考圖會先經(jīng)upload_image_heygen上傳換取公開 URL 再注入請求體——這條上傳路徑同樣復(fù)用了v2 presigned 端點優(yōu)先、失敗回退的容錯思路。7. 斷點續(xù)查長任務(wù)的 Resumable 模式對 5–20 分鐘的生成任務(wù)讓一個進(jìn)程阻塞等待并不劃算進(jìn)程可能重啟、Agent 會話可能中斷。文檔給出的替代方案是生成后立刻持久化video_id進(jìn)程退出之后隨時再查一次狀態(tài)。7.1 保存待處理狀態(tài)interface PendingVideo { videoId: string; createdAt: string; script: string; avatarId: string; voiceId: string; } async function startVideoGeneration(config: VideoGenerateRequest): PromisePendingVideo { const videoId await generateVideo(config); const pending: PendingVideo { videoId, createdAt: new Date().toISOString(), script: config.video_inputs[0].voice.input_text!, avatarId: config.video_inputs[0].character.avatar_id!, voiceId: config.video_inputs[0].voice.voice_id!, }; // Save to file for later retrieval fs.writeFileSync(pending-video.json, JSON.stringify(pending, null, 2)); console.log(Video generation started. ID: ${videoId}); console.log(Check status later with: checkVideoStatus()); return pending; }PendingVideo除了videoId還冗余保存了script、avatarId、voiceId目的是讓后續(xù)查詢進(jìn)程無需重新構(gòu)造請求也能描述這個視頻是什么。7.2 稍后查詢并結(jié)算async function checkVideoStatus(): Promisevoid { if (!fs.existsSync(pending-video.json)) { console.log(No pending video found); return; } const pending: PendingVideo JSON.parse( fs.readFileSync(pending-video.json, utf-8) ); const elapsed Date.now() - new Date(pending.createdAt).getTime(); console.log(Checking video ${pending.videoId} (started ${Math.round(elapsed / 60000)} min ago)...); const status await getVideoStatus(pending.videoId); switch (status.status) { case completed: console.log(Video ready: ${status.video_url}); console.log(Duration: ${status.duration}s); // Clean up pending file fs.unlinkSync(pending-video.json); // Save result fs.writeFileSync(video-result.json, JSON.stringify({ ...pending, videoUrl: status.video_url, thumbnailUrl: status.thumbnail_url, duration: status.duration, title: status.title, createdAt: status.created_at, completedAt: status.completed_at, }, null, 2)); break; case failed: console.error(Video failed: ${status.failure_message}); fs.unlinkSync(pending-video.json); break; default: console.log(Status: ${status.status} - check again in a few minutes); } }注意其結(jié)算語義completed/failed兩個終態(tài)都會清理pending-video.jsoncompleted時額外把交付信息落盤到video-result.json非終態(tài)只做再等幾分鐘的提示不做阻塞。7.3 CLI 友好形態(tài)文檔最后把該模式拆成兩個獨立命令形成典型的兩段式 CLI 體驗// generate-video.ts - Start generation and exit async function main() { const pending await startVideoGeneration(config); console.log(\nVideo ID saved. Run npx tsx check-status.ts to check progress.); process.exit(0); // Exit immediately, dont wait } // check-status.ts - Check and optionally wait async function main() { const args process.argv.slice(2); const shouldWait args.includes(--wait); if (shouldWait) { // Poll until complete (with 20 min timeout) const result await waitForVideo(pending.videoId, apiKey, onProgress, 1200000); console.log(Done: ${result.video_url}); } else { // Just check once and report await checkVideoStatus(); } }--wait參數(shù)提供了第三種行為查詢腳本可以只做看一眼也可以就地切換到 20 分鐘1200000ms超時的阻塞輪詢——這正好對應(yīng)第 2 節(jié)15–20 分鐘超時的建議值。8. 替代方案與最佳實踐清單8.1 Webhook 替代輪詢對于不想維護(hù)輪詢連接的生產(chǎn)系統(tǒng)HeyGen 支持 webhook 推送視頻完成、失敗、翻譯完成、Avatar 訓(xùn)練完成等事件會 POST 到你的端點。完整的事件類型列表、簽名與端點實現(xiàn)含 Express 與 Flask 示例在同目錄的 webhooks.md 中有專門說明Video Agent 端點的callback_idcallback_url參數(shù)對見 video-agent.md 請求字段表即是為該通道預(yù)留的入口。8.2 文檔總結(jié)的五條 Best Practices使用指數(shù)退避——對長任務(wù)逐步增大輪詢間隔poll_heygen的 5s→30s 漸進(jìn)間隔即為此實踐設(shè)置合理超時——大多數(shù)視頻 10 分鐘內(nèi)完成長內(nèi)容上調(diào)至 15–20 分鐘優(yōu)雅處理失敗——利用failure_code/failure_message給出可操作的反饋生產(chǎn)系統(tǒng)優(yōu)先考慮 webhook——比輪詢更省資源緩存視頻 URL——下載用的 URL 有時效性拿到后應(yīng)盡快落盤不要長期持有 URL 反復(fù)引用。9. 小結(jié)這篇參考文檔在 OpenMontage 中的位置video-status.md 是create-video技能Foundation類參考件之一與webhooks.md、assets.md、dimensions.md、quota.md并列見 SKILL.md 的 Reference Files 章節(jié)承擔(dān)拿到 video_id 之后怎么辦這一環(huán)節(jié)的全部知識狀態(tài)機(jī)語義、耗時預(yù)算、輪詢/下載/斷點續(xù)查三套代碼模式。而 tools/video/heygen_video.py 與 tools/video/_shared.py 則證明這些模式不是紙面規(guī)范漸進(jìn)退避輪詢、終態(tài)錯誤上報、框架層重試策略都已在heygen_video工具的調(diào)用鏈中真實運行。對需要接入 HeyGen 或任何異步媒體生成 API 的開發(fā)者本文覆蓋的生成 → 輪詢含進(jìn)度→ 退避下載 → 斷點續(xù)查閉環(huán)可以直接作為實現(xiàn)模板復(fù)用?！久赓M下載鏈接】OpenMontageWorlds first open-source, agentic video production system. 12 production pipelines, 100 tools, 700 agent skill and production-knowledge files. Turn your AI coding assistant into a full video production studio.項目地址: https://gitcode.com/GitHub_Trending/op/OpenMontage創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考