Vector Retrieval Service for Milvus:Kurangi Biaya Pemanggilan AI di Alibaba Cloud Milvus dengan Embedding Cache dan AI Batch

Cara kerja Embedding Cache dan tempat penerapannya

Embedding Cache pertama-tama mencari vektor yang sudah ada berdasarkan teks dan konteks pemanggilan, menggunakan kembali langsung jika ditemukan (cache hit), dan hanya memanggil model untuk konten yang muncul pertama kali. Cache menggunakan kecocokan eksak, sehingga dua potongan teks yang ditulis berbeda akan dihitung secara terpisah. Manfaatnya sepenuhnya berasal dari tingkat duplikasi konten: semakin tinggi duplikasi, semakin banyak permintaan model yang dihemat. Untuk perkiraan orde besaran, lihat Estimasi Biaya.

Jika cache tidak tersedia atau pembacaan timeout, Milvus tetap memanggil model, sehingga operasi tulis dan kueri Anda tidak terganggu.

Aktifkan Embedding Cache

Tambahkan konfigurasi cache ke params Fungsi Embedding. Atur ttl_hours sesuai seberapa sering konten berubah: perpanjang untuk konten yang jarang berubah, seperti judul produk dan FAQ, dan perpendek untuk konten yang cepat berubah agar versi baru segera dihitung ulang.

import json
import uuid
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # Port harus 19530
MILVUS_TOKEN = "root:xxx"
MODEL_NAME = "text-embedding-v4"
VECTOR_DIM = 1024

# Konfigurasi cache: kecocokan eksak, backend Redis, TTL 24 jam
CACHE_CONFIG = json.dumps({
    "enabled": True,
    "exact_cache": {
        "enabled": True,
        "backend": "redis",
        "ttl_hours": 24,
    },
})

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)
collection_name = "ai_embedding_cache_demo"

if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("content", DataType.VARCHAR, max_length=4096)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
schema.add_function(
    Function(
        name="embed_content_with_cache",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["content"],
        output_field_names=["embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": MODEL_NAME,
            "dim": VECTOR_DIM,
            "cache": CACHE_CONFIG,        # ← Aktifkan Embedding Cache
        },
    )
)

index_params = client.prepare_index_params()
index_params.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

# Konten yang mengenai cache saat penulisan tidak menghasilkan pemanggilan model
row = {"content": "Milvus is an open-source vector database."}
client.insert(collection_name, [row])
client.flush(collection_name)

Konten yang mengenai cache saat penulisan tidak menghasilkan pemanggilan model. Untuk mengonfirmasi bahwa cache berlaku, lihat Verifikasi Cache Hit.

Verifikasi cache hit

Konfirmasi cache hit dari respons titik akhir embedding RESTful, yang mengembalikan usage.total_tokens dan request_id. Saat terjadi cache hit, tidak ada pemanggilan model yang benar-benar terjadi, sehingga konsumsi token turun menjadi nol dan tidak ada request ID yang dihasilkan di sisi model.

Dua pemeriksaan intuitif tidak berfungsi:

• Perbandingan vektor — model mengembalikan vektor yang sama untuk input yang sama. Dalam pengujian untuk topik ini, menulis teks yang sama dua kali dengan cache dinonaktifkan sepenuhnya tetap menghasilkan perbedaan elemen-per-elemen sebesar 0.

• Latensi penulisan — overhead inheren dari satu operasi insert plus flush jauh melebihi satu pemanggilan model, dan dalam pengujian penulisan kedua bahkan bisa lebih lambat daripada yang pertama.

Kode berikut mengirimkan teks baru yang sama tiga kali dan melaporkan hasil setiap permintaan. Jalankan dalam skrip atau sesi yang sama dengan langkah sebelumnya, karena kode ini menggunakan kembali impor dan konstanta yang telah didefinisikan.

