向量化緩衝 - Vector Retrieval Service for Milvus

AI_EMBEDDING_CACHE 通過 params.cache=true 啟用精確內容緩衝，相同輸入直接返迴向量以跳過模型調用，適用於 FAQ 匯入或高頻重複查詢情境。緩衝失效時自動降級為Realtime Compute，商務邏輯不應依賴快取命中狀態。

命令格式

REST 介面

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

{
  "model_name": "text-embedding-v4",
  "texts": ["<文本>"],
  "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,
                },
            }),
        },
    )
)

參數說明

參數	說明
`model_name`、`texts`、`dim`	與 AI_EMBEDDING 向量化相同；緩衝只改變產生路徑，不改變模型、欄位或返回格式。
`cache.enabled`	緩衝總開關，預設 `false`。
`cache.timeout_sec`	緩衝讀寫逾時秒數，預設 `5`；緩衝不可用或逾時時會直接調用模型。
`cache.exact_cache.enabled`	Redis 精確緩衝開關，必須與 `cache.enabled` 同時為 `true`。
`cache.exact_cache.backend`	當前僅支援 `redis`。
`cache.exact_cache.ttl_hours`	緩衝有效期間，單位時數，預設 `24`。
`cache.read_latency_degrade.enabled`	是否在連續慢讀或失敗時暫時跳過緩衝後端，預設 `false`。
`cache.read_latency_degrade.threshold_ms`	開啟讀延遲降級時必填，表示慢讀閾值（毫秒）。
`cache.read_latency_degrade.consecutive_slow_reads`、`degrade_duration_sec`	可選。分別表示觸發降級的連續次數和降級期間，預設 `3` 次、`60` 秒。

傳回值說明

返回結構與普通 Embedding 相同，向量位於 data.output.embeddings[].embedding。響應還包含 data.usage（如 total_tokens）與 data.request_id。快取命中時不調用模型，此時 total_tokens 為 0 且不返回 request_id；緩衝不可用或逾時時會降級為直接調用模型，因此業務正確性不應依賴快取命中。

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

樣本：複核客服退款 FAQ 的緩衝複用

客服知識庫在預發和正式匯入階段可能重複處理同一條“退款到賬時間”FAQ。下面的代碼定義一次請求體，再用這份完全相同的內容獨立調用服務兩次（而不是在一次請求中放入兩條相同文本），用於觀察第二次請求是否命中緩衝。

REST 介面

#!/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
)

# 用完全相同的 BODY 獨立請求兩次：首次未命中會調用模型(usage.total_tokens>0 且返回 request_id),
# 第二次命中緩衝直接返迴向量(usage.total_tokens=0 且無 request_id)。
for attempt in 1 2; do
  echo "第 ${attempt} 次使用同一個 BODY 請求 embedding。"
  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("首次寫入同一個 body。")
client.insert(collection_name, [body])
client.flush(collection_name)
print("第二次寫入同一個 body。")
client.insert(collection_name, [body])
client.flush(collection_name)
print(client.query(collection_name, filter="", output_fields=["content"], limit=10))

確認快取命中不要用“兩次向量相同”或單次延遲作為證據。受管理的執行個體推薦對比響應的 data.usage.total_tokens 與 data.request_id：未命中時 total_tokens 大於 0 且返回 request_id，命中時 total_tokens 為 0 且不返回 request_id（說明未再調用模型）。若可訪問底層 Redis，也可在安靜時段先記錄 INFO stats 的 keyspace_hits 基準，再依次執行兩次相同請求對比增量。Redis 故障或逾時會回退到模型調用，不應將快取命中作為業務成功條件，也不應承諾固定延遲改善。