Inférence par lot - Vector Retrieval Service for Milvus - Centre de documentation Alibaba Cloud

L'inférence par lot permet de soumettre des requêtes Chat Completions au format JSONL (texte ou multimodal) sous forme de tâches asynchrones. Utilisez cette fonctionnalité pour les charges de travail hors ligne de longue durée, telles que la synthèse nocturne, l'étiquetage de données historiques, l'évaluation de modèles et l'annotation de données, dont l'exécution peut prendre de quelques minutes à plusieurs heures.

Fonctionnement

Les tâches par lot suivent le modèle d'entrée de fichier Batch OpenAI et de corrélation des résultats : chaque requête utilise un identifiant unique custom_id. Une fois la tâche terminée, les données métier sont réécrites à partir du fichier JSONL de sortie ou d'erreur en utilisant cet identifiant.

@id="p0008b3x8t" Soumission de la tâche — Téléchargez un fichier JSONL encodé en UTF-8 contenant plusieurs requêtes, puis utilisez l'ID de fichier renvoyé pour créer une tâche par lot.

@id="p0010b3x8t" Traitement asynchrone — Le serveur valide et exécute les requêtes ligne par ligne en arrière-plan. L'application interroge les états tels que validating, in_progress et finalizing en utilisant le batch_id.

@id="p0012b3x8t" Téléchargement des résultats — Lorsque la tâche atteint un état terminal (un état indiquant que la tâche est terminée et ne changera plus), téléchargez le fichier output. Si certaines lignes ont échoué, téléchargez le fichier error et utilisez le custom_id pour réécrire les résultats ou identifier les problèmes.

L'inférence par lot convient à l'évaluation hors ligne des modèles, à l'annotation de données historiques, au traitement par lot de contenu et au traitement planifié des actifs. Elle n'est pas adaptée aux requêtes en ligne, telles que les complétions de chat ou de recherche, qui nécessitent des résultats immédiats.

Format des commandes

L'exécution par lot suit quatre étapes : téléchargement du fichier d'entrée → création de la tâche → interrogation du statut → téléchargement des résultats.

Le fichier d'entrée doit être un fichier JSONL encodé en UTF-8. Chaque ligne constitue une requête indépendante contenant un custom_id unique, la valeur fixe method="POST", ainsi qu'un url et un body.model cohérents avec la tâche.

REST API

@id="c0017b3x8t"

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":"<file ID>","endpoint":"/v1/chat/completions","completion_window":"24h"}

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

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

Python

@id="cmdpysecbat01"

Python

@id="cmdpycodbat01"

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

@id="cmdpyhbat0001"

Le script REST dépend de jq.

Paramètres

@id="p0079b3x8t" Paramètre	@id="p0081b3x8t" Description
@id="p0021b3x8t" `file`	@id="p0023b3x8t" Requis pour le téléchargement. Fichier d'entrée JSONL. Le serveur valide le `custom_id`, le `method`, l'`url` et le `body.model` de la première ligne.
@id="p0026b3x8t" `model_name`	@id="p0028b3x8t" Requis pour le téléchargement ; compatible également avec `model`. Doit correspondre au `body.model` de la première ligne du fichier JSONL.
@id="p0031b3x8t" `endpoint`	@id="p0033b3x8t" Requis pour le téléchargement et la création. Utilisez `/v1/chat/completions` pour les Chat Completions ; utilisez `/v1/embeddings` pour les tâches d'embedding de texte. Toutes les lignes d'un même fichier doivent utiliser le même endpoint.
@id="p0036b3x8t" `purpose`	@id="p0038b3x8t" Facultatif pour le téléchargement. Peut être vide ou avoir la valeur `batch`.
@id="p0041b3x8t" `input_file_id`	@id="p0043b3x8t" Requis pour la création d'une tâche. ID de fichier renvoyé par l'API de téléchargement. Les URL OSS ou les identifiants de fichiers externes ne sont pas acceptés.
@id="p0046b3x8t" `completion_window`	@id="p0048b3x8t" Requis pour la création d'une tâche. Fenêtre d'exécution ; prend en charge les valeurs de `24h` à `336h` ou des unités en jours.
@id="p0051b3x8t" `metadata.ds_name`/`metadata.ds_description`	@id="p0053b3x8t" Facultatif. Le nom de la tâche peut contenir jusqu'à 100 caractères ; la description peut contenir jusqu'à 200 caractères.
@id="p0056b3x8t" `batch_id`	@id="p0058b3x8t" Requis pour l'interrogation et le téléchargement. ID de tâche renvoyé par l'API de création.
@id="p0061b3x8t" `file_type`	@id="p0063b3x8t" Requis pour le téléchargement. Utilisez `output` pour télécharger les lignes réussies ; utilisez `error` pour télécharger les lignes ayant échoué.
@id="p0066b3x8t" `provider`	@id="p0068b3x8t" Facultatif ; la valeur par défaut est `aliyun_milvus`. Le téléchargement, la création, l'interrogation et le téléchargement doivent utiliser le même compte Alibaba Cloud, la même région et le même workspace.
@id="p0071b3x8t" `body.enable_thinking`	@id="p0073b3x8t" Facultatif, selon le modèle. Doit se trouver au même niveau que `body.model` dans la ligne JSONL. Pour les modèles avec la fonction de réflexion activée par défaut, définissez explicitement ce paramètre sur `false` pour éviter les jetons de réflexion indésirables. Ne placez pas ce paramètre dans `extra_body`.

