[アップデート] Amazon ElastiCache Serverless for Valkeyでベクトル検索と集計が使えるようになりました！

# acmeテナントのキャッシュだけを対象に、ベクトルが近い3件を探す
FT.SEARCH semantic_cache_idx "@tenant:{acme}=>[KNN 3 @embedding $vec AS score]" PARAMS 2 vec <埋め込みベクトル>

# テナントごとのキャッシュ件数とヒット回数の合計を出す
FT.AGGREGATE semantic_cache_idx "@hits:[-inf +inf]" LOAD 2 @tenant @hits GROUPBY 1 @tenant REDUCE COUNT 0 AS entries REDUCE SUM 1 @hits AS total_hits

aws elasticache create-serverless-cache \
    --serverless-cache-name my-public-cache \
    --engine valkey \
    --major-engine-version 9 \
    --connection-type public \
    --user-group-id default.iam-user-group

git clone https://github.com/yuu551/elasticache-serverless-semantic-cache.git
cd elasticache-serverless-semantic-cache
pnpm install

// ElastiCache Serverless for Valkey 9（パブリックエンドポイント）
const cache = new elasticache.CfnServerlessCache(this, 'Cache', {
  serverlessCacheName: CACHE_NAME,
  engine: 'valkey',
  majorEngineVersion: '9',
  connectionType: 'public',
  userGroupId: 'default.iam-user-group',
});

// AgentCore Runtime V2（VPCなしのパブリックネットワーク）
const runtime = new agentcore.Runtime(this, 'Runtime', {
  runtimeName: 'serverlessSemanticCache',
  agentRuntimeArtifact: agentcore.AgentRuntimeArtifact.fromAsset(
    path.join(import.meta.dirname, '../agent'),
    { platform: Platform.LINUX_ARM64 },
  ),
  platformVersion: agentcore.PlatformVersion.V2,
  networkConfiguration: agentcore.RuntimeNetworkConfiguration.usingPublicNetwork(),
  environmentVariables: {
    VALKEY_HOST: cache.attrEndpointAddress,
    VALKEY_CACHE_NAME: CACHE_NAME,
    VALKEY_USER: CACHE_USER,
  },
});

runtime.addToRolePolicy(new iam.PolicyStatement({
  actions: ['elasticache:Connect'],
  resources: [
    cache.attrArn,
    this.formatArn({ service: 'elasticache', resource: 'user', resourceName: CACHE_USER, arnFormat: cdk.ArnFormat.COLON_RESOURCE_NAME }),
  ],
}));

self._config = GlideClusterClientConfiguration(
    addresses=[NodeAddress(host, 6379)],
    use_tls=True,
    request_timeout=5000,
    credentials=ServerCredentials(
        username=user_id,  # default.iam-user
        iam_config=IamAuthConfig(
            cluster_name=cache_name,
            service=ServiceType.ELASTICACHE,
            region=region,
        ),
    ),
)

async def _get_client(self) -> GlideClusterClient:
    # Runtime V2はスナップショットから復元されるため、
    # 期限のあるIAMトークンを使う接続は最初のリクエストで作る
    if self._client is None:
        self._client = await GlideClusterClient.create(self._config)
        await self._setup_index(self._client)
    return self._client

await client.custom_command([
    "FT.CREATE", INDEX_NAME,
    "ON", "HASH",
    "PREFIX", "1", KEY_PREFIX,
    "SCHEMA",
    "query", "TEXT",
    "tenant", "TAG",
    "hits", "NUMERIC",
    "embedding", "VECTOR", "HNSW", "6",
    "TYPE", "FLOAT32",
    "DIM", str(self.dims),
    "DISTANCE_METRIC", "COSINE",
])

result = await client.custom_command([
    "FT.SEARCH", INDEX_NAME,
    f"@tenant:{{{tenant}}}=>[KNN 3 @embedding $vec AS score]",
    "PARAMS", "2", "vec", vec,
    "RETURN", "1", "score",
])

pnpm cdk deploy

 ✅  ServerlessSemanticCacheStack

✨  Deployment time: 321.4s

Outputs:
ServerlessSemanticCacheStack.CacheEndpoint = semantic-cache-xxxxxx.public.serverless.apne1.cache.amazonaws.com
ServerlessSemanticCacheStack.CacheName = semantic-cache
ServerlessSemanticCacheStack.RuntimeArn = arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/serverlessSemanticCache-xxxxxxxxxx

cd client
uv sync

uv run invoke.py "VPNに接続する方法を教えて" --tenant acme

{
  "response": "VPNの接続は一般的に次の手順で行います。製品によって画面が異なる場合があります。…",
  "cache_hit": false,
  "model_id": "jp.anthropic.claude-haiku-5-5",
  "timing_ms": {
    "embedding": 474.8,
    "vector_search": 3.2,
    "connect_and_index": 2425.6,
    "agent_invocation": 3657.3,
    "total": 6712.6
  }
}
呼び出し全体: 8745.4 ms

