翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# ベストプラクティス
## キャッシュできるデータの選択
セマンティックキャッシュは、レスポンスが比較的安定している反復クエリに適していますが、リアルタイムレスポンスまたは高度に動的なレスポンスはキャッシュの候補として適さないことがよくあります。
既存のアプリケーションコンテキスト (製品 ID、カテゴリ、リージョン、ユーザーセグメントなど) から派生したタグフィルターと数値フィルターを使用して、キャッシュの対象となるクエリとレスポンスを決定し、キャッシュヒットの関連性を向上させます。
## 類似度しきい値の調整
類似度しきい値は、キャッシュヒット率と回答品質のトレードオフを制御します。コスト削減とユースケースの精度のバランスを取るしきい値を選択します。
| Threshold | ヒット率 | 品質リスク | 次の用途に適しています |
| --- | --- | --- | --- |
| 0.95 (厳格) | 低 (～25%) | 非常に小さい | 医療、法律、財務アプリケーション |
| 0.90 (中) | 中 (～55%) | 低 | 一般的なチャットボット |
| 0.80 (バランス) | 高 (～75%) | 低～中 | よくある質問ボット、IT サポート |
| 0.75 (緩和) | 非常に高い (～90%) | 中 | 大量の反復クエリ |
**重要**
より高いしきい値 (0.90～0.95) から開始し、精度をモニタリングしながら徐々に下げます。A/B テストを使用して、ワークロードに最適なバランスを見つけます。
## スタンドアロンクエリと会話
+ **スタンドアロンクエリの場合** – セマンティックキャッシュをユーザークエリテキストに直接適用します。
+ **マルチターン会話の場合** – まず会話メモリを使用して、現在のターンに応答するために必要な主要な事実と最近のメッセージを取得します。次に、raw ダイアログ全体を埋め込む代わりに、現在のユーザーメッセージと取得したコンテキストの組み合わせにセマンティックキャッシュを適用します。
## キャッシュ無効化期間の設定
TTL を使用して、キャッシュミスで再生成される前にキャッシュされたレスポンスが提供される期間を制御します。
| データ型 | 推奨 TTL | 根拠 |
| --- | --- | --- |
| 静的ファクト (ドキュメント、ポリシー) | 24 時間 | 事実の変更頻度が低い |
| 製品情報 | 12～24 時間 | ほとんどのカタログで毎日更新 |
| 一般的なアシスタントレスポンス | 1～4 時間 | 鮮度とヒット率のバランスをとる |
| リアルタイムデータ (料金、インベントリ) | 5～15 分 | データの頻繁な変更 |
| 会話コンテキスト | 30 分 | セッションスコープ、有効期間が短い |
```
# Set TTL with random jitter to spread out cache invalidations
import random
base_ttl = 82800 # ~23 hours
jitter = random.randint(0, 3600) # Up to 1 hour of jitter
valkey_client.expire(cache_key, base_ttl + jitter)
```
**ヒント**
アプリケーションのユースケースと、データまたはモデルの出力が変更される頻度に一致する TTLs を設定します。TTLs を長くするとキャッシュヒット率が上がりますが、古い回答のリスクが高まります。TTLs を短くすると、レスポンスは新しくなりますが、キャッシュヒット率が低くなり、より多くの LLM 推論が必要になります。
## モニタリングとコスト追跡
キャッシュパフォーマンスメトリクスを追跡して、セマンティックキャッシュを経時的に最適化します。
```
def record_cache_event(valkey_client, event_type: str):
"""Track cache hits and misses using atomic counters."""
valkey_client.incr(f"cache:metrics:{event_type}")
# Also track hourly for time-series analysis
from datetime import datetime
hour_key = datetime.now().strftime("%Y%m%d%H")
counter_key = f"cache:metrics:{event_type}:{hour_key}"
valkey_client.incr(counter_key)
valkey_client.expire(counter_key, 86400 * 7) # Keep 7 days
def get_cache_stats(valkey_client) -> dict:
"""Get current cache performance metrics."""
hits = int(valkey_client.get("cache:metrics:hit") or 0)
misses = int(valkey_client.get("cache:metrics:miss") or 0)
total = hits + misses
hit_rate = hits / total if total > 0 else 0
avg_cost_per_call = 0.015 # Example: ~$0.015 per LLM call
savings = hits * avg_cost_per_call
return {
"total_requests": total,
"hits": hits,
"misses": misses,
"hit_rate": round(hit_rate, 3),
"estimated_savings_usd": round(savings, 2),
}
```
## メモリ管理
+ **maxmemory ポリシー**の設定 – クラスターがメモリ制限に達したときにleast-recently-usedキャッシュエントリを自動的に削除するように ElastiCache クラスター`maxmemory-policy allkeys-lru`で を設定します。
+ **容量の計画** – 各キャッシュエントリには通常、約 4～6 KB (埋め込みディメンション x 4 バイト \+ クエリテキスト \+ レスポンステキスト) が必要です。1 GB の ElastiCache インスタンスには、約 170,000 個のキャッシュされたエントリを保存できます。
+ **古いデータにキャッシュ無効化を使用する** – 基盤となるデータが変更された場合は、テキスト検索を使用して関連するキャッシュエントリを検索して無効にします。
```
def invalidate_by_topic(valkey_client, topic_keyword: str):
"""Remove cached entries matching a topic after a data update."""
results = valkey_client.execute_command(
"FT.SEARCH", "semantic_cache",
f"@query:{topic_keyword}",
"NOCONTENT", # Only return keys, not fields
)
if results[0] > 0:
keys = results[1:]
for key in keys:
valkey_client.delete(key)
print(f"Invalidated {len(keys)} cached entries for '{topic_keyword}'")
```