Valeurs de retour

Après la création réussie d'une tâche, data.id est renvoyé en tant que batch_id. Les états intermédiaires incluent validating, in_progress, finalizing et cancelling. Les états terminaux (états indiquant que la tâche est terminée et ne changera plus) incluent completed, failed, expired et cancelled.

Les tâches par lot sont traitées de manière asynchrone dans une file d'attente. Le temps écoulé entre la création et l'atteinte d'un état terminal varie généralement de quelques minutes à plusieurs heures, période pendant laquelle le statut reste in_progress. Vous pouvez appeler l'API d'interrogation POST /v2/vectordb/ai/batch/jobs/describe (avec le batch_id) à tout moment pour obtenir le dernier status. Si le sondage du script exemple se termine avant que la tâche ne soit achevée, vous pouvez interroger à nouveau ultérieurement en utilisant le batch_id une fois que le statut passe à completed, puis télécharger les résultats. L'API de téléchargement renvoie directement le contenu JSONL ; chaque ligne conserve le custom_id d'entrée pour permettre la réécriture des résultats dans les données métier. Les tâches par lot ne conviennent pas aux interactions en ligne nécessitant des réponses immédiates.

Exemple 1 : Génération nocturne de résumés pour une base de connaissances (texte)

Une équipe opérationnelle doit générer par lot des résumés pour les articles ajoutés dans la journée. Chaque ligne contient une requête pour un article. Les paramètres clés sont body.model=qwen3.7-max, url=/v1/chat/completions et completion_window=24h.

REST API

@id="c0089b3x8t"

#!/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 "Save the sample JSONL requests to a file first, then set AIFUNC_AI_BATCH_INPUT_FILE to the file path." >&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

@id="pt0094b3x8t"

Python

@id="c0090b3x8t"

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 English 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 an English 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 English 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 English 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 English 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)

@id="h0095b3x8t"

Résultat attendu : une fois la tâche passée à l'état completed, chaque ligne du fichier output.jsonl téléchargé conserve son custom_id. Par exemple :

{"custom_id":"milvus-summary-1","response":{"status_code":200,"body":{"choices":[{"message":{"role":"assistant","content":"Milvus is a vector database that serves RAG, recommendation, and multimodal retrieval scenarios."}}]}}}

Côté application, réécrivez les résumés dans les articles correspondants en utilisant le custom_id. Si un fichier error.jsonl est produit, relancez uniquement les articles ayant échoué.

Exemple 2 : Génération par lot de descriptions d'images pour des cas de service animalier (image)

