企业官网建设流程全解析

OpenMontage 中 HeyGen 视频状态轮询实战：从 poll 模式到断点续查的完整实现

【免费下载链接】OpenMontageWorld's 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 API：GET /v2/videos/{video_id}，需要自行处理状态机与重试。

SKILL.md的"Tool Selection"表格进一步明确了这一优先级：有mcp__heygen__*工具时优先用它（自动处理鉴权与请求格式），没有时才回退到裸 HTTP 调用。

1.1 查询接口的最小实现

文档给出三种语言的直接调用示例。curl 版本如下，注意鉴权走X-Api-Key请求头，key 来自环境变量HEYGEN_API_KEY：

curl -X GET "https://api.heygen.com/v2/videos/YOUR_VIDEO_ID" \ -H "X-Api-Key: $HEYGEN_API_KEY"

TypeScript 版本定义了完整的响应结构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): Promise<VideoStatusResponse["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( f"https://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 四种状态及其语义

轮询逻辑的本质就是一个针对这四个状态的状态机：completed返回 URL、failed抛错、其余状态 sleep 后重试。

2.2 生成耗时与影响因素

文档给出的经验值是：视频生成通常需要5–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(f"HeyGen generation failed: {data.get('error', 'Unknown')}")，见 tools/video/_shared.py）。

4. 轮询实现：从基础循环到进度回调

4.1 基础轮询

核心是一个"截止时刻 + 固定间隔"循环：记录startTime，每轮查一次状态，completed返回video_url，failed抛错，pending/processing则 sleep 后继续；超出maxWaitMs后抛超时错误。

async function waitForVideo( videoId: string, maxWaitMs = 600000, // 10 minutes pollIntervalMs = 5000 // 5 seconds ): Promise<string> { 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 ): Promise<string> { 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(f"Status: {status}, Elapsed: {elapsed}s") video_url = wait_for_video(video_id, on_progress=progress_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, headers=headers, timeout=30) 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(f"HeyGen 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(f"HeyGen 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 带重试的下载（TypeScript）

async function downloadVideoWithRetry( videoUrl: string, outputPath = "./output/video.mp4", maxRetries = 5, initialDelayMs = 2000 ): Promise<void> { 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 → 32s（initialDelayMs * 2^attempt，最多 5 次）。

5.2 带重试的下载（Python）

Python 版本使用stream=True分块写入（chunk_size=8192），避免大文件一次性占用内存；重试逻辑与 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, stream=True, timeout=60) response.raise_for_status() with open(output_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) print(f"Video downloaded to {output_path}") return except Exception as e: last_error = e delay = initial_delay * (2 ** attempt) # Exponential backoff print(f"Download attempt {attempt + 1} failed, retrying in {delay}s...") time.sleep(delay) raise Exception(f"Failed 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, timeout=120)一次性下载，未做应用层重试——但这并不冲突，因为 tools/video/heygen_video.py 声明了retry_policy = RetryPolicy(max_retries=2, backoff_seconds=10.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): Promise<string> { // 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/executions（workflow_type: "GenerateVideoNode"，返回execution_id），第 2 步调用内部poll_heygen(execution_id, api_key, timeout=600)，第 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): Promise<PendingVideo> { 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(): Promise<void> { 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.json，completed时额外把交付信息落盘到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, don't 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_id+callback_url参数对（见 video-agent.md 请求字段表）即是为该通道预留的入口。

8.2 文档总结的五条 Best Practices

1. 使用指数退避——对长任务逐步增大轮询间隔（poll_heygen的 5s→30s 渐进间隔即为此实践）；
2. 设置合理超时——大多数视频 10 分钟内完成，长内容上调至 15–20 分钟；
3. 优雅处理失败——利用failure_code/failure_message给出可操作的反馈；
4. 生产系统优先考虑 webhook——比轮询更省资源；
5. 缓存视频 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 的开发者，本文覆盖的"生成 → 轮询（含进度）→ 退避下载 → 断点续查"闭环可以直接作为实现模板复用。

【免费下载链接】OpenMontageWorld's 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