视频生成与编辑 - 向量检索服务 Milvus 版

命令格式

REST 接口

先调用 POST /v2/vectordb/ai/video_edit 创建异步任务：

{
  "model_name": "<model_name>",
  "prompt": "<instruction>",
  "media": [
    {"type": "first_frame | video | reference_image", "url": "<media_url>"}
  ],
  "params": {
    "resolution": "720P",
    "audio_setting": "none | origin",
    "watermark": false,
    "timeout_sec": 180
  }
}

再调用 POST /v2/vectordb/ai/tasks/describe 查询单个任务：

{
  "provider": "aliyun_milvus",
  "model_name": "<model_name>",
  "task_id": "<task_id>"
}

批量创建时，以 media_inputs 替换 media，每个内部数组是一项任务，且二者不能同时传入。

Python

create_status, create_data = post_json(
    "/v2/vectordb/ai/video_edit",
    {
        "model_name": "<model_name>",
        "prompt": "<instruction>",
        "media": [{"type": "video", "url": "<media_url>"}],
        "params": {"resolution": "720P", "audio_setting": "origin"},
    },
)
describe_status, describe_data = post_json(
    "/v2/vectordb/ai/tasks/describe",
    {
        "provider": "aliyun_milvus",
        "model_name": "<model_name>",
        "task_id": create_data["data"]["output"]["task_id"],
    },
)

参数说明

参数	说明
`model_name`	必填。可使用 `happyhorse-1.1-t2v`（文生视频）、`happyhorse-1.1-i2v` / `happyhorse-1.1-r2v`（图片输入）或 `happyhorse-1.0-video-edit`（视频编辑）。创建和查询必须传入相同模型名。
`prompt`	可选，描述视频内容或编辑目标。
`media`	按模型输入要求传入。文生视频可省略；首帧生视频传一个 `first_frame`；视频编辑传一个 `video`，可额外传入不超过 4 个 `reference_image`。
`media_inputs`	批量创建使用的二维媒体数组；不能与 `media` 同时传入。
`task_id`	查询时必填。任务查询接口一次只支持一个 `task_id`，不支持 `task_ids` 批量查询。
`provider`	查询时可选，固定为 `aliyun_milvus`。
`resolution`、`audio_setting`、`watermark`	可选，分别控制分辨率、音频策略和水印；支持范围以模型能力为准。
`timeout_sec`、`max_concurrency`	可选，控制请求超时和批量创建并发；`max_concurrency` 范围为 `1`–`64`。
`negative_prompt`、`prompt_extend` 等	会按模型能力透传。媒体 URL 不能为空，且必须能被模型服务访问。

返回值说明

创建成功会立即返回 task_id 和初始 task_status。随后轮询查询接口，状态为 SUCCEEDED 时返回 video_url;FAILED 或 CANCELED 表示任务结束但未生成结果。

{
  "code": 0,
  "data": {
    "output": {
      "task_id": "<task_id>",
      "task_status": "SUCCEEDED",
      "video_url": "https://<result-host>/videos/result.mp4"
    },
    "usage": {"duration": 5, "output_video_duration": 5, "video_count": 1}
  }
}

示例一：营销文案生成产品概念短片（文本）

运营人员仅提供营销文案，生成产品概念短片。将 MILVUS_REST_BASE_URL、MILVUS_AUTH_TOKEN 占位值替换为实际集群地址和 token,REST 脚本依赖 jq。

REST 接口

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST \
    "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

MODEL_NAME="happyhorse-1.1-t2v"
PROMPT="${AIFUNC_VIDEO_EDIT_PROMPT:-A stylish cat astronaut walks slowly on the moon, cinematic lighting.}"

