All Products
Search

This Product

    Vector Retrieval Service for Milvus:Inferensi Batch

    Document Center

    Vector Retrieval Service for Milvus:Inferensi Batch

    Last Updated:Aug 04, 2026

    Inferensi Batch mengirimkan permintaan teks dalam format JSONL atau permintaan Chat Completions multimodal sebagai tugas batch asinkron. Fitur ini cocok untuk beban kerja offline jangka panjang seperti Ringkasan semalam, pelabelan data historis, evaluasi model, dan anotasi data yang dapat memakan waktu beberapa menit hingga jam.

    Cara kerja

    Tugas batch mengikuti pola input file dan korelasi hasil OpenAI Batch: setiap permintaan menggunakan custom_id unik, dan setelah selesai, data bisnis ditulis balik dari output atau error JSONL menggunakan pengidentifikasi tersebut.

    • Kirimkan tugas — Unggah file JSONL ber-encoding UTF-8 yang berisi beberapa permintaan, lalu gunakan ID file yang dikembalikan untuk membuat tugas batch.

    • Pemrosesan asinkron — Server memvalidasi dan mengeksekusi permintaan baris per baris di latar belakang. Aplikasi melakukan kueri status seperti validating, in_progress, dan finalizing menggunakan batch_id.

    • Unduh hasil — Setelah tugas mencapai status terminal (status yang menandakan tugas telah selesai dan tidak akan berubah lagi), unduh output. Jika terdapat baris yang gagal, unduh error dan gunakan custom_id untuk menulis balik hasil atau melokalisasi masalah.

    Inferensi Batch cocok untuk evaluasi model offline, anotasi data historis, pemrosesan batch konten, dan pemrosesan aset terjadwal. Fitur ini tidak cocok untuk permintaan online seperti chat atau search completions yang memerlukan respons segera.

    Format perintah

    Eksekusi batch mengikuti empat langkah: unggah file input → buat tugas → kueri status → unduh hasil.

    File input harus berformat JSONL ber-encoding UTF-8. Setiap baris merupakan permintaan independen yang berisi custom_id unik, nilai tetap method="POST", serta url dan body.model yang konsisten dengan tugas.

    REST API

    REST API

    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

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

    Skrip REST bergantung pada jq.

    Parameter

    ParameterDeskripsi
    fileDiperlukan saat unggah. File input JSONL. Server memvalidasi custom_id, method, url, dan body.model pada baris pertama.
    model_nameDiperlukan saat unggah; juga kompatibel dengan model. Harus sesuai dengan body.model pada baris pertama JSONL.
    endpointDiperlukan saat unggah dan pembuatan. Gunakan /v1/chat/completions untuk Chat Completions; gunakan /v1/embeddings untuk tugas text embedding. Semua baris dalam file yang sama harus menggunakan endpoint yang sama.
    purposeOpsional saat unggah. Hanya boleh kosong atau batch.
    input_file_idDiperlukan saat membuat tugas. ID file yang dikembalikan oleh API unggah. URL OSS atau pengidentifikasi file eksternal tidak diterima.
    completion_windowDiperlukan saat membuat tugas. Jendela penyelesaian; mendukung 24h hingga 336h atau satuan hari.
    metadata.ds_name/metadata.ds_descriptionOpsional. Nama tugas maksimal 100 karakter; deskripsi maksimal 200 karakter.
    batch_idDiperlukan saat kueri dan unduh. ID tugas yang dikembalikan oleh API pembuatan.
    file_typeDiperlukan saat unduh. Gunakan output untuk mengunduh baris yang berhasil; gunakan error untuk mengunduh baris yang gagal.
    providerOpsional; default ke aliyun_milvus. Unggah, pembuatan, kueri, dan unduh harus menggunakan Akun Alibaba Cloud, Wilayah, dan ruang kerja yang sama.
    body.enable_thinkingOpsional, tergantung model. Harus berada pada level yang sama dengan body.model dalam baris JSONL. Untuk model yang secara default mengaktifkan thinking, atur eksplisit parameter ini ke false untuk menghindari token thinking yang tidak diinginkan. Jangan letakkan parameter ini di dalam extra_body.

    Nilai kembalian

    Setelah tugas berhasil dibuat, data.id dikembalikan sebagai batch_id. Status polling mencakup validating, in_progress, finalizing, dan cancelling. Status terminal (status yang menandakan tugas telah selesai dan tidak akan berubah lagi) mencakup completed, failed, expired, dan cancelled.

    Tugas batch diproses secara asinkron dalam antrian. Waktu dari pembuatan hingga mencapai status terminal biasanya berkisar antara beberapa menit hingga beberapa jam, selama itu status tetap in_progress. Anda dapat memanggil API kueri POST /v2/vectordb/ai/batch/jobs/describe (dengan batch_id) kapan saja untuk mendapatkan status terbaru. Jika polling skrip contoh berakhir sebelum tugas selesai, Anda dapat melakukan kueri ulang nanti menggunakan batch_id begitu status berubah menjadi completed, lalu mengunduh hasilnya. API unduh mengembalikan konten JSONL secara langsung; setiap baris mempertahankan custom_id input untuk menulis hasil balik ke data bisnis. Tugas batch tidak cocok untuk interaksi online yang memerlukan respons segera.

    Contoh 1: Generasi Ringkasan basis pengetahuan batch semalam (teks)

    Tim operasional perlu menghasilkan Ringkasan secara batch untuk artikel yang ditambahkan hari itu. Setiap baris berisi satu permintaan artikel. Parameter utama adalah body.model=qwen3.7-max, url=/v1/chat/completions, dan completion_window=24h.

    REST API

    REST API

    #!/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 "Simpan contoh permintaan JSONL ke file terlebih dahulu, lalu atur AIFUNC_AI_BATCH_INPUT_FILE ke path file tersebut." >&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 "Output batch diunduh ke: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID berakhir dengan status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Status tidak dikenal: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID tidak selesai setelah $MAX_POLL_ATTEMPTS kali pemeriksaan." >&2
