API Runway ML de génération de vidéo à partir de texte en Python : un exemple exécutable et la migration vers hiapi

Vous voulez appeler un modèle de génération de vidéo à partir de texte depuis Python, et Runway Gen-3 est le point de départ évident. Ce guide présente un exemple Python minimal et exécutable qui illustre le schéma asynchrone propre à Runway (créer une tâche → interroger → télécharger), puis transpose exactement le même code vers wan2.7-video/text-to-video de hiapi — une solution de remplacement directe qui utilise la même structure /v1/tasks.

Ce dont vous avez besoin

• Python 3.9+ et la bibliothèque requests (pip install requests).
• Une clé API. Pour hiapi : récupérez-en une depuis votre tableau de bord hiapi et exportez-la dans HIAPI_API_KEY (elle commence par sk-).
• Un prompt. Les modèles vidéo acceptent un seul prompt texte ainsi que quelques paramètres (format d'image, durée en secondes).

Partout ci-dessous, l'en-tête d'authentification est Authorization: Bearer sk-<key>. Si vous recevez un HTTP 401 avec permission_denied, cela signifie que la clé est erronée ou qu'elle n'a pas accès au modèle choisi.

Exemple minimal exécutable (schéma asynchrone de style Runway)

Les fournisseurs de génération de vidéo à partir de texte ne renvoient pas la vidéo de manière synchrone — la génération prend de 30 à 120 secondes. La structure standard est la suivante :

1. POST /tasks avec l'id du modèle + les entrées → renvoie un taskId.
2. GET /tasks/{id} → interroger jusqu'à ce que status vaille success ou fail.
3. Lire output[0].url et télécharger le MP4 (cette URL est valable pour une courte durée).

Voici un script Python autonome qui implémente cette boucle :

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")

Ce script est volontairement simple : pas de SDK, pas de code de collage asynchrone, seulement requests. Une fois qu'il fonctionne, vous pouvez conserver exactement le même flux de contrôle et changer de fournisseur en modifiant seulement deux chaînes de caractères.

Migration vers hiapi (wan2.7-video/text-to-video)

hiapi expose le même contrat de tâche asynchrone sur https://api.hiapi.ai/v1/tasks. Les seules différences dans le code d'appel sont :

1. API_BASE devient https://api.hiapi.ai/v1.
2. L'id du modèle est wan2.7-video/text-to-video (un id nu, pas une URL).
3. La clé d'authentification est votre clé hiapi sk-….

Concrètement, seule create_task change — wait_task et download restent identiques :

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"]

La structure de réponse est identique : à la création, {"data": {"taskId": "..."}} est renvoyé ; lors de l'interrogation, on obtient :

{
  "data": {
    "status": "success",
    "output": [{"url": "https://…/result.mp4", "expireAt": 1750000000}],
    "error": null
  }
}

Ainsi, les fonctions wait_task(...) et download(...) de la section précédente fonctionnent sans modification. La migration complète se résume à deux lignes : changer API_BASE et changer model. Tout le reste demeure inchangé.

output[].url est valable pour une courte durée (quelques heures, la réponse contient expireAt). Traitez-la comme une redirection — téléchargez immédiatement les octets et stockez-les vous-même (S3, R2, votre CDN). N'intégrez jamais l'URL brute dans une page.

Bonnes pratiques de production

L'exemple minimal ci-dessus convient pour une démonstration, mais avant de passer en production, quatre éléments supplémentaires s'imposent.

Privilégier les callbacks à l'interrogation répétée quand c'est possible

Une interrogation toutes les 5 secondes gaspille des requêtes et immobilise un worker pendant une à deux minutes. Si votre environnement dispose d'une URL de webhook publique, fournissez-la à l'API pour qu'elle vous rappelle :

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,
)

Votre webhook reçoit l'objet tâche complet (de même structure que la réponse d'interrogation), de sorte que l'étape de téléchargement reste identique.

Quand interroger, quand utiliser un callback :

• Interrogation : scripts, notebooks, tâches par lots, et tout contexte où vous ne voulez pas exposer un écouteur HTTP.
• Callback : applications web, workers de file d'attente, et tout environnement disposant d'un point d'accès public. Toujours avec idempotence (section suivante) — le fournisseur peut réessayer.

