Vector Retrieval Service for Milvus:Embedding Cache と AI Batch を利用した Alibaba Cloud Milvus の AI 呼び出しコストの削減

Embedding Cache の仕組みと適用範囲

Embedding Cache は、まずテキストと呼び出しコンテキストによって既存のベクターを検索し、ヒットした場合は直接再利用し、初めて出現するコンテンツに対してのみモデルを呼び出します。キャッシュは完全一致を使用するため、記述が異なる 2 つのテキストは別々に計算されます。したがって、メリットは完全にコンテンツの重複率に依存します。重複が多ければ多いほど、より多くのモデルリクエストを節約できます。おおよその見積もりについては、「コスト見積もり」をご参照ください。

キャッシュが利用できない場合や読み取りがタイムアウトした場合でも、Milvus はモデルの呼び出しを続行するため、書き込みやクエリが中断されることはありません。

Embedding Cache の有効化

Embedding Function の params に cache 構成を追加します。コンテンツの変更頻度に応じて ttl_hours を設定します。プロダクトのタイトルやよくある質問など、変更頻度の低いコンテンツには長く設定し、変更の速いコンテンツには短く設定して、新しいバージョンがより早く再計算されるようにします。

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 時間の 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,        # ← 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)

書き込み中にキャッシュにヒットしたコンテンツは、モデル呼び出しを生成しません。キャッシュが有効になっていることを確認するには、「キャッシュヒットの確認」をご参照ください。

キャッシュヒットの確認

キャッシュヒットは、RESTful 埋め込みエンドポイントのレスポンスから確認します。このレスポンスは usage.total_tokens と request_id を返します。キャッシュヒットの場合、実際にはモデル呼び出しは発生しないため、トークン消費量はゼロになり、モデル側でリクエスト ID は生成されません。

直感的に考えられる 2 つの確認方法は機能しません：

• ベクター比較 — モデルは同じ入力に対して同じベクターを返します。このトピックのテストでは、キャッシュを完全に無効にして同じテキストを 2 回書き込んでも、要素ごとの差は 0 でした。

• 書き込みレイテンシ — 1 回の insert と flush の固有のオーバーヘッドは、1 回のモデル呼び出しをはるかに超えており、テストでは 2 回目の書き込みが 1 回目よりも遅くなることさえありました。

以下のコードは、同じ新しいテキストを 3 回送信し、各リクエストの結果を報告します。前のステップで定義されたインポートと定数を再利用するため、同じスクリプトまたはセッションで実行してください。

# ==================== キャッシュが実際にヒットしたかどうかの確認 ====================
# 基準：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"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"リクエスト {i}：total_tokens={total_tokens} "
          f"request_id={'（空）' if not request_id else request_id} "
          f"-> {'キャッシュヒット' if hit else 'キャッシュミス、モデルが呼び出されました'}")

以下の表は、同じ新しいテキストに対する 3 回連続のリクエストの測定結果です：

両方のフィールドはレスポンスボディで直接利用でき、追加の権限は必要ありません。サーバー側で相互検証するには、コンソールで埋め込みモデルの呼び出し回数がどのように変化するかを監視します。

AI Batch の仕組みと適用範囲

AI Batch は、JSONL に書き込まれたリクエストを受け取り、非同期で受理し、オフラインでバッチ処理します。e コマースにおける典型的な運用モデルでは、日中のモデルクォータは検索や質問を行うユーザーのために確保されます。そして、その日に蓄積されたプロダクトデータやサポートの会話は夜間に処理されます。例えば、プロダクト説明の一括生成、タグの入力、長い会話の要約などです。このデータに対する結果は、リアルタイム料金を支払うことなく、翌営業日までに業務システムに書き戻されます。

入力ファイルの準備とバッチジョブの実行

入力は JSONL 形式で、1 行が 1 つのジョブに対応します。custom_id を使用して各ジョブを元のデータにリンクさせることで、結果を特定のレコードにマッピングできます：

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

全プロセスは、JSONL ファイルのアップロード、ジョブの作成、ステータスのポーリング、結果のダウンロードの 4 つのステップで構成されます。AI Batch は RESTful API を使用するため、以下のブロックでは、認証済みリクエスト、マルチパートアップロード、結果ダウンロードのための定数とヘルパー関数を定義します。4 つのステップそれぞれでこれらの関数を再利用するため、最初にこのブロックを実行してください。

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 API を使用します。base_url をインスタンスアドレス（ポート 19530）に設定します
MILVUS_BASE_URL = "http://c-xxx.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):
    """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):
    """結果ファイルをダウンロードします。注意：file_id ではなく batch_id + file_type を使用します"""
    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) バッチジョブを作成し、完了ウィンドウを宣言
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_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 分に 1 回のポーリングで十分です

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

結果の確認と利用

以下の表は、4 つの操作の測定されたレスポンスを示しています：

ダウンロード操作は、ファイル ID ではなく batch_id と file_type でファイルを特定します。これは間違いやすい点です。

ジョブが完了した後も、結果ファイルには元の custom_id が保持されているため、アプリケーションは各要約を対応するレコードに書き戻すことができます。データの一部が失敗した場合は、error_file_id に対応するエラーファイルのみをダウンロードし、失敗した行をリトライします。完了した結果はそのまま保持され、バッチ全体を再実行する必要はありません。