Reduce AI invocation costs on Alibaba Cloud Milvus with Embedding Cache and AI Batch

How Embedding Cache works and where it applies

Embedding Cache first looks up an existing vector by the text and the invocation context, reuses it directly on a hit, and invokes the model only for content that appears for the first time. The cache uses exact match, so two pieces of text are computed separately if they are written differently. The benefit therefore comes entirely from the content duplication rate: the more duplication, the more model requests you save. For an order-of-magnitude estimate, see Cost estimation.

If the cache is unavailable or a read times out, Milvus continues to invoke the model, so your writes and queries are not interrupted.

Enable Embedding Cache

Add the cache configuration to params of the Embedding Function. Set ttl_hours according to how often the content changes: extend it for content that changes infrequently, such as product titles and FAQs, and shorten it for fast-changing content so that new versions are recomputed sooner.

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"  # The port must be 19530
MILVUS_TOKEN = "root:xxx"
MODEL_NAME = "text-embedding-v4"
VECTOR_DIM = 1024

# Cache configuration: exact match, Redis backend, 24-hour TTL
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,        # ← Enable 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)

# Content that hits the cache during a write generates no model invocation
row = {"content": "Milvus is an open-source vector database."}
client.insert(collection_name, [row])
client.flush(collection_name)

Content that hits the cache during a write generates no model invocation. To confirm that the cache takes effect, see Verify a cache hit.

Verify a cache hit

Confirm a cache hit from the response of the RESTful embedding endpoint, which returns usage.total_tokens and request_id. On a cache hit, no model invocation actually happens, so token consumption drops to zero and no request ID is generated on the model side.

Two intuitive checks do not work:

• Vector comparison — the model returns the same vector for the same input. In the tests for this topic, writing the same text twice with the cache completely disabled still produced an element-wise difference of 0.

• Write latency — the inherent overhead of one insert plus flush far exceeds a single model invocation, and in tests the second write can even be slower than the first.

The following code sends the same new text three times and reports the result of each request. Run it in the same script or session as the previous step, because it reuses the imports and constants defined there.

# ==================== Verify whether the cache is actually hit ====================
# Criteria: whether usage.total_tokens drops to zero and whether request_id is empty.
# When both hold, no model invocation actually happened, which means a 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"))

# Use a brand-new piece of text so that the first request is guaranteed to 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'}")

The following table lists the measured results of three consecutive requests for the same new text:

Both fields are directly available in the response body, and no extra permissions are required. To cross-validate on the server side, watch how the number of embedding model invocations changes in the console.

How AI Batch works and where it applies

AI Batch takes requests written to JSONL, accepts them asynchronously, and runs them offline in batches. A typical operating model in e-commerce reserves the model quota during the day for users who are searching and asking questions. The day's accumulated product data and support conversations are then processed at night: generating product descriptions in bulk, filling in tags, and turning long conversations into summaries. The results are written back to the business systems before the next working day, without paying real-time prices for this data.

Prepare the input file and run the batch job

The input is JSONL, with one line per job. Use custom_id to link each job to the original data so that results can be mapped back to specific records:

{"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}}

The full process has four steps: upload the JSONL file, create the job, poll the status, and download the results. AI Batch uses the RESTful API, so the following block defines the constants and the helper functions for authenticated requests, multipart upload, and result download. Run this block first, because each of the four steps reuses these functions.

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 uses the RESTful API. Set base_url to the instance address (port 19530)
MILVUS_BASE_URL = "http://c-xxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "root:xxx"
MODEL_NAME = "qwen3.7-max"     # Must match body.model on every line of 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):
    """Upload input.jsonl as 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):
    """Download the result file. Note: use batch_id + file_type, not 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) Upload the JSONL input file
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) Create the Batch job and declare the completion window
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) Poll the job status. Batch jobs run asynchronously on an hourly scale. In production,
#    record the batch_id and poll later from a scheduled task instead of polling intensively in the 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 once every 5 minutes is enough

# 4) Download the results after completion. Failed lines are in error_file_id and can be retried separately
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"))

Confirm and use the results

The following table lists the measured responses of the four operations:

The download operation locates the file by batch_id plus file_type rather than by file ID, which is easy to get wrong.

After the job completes, the result file still carries the original custom_id, so your application can write each summary back to the matching record. If part of the data fails, download only the error file corresponding to error_file_id and retry the failed lines. Completed results are kept as they are, and there is no need to rerun the whole batch.