# ==================== Verifikasi apakah cache benar-benar terkena ====================
# Kriteria: apakah usage.total_tokens turun menjadi nol dan apakah request_id kosong.
# Jika keduanya terpenuhi, tidak ada pemanggilan model yang benar-benar terjadi, artinya cache hit.

def post_json(path, body, timeout=180):
    request = Request(
        f"{MILVUS_URI.rstrip('/')}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_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"))

# Gunakan teks yang benar-benar baru agar permintaan pertama pasti miss
text = f"cache hit verification {uuid.uuid4().hex[:12]}"
body = {
    "model_name": MODEL_NAME,
    "texts": [text],
    "params": {"dim": VECTOR_DIM, "cache": CACHE_CONFIG},
}

for i in (1, 2, 3):
    status, data = post_json("/v2/vectordb/ai/embedding", body)
    assert status == 200 and data.get("code") == 0, data
    usage = data["data"].get("usage", {})
    request_id = data["data"].get("request_id", "")
    total_tokens = usage.get("total_tokens")
    hit = (total_tokens == 0) and not request_id
    print(f"Request {i}: total_tokens={total_tokens} "
          f"request_id={'(empty)' if not request_id else request_id} "
          f"-> {'cache hit' if hit else 'cache miss, the model was invoked'}")

Tabel berikut mencantumkan hasil pengukuran dari tiga permintaan berturut-turut untuk teks baru yang sama:

Kedua bidang tersebut tersedia langsung dalam badan respons dan tidak memerlukan izin tambahan. Untuk validasi silang di sisi server, pantau perubahan jumlah pemanggilan model embedding di Konsol.

Cara kerja AI Batch dan tempat penerapannya

AI Batch menerima permintaan yang ditulis dalam format JSONL, menerimanya secara asinkron, dan menjalankannya secara offline dalam batch. Model operasional khas di e-commerce menyisihkan kuota model selama hari kerja untuk pengguna yang sedang mencari dan mengajukan pertanyaan. Data produk dan percakapan dukungan yang terakumulasi sepanjang hari kemudian diproses pada malam hari: menghasilkan deskripsi produk secara massal, mengisi tag, dan mengubah percakapan panjang menjadi ringkasan. Hasilnya ditulis kembali ke sistem bisnis sebelum hari kerja berikutnya, tanpa membayar harga real-time untuk data tersebut.

Siapkan file input dan jalankan pekerjaan batch

Input berupa JSONL, dengan satu baris per pekerjaan. Gunakan custom_id untuk menghubungkan setiap pekerjaan dengan data aslinya sehingga hasilnya dapat dipetakan kembali ke catatan tertentu:

{"custom_id":"article-001","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen3.7-max","messages":[{"role":"user","content":"Generate a one-sentence summary for this knowledge base article..."}],"enable_thinking":false}}
{"custom_id":"article-002","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen3.7-max","messages":[{"role":"user","content":"Generate a one-sentence summary for this knowledge base article..."}],"enable_thinking":false}}

Proses lengkap terdiri dari empat langkah: unggah file JSONL, buat pekerjaan, polling status, dan unduh hasil. AI Batch menggunakan RESTful API, sehingga blok berikut mendefinisikan konstanta dan fungsi pembantu untuk permintaan terautentikasi, unggah multi-bagian, dan pengunduhan hasil. Jalankan blok ini terlebih dahulu, karena keempat langkah tersebut menggunakan kembali fungsi-fungsi ini.

import json
import shutil
import tempfile
import time
import uuid
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, urlopen

# AI Batch menggunakan RESTful API. Setel base_url ke alamat instans (port 19530)
MILVUS_BASE_URL = "http://c-xxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "root:xxx"
MODEL_NAME = "qwen3.7-max"     # Harus sesuai dengan body.model di setiap baris input.jsonl
PROVIDER = "aliyun_milvus"
ENDPOINT = "/v1/chat/completions"