BODY=$(cat <<JSON
{"model_name":"$MODEL_NAME","prompt":"$PROMPT","params":{"resolution":"720P","audio_setting":"none","watermark":false,"timeout_sec":180}}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/video_edit" "$BODY")"
echo "$RESPONSE_BODY" | jq .
TASK_ID="$(echo "$RESPONSE_BODY" | jq -r '.data.output.task_id // empty')"
[ -n "$TASK_ID" ] || exit 1
post_json "/v2/vectordb/ai/tasks/describe" "{\"provider\":\"aliyun_milvus\",\"model_name\":\"$MODEL_NAME\",\"task_id\":\"$TASK_ID\"}" | jq .

Python

from __future__ import annotations

import json
import os
import sys
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

MILVUS_REST_BASE_URL = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN = "<yourUsername>:<yourPassword>"

def post_json(path: str, body: dict[str, Any], timeout: int = 120) -> tuple[int, dict[str, Any]]:
    base_url = MILVUS_REST_BASE_URL.rstrip("/")
    token = MILVUS_AUTH_TOKEN
    request = Request(
        f"{base_url}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            status = response.status
            raw = response.read().decode("utf-8")
    except HTTPError as exc:
        status = exc.code
        raw = exc.read().decode("utf-8")
    return status, json.loads(raw)

MODEL_NAME = "happyhorse-1.1-t2v"
PROMPT = os.getenv("AIFUNC_VIDEO_EDIT_PROMPT", "A stylish cat astronaut walks slowly on the moon, cinematic lighting.")

status, data = post_json(
    "/v2/vectordb/ai/video_edit",
    {"model_name": MODEL_NAME, "prompt": PROMPT, "params": {"resolution": "720P", "audio_setting": "none", "watermark": False, "timeout_sec": 180}},
)
print(json.dumps(data, ensure_ascii=False, indent=2))
if status != 200 or data.get("code") != 0:
    sys.exit(1)
task_id = data.get("data", {}).get("output", {}).get("task_id")
if not task_id:
    sys.exit(1)
status, data = post_json("/v2/vectordb/ai/tasks/describe", {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "task_id": task_id})
print(json.dumps(data, ensure_ascii=False, indent=2))
if status != 200 or data.get("code") != 0:
    sys.exit(1)

两种调用均在终态 SUCCEEDED 时返回 data.output.video_url。

示例二：服装造型参考图生成展示短视频（图片）

服装运营团队基于造型参考图生成用于素材初审的展示短视频。默认素材是公开的服装造型参考图；自有素材可通过 AIFUNC_VIDEO_EDIT_IMAGE_URL 覆盖。

REST 接口

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST \
    "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

MODEL_NAME="happyhorse-1.1-i2v"
IMAGE_URL="${AIFUNC_VIDEO_EDIT_IMAGE_URL:-https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp}"

BODY=$(cat <<JSON
{"model_name":"$MODEL_NAME","prompt":"Animate the subject in the first-frame image with gentle camera movement.","media":[{"type":"first_frame","url":"$IMAGE_URL"}],"params":{"resolution":"720P","audio_setting":"none","watermark":false,"timeout_sec":180}}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/video_edit" "$BODY")"
echo "$RESPONSE_BODY" | jq .
TASK_ID="$(echo "$RESPONSE_BODY" | jq -r '.data.output.task_id // empty')"
[ -n "$TASK_ID" ] || exit 1
post_json "/v2/vectordb/ai/tasks/describe" "{\"provider\":\"aliyun_milvus\",\"model_name\":\"$MODEL_NAME\",\"task_id\":\"$TASK_ID\"}" | jq .

Python

from __future__ import annotations

import json
import os
import sys
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

MILVUS_REST_BASE_URL = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN = "<yourUsername>:<yourPassword>"

def post_json(path: str, body: dict[str, Any], timeout: int = 120) -> tuple[int, dict[str, Any]]:
    base_url = MILVUS_REST_BASE_URL.rstrip("/")
    token = MILVUS_AUTH_TOKEN
    request = Request(
        f"{base_url}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            status = response.status
            raw = response.read().decode("utf-8")
    except HTTPError as exc:
        status = exc.code
        raw = exc.read().decode("utf-8")
    return status, json.loads(raw)

MODEL_NAME = "happyhorse-1.1-i2v"
IMAGE_URL = os.getenv("AIFUNC_VIDEO_EDIT_IMAGE_URL", "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp")

status, data = post_json(
    "/v2/vectordb/ai/video_edit",
    {"model_name": MODEL_NAME, "prompt": "Animate the subject in the first-frame image with gentle camera movement.", "media": [{"type": "first_frame", "url": IMAGE_URL}], "params": {"resolution": "720P", "audio_setting": "none", "watermark": False, "timeout_sec": 180}},
)
print(json.dumps(data, ensure_ascii=False, indent=2))
if status != 200 or data.get("code") != 0:
    sys.exit(1)
task_id = data.get("data", {}).get("output", {}).get("task_id")
if not task_id:
    sys.exit(1)
status, data = post_json("/v2/vectordb/ai/tasks/describe", {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "task_id": task_id})
print(json.dumps(data, ensure_ascii=False, indent=2))
if status != 200 or data.get("code") != 0:
    sys.exit(1)

两种调用均在终态 SUCCEEDED 时返回非空的服装展示视频 URL。生成的镜头和动效具有非确定性，应在发布前人工复核款式和品牌要求。

示例三：马头角色概念片替换服装（视频）

品牌创意团队将角色概念片中的服装替换为参考图中的黑白条纹毛衣，用于内部创意评审。可通过 AIFUNC_VIDEO_EDIT_VIDEO_URL 和 AIFUNC_VIDEO_EDIT_IMAGE_URL 覆盖默认素材。

REST 接口

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST \
    "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

MODEL_NAME="happyhorse-1.0-video-edit"
VIDEO_URL="${AIFUNC_VIDEO_EDIT_VIDEO_URL:-https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4}"
IMAGE_URL="${AIFUNC_VIDEO_EDIT_IMAGE_URL:-https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp}"

BODY=$(cat <<JSON
{"model_name":"$MODEL_NAME","prompt":"Make the character in the video wear the striped sweater from the reference image.","media":[{"type":"video","url":"$VIDEO_URL"},{"type":"reference_image","url":"$IMAGE_URL"}],"params":{"resolution":"720P","audio_setting":"origin","watermark":false,"timeout_sec":180}}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/video_edit" "$BODY")"
echo "$RESPONSE_BODY" | jq .
TASK_ID="$(echo "$RESPONSE_BODY" | jq -r '.data.output.task_id // empty')"
[ -n "$TASK_ID" ] || exit 1
post_json "/v2/vectordb/ai/tasks/describe" "{\"provider\":\"aliyun_milvus\",\"model_name\":\"$MODEL_NAME\",\"task_id\":\"$TASK_ID\"}" | jq .

Python

from __future__ import annotations

import json
import os
import sys
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

MILVUS_REST_BASE_URL = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN = "<yourUsername>:<yourPassword>"

def post_json(path: str, body: dict[str, Any], timeout: int = 120) -> tuple[int, dict[str, Any]]:
    base_url = MILVUS_REST_BASE_URL.rstrip("/")
    token = MILVUS_AUTH_TOKEN
    request = Request(
        f"{base_url}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            status = response.status
            raw = response.read().decode("utf-8")
    except HTTPError as exc:
        status = exc.code
        raw = exc.read().decode("utf-8")
    return status, json.loads(raw)

MODEL_NAME = "happyhorse-1.0-video-edit"
VIDEO_URL = os.getenv("AIFUNC_VIDEO_EDIT_VIDEO_URL", "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4")
IMAGE_URL = os.getenv("AIFUNC_VIDEO_EDIT_IMAGE_URL", "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp")

status, data = post_json(
    "/v2/vectordb/ai/video_edit",
    {"model_name": MODEL_NAME, "prompt": "Make the character in the video wear the striped sweater from the reference image.", "media": [{"type": "video", "url": VIDEO_URL}, {"type": "reference_image", "url": IMAGE_URL}], "params": {"resolution": "720P", "audio_setting": "origin", "watermark": False, "timeout_sec": 180}},
)
print(json.dumps(data, ensure_ascii=False, indent=2))
if status != 200 or data.get("code") != 0:
    sys.exit(1)
task_id = data.get("data", {}).get("output", {}).get("task_id")
if not task_id:
    sys.exit(1)
status, data = post_json("/v2/vectordb/ai/tasks/describe", {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "task_id": task_id})
print(json.dumps(data, ensure_ascii=False, indent=2))
if status != 200 or data.get("code") != 0:
    sys.exit(1)

两种调用均在终态 SUCCEEDED 时返回非空的创意服装替换视频 URL。替换效果具有非确定性，不应将示例结果当作真实产品上身效果。

异常处理与媒体安全

FAILED 或 CANCELED 是终态，应停止轮询并保留完整响应中的错误信息；不要无上限自动重试。
达到本地轮询次数上限时，客户端应记录 task_id 并按业务策略稍后继续查询，不能将本地超时当作服务端已失败。
创建任务失败时，先确认媒体 URL 可由模型服务直接访问、未过期且格式受模型支持，再检查模型与 media.type 是否匹配。
使用自有人物、商品、声音或品牌素材前，请确认已取得必要授权。建议使用短时效、最小权限的媒体 URL，并按保留策略删除原始素材和未采用的生成结果。