团队做短视频素材，Kling、Veo、Seedance 都在用，但每家网页端只能一条条手动生成。写了个 Python 脚本做批量调度，同一个 prompt 同时丢给多个模型，生成完自动归档，记录一下。

一、需求

每周十几条素材，不同模型擅长的风格不一样（Kling 人物自然、Veo 画面电影感、Seedance 节奏好），同一个 prompt 丢给三个模型让运营挑最好的，这是我们的工作流。手动在三个网页端分别操作效率太低。

所以要做的就是：输入一批 prompt → 自动分发到多个模型 → 生成完下载归档。

查了一圈发现 ai.tikhub.io 把这几家模型都聚合到了 /v1/videos 接口下，一个 Key 能调所有，省去了管理多套 API 的麻烦。

二、核心代码

video_gen/client.py — 提交 + 轮询 + 下载

import os, time, requests

class VideoGenClient:
    def __init__(self):
        self.base_url = os.getenv("AI_BASE_URL", "https://ai.tikhub.io")
        self.api_key = os.getenv("AI_API_KEY")
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }

    def submit_task(self, prompt: str, model: str = "kling-v3") -> dict:
        resp = requests.post(
            f"{self.base_url}/v1/videos",
            headers=self.headers,
            json={"model": model, "prompt": prompt}, timeout=30
        )
        if resp.status_code != 200:
            raise RuntimeError(f"提交失败 [{model}]: {resp.text}")
        return resp.json()

    def poll_result(self, task_id: str, max_wait=600, interval=15) -> dict:
        elapsed = 0
        while elapsed < max_wait:
            data = requests.get(
                f"{self.base_url}/v1/videos/{task_id}",
                headers=self.headers, timeout=15
            ).json()
            if data.get("status") == "completed": return data
            if data.get("status") == "failed":
                raise RuntimeError(f"生成失败: {data.get('error')}")
            time.sleep(interval); elapsed += interval
        raise TimeoutError(f"任务超时（{max_wait}s）")

    def download_video(self, url: str, save_path: str):
        resp = requests.get(url, stream=True, timeout=120)
        with open(save_path, "wb") as f:
            for chunk in resp.iter_content(8192): f.write(chunk)

video_gen/batch.py — 并发批量调度

import csv, os
from concurrent.futures import ThreadPoolExecutor, as_completed
from .client import VideoGenClient

MODELS = ["kling-v3", "veo-3.1-generate-001", "seedance-1-5-pro"]

def generate_one(client, prompt_id, prompt, model, output_dir):
    """单个任务：提交 → 轮询 → 下载"""
    print(f"[{prompt_id}] 提交到 {model}...")
    result = client.submit_task(prompt, model=model)
    task_id = result.get("id") or result.get("task_id")

    completed = client.poll_result(task_id)
    video_url = completed.get("video_url") or completed.get("url")

    model_dir = os.path.join(output_dir, model)
    os.makedirs(model_dir, exist_ok=True)
    save_path = os.path.join(model_dir, f"{prompt_id}.mp4")
    client.download_video(video_url, save_path)
    print(f"[{prompt_id}] {model} 完成 → {save_path}")
    return {"prompt_id": prompt_id, "model": model, "path": save_path}

def batch_generate(csv_path, output_dir="./output"):
    client = VideoGenClient()
    prompts = []
    with open(csv_path, "r", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            prompts.append(row)

    print(f"共 {len(prompts)} prompt × {len(MODELS)} 模型 = {len(prompts)*len(MODELS)} 个任务\n")

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {}
        for p in prompts:
            for model in MODELS:
                future = executor.submit(generate_one, client, p["id"], p["prompt"], model, output_dir)
                futures[future] = (p["id"], model)

        success, failed = 0, 0
        for future in as_completed(futures):
            try: future.result(); success += 1
            except Exception as e:
                pid, m = futures[future]
                print(f"[{pid}] {m} 失败: {e}"); failed += 1

    print(f"\n=== {success} 成功 / {failed} 失败 ===")

三、运行效果

准备 prompts.csv：

id,prompt
001,一只橘猫在钢琴上弹奏月光奏鸣曲，电影质感，暖色灯光
002,航拍日落时分的海边公路，一辆红色敞篷车飞驰，4K画质
003,一个咖啡师在吧台做拿铁拉花的特写镜头，浅景深

export AI_BASE_URL="https://ai.tikhub.io" && export AI_API_KEY="你的Key"
python run.py prompts.csv

生成完自动按模型归档：

output/
├── kling-v3/          001.mp4  002.mp4  003.mp4
├── veo-3.1-generate-001/  001.mp4  002.mp4  003.mp4
└── seedance-1-5-pro/      001.mp4  002.mp4  003.mp4

运营按编号横向对比三个模型的效果，挑最好的即可。

四、后续优化：LLM 自动扩写 prompt

有些原始 prompt 太简单（“猫弹钢琴”），视频效果不好。加了一步预处理，用 Chat API 把简短描述扩写成详细的视频 prompt。因为同一个 API 入口同时支持 Chat 和视频端点，直接复用：

def enhance_prompt(raw_prompt: str) -> str:
    resp = requests.post(
        f"{BASE_URL}/v1/chat/completions",
        headers=HEADERS,
        json={
            "model": "gpt-4o",
            "messages": [{
                "role": "system",
                "content": "将简短描述扩写成详细的视频生成 prompt，包含构图、镜头、光线、色调。只输出英文。"
            }, {"role": "user", "content": raw_prompt}],
            "max_tokens": 300
        }
    )
    return resp.json()["choices"][0]["message"]["content"]

五、踩坑和模型体会

轮询间隔别太短。 5 秒一次会被限流，15–20 秒够了。并发数控制在 4 比较稳。

prompt 建议用英文。 多数视频模型对英文理解更好，中文偶尔语义偏差，所以上面才加了 LLM 扩写步骤。

模型方面：Kling v3 人物动态最自然（2–3 分钟/条），Veo 3.1 画面质感最好但最慢（5 分钟+），Seedance 1.5 Pro 速度最快适合出量。我们的策略是日常用 Seedance，重点内容 Kling + Veo 对比出品质。

参考资源：

• OpenAI Sora API：platform.openai.com/docs
• TikHub AI API：ai.tikhub.io
• concurrent.futures 文档：docs.python.org/3/library/concurrent.futures.html