exit 1

    Python

    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", "Untuk basis pengetahuan dukungan, ringkas dalam satu kalimat Bahasa Mandarin: Milvus menyimpan dan mencari penyematan vektor untuk RAG, rekomendasi, dan pengambilan multimodal.", enable_thinking=False),
            batch_request("milvus-summary-2", "Ubah catatan insiden ini menjadi judul FAQ Bahasa Mandarin dan jawaban satu kalimat: setelah dokumen diperbarui, hasilkan ulang penyematan sebelum pengguna mencari konten baru.", enable_thinking=False),
            batch_request("milvus-summary-3", "Buat deskripsi katalog satu kalimat Bahasa Mandarin untuk kasus RAG perusahaan yang mengambil cuplikan kebijakan sebelum asisten menjawab.", 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": "Untuk pustaka media layanan hewan peliharaan, kembalikan keterangan aksesibilitas Bahasa Mandarin yang ringkas dan hingga tiga tag subjek yang dapat dicari untuk foto kasus yang diunggah ini."},
        ])]
    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": "Untuk pustaka aset pemasaran, kembalikan ringkasan adegan satu kalimat Bahasa Mandarin, tiga kata kunci pengambilan, dan apakah diperlukan tinjauan keselamatan merek manual."},
        ])]
    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": "Untuk arsip sapaan hotline layanan pelanggan, transkripsikan pesan sambutan dan nilai apakah tujuan, ketersediaan layanan, serta pemberitahuan rekaman atau privasi jelas; tandai detail yang tidak ada dalam audio sebagai belum dikonfirmasi."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE harus salah satu dari text, image, video, atau 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": "Contoh AI Batch"},
    })
    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"Output batch diunduh ke: {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} berakhir dengan 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)

    Hasil yang diharapkan: setelah tugas mencapai status completed, setiap baris dalam output.jsonl yang diunduh mempertahankan custom_id-nya. Contohnya:

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

    Sisi aplikasi menulis balik Ringkasan ke artikel yang sesuai menggunakan custom_id. Jika dihasilkan error.jsonl, coba ulang hanya artikel yang gagal.

    Contoh 2: Generasi deskripsi gambar kasus layanan hewan peliharaan batch (gambar)

    Platform layanan hewan peliharaan perlu menghasilkan deskripsi aksesibilitas dan tag pengambilan untuk gambar kasus historis yang diunggah oleh penjual. Setiap baris content mencakup image_url dan instruksi teks. Gunakan model multimodal yang mendukung pemahaman gambar; model_name yang ditentukan saat pengunggahan harus sesuai.

    REST API

    REST API

    #!/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 "Simpan contoh permintaan JSONL ke file terlebih dahulu, lalu atur AIFUNC_AI_BATCH_INPUT_FILE ke path file tersebut." >&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 "Output batch diunduh ke: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID berakhir dengan status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Status tidak dikenal: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID tidak selesai setelah $MAX_POLL_ATTEMPTS kali pemeriksaan." >&2
