批量推理 - 向量检索服务 Milvus 版

AI_BATCH 支持将 JSONL格式的文本或多模态 Chat Completions请求作为异步批量任务提交，适用于夜间摘要、历史数据打标、模型评测及数据标注等分钟至小时级的长耗时处理场景。

工作原理

批量任务遵循 OpenAI Batch 的文件输入和结果关联方式：每行请求使用唯一 custom_id，完成后从输出或错误 JSONL 中按该标识回写业务数据。

提交任务：上传包含多条请求的 UTF-8 JSONL 文件，再使用返回的文件 ID 创建 Batch 任务。
异步处理：服务端在后台校验并逐行执行请求；业务侧通过 batch_id 查询 validating、in_progress、finalizing 等状态。
下载结果：任务进入终态后下载 output；如有失败行，再下载 error，并按 custom_id 回写结果或定位问题。

适合离线模型评测、历史数据标注、内容批处理和定时素材加工；不适合聊天、搜索补全等需要立即展示结果的在线请求。

命令格式

Batch 按“上传输入文件 → 创建任务 → 查询状态 → 下载结果”四步执行：

REST 接口

POST /v2/vectordb/ai/batch/files/upload
multipart/form-data: file, model_name, endpoint, purpose=batch

POST /v2/vectordb/ai/batch/jobs/create
{"input_file_id":"<文件ID>","endpoint":"/v1/chat/completions","completion_window":"24h"}

POST /v2/vectordb/ai/batch/jobs/describe
{"batch_id":"<任务ID>"}

POST /v2/vectordb/ai/batch/files/content
{"batch_id":"<任务ID>","file_type":"output|error"}

Python

upload_status, upload_data = post_multipart_upload("<input.jsonl>")
create_status, create_data = post_json(
    "/v2/vectordb/ai/batch/jobs/create",
    {
        "input_file_id": upload_data["data"]["id"],
        "endpoint": "<endpoint>",
        "completion_window": "24h",
    },
)
describe_status, describe_data = post_json(
    "/v2/vectordb/ai/batch/jobs/describe",
    {"batch_id": create_data["data"]["id"]},
)
content_status, content_data = post_json(
    "/v2/vectordb/ai/batch/files/content",
    {"batch_id": create_data["data"]["id"], "file_type": "output"},
)

输入文件必须是 UTF-8 JSONL。每行是一条独立请求，包含唯一 custom_id、固定值 method="POST"、与任务一致的 url 和 body.model。REST 脚本依赖 jq。

参数说明

参数	说明
`file`	上传时必填。JSONL 输入文件；服务端会快速校验第一行的 `custom_id`、`method`、`url` 和 `body.model`。
`model_name`	上传时必填，也兼容 `model`。必须与 JSONL 第一行的 `body.model` 一致。
`endpoint`	上传和创建时必填。Chat Completions 使用 `/v1/chat/completions`，文本向量任务使用 `/v1/embeddings`；同一文件的各行必须一致。
`purpose`	上传时可选，只能为空或 `batch`。
`input_file_id`	创建任务必填。由上传接口返回的文件 ID；不接受 OSS URL 或外部文件标识。
`completion_window`	创建任务必填。完成窗口，支持 `24h`~`336h` 或天单位。
`metadata.ds_name` / `metadata.ds_description`	可选。任务名称最长 100 个字符，描述最长 200 个字符。
`batch_id`	查询和下载必填。创建接口返回的任务 ID。
`file_type`	下载必填。`output` 下载成功行，`error` 下载失败行。
`provider`	可选，默认 `aliyun_milvus`。上传、创建、查询和下载必须使用同一百炼账号、地域和工作空间。
`body.enable_thinking`	按模型选填。应与 JSONL 行中的 `body.model` 同级；对于默认启用思考的模型，显式设为 `false` 可避免不需要的思考 Token。不要将该参数放入 `extra_body`。

返回值说明

创建任务成功后返回 data.id，即 batch_id。轮询状态包括 validating、in_progress、finalizing、cancelling，终态包括 completed、failed、expired、cancelled。

批量任务采用排队方式异步处理，从创建到进入终态通常需要数分钟到数小时，其间状态持续为 in_progress。可随时调用查询接口 POST /v2/vectordb/ai/batch/jobs/describe（传入 batch_id）获取最新 status；若示例脚本的轮询在任务完成前结束，可稍后凭 batch_id 重新查询，待状态变为 completed 后再下载结果。下载接口直接返回 JSONL 原文；每一行保留输入时的 custom_id，用于将结果回写到业务数据。批量任务不适合等待即时返回的在线交互。

示例一：夜间批量生成知识库摘要（文本）

运营团队需要为当天新增文章批量生成摘要。每行一个文章请求，关键参数是 body.model=qwen3.7-max、url=/v1/chat/completions 和 completion_window=24h。

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="qwen3.7-max"
INPUT_FILE="${AIFUNC_AI_BATCH_INPUT_FILE:-}"
POLL_INTERVAL_SEC="${AIFUNC_AI_BATCH_POLL_INTERVAL_SEC:-5}"
MAX_POLL_ATTEMPTS="${AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS:-120}"

if [ -z "$INPUT_FILE" ]; then
  echo "请先将本示例 JSONL 请求保存为文件，并通过 AIFUNC_AI_BATCH_INPUT_FILE 指定路径。" >&2
  exit 1
fi

