Embedding cache-Vector Retrieval Service for Milvus(Milvus)-阿里云帮助中心

AI_EMBEDDING_CACHE enables params.cache=true for exact-content caching. Identical inputs return vectors directly without a model call. This is suitable for FAQ imports and frequently repeated queries. If the cache is unavailable, the service falls back to real-time computation. Business logic should not depend on cache hits.

Syntax

REST API

POST /v2/vectordb/ai/embedding
Content-Type: application/json

{
  "model_name": "text-embedding-v4",
  "texts": ["<text>"],
  "params": {
    "dim": 1024,
    "cache": {
      "enabled": true,
      "exact_cache": {"enabled": true, "backend": "redis", "ttl_hours": 24}
    }
  }
}

Python

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=1024)
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": "text-embedding-v4",
            "dim": 1024,
            "cache": json.dumps({
                "enabled": True,
                "exact_cache": {
                    "enabled": True,
                    "backend": "redis",
                    "ttl_hours": 24,
                },
            }),
        },
    )
)

Parameters

Parameter	Description
`model_name`,`texts`,`dim`	Same as AI_EMBEDDING. Caching changes only the generation path, not the model, fields, or response format.
`cache.enabled`	The master cache switch. Default: `false`.
`cache.timeout_sec`	The cache read/write timeout in seconds. Default: `5`. If the cache is unavailable or times out, the model is called directly.
`cache.exact_cache.enabled`	The Redis exact-cache switch. This and `cache.enabled` must both be `true`.
`cache.exact_cache.backend`	Currently, only `redis`.
`cache.exact_cache.ttl_hours`	The cache lifetime in hours. Default: `24`.
`cache.read_latency_degrade.enabled`	Whether to temporarily bypass the cache backend after consecutive slow reads or failures. Default: `false`.
`cache.read_latency_degrade.threshold_ms`	Required when read-latency degradation is enabled. The slow-read threshold in milliseconds.
`cache.read_latency_degrade.consecutive_slow_reads`,`degrade_duration_sec`	Optional. The number of consecutive slow reads that triggers degradation and the degradation duration. Defaults: `3` times and `60` seconds.

Return values

The response has the same structure as a regular Embedding response. Vectors are at data.output.embeddings[].embedding. The response also includes data.usage (such as total_tokens) and data.request_id. On a cache hit, the model is not called: total_tokens is 0 and no request_id is returned. If the cache is unavailable or times out, the model is called directly. Business correctness should therefore not depend on cache hits.

{"code":0,"data":{"output":{"embeddings":[{"text_index":0,"embedding":[0.01,0.02]}]},"usage":{"total_tokens":22},"request_id":"xxxxxxxx"}}

Example: Verify cache reuse for a customer service refund FAQ

A customer service knowledge base may process the same "refund arrival time" FAQ during staging and production imports. The following code defines one request body and sends the exact same content in two independent calls, rather than including two identical texts in a single request, to observe whether the second request hits the cache.

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="text-embedding-v4"

BODY=$(cat <<JSON
{
  "model_name": "$MODEL_NAME",
  "texts": [
    "Milvus is an open-source vector database.",
    "Milvus is an open-source vector database."
  ],
  "params": {
    "dim": 1024,
    "cache": {
      "enabled": true,
      "exact_cache": {"enabled": true, "backend": "redis", "ttl_hours": 24}
    }
  }
}
JSON
)

# Send the same BODY in two independent requests: a first cache miss calls the model (usage.total_tokens > 0 and request_id is returned),
# a second cache hit returns vectors directly (usage.total_tokens = 0 and no request_id).
for attempt in 1 2; do
  echo "Attempt ${attempt}: request embeddings with the same BODY."
  RESPONSE_BODY="$(post_json "/v2/vectordb/ai/embedding" "$BODY")"
  if command -v jq >/dev/null 2>&1; then
    echo "$RESPONSE_BODY" | jq '{code, total_tokens: .data.usage.total_tokens, request_id: .data.request_id}'
    [ "$(echo "$RESPONSE_BODY" | jq -r '.code // -1')" = "0" ] || exit 1
  else
    echo "$RESPONSE_BODY"
  fi
done

Python

from __future__ import annotations
import json
import os
from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "<yourUsername>:<yourPassword>"

def add_id(schema: Any) -> None:
    schema.add_field("id", DataType.INT64, is_primary=True)

MODEL_NAME = "text-embedding-v4"
VECTOR_DIM = int(os.getenv("AIFUNC_EMBEDDING_DIM", "1024"))
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

collection_name = "simple_ai_embedding_cache"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
add_id(schema)
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": json.dumps(
                {"enabled": True, "exact_cache": {"enabled": True, "backend": "redis", "ttl_hours": 24}}
            ),
        },
    )
)
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)
body = {"content": "Milvus is an open-source vector database."}
print("Insert the same body for the first time.")
client.insert(collection_name, [body])
client.flush(collection_name)
print("Insert the same body for the second time.")
client.insert(collection_name, [body])
client.flush(collection_name)
print(client.query(collection_name, filter="", output_fields=["content"], limit=10))

Do not use identical vectors in two responses or the latency of one request as evidence of a cache hit. For managed instances, compare the response fields data.usage.total_tokens and data.request_id: on a cache miss, total_tokens is greater than 0 and request_id is returned; on a cache hit, total_tokens is 0 and no request_id is returned, indicating that the model was not called again. If you can access the underlying Redis instance, first record the INFO stats keyspace_hits baseline during a quiet period, and then send two identical requests in sequence to compare the increase. Redis failures or timeouts fall back to model calls. Do not make cache hits a business success condition or promise a fixed latency improvement.