Vector Retrieval Service for Milvus：通過Embedding Cache與AI Batch降低阿里雲Milvus的AI調用成本

工作方式與適用範圍

Embedding Cache 會先按文本與調用上下文尋找已有向量，命中後直接複用，只有首次出現的新內容才真正調用模型。緩衝採用精確匹配——兩段文字只要寫法不同，就會分別計算。因此它的收益完全來自內容重複率：重複越多，省下的模型請求越多。

以日均 10 萬次向量化、平均每次 50 Token、快取命中率 50% 估算，命中的那一半請求不再產生任何模型調用與等待，整體 Token 消耗大約下降一半。

開啟緩衝

在 Embedding Function 的 params 中增加 cache 配置即可。ttl_hours 按內容更新頻率設定：商品標題、FAQ 這類更新不頻繁的內容可適當延長，內容變化快時縮短，讓新版本更快重新計算。

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"  # 連接埠必須寫 19530
MILVUS_TOKEN = "root:xxx"
MODEL_NAME = "text-embedding-v4"
VECTOR_DIM = 1024

# 緩衝配置：精確匹配，Redis 後端，有效期間 24 小時
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,        # ← 開啟 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)

# 寫入時命中緩衝的內容不會再產生模型調用
row = {"content": "Milvus is an open-source vector database."}
client.insert(collection_name, [row])
client.flush(collection_name)

如何確認緩衝真正命中

這一步容易踩坑，需要特別說明。比較兩次寫入得到的向量是否一致，無法判斷緩衝是否命中——Embedding 模型對相同輸入本身就是確定性輸出，實測在完全不開啟緩衝的情況下，兩次寫入同一文本得到的向量逐元素差值同樣為 0。同理，寫入耗時也不能作為判據：一次 insert 加 flush 的固有開銷遠大於單次模型調用，實測第二次寫入甚至可能比第一次更慢。

可靠的判據是響應中的 usage.total_tokens 與 request_id：命中緩衝時本次沒有真正調用模型，因此 Token 消耗歸零、且不會產生模型側的請求 ID。

# ==================== 驗證緩衝是否真正命中 ====================
# 判據：usage.total_tokens 是否歸零、request_id 是否為空白。
# 二者同時成立說明本次沒有真正調用模型，即命中了緩衝。

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"))


# 用一段全新文本，確保第一次必然未命中
text = f"快取命中驗證 {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"第 {i} 次: total_tokens={total_tokens} "
          f"request_id={'(空)' if not request_id else request_id} "
          f"-> {'命中緩衝' if hit else '未命中，調用了模型'}")

對同一段新文本連續請求三次的實測結果：

典型情境

一家電商白天把模型額度留給正在搜尋和諮詢的使用者，夜裡再集中處理當天積累的商品資料與客服會話：批量產生商品描述、補齊標籤、把長對話整理成摘要。第二天營運上班時結果已寫回業務系統——使用者沒有多等一秒，也不必按即時價格處理這批資料。

準備輸入檔案

輸入為 JSONL，一行對應一條任務，用 custom_id 關聯未經處理資料，便於結果回寫時對應到具體記錄：

{"custom_id":"article-001","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen3.7-max","messages":[{"role":"user","content":"請為這篇知識庫文章產生一句話摘要……"}],"enable_thinking":false}}
{"custom_id":"article-002","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen3.7-max","messages":[{"role":"user","content":"請為這篇知識庫文章產生一句話摘要……"}],"enable_thinking":false}}

提交與回查

完整流程為四步：上傳 JSONL、建立任務、回查狀態、下載結果。

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 走 RESTful 介面，base_url 用執行個體地址（連接埠 19530）
MILVUS_BASE_URL = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "root:xxx"
MODEL_NAME = "qwen3.7-max"     # 必須與 input.jsonl 裡每行 body.model 一致
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):
    """multipart 上傳 input.jsonl"""
    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):
    """下載結果檔案。注意用 batch_id + file_type，不是 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) 上傳 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"上傳失敗：HTTP={status} message={data.get('message')}")
print(f"input_file_id = {input_file_id}")

# 2) 建立 Batch 任務，聲明完成視窗
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"建立失敗：HTTP={status} message={data.get('message')}")
print(f"batch_id = {batch_id}")

# 3) 回查任務狀態。Batch 為小時級非同步任務，生產環境建議記錄 batch_id
#    後由定時任務稍後回查，不要在前台密集輪詢。
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)      # 5 分鐘回查一次即可

# 4) 完成後下載結果；失敗行在 error_file_id 中，可單獨重試
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"))

四個介面的實測返回：

任務完成後結果檔案仍帶回原來的 custom_id，業務可以把摘要準確寫回對應記錄。部分資料失敗時，只需下載 error_file_id 對應的錯誤檔案並重試失敗行，已完成的結果直接保留，無需整批重做。