download_batch_file() {
  local file_type="$1"; local output_file="$2"; local content_body
  content_body="$(jq -nc --arg batch_id "$BATCH_ID" --arg file_type "$file_type" '{provider:"aliyun_milvus", batch_id:$batch_id, file_type:$file_type}')"
  curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/content" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" -H "Content-Type: application/json" \
    -d "$content_body" --output "$output_file"
}

UPLOAD_RESPONSE="$(curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/upload" \
  -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
  -F "provider=aliyun_milvus" -F "model_name=$MODEL_NAME" -F "endpoint=/v1/chat/completions" \
  -F "purpose=batch" -F "file=@$INPUT_FILE;type=application/jsonl")"
INPUT_FILE_ID="$(echo "$UPLOAD_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$INPUT_FILE_ID" ] || exit 1

CREATE_BODY="$(jq -nc --arg input_file_id "$INPUT_FILE_ID" '{provider:"aliyun_milvus", input_file_id:$input_file_id, endpoint:"/v1/chat/completions", completion_window:"24h", metadata:{ds_name:"milvus-ai-function-example", ds_description:"AI Batch REST example"}}')"
CREATE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/create" "$CREATE_BODY")"
BATCH_ID="$(echo "$CREATE_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$BATCH_ID" ] || exit 1

for ((attempt = 1; attempt <= MAX_POLL_ATTEMPTS; attempt++)); do
  DESCRIBE_BODY="$(jq -nc --arg batch_id "$BATCH_ID" '{provider:"aliyun_milvus", batch_id:$batch_id}')"
  DESCRIBE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/describe" "$DESCRIBE_BODY")"
  BATCH_STATUS="$(echo "$DESCRIBE_RESPONSE" | jq -r '.data.status // empty')"
  case "$BATCH_STATUS" in
    completed)
      OUTPUT_FILE="${AIFUNC_AI_BATCH_OUTPUT_FILE:-./ai_batch_output_${BATCH_ID}.jsonl}"
      download_batch_file "output" "$OUTPUT_FILE"
      echo "Batch output downloaded to: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID ended with status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Unknown status: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID did not finish after $MAX_POLL_ATTEMPTS checks." >&2
exit 1

Python

from __future__ import annotations

import json
import os
import shutil
import sys
import tempfile
import time
import uuid
from pathlib import Path
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>"

MODEL_NAME = "qwen3.7-max"
INPUT_MODE = os.getenv("AIFUNC_AI_BATCH_INPUT_MODE", "text").lower()
POLL_INTERVAL_SEC = int(os.getenv("AIFUNC_AI_BATCH_POLL_INTERVAL_SEC", "5"))
MAX_POLL_ATTEMPTS = int(os.getenv("AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS", "120"))


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


def batch_request(custom_id: str, content: Any, *, enable_thinking: bool | None = None) -> dict[str, Any]:
    body: dict[str, Any] = {"model": MODEL_NAME, "messages": [{"role": "user", "content": content}]}
    if enable_thinking is not None:
        body["enable_thinking"] = enable_thinking
    return {"custom_id": custom_id, "method": "POST", "url": "/v1/chat/completions", "body": body}


