教程
Python 中的 Runway ML 文本生成视频 API：一个可运行的示例以及迁移到 hiapi
你想从 Python 调用文本生成视频模型，Runway Gen-3 是显而易见的起点。本指南展示一个最小的可运行 Python 示例，演示 Runway 风格的异步模式（创建任务 → 轮询 → 下载），然后将完全相同的代码映射到 hiapi 的 wan2.7-video/text-to-video —— 一个使用相同 /v1/tasks 结构的直接替代方案。
你需要准备什么
- Python 3.9+ 和
requests库（pip install requests）。 - 一个 API 密钥。对于 hiapi：从你的 hiapi 控制面板 获取一个，并将其导出为
HIAPI_API_KEY（它以sk-开头）。 - 一个提示词。视频模型接受单个文本提示词以及少量参数（宽高比、时长秒数）。
下面各处的认证头都是
Authorization: Bearer sk-<key>。如果你收到 HTTP 401 和permission_denied，说明密钥错误或没有所选模型的访问权限。
最小可运行示例（Runway 风格异步模式）
文本生成视频提供商不会同步返回视频 —— 生成需要 30–120 秒。标准结构是：
POST /tasks，带上模型 id + 输入 → 返回一个taskId。GET /tasks/{id}→ 轮询直到status为success或fail。- 读取
output[0].url并下载 MP4（该 URL 是短期有效的）。
下面是一个自包含的 Python 脚本，实现了这个循环：
import os
import time
import requests
API_BASE = "https://api.runwayml.com/v1" # 或任何使用相同结构的提供商
API_KEY = os.environ["RUNWAY_API_KEY"]
def create_task(prompt: str, aspect_ratio: str = "16:9", duration: int = 5) -> str:
resp = requests.post(
f"{API_BASE}/tasks",
headers={"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"},
json={
"model": "gen3a_turbo",
"input": {
"prompt": prompt,
"aspect_ratio": aspect_ratio,
"duration": duration,
},
},
timeout=60,
)
resp.raise_for_status()
return resp.json()["data"]["taskId"]
def wait_task(task_id: str, timeout_s: int = 600, interval: int = 5) -> dict:
deadline = time.time() + timeout_s
while time.time() < deadline:
resp = requests.get(
f"{API_BASE}/tasks/{task_id}",
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=30,
)
resp.raise_for_status()
task = resp.json()["data"]
if task["status"] == "success":
return task
if task["status"] == "fail":
raise RuntimeError(f"task failed: {task.get('error')}")
time.sleep(interval)
raise TimeoutError(f"task {task_id} not done after {timeout_s}s")
def download(url: str, path: str) -> None:
with requests.get(url, stream=True, timeout=120) as r:
r.raise_for_status()
with open(path, "wb") as f:
for chunk in r.iter_content(chunk_size=1 << 20):
f.write(chunk)
if __name__ == "__main__":
tid = create_task("a cat walking on a beach at sunset, cinematic")
print(f"task {tid} created, polling…")
task = wait_task(tid)
download(task["output"][0]["url"], "out.mp4")
print("wrote out.mp4")
这个脚本刻意保持简单：没有 SDK，没有异步胶水代码，只有 requests。一旦它跑通，你可以保留完全相同的控制流，只更换两个字符串就能切换提供商。
迁移到 hiapi（wan2.7-video/text-to-video）
hiapi 在 https://api.hiapi.ai/v1/tasks 暴露相同的异步任务契约。调用代码中唯一的区别是：
API_BASE变为https://api.hiapi.ai/v1。- 模型 id 是
wan2.7-video/text-to-video（裸 id，不是 URL）。 - 认证密钥是你的 hiapi
sk-…密钥。
具体来说，只有 create_task 会变化 —— wait_task 和 download 完全相同：
import os
import requests
API_BASE = "https://api.hiapi.ai/v1"
API_KEY = os.environ["HIAPI_API_KEY"]
def create_task(prompt: str, aspect_ratio: str = "16:9", duration: int = 5) -> str:
resp = requests.post(
f"{API_BASE}/tasks",
headers={"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"},
json={
"model": "wan2.7-video/text-to-video",
"input": {
"prompt": prompt,
"ratio": aspect_ratio, # wan2.7 称之为 ratio：16:9、9:16、1:1、4:3、3:4
"duration": duration, # 2-15 秒
"resolution": "720P", # 720P $0.10/秒；默认是 1080P，$0.167/秒
},
},
timeout=60,
)
resp.raise_for_status()
return resp.json()["data"]["taskId"]
响应结构完全相同：创建时返回 {"data": {"taskId": "..."}}，轮询时返回：
{
"data": {
"status": "success",
"output": [{"url": "https://…/result.mp4", "expireAt": 1750000000}],
"error": null
}
}
因此上一节的 wait_task(...) 和 download(...) 无需修改即可工作。完整迁移只有两行：更改 API_BASE 和更改 model。周围的一切保持不变。
output[].url是短期有效的（几小时，响应中带有expireAt）。把它当作重定向来处理 —— 立即下载字节，自行存储（S3、R2、你的 CDN）。永远不要把原始 URL 嵌入页面。
生产级模式
上面的最小示例适合演示，但在上线前你还需要额外四件事。
尽可能使用回调而不是轮询
每 5 秒轮询会浪费请求，并让一个 worker 被占用一两分钟。如果你的环境有公开的 webhook URL，把它交给 API，让它回调你：
resp = requests.post(
f"{API_BASE}/tasks",
headers={"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"},
json={
"model": "wan2.7-video/text-to-video",
"input": {"prompt": prompt, "ratio": "16:9", "duration": 5, "resolution": "720P"},
"callback": {
"url": "https://yourapp.com/hooks/hiapi",
"when": "final", # 仅在状态为 success 或 fail 时回调我们
},
},
timeout=60,
)
你的 webhook 会收到完整的任务对象（与轮询响应的结构相同），因此下载步骤完全相同。
何时轮询 vs 回调：
- 轮询：脚本、notebook、批处理任务，以及任何你不想暴露 HTTP 监听器的地方。
- 回调：Web 应用、队列 worker，以及任何存在公开端点的地方。始终配合幂等性（下一节）—— 提供商可能会重试。
让提交具备幂等性
如果你的代码重试 create_task（网络抖动、循环中途崩溃），很容易重复提交相同提示词并支付两次费用。廉价的修复方法：将输入哈希为一个稳定的客户端密钥，并在调用前在自己的表中去重：
import hashlib, json
def submit_key(prompt: str, aspect_ratio: str, duration: int) -> str:
payload = json.dumps({"p": prompt, "a": aspect_ratio, "d": duration},
sort_keys=True)
return hashlib.sha256(payload.encode()).hexdigest()
将该密钥作为 Idempotency-Key 请求头发送（最多 255 字节）。hiapi 每个账户和密钥只创建一个任务，使用相同密钥的重试会返回原始 taskId，而不是重复计费。在此基础上将 (submit_key → task_id) 存储在你自己的数据库中，当以后需要查找任务时仍然有帮助。
精确处理错误
你会从 hiapi 看到两种错误封装：
{ "error": {
"code": "permission_denied",
"message": "This API key cannot use the selected model.",
"request_id": "20260626…",
"type": "hiapi_error"
} }
……出现在认证/权限失败时（HTTP 401/403），以及任务最终失败时：
{ "data": { "status": "fail",
"error": {"code": "...", "message": "..."} } }
区别对待它们：创建时的 permission_denied 意味着你的密钥错误或不覆盖该模型 —— 重试没有意义。轮询后任务 status: "fail" 可能是暂时性模型错误，值得用相同提示词重试一次。始终记录 request_id，以便支持人员追踪。
定价和配额
视频生成按输出秒数计价，而不是按请求计价，并且因分辨率和模型层级而异。在计算单位经济性之前，请查看 hiapi 定价页面 上的实时数字 —— 不要在仓库中硬编码美元数值。
相关阅读
- Wan 2.7 文本生成视频模型页面 —— 模型卡、支持的
aspect_ratio/duration值，以及一个发出完全相同/v1/tasks调用的 playground。 - hiapi 文档 —— 异步任务契约的完整参考（
/v1/tasks创建、轮询、回调 schema、错误码）。 - 模型目录 —— 共享相同调用结构的图像、视频和音频模型。
常见问题
问：视频模型和图像模型需要不同的 SDK 吗？
答：不需要。两者都通过 POST /v1/tasks 带上 {model, input}，并且都返回 {data:{taskId}}。只有每个模型的 input 字段不同 —— 视频接受 duration，图像不接受。轮询循环完全相同。
问：为什么 output[].url 在几小时后出现 404？
答：该 URL 是签名的且会过期（响应中的 expireAt 是 Unix 时间戳）。立即下载字节，推送到你自己的存储，并从那里提供服务。
问：如何在不重写的情况下从 Runway Gen-3 切换到 Wan 2.7？
答：如果你的提供商使用相同的异步任务结构（创建 → 轮询 → 输出 URL），更改两个字符串：API_BASE 和 model id。将它们包装在环境变量中（API_BASE、MODEL_ID），你就可以用同一份代码进行 A/B 测试。
问：尽管我的密钥看起来正确，却收到 permission_denied。
答：检查两件事：(1) 密钥有权访问特定模型（某些模型需要付费层级），以及 (2) Authorization 头正好是 Bearer sk-... 并带有空格 —— 不是 Bearer: sk-...，也不是原始密钥。错误正文中的完整 request_id 是支持人员追踪所需的。
问：我应该轮询还是使用回调？
答：在任何有公开 HTTP 端点的环境中使用回调（Web 应用、队列 worker）。在脚本、cron 任务和 notebook 中使用轮询 —— 任何搭建 webhook 显得愚蠢的地方。无论哪种方式，都要存储 taskId，以便崩溃的 worker 可以通过轮询恢复，而不是重新提交。
