向量化缓存 - 向量检索服务 Milvus 版

AI_EMBEDDING_CACHE 通过 params.cache=true 启用精确内容缓存，相同输入直接返回向量以跳过模型调用，适用于 FAQ 导入或高频重复查询场景。缓存失效时自动降级为实时计算，业务逻辑不应依赖缓存命中状态。

命令格式

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 故障或超时会回退到模型调用，不应将缓存命中作为业务成功条件，也不应承诺固定延迟改善。