exit 1

    Python

    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", "Untuk basis pengetahuan dukungan, ringkas dalam satu kalimat Bahasa Mandarin: Milvus menyimpan dan mencari penyematan vektor untuk RAG, rekomendasi, dan pengambilan multimodal.", enable_thinking=False),
            batch_request("milvus-summary-2", "Ubah catatan insiden ini menjadi judul FAQ Bahasa Mandarin dan jawaban satu kalimat: setelah dokumen diperbarui, hasilkan ulang penyematan sebelum pengguna mencari konten baru.", enable_thinking=False),
            batch_request("milvus-summary-3", "Buat deskripsi katalog satu kalimat Bahasa Mandarin untuk kasus RAG perusahaan yang mengambil cuplikan kebijakan sebelum asisten menjawab.", 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": "Untuk pustaka media layanan hewan peliharaan, kembalikan keterangan aksesibilitas Bahasa Mandarin yang ringkas dan hingga tiga tag subjek yang dapat dicari untuk foto kasus yang diunggah ini."},
        ])]
    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": "Untuk pustaka aset pemasaran, kembalikan ringkasan adegan satu kalimat Bahasa Mandarin, tiga kata kunci pengambilan, dan apakah diperlukan tinjauan keselamatan merek manual."},
        ])]
    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": "Untuk arsip sapaan hotline layanan pelanggan, transkripsikan pesan sambutan dan nilai apakah tujuan, ketersediaan layanan, serta pemberitahuan rekaman atau privasi jelas; tandai detail yang tidak ada dalam audio sebagai belum dikonfirmasi."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE harus salah satu dari text, image, video, atau 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": "Contoh AI Batch"},
    })
    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"Output batch diunduh ke: {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} berakhir dengan 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)

    Hasil yang diharapkan: image-output.jsonl mengembalikan deskripsi dan tag gambar menggunakan custom_id=milvus-image-1 yang sama. Sisi aplikasi dapat menulis teks ke bidang pustaka kasus dan menggabungkannya dengan vektor gambar untuk pengambilan gambar-teks. Jika tugas selesai tetapi memiliki baris yang gagal, unduh file hasil file_type=error.

    Contoh 3: Generasi Ringkasan materi video batch (video)

    Tim operasi perlu menghasilkan ringkasan satu kalimat untuk video pendek historis guna mendukung pengambilan aset video. Setiap baris meneruskan URL video melalui blok konten video. Gunakan model multimodal yang mendukung pemahaman video.

    REST API

    REST API

    #!/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 "Simpan contoh permintaan JSONL ke file terlebih dahulu, lalu atur AIFUNC_AI_BATCH_INPUT_FILE ke path file tersebut." >&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 "Output batch diunduh ke: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID berakhir dengan status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Status tidak dikenal: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID tidak selesai setelah $MAX_POLL_ATTEMPTS kali pemeriksaan." >&2