def post_json(path, body, timeout=180):
    request = Request(
        f"{MILVUS_BASE_URL.rstrip('/')}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_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 upload_input_file(input_file):
    """Unggah input.jsonl sebagai multipart"""
    boundary = f"----milvus-ai-batch-{uuid.uuid4().hex}"
    fields = {"provider": PROVIDER, "model_name": MODEL_NAME,
              "endpoint": ENDPOINT, "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 fh:
            shutil.copyfileobj(fh, payload)
        payload.write(b"\r\n")
        payload.write(f"--{boundary}--\r\n".encode())

        length = payload.tell()
        payload.seek(0)
        request = Request(
            f"{MILVUS_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/upload",
            data=payload,
            headers={"Authorization": f"Bearer {MILVUS_TOKEN}",
                     "Content-Type": f"multipart/form-data; boundary={boundary}",
                     "Content-Length": str(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, file_type, output_file):
    """Unduh file hasil. Catatan: gunakan batch_id + file_type, bukan file_id"""
    request = Request(
        f"{MILVUS_BASE_URL.rstrip('/')}/v2/vectordb/ai/batch/files/content",
        data=json.dumps({"provider": PROVIDER, "batch_id": batch_id,
                         "file_type": file_type}, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_TOKEN}",
                 "Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=600) as response, output_file.open("wb") as out:
        shutil.copyfileobj(response, out)

# 1) Unggah file input JSONL
status, data = upload_input_file("input.jsonl")
input_file_id = (data.get("data") or {}).get("id")
if status != 200 or not input_file_id:
    raise SystemExit(f"Upload failed: HTTP={status} message={data.get('message')}")
print(f"input_file_id = {input_file_id}")

# 2) Buat pekerjaan Batch dan deklarasikan jendela penyelesaian
status, data = post_json("/v2/vectordb/ai/batch/jobs/create", {
    "provider": PROVIDER,
    "input_file_id": input_file_id,
    "endpoint": ENDPOINT,
    "completion_window": "24h",
})
batch_id = (data.get("data") or {}).get("id")
if status != 200 or not batch_id:
    raise SystemExit(f"Creation failed: HTTP={status} message={data.get('message')}")
print(f"batch_id = {batch_id}")

# 3) Polling status pekerjaan. Pekerjaan batch berjalan asinkron berskala jam. Di produksi,
#    catat batch_id dan lakukan polling nanti dari tugas terjadwal, bukan polling intensif di foreground.
while True:
    status, data = post_json("/v2/vectordb/ai/batch/jobs/describe",
                             {"provider": PROVIDER, "batch_id": batch_id})
    batch = data.get("data") or {}
    batch_status = batch.get("status")
    print(f"{batch_status}  {batch.get('request_counts')}")
    if batch_status in {"completed", "failed", "expired", "cancelled"}:
        break
    time.sleep(300)      # Polling setiap 5 menit sudah cukup

# 4) Unduh hasil setelah selesai. Baris yang gagal ada di error_file_id dan dapat dicoba ulang secara terpisah
if batch_status == "completed":
    download_batch_file(batch_id, "output", Path("output.jsonl"))
    if batch.get("error_file_id"):
        download_batch_file(batch_id, "error", Path("error.jsonl"))

Konfirmasi dan gunakan hasil

Tabel berikut mencantumkan respons terukur dari keempat operasi:

Operasi unduh menemukan file berdasarkan batch_id ditambah file_type, bukan berdasarkan ID file, yang mudah salah.

Setelah pekerjaan selesai, file hasil tetap membawa custom_id asli, sehingga aplikasi Anda dapat menulis setiap ringkasan kembali ke catatan yang sesuai. Jika sebagian data gagal, unduh hanya file error yang sesuai dengan error_file_id dan coba ulang baris yang gagal. Hasil yang telah selesai tetap dipertahankan, dan tidak perlu menjalankan ulang seluruh batch.