def default_requests(input_mode: str) -> list[dict[str, Any]]:
    if input_mode == "text":
        return [
            batch_request("milvus-summary-1", "For a support knowledge base, summarize in one Chinese sentence: Milvus stores and searches vector embeddings for RAG, recommendation, and multimodal retrieval.", enable_thinking=False),
            batch_request("milvus-summary-2", "Turn this incident note into a Chinese FAQ title and one-sentence answer: after documents are updated, regenerate embeddings before users search the new content.", enable_thinking=False),
            batch_request("milvus-summary-3", "Create a one-sentence Chinese catalog description for an enterprise RAG case that retrieves policy passages before the assistant answers.", enable_thinking=False),
        ]
    if input_mode == "image":
        return [batch_request("milvus-image-1", [
            {"type": "image_url", "image_url": {"url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"}},
            {"type": "text", "text": "For a pet-service media library, return a concise Chinese accessibility caption and up to three searchable subject tags for this uploaded case photo."},
        ])]
    if input_mode == "video":
        return [batch_request("milvus-video-1", [
            {"type": "video", "video": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4"},
            {"type": "text", "text": "For a marketing asset library, return a one-sentence Chinese scene summary, three retrieval keywords, and whether manual brand-safety review is needed."},
        ])]
    if input_mode == "audio":
        return [batch_request("milvus-audio-1", [
            {"type": "input_audio", "input_audio": {"data": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3", "format": "mp3"}},
            {"type": "text", "text": "For a customer-service hotline greeting archive, transcribe the welcome message and assess whether its purpose, service availability, and recording or privacy notice are clear; mark details not present in the audio as unconfirmed."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE must be one of text, image, video, or audio", file=sys.stderr)
    sys.exit(1)


def post_multipart_upload(input_file: str) -> tuple[int, dict[str, Any]]:
    boundary = f"----milvus-ai-batch-{uuid.uuid4().hex}"
    fields = {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "endpoint": "/v1/chat/completions", "purpose": "batch"}
    with tempfile.TemporaryFile(mode="w+b") as payload:
        for name, value in fields.items():
            payload.write(f"--{boundary}\r\n".encode())
            payload.write(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode())
            payload.write(value.encode())
            payload.write(b"\r\n")
        payload.write(f"--{boundary}\r\n".encode())
        payload.write(b'Content-Disposition: form-data; name="file"; filename="input.jsonl"\r\n')
        payload.write(b"Content-Type: application/jsonl\r\n\r\n")
        with open(input_file, "rb") as input_handle:
            shutil.copyfileobj(input_handle, payload)
        payload.write(b"\r\n")
        payload.write(f"--{boundary}--\r\n".encode())
        payload_length = payload.tell()
        payload.seek(0)
        request = Request(
            f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/upload",
            data=payload,
            headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Content-Length": str(payload_length)},
            method="POST",
        )
        try:
            with urlopen(request, timeout=600) as response:
                return response.status, json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            return exc.code, json.loads(exc.read().decode("utf-8"))


def download_batch_file(batch_id: str, file_type: str, output_file: Path) -> None:
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/content",
        data=json.dumps({"provider": "aliyun_milvus", "batch_id": batch_id, "file_type": file_type}, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=600) as response, output_file.open("wb") as output:
        shutil.copyfileobj(response, output)


temporary_input = None
INPUT_FILE = os.getenv("AIFUNC_AI_BATCH_INPUT_FILE")
if INPUT_FILE is None:
    temporary_input = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False, encoding="utf-8")
    for request in default_requests(INPUT_MODE):
        temporary_input.write(json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n")
    temporary_input.close()
    INPUT_FILE = temporary_input.name

try:
    status, data = post_multipart_upload(INPUT_FILE)
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    input_file_id = data.get("data", {}).get("id")

    status, data = post_json("/v2/vectordb/ai/batch/jobs/create", {
        "provider": "aliyun_milvus", "input_file_id": input_file_id, "endpoint": "/v1/chat/completions",
        "completion_window": "24h", "metadata": {"ds_name": "milvus-ai-function-example", "ds_description": "AI Batch example"},
    })
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    batch_id = data.get("data", {}).get("id")

    for attempt in range(1, MAX_POLL_ATTEMPTS + 1):
        status, data = post_json("/v2/vectordb/ai/batch/jobs/describe", {"provider": "aliyun_milvus", "batch_id": batch_id})
        batch_data = data.get("data", {})
        batch_status = batch_data.get("status")
        if batch_status == "completed":
            output_file = Path(os.getenv("AIFUNC_AI_BATCH_OUTPUT_FILE", f"ai_batch_output_{batch_id}.jsonl"))
            download_batch_file(batch_id, "output", output_file)
            print(f"Batch output downloaded to: {output_file}")
            if batch_data.get("error_file_id"):
                download_batch_file(batch_id, "error", Path(f"ai_batch_errors_{batch_id}.jsonl"))
            break
        if batch_status in {"failed", "expired", "cancelled"}:
            print(f"Batch {batch_id} ended with status: {batch_status}", file=sys.stderr)
            sys.exit(1)
        if attempt < MAX_POLL_ATTEMPTS:
            time.sleep(POLL_INTERVAL_SEC)
finally:
    if temporary_input is not None:
        os.unlink(temporary_input.name)

预期结果：任务进入 completed 后下载的 output.jsonl 中每一行都保留 custom_id，例如

{"custom_id":"milvus-summary-1","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"Milvus 是服务于 RAG、推荐和多模态检索场景的向量数据库。"}}]}}}

业务侧按 custom_id 将摘要回写到相应文章；如产生 error.jsonl，只重试其中失败的文章。

示例二：批量生成宠物服务案例图片描述（图片）

宠物服务平台需要为商家上传的历史案例图片生成无障碍描述和检索标签。每行的 content 包含 image_url 与文本指令，使用支持图像理解的多模态模型，上传时的 model_name 必须一致。

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="qwen3-vl-plus"
INPUT_FILE="${AIFUNC_AI_BATCH_INPUT_FILE:-}"
POLL_INTERVAL_SEC="${AIFUNC_AI_BATCH_POLL_INTERVAL_SEC:-5}"
MAX_POLL_ATTEMPTS="${AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS:-120}"

if [ -z "$INPUT_FILE" ]; then
  echo "请先将本示例 JSONL 请求保存为文件，并通过 AIFUNC_AI_BATCH_INPUT_FILE 指定路径。" >&2
  exit 1
fi

download_batch_file() {
  local file_type="$1"; local output_file="$2"; local content_body
  content_body="$(jq -nc --arg batch_id "$BATCH_ID" --arg file_type "$file_type" '{provider:"aliyun_milvus", batch_id:$batch_id, file_type:$file_type}')"
  curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/content" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" -H "Content-Type: application/json" \
    -d "$content_body" --output "$output_file"
}

UPLOAD_RESPONSE="$(curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/upload" \
  -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
  -F "provider=aliyun_milvus" -F "model_name=$MODEL_NAME" -F "endpoint=/v1/chat/completions" \
  -F "purpose=batch" -F "file=@$INPUT_FILE;type=application/jsonl")"
INPUT_FILE_ID="$(echo "$UPLOAD_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$INPUT_FILE_ID" ] || exit 1

CREATE_BODY="$(jq -nc --arg input_file_id "$INPUT_FILE_ID" '{provider:"aliyun_milvus", input_file_id:$input_file_id, endpoint:"/v1/chat/completions", completion_window:"24h", metadata:{ds_name:"milvus-ai-function-example", ds_description:"AI Batch REST example"}}')"
CREATE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/create" "$CREATE_BODY")"
BATCH_ID="$(echo "$CREATE_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$BATCH_ID" ] || exit 1

for ((attempt = 1; attempt <= MAX_POLL_ATTEMPTS; attempt++)); do
  DESCRIBE_BODY="$(jq -nc --arg batch_id "$BATCH_ID" '{provider:"aliyun_milvus", batch_id:$batch_id}')"
  DESCRIBE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/describe" "$DESCRIBE_BODY")"
  BATCH_STATUS="$(echo "$DESCRIBE_RESPONSE" | jq -r '.data.status // empty')"
  case "$BATCH_STATUS" in
    completed)
      OUTPUT_FILE="${AIFUNC_AI_BATCH_OUTPUT_FILE:-./ai_batch_output_${BATCH_ID}.jsonl}"
      download_batch_file "output" "$OUTPUT_FILE"
      echo "Batch output downloaded to: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID ended with status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Unknown status: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID did not finish after $MAX_POLL_ATTEMPTS checks." >&2
exit 1

Python

from __future__ import annotations

import json
import os
import shutil
import sys
import tempfile
import time
import uuid
from pathlib import Path
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>"

MODEL_NAME = "qwen3-vl-plus"
INPUT_MODE = os.getenv("AIFUNC_AI_BATCH_INPUT_MODE", "image").lower()
POLL_INTERVAL_SEC = int(os.getenv("AIFUNC_AI_BATCH_POLL_INTERVAL_SEC", "5"))
MAX_POLL_ATTEMPTS = int(os.getenv("AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS", "120"))


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


def batch_request(custom_id: str, content: Any, *, enable_thinking: bool | None = None) -> dict[str, Any]:
    body: dict[str, Any] = {"model": MODEL_NAME, "messages": [{"role": "user", "content": content}]}
    if enable_thinking is not None:
        body["enable_thinking"] = enable_thinking
    return {"custom_id": custom_id, "method": "POST", "url": "/v1/chat/completions", "body": body}


def default_requests(input_mode: str) -> list[dict[str, Any]]:
    if input_mode == "text":
        return [
            batch_request("milvus-summary-1", "For a support knowledge base, summarize in one Chinese sentence: Milvus stores and searches vector embeddings for RAG, recommendation, and multimodal retrieval.", enable_thinking=False),
            batch_request("milvus-summary-2", "Turn this incident note into a Chinese FAQ title and one-sentence answer: after documents are updated, regenerate embeddings before users search the new content.", enable_thinking=False),
            batch_request("milvus-summary-3", "Create a one-sentence Chinese catalog description for an enterprise RAG case that retrieves policy passages before the assistant answers.", enable_thinking=False),
        ]
    if input_mode == "image":
        return [batch_request("milvus-image-1", [
            {"type": "image_url", "image_url": {"url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"}},
            {"type": "text", "text": "For a pet-service media library, return a concise Chinese accessibility caption and up to three searchable subject tags for this uploaded case photo."},
        ])]
    if input_mode == "video":
        return [batch_request("milvus-video-1", [
            {"type": "video", "video": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4"},
            {"type": "text", "text": "For a marketing asset library, return a one-sentence Chinese scene summary, three retrieval keywords, and whether manual brand-safety review is needed."},
        ])]
    if input_mode == "audio":
        return [batch_request("milvus-audio-1", [
            {"type": "input_audio", "input_audio": {"data": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3", "format": "mp3"}},
            {"type": "text", "text": "For a customer-service hotline greeting archive, transcribe the welcome message and assess whether its purpose, service availability, and recording or privacy notice are clear; mark details not present in the audio as unconfirmed."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE must be one of text, image, video, or audio", file=sys.stderr)
    sys.exit(1)


def post_multipart_upload(input_file: str) -> tuple[int, dict[str, Any]]:
    boundary = f"----milvus-ai-batch-{uuid.uuid4().hex}"
    fields = {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "endpoint": "/v1/chat/completions", "purpose": "batch"}
    with tempfile.TemporaryFile(mode="w+b") as payload:
        for name, value in fields.items():
            payload.write(f"--{boundary}\r\n".encode())
            payload.write(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode())
            payload.write(value.encode())
            payload.write(b"\r\n")
        payload.write(f"--{boundary}\r\n".encode())
        payload.write(b'Content-Disposition: form-data; name="file"; filename="input.jsonl"\r\n')
        payload.write(b"Content-Type: application/jsonl\r\n\r\n")
        with open(input_file, "rb") as input_handle:
            shutil.copyfileobj(input_handle, payload)
        payload.write(b"\r\n")
        payload.write(f"--{boundary}--\r\n".encode())
        payload_length = payload.tell()
        payload.seek(0)
        request = Request(
            f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/upload",
            data=payload,
            headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Content-Length": str(payload_length)},
            method="POST",
        )
        try:
            with urlopen(request, timeout=600) as response:
                return response.status, json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            return exc.code, json.loads(exc.read().decode("utf-8"))


def download_batch_file(batch_id: str, file_type: str, output_file: Path) -> None:
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/content",
        data=json.dumps({"provider": "aliyun_milvus", "batch_id": batch_id, "file_type": file_type}, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=600) as response, output_file.open("wb") as output:
        shutil.copyfileobj(response, output)


temporary_input = None
INPUT_FILE = os.getenv("AIFUNC_AI_BATCH_INPUT_FILE")
if INPUT_FILE is None:
    temporary_input = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False, encoding="utf-8")
    for request in default_requests(INPUT_MODE):
        temporary_input.write(json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n")
    temporary_input.close()
    INPUT_FILE = temporary_input.name

try:
    status, data = post_multipart_upload(INPUT_FILE)
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    input_file_id = data.get("data", {}).get("id")

    status, data = post_json("/v2/vectordb/ai/batch/jobs/create", {
        "provider": "aliyun_milvus", "input_file_id": input_file_id, "endpoint": "/v1/chat/completions",
        "completion_window": "24h", "metadata": {"ds_name": "milvus-ai-function-example", "ds_description": "AI Batch example"},
    })
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    batch_id = data.get("data", {}).get("id")

    for attempt in range(1, MAX_POLL_ATTEMPTS + 1):
        status, data = post_json("/v2/vectordb/ai/batch/jobs/describe", {"provider": "aliyun_milvus", "batch_id": batch_id})
        batch_data = data.get("data", {})
        batch_status = batch_data.get("status")
        if batch_status == "completed":
            output_file = Path(os.getenv("AIFUNC_AI_BATCH_OUTPUT_FILE", f"ai_batch_output_{batch_id}.jsonl"))
            download_batch_file(batch_id, "output", output_file)
            print(f"Batch output downloaded to: {output_file}")
            if batch_data.get("error_file_id"):
                download_batch_file(batch_id, "error", Path(f"ai_batch_errors_{batch_id}.jsonl"))
            break
        if batch_status in {"failed", "expired", "cancelled"}:
            print(f"Batch {batch_id} ended with status: {batch_status}", file=sys.stderr)
            sys.exit(1)
        if attempt < MAX_POLL_ATTEMPTS:
            time.sleep(POLL_INTERVAL_SEC)
finally:
    if temporary_input is not None:
        os.unlink(temporary_input.name)

预期结果：image-output.jsonl 使用同一 custom_id=milvus-image-1 返回图片描述和标签。业务侧可将文本写入案例库字段，再结合图片向量实现图文检索；若任务完成但有失败行，仍需下载 file_type=error 的结果文件。

示例三：批量生成视频素材摘要（视频）

运营团队需要为历史短视频生成一句摘要，供视频素材检索。每行通过 video 内容块传入视频 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="qwen3-vl-plus"
INPUT_FILE="${AIFUNC_AI_BATCH_INPUT_FILE:-}"
POLL_INTERVAL_SEC="${AIFUNC_AI_BATCH_POLL_INTERVAL_SEC:-5}"
MAX_POLL_ATTEMPTS="${AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS:-120}"

if [ -z "$INPUT_FILE" ]; then
  echo "请先将本示例 JSONL 请求保存为文件，并通过 AIFUNC_AI_BATCH_INPUT_FILE 指定路径。" >&2
  exit 1
fi

download_batch_file() {
  local file_type="$1"; local output_file="$2"; local content_body
  content_body="$(jq -nc --arg batch_id "$BATCH_ID" --arg file_type "$file_type" '{provider:"aliyun_milvus", batch_id:$batch_id, file_type:$file_type}')"
  curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/content" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" -H "Content-Type: application/json" \
    -d "$content_body" --output "$output_file"
}

UPLOAD_RESPONSE="$(curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/upload" \
  -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
  -F "provider=aliyun_milvus" -F "model_name=$MODEL_NAME" -F "endpoint=/v1/chat/completions" \
  -F "purpose=batch" -F "file=@$INPUT_FILE;type=application/jsonl")"
INPUT_FILE_ID="$(echo "$UPLOAD_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$INPUT_FILE_ID" ] || exit 1

CREATE_BODY="$(jq -nc --arg input_file_id "$INPUT_FILE_ID" '{provider:"aliyun_milvus", input_file_id:$input_file_id, endpoint:"/v1/chat/completions", completion_window:"24h", metadata:{ds_name:"milvus-ai-function-example", ds_description:"AI Batch REST example"}}')"
CREATE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/create" "$CREATE_BODY")"
BATCH_ID="$(echo "$CREATE_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$BATCH_ID" ] || exit 1

for ((attempt = 1; attempt <= MAX_POLL_ATTEMPTS; attempt++)); do
  DESCRIBE_BODY="$(jq -nc --arg batch_id "$BATCH_ID" '{provider:"aliyun_milvus", batch_id:$batch_id}')"
  DESCRIBE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/describe" "$DESCRIBE_BODY")"
  BATCH_STATUS="$(echo "$DESCRIBE_RESPONSE" | jq -r '.data.status // empty')"
  case "$BATCH_STATUS" in
    completed)
      OUTPUT_FILE="${AIFUNC_AI_BATCH_OUTPUT_FILE:-./ai_batch_output_${BATCH_ID}.jsonl}"
      download_batch_file "output" "$OUTPUT_FILE"
      echo "Batch output downloaded to: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID ended with status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Unknown status: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID did not finish after $MAX_POLL_ATTEMPTS checks." >&2
exit 1

Python

from __future__ import annotations

import json
import os
import shutil
import sys
import tempfile
import time
import uuid
from pathlib import Path
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>"

MODEL_NAME = "qwen3-vl-plus"
INPUT_MODE = os.getenv("AIFUNC_AI_BATCH_INPUT_MODE", "video").lower()
POLL_INTERVAL_SEC = int(os.getenv("AIFUNC_AI_BATCH_POLL_INTERVAL_SEC", "5"))
MAX_POLL_ATTEMPTS = int(os.getenv("AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS", "120"))


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


def batch_request(custom_id: str, content: Any, *, enable_thinking: bool | None = None) -> dict[str, Any]:
    body: dict[str, Any] = {"model": MODEL_NAME, "messages": [{"role": "user", "content": content}]}
    if enable_thinking is not None:
        body["enable_thinking"] = enable_thinking
    return {"custom_id": custom_id, "method": "POST", "url": "/v1/chat/completions", "body": body}


def default_requests(input_mode: str) -> list[dict[str, Any]]:
    if input_mode == "text":
        return [
            batch_request("milvus-summary-1", "For a support knowledge base, summarize in one Chinese sentence: Milvus stores and searches vector embeddings for RAG, recommendation, and multimodal retrieval.", enable_thinking=False),
            batch_request("milvus-summary-2", "Turn this incident note into a Chinese FAQ title and one-sentence answer: after documents are updated, regenerate embeddings before users search the new content.", enable_thinking=False),
            batch_request("milvus-summary-3", "Create a one-sentence Chinese catalog description for an enterprise RAG case that retrieves policy passages before the assistant answers.", enable_thinking=False),
        ]
    if input_mode == "image":
        return [batch_request("milvus-image-1", [
            {"type": "image_url", "image_url": {"url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"}},
            {"type": "text", "text": "For a pet-service media library, return a concise Chinese accessibility caption and up to three searchable subject tags for this uploaded case photo."},
        ])]
    if input_mode == "video":
        return [batch_request("milvus-video-1", [
            {"type": "video", "video": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4"},
            {"type": "text", "text": "For a marketing asset library, return a one-sentence Chinese scene summary, three retrieval keywords, and whether manual brand-safety review is needed."},
        ])]
    if input_mode == "audio":
        return [batch_request("milvus-audio-1", [
            {"type": "input_audio", "input_audio": {"data": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3", "format": "mp3"}},
            {"type": "text", "text": "For a customer-service hotline greeting archive, transcribe the welcome message and assess whether its purpose, service availability, and recording or privacy notice are clear; mark details not present in the audio as unconfirmed."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE must be one of text, image, video, or audio", file=sys.stderr)
    sys.exit(1)


def post_multipart_upload(input_file: str) -> tuple[int, dict[str, Any]]:
    boundary = f"----milvus-ai-batch-{uuid.uuid4().hex}"
    fields = {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "endpoint": "/v1/chat/completions", "purpose": "batch"}
    with tempfile.TemporaryFile(mode="w+b") as payload:
        for name, value in fields.items():
            payload.write(f"--{boundary}\r\n".encode())
            payload.write(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode())
            payload.write(value.encode())
            payload.write(b"\r\n")
        payload.write(f"--{boundary}\r\n".encode())
        payload.write(b'Content-Disposition: form-data; name="file"; filename="input.jsonl"\r\n')
        payload.write(b"Content-Type: application/jsonl\r\n\r\n")
        with open(input_file, "rb") as input_handle:
            shutil.copyfileobj(input_handle, payload)
        payload.write(b"\r\n")
        payload.write(f"--{boundary}--\r\n".encode())
        payload_length = payload.tell()
        payload.seek(0)
        request = Request(
            f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/upload",
            data=payload,
            headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Content-Length": str(payload_length)},
            method="POST",
        )
        try:
            with urlopen(request, timeout=600) as response:
                return response.status, json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            return exc.code, json.loads(exc.read().decode("utf-8"))


def download_batch_file(batch_id: str, file_type: str, output_file: Path) -> None:
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/content",
        data=json.dumps({"provider": "aliyun_milvus", "batch_id": batch_id, "file_type": file_type}, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=600) as response, output_file.open("wb") as output:
        shutil.copyfileobj(response, output)


temporary_input = None
INPUT_FILE = os.getenv("AIFUNC_AI_BATCH_INPUT_FILE")
if INPUT_FILE is None:
    temporary_input = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False, encoding="utf-8")
    for request in default_requests(INPUT_MODE):
        temporary_input.write(json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n")
    temporary_input.close()
    INPUT_FILE = temporary_input.name

try:
    status, data = post_multipart_upload(INPUT_FILE)
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    input_file_id = data.get("data", {}).get("id")

    status, data = post_json("/v2/vectordb/ai/batch/jobs/create", {
        "provider": "aliyun_milvus", "input_file_id": input_file_id, "endpoint": "/v1/chat/completions",
        "completion_window": "24h", "metadata": {"ds_name": "milvus-ai-function-example", "ds_description": "AI Batch example"},
    })
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    batch_id = data.get("data", {}).get("id")

    for attempt in range(1, MAX_POLL_ATTEMPTS + 1):
        status, data = post_json("/v2/vectordb/ai/batch/jobs/describe", {"provider": "aliyun_milvus", "batch_id": batch_id})
        batch_data = data.get("data", {})
        batch_status = batch_data.get("status")
        if batch_status == "completed":
            output_file = Path(os.getenv("AIFUNC_AI_BATCH_OUTPUT_FILE", f"ai_batch_output_{batch_id}.jsonl"))
            download_batch_file(batch_id, "output", output_file)
            print(f"Batch output downloaded to: {output_file}")
            if batch_data.get("error_file_id"):
                download_batch_file(batch_id, "error", Path(f"ai_batch_errors_{batch_id}.jsonl"))
            break
        if batch_status in {"failed", "expired", "cancelled"}:
            print(f"Batch {batch_id} ended with status: {batch_status}", file=sys.stderr)
            sys.exit(1)
        if attempt < MAX_POLL_ATTEMPTS:
            time.sleep(POLL_INTERVAL_SEC)
finally:
    if temporary_input is not None:
        os.unlink(temporary_input.name)

预期结果：video-output.jsonl 以 custom_id=milvus-video-1 返回视频摘要。将摘要写入视频素材元数据后，可结合视频向量和关键词完成检索；终态带有 error_file_id 时，应保留错误文件以定位失败任务。

示例四：夜间归档客服热线欢迎语（音频）

客服运营团队会定期归档各条服务热线的欢迎语，转写其内容并检查热线用途、服务时间、录音或隐私告知是否清晰，再按欢迎语版本 ID 回写配置中心。每行通过 input_audio 传入一段 MP3 欢迎语，使用支持音频理解的模型，并以唯一 custom_id 关联原始版本。

音频和转写可能包含个人信息或内部服务配置。运行前应确认采集授权及脱敏要求，输出仅保留到完成回写和抽检所需的期限，之后按组织的数据保留策略安全删除；不要将生产录音或结果提交到代码仓库。

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="qwen3.5-omni-plus"
INPUT_FILE="${AIFUNC_AI_BATCH_INPUT_FILE:-}"
POLL_INTERVAL_SEC="${AIFUNC_AI_BATCH_POLL_INTERVAL_SEC:-5}"
MAX_POLL_ATTEMPTS="${AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS:-120}"

if [ -z "$INPUT_FILE" ]; then
  echo "请先将本示例 JSONL 请求保存为文件，并通过 AIFUNC_AI_BATCH_INPUT_FILE 指定路径。" >&2
  exit 1
fi

download_batch_file() {
  local file_type="$1"; local output_file="$2"; local content_body
  content_body="$(jq -nc --arg batch_id "$BATCH_ID" --arg file_type "$file_type" '{provider:"aliyun_milvus", batch_id:$batch_id, file_type:$file_type}')"
  curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/content" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" -H "Content-Type: application/json" \
    -d "$content_body" --output "$output_file"
}

UPLOAD_RESPONSE="$(curl --fail --silent --show-error -X POST "$MILVUS_REST_BASE_URL/v2/vectordb/ai/batch/files/upload" \
  -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
  -F "provider=aliyun_milvus" -F "model_name=$MODEL_NAME" -F "endpoint=/v1/chat/completions" \
  -F "purpose=batch" -F "file=@$INPUT_FILE;type=application/jsonl")"
INPUT_FILE_ID="$(echo "$UPLOAD_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$INPUT_FILE_ID" ] || exit 1

CREATE_BODY="$(jq -nc --arg input_file_id "$INPUT_FILE_ID" '{provider:"aliyun_milvus", input_file_id:$input_file_id, endpoint:"/v1/chat/completions", completion_window:"24h", metadata:{ds_name:"milvus-ai-function-example", ds_description:"AI Batch REST example"}}')"
CREATE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/create" "$CREATE_BODY")"
BATCH_ID="$(echo "$CREATE_RESPONSE" | jq -r '.data.id // empty')"
[ -n "$BATCH_ID" ] || exit 1

for ((attempt = 1; attempt <= MAX_POLL_ATTEMPTS; attempt++)); do
  DESCRIBE_BODY="$(jq -nc --arg batch_id "$BATCH_ID" '{provider:"aliyun_milvus", batch_id:$batch_id}')"
  DESCRIBE_RESPONSE="$(post_json "/v2/vectordb/ai/batch/jobs/describe" "$DESCRIBE_BODY")"
  BATCH_STATUS="$(echo "$DESCRIBE_RESPONSE" | jq -r '.data.status // empty')"
  case "$BATCH_STATUS" in
    completed)
      OUTPUT_FILE="${AIFUNC_AI_BATCH_OUTPUT_FILE:-./ai_batch_output_${BATCH_ID}.jsonl}"
      download_batch_file "output" "$OUTPUT_FILE"
      echo "Batch output downloaded to: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID ended with status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Unknown status: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID did not finish after $MAX_POLL_ATTEMPTS checks." >&2
exit 1

Python

from __future__ import annotations

import json
import os
import shutil
import sys
import tempfile
import time
import uuid
from pathlib import Path
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>"

MODEL_NAME = "qwen3.5-omni-plus"
INPUT_MODE = os.getenv("AIFUNC_AI_BATCH_INPUT_MODE", "audio").lower()
POLL_INTERVAL_SEC = int(os.getenv("AIFUNC_AI_BATCH_POLL_INTERVAL_SEC", "5"))
MAX_POLL_ATTEMPTS = int(os.getenv("AIFUNC_AI_BATCH_MAX_POLL_ATTEMPTS", "120"))


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


def batch_request(custom_id: str, content: Any, *, enable_thinking: bool | None = None) -> dict[str, Any]:
    body: dict[str, Any] = {"model": MODEL_NAME, "messages": [{"role": "user", "content": content}]}
    if enable_thinking is not None:
        body["enable_thinking"] = enable_thinking
    return {"custom_id": custom_id, "method": "POST", "url": "/v1/chat/completions", "body": body}


def default_requests(input_mode: str) -> list[dict[str, Any]]:
    if input_mode == "text":
        return [
            batch_request("milvus-summary-1", "For a support knowledge base, summarize in one Chinese sentence: Milvus stores and searches vector embeddings for RAG, recommendation, and multimodal retrieval.", enable_thinking=False),
            batch_request("milvus-summary-2", "Turn this incident note into a Chinese FAQ title and one-sentence answer: after documents are updated, regenerate embeddings before users search the new content.", enable_thinking=False),
            batch_request("milvus-summary-3", "Create a one-sentence Chinese catalog description for an enterprise RAG case that retrieves policy passages before the assistant answers.", enable_thinking=False),
        ]
    if input_mode == "image":
        return [batch_request("milvus-image-1", [
            {"type": "image_url", "image_url": {"url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"}},
            {"type": "text", "text": "For a pet-service media library, return a concise Chinese accessibility caption and up to three searchable subject tags for this uploaded case photo."},
        ])]
    if input_mode == "video":
        return [batch_request("milvus-video-1", [
            {"type": "video", "video": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4"},
            {"type": "text", "text": "For a marketing asset library, return a one-sentence Chinese scene summary, three retrieval keywords, and whether manual brand-safety review is needed."},
        ])]
    if input_mode == "audio":
        return [batch_request("milvus-audio-1", [
            {"type": "input_audio", "input_audio": {"data": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3", "format": "mp3"}},
            {"type": "text", "text": "For a customer-service hotline greeting archive, transcribe the welcome message and assess whether its purpose, service availability, and recording or privacy notice are clear; mark details not present in the audio as unconfirmed."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE must be one of text, image, video, or audio", file=sys.stderr)
    sys.exit(1)


def post_multipart_upload(input_file: str) -> tuple[int, dict[str, Any]]:
    boundary = f"----milvus-ai-batch-{uuid.uuid4().hex}"
    fields = {"provider": "aliyun_milvus", "model_name": MODEL_NAME, "endpoint": "/v1/chat/completions", "purpose": "batch"}
    with tempfile.TemporaryFile(mode="w+b") as payload:
        for name, value in fields.items():
            payload.write(f"--{boundary}\r\n".encode())
            payload.write(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode())
            payload.write(value.encode())
            payload.write(b"\r\n")
        payload.write(f"--{boundary}\r\n".encode())
        payload.write(b'Content-Disposition: form-data; name="file"; filename="input.jsonl"\r\n')
        payload.write(b"Content-Type: application/jsonl\r\n\r\n")
        with open(input_file, "rb") as input_handle:
            shutil.copyfileobj(input_handle, payload)
        payload.write(b"\r\n")
        payload.write(f"--{boundary}--\r\n".encode())
        payload_length = payload.tell()
        payload.seek(0)
        request = Request(
            f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/upload",
            data=payload,
            headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": f"multipart/form-data; boundary={boundary}", "Content-Length": str(payload_length)},
            method="POST",
        )
        try:
            with urlopen(request, timeout=600) as response:
                return response.status, json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            return exc.code, json.loads(exc.read().decode("utf-8"))


def download_batch_file(batch_id: str, file_type: str, output_file: Path) -> None:
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/content",
        data=json.dumps({"provider": "aliyun_milvus", "batch_id": batch_id, "file_type": file_type}, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_AUTH_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=600) as response, output_file.open("wb") as output:
        shutil.copyfileobj(response, output)


temporary_input = None
INPUT_FILE = os.getenv("AIFUNC_AI_BATCH_INPUT_FILE")
if INPUT_FILE is None:
    temporary_input = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False, encoding="utf-8")
    for request in default_requests(INPUT_MODE):
        temporary_input.write(json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n")
    temporary_input.close()
    INPUT_FILE = temporary_input.name

try:
    status, data = post_multipart_upload(INPUT_FILE)
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    input_file_id = data.get("data", {}).get("id")

    status, data = post_json("/v2/vectordb/ai/batch/jobs/create", {
        "provider": "aliyun_milvus", "input_file_id": input_file_id, "endpoint": "/v1/chat/completions",
        "completion_window": "24h", "metadata": {"ds_name": "milvus-ai-function-example", "ds_description": "AI Batch example"},
    })
    if status != 200 or data.get("code") != 0:
        sys.exit(1)
    batch_id = data.get("data", {}).get("id")

    for attempt in range(1, MAX_POLL_ATTEMPTS + 1):
        status, data = post_json("/v2/vectordb/ai/batch/jobs/describe", {"provider": "aliyun_milvus", "batch_id": batch_id})
        batch_data = data.get("data", {})
        batch_status = batch_data.get("status")
        if batch_status == "completed":
            output_file = Path(os.getenv("AIFUNC_AI_BATCH_OUTPUT_FILE", f"ai_batch_output_{batch_id}.jsonl"))
            download_batch_file(batch_id, "output", output_file)
            print(f"Batch output downloaded to: {output_file}")
            if batch_data.get("error_file_id"):
                download_batch_file(batch_id, "error", Path(f"ai_batch_errors_{batch_id}.jsonl"))
            break
        if batch_status in {"failed", "expired", "cancelled"}:
            print(f"Batch {batch_id} ended with status: {batch_status}", file=sys.stderr)
            sys.exit(1)
        if attempt < MAX_POLL_ATTEMPTS:
            time.sleep(POLL_INTERVAL_SEC)
finally:
    if temporary_input is not None:
        os.unlink(temporary_input.name)

预期结果：call-greeting-output.jsonl 或 call-greeting-error.jsonl 中恰有一行 custom_id=milvus-audio-1。成功结果经有限范围校验后写入配置中心并交由客服运营人员抽检；失败行保留原 custom_id，可在修复音频权限、MP3 格式或模型参数后单独重试。