exit 1

    Python

    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", "Untuk basis pengetahuan dukungan, ringkas dalam satu kalimat Bahasa Mandarin: Milvus menyimpan dan mencari penyematan vektor untuk RAG, rekomendasi, dan pengambilan multimodal.", enable_thinking=False),
            batch_request("milvus-summary-2", "Ubah catatan insiden ini menjadi judul FAQ Bahasa Mandarin dan jawaban satu kalimat: setelah dokumen diperbarui, hasilkan ulang penyematan sebelum pengguna mencari konten baru.", enable_thinking=False),
            batch_request("milvus-summary-3", "Buat deskripsi katalog satu kalimat Bahasa Mandarin untuk kasus RAG perusahaan yang mengambil cuplikan kebijakan sebelum asisten menjawab.", 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": "Untuk pustaka media layanan hewan peliharaan, kembalikan keterangan aksesibilitas Bahasa Mandarin yang ringkas dan hingga tiga tag subjek yang dapat dicari untuk foto kasus yang diunggah ini."},
        ])]
    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": "Untuk pustaka aset pemasaran, kembalikan ringkasan adegan satu kalimat Bahasa Mandarin, tiga kata kunci pengambilan, dan apakah diperlukan tinjauan keselamatan merek manual."},
        ])]
    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": "Untuk arsip sapaan hotline layanan pelanggan, transkripsikan pesan sambutan dan nilai apakah tujuan, ketersediaan layanan, serta pemberitahuan rekaman atau privasi jelas; tandai detail yang tidak ada dalam audio sebagai belum dikonfirmasi."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE harus salah satu dari text, image, video, atau 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": "Contoh AI Batch"},
    })
    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"Output batch diunduh ke: {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} berakhir dengan 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)

    Hasil yang diharapkan: video-output.jsonl mengembalikan Ringkasan video dengan custom_id=milvus-video-1. Setelah menulis Ringkasan ke metadata aset video, Anda dapat menggabungkan vektor video dan kata kunci untuk pengambilan. Jika status terminal mencakup error_file_id, simpan file error untuk melokalisasi tugas yang gagal.

    Contoh 4: Pengarsipan semalam sapaan hotline layanan pelanggan (audio)

    Tim operasional layanan pelanggan secara berkala mengarsipkan sapaan setiap hotline layanan, mentranskripsikan kontennya dan memeriksa apakah tujuan hotline, jam layanan, serta pemberitahuan rekaman atau privasi jelas, lalu menulis hasilnya kembali ke pusat konfigurasi menggunakan ID versi sapaan. Setiap baris meneruskan sapaan MP3 melalui input_audio menggunakan model yang mendukung Pemahaman audio, dengan custom_id unik untuk menghubungkan versi aslinya.

    Audio dan transkrip dapat berisi informasi pribadi atau konfigurasi layanan internal. Konfirmasi otorisasi pengumpulan dan persyaratan desensitisasi sebelum menjalankan. Simpan output hanya untuk periode yang diperlukan guna menyelesaikan tulis balik dan pemeriksaan sampel, lalu hapus secara aman sesuai Kebijakan retensi data organisasi Anda. Jangan commit rekaman produksi atau hasilnya ke repositori kode.

    REST API

    REST API

    #!/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 "Simpan contoh permintaan JSONL ke file terlebih dahulu, lalu atur AIFUNC_AI_BATCH_INPUT_FILE ke path file tersebut." >&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 "Output batch diunduh ke: $OUTPUT_FILE"
      exit 0 ;;
    failed|expired|cancelled)
      echo "Batch $BATCH_ID berakhir dengan status: $BATCH_STATUS" >&2; exit 1 ;;
    validating|in_progress|finalizing|cancelling)
      [ "$attempt" -lt "$MAX_POLL_ATTEMPTS" ] && sleep "$POLL_INTERVAL_SEC" ;;
    *) echo "Status tidak dikenal: ${BATCH_STATUS:-empty}" >&2; exit 1 ;;
  esac
done
echo "Batch $BATCH_ID tidak selesai setelah $MAX_POLL_ATTEMPTS kali pemeriksaan." >&2
exit 1

    Python

    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", "Untuk basis pengetahuan dukungan, ringkas dalam satu kalimat Bahasa Mandarin: Milvus menyimpan dan mencari penyematan vektor untuk RAG, rekomendasi, dan pengambilan multimodal.", enable_thinking=False),
            batch_request("milvus-summary-2", "Ubah catatan insiden ini menjadi judul FAQ Bahasa Mandarin dan jawaban satu kalimat: setelah dokumen diperbarui, hasilkan ulang penyematan sebelum pengguna mencari konten baru.", enable_thinking=False),
            batch_request("milvus-summary-3", "Buat deskripsi katalog satu kalimat Bahasa Mandarin untuk kasus RAG perusahaan yang mengambil cuplikan kebijakan sebelum asisten menjawab.", 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": "Untuk pustaka media layanan hewan peliharaan, kembalikan keterangan aksesibilitas Bahasa Mandarin yang ringkas dan hingga tiga tag subjek yang dapat dicari untuk foto kasus yang diunggah ini."},
        ])]
    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": "Untuk pustaka aset pemasaran, kembalikan ringkasan adegan satu kalimat Bahasa Mandarin, tiga kata kunci pengambilan, dan apakah diperlukan tinjauan keselamatan merek manual."},
        ])]
    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": "Untuk arsip sapaan hotline layanan pelanggan, transkripsikan pesan sambutan dan nilai apakah tujuan, ketersediaan layanan, serta pemberitahuan rekaman atau privasi jelas; tandai detail yang tidak ada dalam audio sebagai belum dikonfirmasi."},
        ])]
    print("AIFUNC_AI_BATCH_INPUT_MODE harus salah satu dari text, image, video, atau 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": "Contoh AI Batch"},
    })
    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"Output batch diunduh ke: {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} berakhir dengan 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)

    Hasil yang diharapkan: call-greeting-output.jsonl atau call-greeting-error.jsonl berisi tepat satu baris dengan custom_id=milvus-audio-1. Setelah verifikasi lingkup terbatas, hasil yang berhasil ditulis ke pusat konfigurasi untuk pemeriksaan sampel oleh staf operasional layanan pelanggan. Baris yang gagal mempertahankan custom_id asli dan dapat dicoba ulang secara individual setelah memperbaiki izin audio, format MP3, atau parameter model.