uv run invoke.py "VPNに接続する方法を教えて" --tenant acme
uv run invoke.py "VPNのつなぎ方を知りたい" --tenant acme

"cache_hit": true,
"similarity_score": 1.0,
"timing_ms": { "embedding": 553.8, "vector_search": 3.1, "connect_and_index": 1879.6, "total": 2440.8 }
呼び出し全体: 4403.4 ms

"cache_hit": true,
"similarity_score": 0.8661,
"timing_ms": { "embedding": 523.4, "vector_search": 3.2, "connect_and_index": 2135.2, "total": 2666.8 }
呼び出し全体: 5119.3 ms

SESSION=$(uuidgen)
uv run invoke.py "経費精算の締め日はいつ？" --tenant acme -s $SESSION
uv run invoke.py "経費精算の締め日はいつ？" --tenant acme -s $SESSION
uv run invoke.py "経費の精算はいつまでに出せばいい？" --tenant acme -s $SESSION

"cache_hit": false,
"timing_ms": { "embedding": 190.2, "vector_search": 3.0, "connect_and_index": 210.2, "agent_invocation": 2654.4, "total": 3190.2 }
呼び出し全体: 5214.2 ms

"cache_hit": true,
"similarity_score": 1.0,
"timing_ms": { "embedding": 160.4, "vector_search": 3.0, "total": 167.8 }
呼び出し全体: 411.0 ms

"cache_hit": true,
"similarity_score": 0.8623,
"timing_ms": { "embedding": 146.5, "vector_search": 2.9, "total": 153.7 }
呼び出し全体: 398.0 ms

SESSION=$(uuidgen)
uv run invoke.py "VPNに接続する方法を教えて" --tenant globex -s $SESSION
uv run invoke.py "VPNのつなぎ方を知りたい" --tenant globex -s $SESSION

"cache_hit": false,
"model_id": "jp.anthropic.claude-haiku-5-5",

"cache_hit": true,
"similarity_score": 0.8661,
"cache_key": "cache:globex:b32d0012f03439a593d0514c8135ef0a",

# 全文検索でキャッシュ済みの質問文をキーワードで探す
command = ["FT.SEARCH", INDEX_NAME, f"@query:{args.query}",
           "RETURN", "3", "query", "tenant", "hits"]

uv run cache_ops.py text 'VPN'
uv run cache_ops.py text 'VPN*'

== text VPN
[ 0 ]
(35.2 ms)

== text VPN*
[
  2,
  {
    "cache:globex:b32d0012f03439a593d0514c8135ef0a": { "query": "VPNに接続する方法を教えて", "tenant": "globex", "hits": "1" },
    "cache:acme:b32d0012f03439a593d0514c8135ef0a": { "query": "VPNに接続する方法を教えて", "tenant": "acme", "hits": "2" }
  }
]
(45.7 ms)

punctuation: ,.<>{}[]"':;!@#$%^&*()-+=~/\|?

# テナントごとのキャッシュ件数とヒット回数の合計を集計する
command = ["FT.AGGREGATE", INDEX_NAME, "@hits:[-inf +inf]",
           "LOAD", "2", "@tenant", "@hits",
           "GROUPBY", "1", "@tenant",
           "REDUCE", "COUNT", "0", "AS", "entries",
           "REDUCE", "SUM", "1", "@hits", "AS", "total_hits",
           "SORTBY", "2", "@total_hits", "DESC"]

uv run cache_ops.py stats

[
  { "tenant": "acme", "entries": "2", "total_hits": "4" },
  { "tenant": "globex", "entries": "2", "total_hits": "1" }
]
(55.0 ms)

uv run cache_ops.py ranking

[
  { "hits": "2", "query": "VPNに接続する方法を教えて", "tenant": "acme" },
  { "hits": "2", "query": "経費精算の締め日はいつ？", "tenant": "acme" },
  { "hits": "1", "query": "VPNに接続する方法を教えて", "tenant": "globex" }
]
(122.2 ms)

export VALKEY_HOST=<スタック出力のCacheEndpoint> VALKEY_CACHE_NAME=semantic-cache
uv run janome_search.py setup
uv run janome_search.py search 締め日
uv run janome_search.py cleanup

@query:締め日*            -> 0件
@tokens:締め日            -> 1件 ["経費精算の締め日はいつ？"]
@tokens:つなぐ            -> 1件 ["VPNのつなぎ方を知りたい"]
@tokens:申請 @tokens:方法 -> 1件 ["有給休暇の申請方法は？"]

# __init__（起動時・スナップショット前）でTokenizerを作っておく
self.tokenizer = Tokenizer()

def tokenize(self, text: str) -> str:
    """内容語の基本形を空白区切りにする（例: 経費 精算 締め日 いる）"""
    return " ".join(t.base_form for t in self.tokenizer.tokenize(text)
                    if t.part_of_speech.split(",")[0] in CONTENT_POS)

# store()で保存するフィールドに分かち書きを足す
await client.hset(key, {
    "query": query,
    "tokens": self.tokenize(query),
    # answer・tenant・hits・embeddingは今までどおり
})

pnpm cdk destroy