Rendre la soumission idempotente

Si votre code réessaie create_task (instabilité réseau, plantage en cours de boucle), il est facile de soumettre deux fois le même prompt et de payer deux fois. Correctif peu coûteux : hacher les entrées en une clé cliente stable et dédupliquer dans votre propre table avant l'appel :

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()

Envoyez cette clé dans l'en-tête de requête Idempotency-Key (255 octets maximum). hiapi ne crée qu'une seule tâche par compte et par clé ; un nouvel essai avec la même clé renvoie le taskId d'origine au lieu d'être facturé de nouveau. Stocker en plus (submit_key → task_id) dans votre propre base de données vous sera utile pour retrouver des tâches plus tard.

Traiter les erreurs avec précision

Vous rencontrerez deux enveloppes d'erreur venant de hiapi :

{ "error": {
    "code": "permission_denied",
    "message": "This API key cannot use the selected model.",
    "request_id": "20260626…",
    "type": "hiapi_error"
} }

… renvoyée en cas d'échec d'authentification/permission (HTTP 401/403), et, lorsque la tâche finit par échouer :

{ "data": { "status": "fail",
            "error": {"code": "...", "message": "..."} } }

Distinguez-les : un permission_denied à la création signifie que votre clé est erronée ou ne couvre pas ce modèle — réessayer ne sert à rien. Un status: "fail" après interrogation peut être une erreur temporaire du modèle, qui mérite un nouvel essai avec le même prompt. Enregistrez toujours le request_id pour permettre au support de tracer.

Tarification et quotas

La génération vidéo est facturée à la seconde de sortie, et non à la requête, et varie selon la résolution et le niveau de modèle. Avant de calculer votre économie unitaire, consultez les chiffres à jour sur la page de tarification hiapi — ne codez pas en dur des montants en dollars dans votre dépôt.

Lectures associées

• Page du modèle Wan 2.7 de génération de vidéo à partir de texte — fiche modèle, valeurs aspect_ratio/duration prises en charge, et un playground qui émet exactement le même appel /v1/tasks.
• Documentation hiapi — référence complète du contrat de tâche asynchrone (création /v1/tasks, interrogation, schéma de callback, codes d'erreur).
• Catalogue de modèles — modèles d'image, de vidéo et d'audio partageant la même structure d'appel.

Questions fréquentes

Q : Les modèles vidéo et les modèles d'image nécessitent-ils des SDK différents ? R : Non. Les deux passent par POST /v1/tasks avec {model, input}, et les deux renvoient {data:{taskId}}. Seuls les champs input diffèrent d'un modèle à l'autre — la vidéo accepte duration, l'image non. La boucle d'interrogation est exactement la même.

Q : Pourquoi output[].url renvoie-t-elle un 404 après quelques heures ? R : Cette URL est signée et expire (le expireAt de la réponse est un horodatage Unix). Téléchargez immédiatement les octets, poussez-les vers votre propre stockage et servez-les depuis là.

Q : Comment passer de Runway Gen-3 à Wan 2.7 sans réécrire le code ? R : Si votre fournisseur utilise la même structure de tâche asynchrone (création → interrogation → URL de sortie), changez deux chaînes : API_BASE et l'id model. Placez-les dans des variables d'environnement (API_BASE, MODEL_ID) et vous pourrez faire des tests A/B avec le même code.

Q : Je reçois permission_denied alors que ma clé semble correcte. R : Vérifiez deux choses : (1) que la clé a accès au modèle spécifique (certains modèles exigent un niveau payant), et (2) que l'en-tête Authorization est bien Bearer sk-... avec un espace — pas Bearer: sk-..., ni la clé brute. Le request_id complet dans le corps de l'erreur est nécessaire au support pour tracer.

Q : Faut-il interroger ou utiliser un callback ? R : Utilisez un callback dans tout environnement disposant d'un point d'accès HTTP public (applications web, workers de file d'attente). Utilisez l'interrogation dans les scripts, les tâches cron et les notebooks — partout où monter un webhook serait déraisonnable. Dans les deux cas, stockez le taskId afin qu'un worker planté puisse reprendre par interrogation au lieu de resoumettre.