OpenMontage 中 HeyGen 视频状态轮询实战从 poll 模式到断点续查的完整实现【免费下载链接】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 展开系统讲解 HeyGen 异步视频生成的状态轮询机制状态类型语义、生成耗时预估、completed/failed两种响应结构、带进度回调的轮询实现、带指数退避的下载重试以及适合长任务的断点续查模式。读完本篇你可以为任何接入 HeyGen API或其他异步媒体生成服务的流程编写生产级轮询逻辑并理解 OpenMontage 工具层heygen_video是如何在源码中落地同一套轮询策略的。1. 背景为什么状态轮询是异步视频生成的必经环节HeyGen 的视频生成是完全异步的提交请求后服务端只返回一个video_id真正的渲染在后台排队执行。客户端必须反复查询状态接口直到拿到video_url或确认失败。这正是 SKILL.md 中Default Workflow的第 3 步——Callmcp__heygen__get_videowith the returned video_id to poll status and get the download URL。文档给出了两条查询路径MCP 工具首选若 HeyGen MCP 服务器已连接直接使用mcp__heygen__get_video并传入videoId参数。它一次性返回 status、video_url、thumbnail_url、duration、title、gif_url、captioned_video_url等全部元数据直接调用 REST APIGET /v2/videos/{video_id}需要自行处理状态机与重试。SKILL.md的Tool Selection表格进一步明确了这一优先级有mcp__heygen__*工具时优先用它自动处理鉴权与请求格式没有时才回退到裸 HTTP 调用。1.1 查询接口的最小实现文档给出三种语言的直接调用示例。curl 版本如下注意鉴权走X-Api-Key请求头key 来自环境变量HEYGEN_API_KEYcurl -X GET https://api.heygen.com/v2/videos/YOUR_VIDEO_ID \ -H X-Api-Key: $HEYGEN_API_KEYTypeScript 版本定义了完整的响应结构VideoStatusResponse其中data字段包含id、status四态之一、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字段非空即代表调用失败如 404、鉴权失败应直接抛错业务状态含失败态则放在data.status中表达。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. 状态类型与耗时预估2.1 四种状态及其语义Status含义客户端行为pending视频已入队等待处理继续轮询processing视频正在生成继续轮询completed视频可下载读取video_url下载failed生成失败读取failure_message定位原因并终止轮询逻辑的本质就是一个针对这四个状态的状态机completed返回 URL、failed抛错、其余状态 sleep 后重试。2.2 生成耗时与影响因素文档给出的经验值是视频生成通常需要5–15 分钟高峰负载或长脚本场景可能超过 20 分钟。影响耗时的主要因素因素影响脚本长度脚本越长处理时间显著增加分辨率1080p 比 720p 慢Avatar 复杂度部分 avatar 渲染更快队列负载高峰时段可能等待 15–20 分钟以上多场景每个场景都增加处理时间基于此文档给出的工程建议是超时设置为 15–20 分钟900,000–1,200,000 ms语音脚本超过 2 分钟时应预期 15 分钟以上的等待长视频建议改用异步模式保存video_id稍后再查见第 5 节。这一经验值与 OpenMontage 源码中的实际实现一致heygen_video工具的轮询默认超时为 600 秒见 poll_heygen 的timeout: int 600参数——它对应约 10 分钟的基础预算而文档建议对长内容上调到 15–20 分钟两者并不矛盾前者是工具链的保守默认后者是面向长脚本的上限建议。3. 响应格式详解理解响应 JSON 的两类形态是编写正确状态处理代码的前提。3.1 completed 响应成功时data中携带全部交付元数据。文档示例{ 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 分钟可以看到实际渲染时长可用这两个字段做生成耗时统计。3.2 failed 响应失败时响应结构不变但只有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是机器可读的错误分类failure_message是可直接展示给用户的文本。轮询代码在failed分支应优先把两者都记录下来——OpenMontage 源码中的poll_heygen同样遵循失败即抛异常并携带错误详情的原则raise RuntimeError(fHeyGen generation failed: {data.get(error, Unknown)})见 tools/video/_shared.py。4. 轮询实现从基础循环到进度回调4.1 基础轮询核心是一个截止时刻 固定间隔循环记录startTime每轮查一次状态completed返回video_urlfailed抛错pending/processing则 sleep 后继续超出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); }默认参数为 10 分钟超时、5 秒轮询间隔——注意这里的maxWaitMs默认值偏保守对长视频应按第 2 节的建议显式传入更大的值。4.2 带进度回调的轮询在 Agent 或 CLI 场景中用户需要看到还在跑已等待 X 秒这类反馈。做法是引入ProgressCallback (status, elapsed) void在每次轮询后把当前状态与已耗时传给回调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]参数为当前状态与已等待秒数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 源码中的真实轮询实现上述示例之外仓库的工具层给出了一个可直接借鉴的生产实现。tools/video/_shared.py 中的poll_heygen有两个值得注意的设计渐进式退避。间隔不是固定的 5 秒而是从 5.0 秒开始每轮乘以 1.2上限 30 秒interval min(interval * 1.2, 30.0)。这正好实践了文档Best Practices第 1 条——对长任务增大轮询间隔——避免在 15 分钟级的任务里做无谓的高频请求。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)响应结构兼容。completed分支同时尝试output.video.video_url与output.video_url两条路径取值并在都取不到时抛出带完整响应体的错误Completed but no video_url in output——这是一种防御性写法应对服务端响应结构在不同端点版本间的差异。从源码结构看poll_heygen服务于 Workflow 端点/v1/workflows/executions/{id}而文档示例针对的是标准video_id状态端点/v2/videos/{id}两者状态机语义completed / failed / 轮询 / 超时完全一致可视为同一套模式在不同端点上的落地。5. 下载阶段completed 不等于立即可下载文档特别强调了一个容易被忽略的事实状态显示completed之后video_url可能仍短暂不可用文件还在向 CDN 分发因此下载必须带重试与指数退避。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避免大文件一次性占用内存重试逻辑与 TypeScript 版一一对应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时抛错。对比仓库实现generate_heygen_video在 tools/video/_shared.py 中拿到video_url后直接requests.get(video_url, timeout120)一次性下载未做应用层重试——但这并不冲突因为 tools/video/heygen_video.py 声明了retry_policy RetryPolicy(max_retries2, backoff_seconds10.0, retryable_errors[rate_limit, timeout, server_error])由工具框架层对execute整体重试兜底。这说明同一份文档知识在 OpenMontage 中有两种落地方式轮询工具自行实现退避poll_heygen下载重试则委托给工具框架的 RetryPolicy。6. 完整工作流生成 → 轮询 → 下载把前述环节串起来就是一个端到端的一次性流程。以下示例以 Video Agent 生成端点为起点该端点返回data.video_id与 video-agent.md 中的响应示例一致再进入轮询与下载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 步调用内部poll_heygen(execution_id, api_key, timeout600)第 3 步把结果写入output_path并以ToolResult返回execution_id、provider_variant、aspect_ratio等元数据见 tools/video/_shared.py。另外image_to_video场景下本地参考图会先经upload_image_heygen上传换取公开 URL 再注入请求体——这条上传路径同样复用了v2 presigned 端点优先、失败回退的容错思路。7. 断点续查长任务的 Resumable 模式对 5–20 分钟的生成任务让一个进程阻塞等待并不划算进程可能重启、Agent 会话可能中断。文档给出的替代方案是生成后立刻持久化video_id进程退出之后随时再查一次状态。7.1 保存待处理状态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目的是让后续查询进程无需重新构造请求也能描述这个视频是什么。7.2 稍后查询并结算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); } }注意其结算语义completed/failed两个终态都会清理pending-video.jsoncompleted时额外把交付信息落盘到video-result.json非终态只做再等几分钟的提示不做阻塞。7.3 CLI 友好形态文档最后把该模式拆成两个独立命令形成典型的两段式 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参数提供了第三种行为查询脚本可以只做看一眼也可以就地切换到 20 分钟1200000ms超时的阻塞轮询——这正好对应第 2 节15–20 分钟超时的建议值。8. 替代方案与最佳实践清单8.1 Webhook 替代轮询对于不想维护轮询连接的生产系统HeyGen 支持 webhook 推送视频完成、失败、翻译完成、Avatar 训练完成等事件会 POST 到你的端点。完整的事件类型列表、签名与端点实现含 Express 与 Flask 示例在同目录的 webhooks.md 中有专门说明Video Agent 端点的callback_idcallback_url参数对见 video-agent.md 请求字段表即是为该通道预留的入口。8.2 文档总结的五条 Best Practices使用指数退避——对长任务逐步增大轮询间隔poll_heygen的 5s→30s 渐进间隔即为此实践设置合理超时——大多数视频 10 分钟内完成长内容上调至 15–20 分钟优雅处理失败——利用failure_code/failure_message给出可操作的反馈生产系统优先考虑 webhook——比轮询更省资源缓存视频 URL——下载用的 URL 有时效性拿到后应尽快落盘不要长期持有 URL 反复引用。9. 小结这篇参考文档在 OpenMontage 中的位置video-status.md 是create-video技能Foundation类参考件之一与webhooks.md、assets.md、dimensions.md、quota.md并列见 SKILL.md 的 Reference Files 章节承担拿到 video_id 之后怎么办这一环节的全部知识状态机语义、耗时预算、轮询/下载/断点续查三套代码模式。而 tools/video/heygen_video.py 与 tools/video/_shared.py 则证明这些模式不是纸面规范渐进退避轮询、终态错误上报、框架层重试策略都已在heygen_video工具的调用链中真实运行。对需要接入 HeyGen 或任何异步媒体生成 API 的开发者本文覆盖的生成 → 轮询含进度→ 退避下载 → 断点续查闭环可以直接作为实现模板复用。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考