Une plateforme de services pour animaux de compagnie doit générer des descriptions d'accessibilité et des tags de récupération pour les images de cas historiques téléchargées par les marchands. Le champ content de chaque ligne inclut un image_url et une instruction textuelle. Utilisez un modèle multimodal prenant en charge la compréhension d'image ; le model_name spécifié lors du téléchargement doit correspondre.

REST API

@id="c0100b3x8t"

#!/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 "Save the sample JSONL requests to a file first, then set AIFUNC_AI_BATCH_INPUT_FILE to the file path." >&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

@id="pt0105b3x8t"

Python

@id="c0101b3x8t"

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 English 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 an English 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 English 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 English 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 English 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)

@id="h0106b3x8t"

Résultat attendu : le fichier image-output.jsonl renvoie la description de l'image et les tags en utilisant le même custom_id=milvus-image-1. Côté application, vous pouvez inscrire le texte dans les champs de la bibliothèque de cas et le combiner avec les vecteurs d'image pour la récupération image-texte. Si la tâche se termine mais comporte des lignes ayant échoué, téléchargez le fichier de résultat avec file_type=error.

Exemple 3 : Génération de résumés par lot pour des vidéos (vidéo)

Une équipe opérationnelle doit générer un résumé d'une phrase pour des vidéos courtes historiques afin de faciliter la recherche dans les ressources vidéo. Chaque ligne transmet une URL de vidéo via un bloc de contenu video. Utilisez un modèle multimodal prenant en charge la compréhension vidéo.

REST API

@id="c0111b3x8t"

#!/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 "Save the sample JSONL requests to a file first, then set AIFUNC_AI_BATCH_INPUT_FILE to the file path." >&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

@id="pt0116b3x8t"

Python

@id="c0112b3x8t"

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 English 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 an English 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 English 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 English 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 English 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)

@id="h0117b3x8t"

Résultat attendu : le fichier video-output.jsonl renvoie le résumé vidéo avec custom_id=milvus-video-1. Après avoir intégré le résumé dans les métadonnées de la ressource vidéo, vous pouvez combiner les vecteurs vidéo et les mots-clés pour la recherche. Si l'état final inclut un error_file_id, conservez le fichier d'erreur pour identifier les tâches ayant échoué.

Exemple 4 : Archivage nocturne des messages d'accueil des lignes d'assistance client (audio)

Une équipe opérationnelle du service client archive périodiquement les messages d'accueil de chaque ligne d'assistance. Elle transcrit le contenu et vérifie si l'objectif de la ligne, les horaires de service ainsi que les mentions d'enregistrement ou de confidentialité sont clairs, puis écrit les résultats dans le centre de configuration à l'aide de l'ID de version du message d'accueil. Chaque ligne transmet un message d'accueil MP3 via input_audio en utilisant un modèle prenant en charge la compréhension audio, avec un custom_id unique pour corréler la version d'origine.

Les fichiers audio et les transcriptions peuvent contenir des informations personnelles ou des configurations internes de service. Vérifiez les autorisations de collecte et les exigences de masquage des données avant l'exécution. Conservez les résultats uniquement pendant la période nécessaire pour effectuer l'écriture inverse et les contrôles ponctuels, puis supprimez-les de manière sécurisée conformément à la politique de conservation des données de votre organisation. Ne commitez pas les enregistrements de production ni les résultats dans un dépôt de code.

REST API

@id="c0123b3x8t"

#!/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 "Save the sample JSONL requests to a file first, then set AIFUNC_AI_BATCH_INPUT_FILE to the file path." >&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

@id="pt0128b3x8t"

Python

@id="c0124b3x8t"

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 English 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 an English 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 English 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 English 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 English 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)

@id="h0129b3x8t"

Résultat attendu : les fichiers call-greeting-output.jsonl ou call-greeting-error.jsonl contiennent exactement une ligne avec custom_id=milvus-audio-1. Après une vérification à portée limitée, les résultats réussis sont écrits dans le centre de configuration pour un contrôle ponctuel par le personnel opérationnel du service client. Les lignes ayant échoué conservent le custom_id d'origine et peuvent être relancées individuellement après correction des autorisations audio, du format MP3 ou des paramètres du modèle.