視頻產生與編輯 - Vector Retrieval Service for 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，並按保留原則刪除原始素材和未採用的產生結果。