翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

ElastiCache for Valkey を使用したセマンティックキャッシュの実装

次のチュートリアルでは、Amazon Bedrock で ElastiCache for Valkey を使用して、リードスルーセマンティックキャッシュを実装する方法を示します。

ステップ 1: ElastiCache for Valkey クラスターを作成する

以下を使用して、バージョン 8.2 以降の ElastiCache for Valkey クラスターを作成します AWS CLI。

aws elasticache create-replication-group \
  --replication-group-id "valkey-semantic-cache" \
  --cache-node-type cache.r7g.large \
  --engine valkey \
  --engine-version 8.2 \
  --num-node-groups 1 \
  --replicas-per-node-group 1

ステップ 2: クラスターに接続し、埋め込みを設定する

Amazon EC2 インスタンスで実行されているアプリケーションコードから、ElastiCache クラスターに接続し、埋め込みモデルを設定します。

from valkey.cluster import ValkeyCluster
from langchain_aws import BedrockEmbeddings

# Connect to ElastiCache for Valkey
valkey_client = ValkeyCluster(
    host="mycluster.xxxxxx.clustercfg.use1.cache.amazonaws.com",  # Your cluster endpoint
    port=6379,
    decode_responses=False
)

# Set up Amazon Bedrock Titan embeddings
embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1"
)

ホスト値を ElastiCache クラスターの設定エンドポイントに置き換えます。クラスターエンドポイントを検索する手順については、ElastiCache クラスターへのアクセス」を参照してください。

ステップ 3: セマンティックキャッシュのベクトルインデックスを作成する

ベクトル検索の COSINE 距離を持つ HNSW インデックスを使用してクエリを自動的に埋め込む ValkeyStore を設定します。

from langgraph_checkpoint_aws import ValkeyStore
from hashlib import md5

store = ValkeyStore(
    client=valkey_client,
    index={
        "collection_name": "semantic_cache",
        "embed": embeddings,
        "fields": ["query"],           # Fields to vectorize
        "index_type": "HNSW",          # Vector search algorithm
        "distance_metric": "COSINE",   # Similarity metric
        "dims": 1024                   # Titan V2 produces 1024-d vectors
    }
)
store.setup()

def cache_key_for_query(query: str):
    """Generate a deterministic cache key for a query."""
    return md5(query.encode("utf-8")).hexdigest()

ステップ 4: キャッシュ検索および更新関数を実装する

関数を作成してキャッシュで意味的に類似したクエリを検索し、新しいクエリとレスポンスのペアを保存します。

def search_cache(user_message: str, k: int = 3, min_similarity: float = 0.8):
    """Look up a semantically similar cached response from ElastiCache."""
    hits = store.search(
        namespace="semantic-cache",
        query=user_message,
        limit=k
    )
    if not hits:
        return None

    # Sort by similarity score (highest first)
    hits = sorted(hits, key=lambda h: h["score"], reverse=True)
    top_hit = hits[0]
    score = top_hit["score"]

    if score < min_similarity:
        return None  # Below similarity threshold

    return top_hit["value"]["answer"]  # Return cached answer


def store_cache(user_message: str, result_message: str):
    """Store a new query-response pair in the semantic cache."""
    key = cache_key_for_query(user_message)
    store.put(
        namespace="semantic-cache",
        key=key,
        value={
            "query": user_message,
            "answer": result_message
        }
    )

ステップ 5: 読み取りキャッシュパターンを実装する

キャッシュをアプリケーションのリクエスト処理に統合します。

import time

def handle_query(user_message: str) -> dict:
    """Handle a user query with read-through semantic cache."""
    start = time.time()

    # Step 1: Search the semantic cache
    cached_response = search_cache(user_message, min_similarity=0.8)

    if cached_response:
        # Cache hit - return cached response
        elapsed = (time.time() - start) * 1000
        return {
            "response": cached_response,
            "source": "cache",
            "latency_ms": round(elapsed, 1),
        }

    # Step 2: Cache miss - invoke LLM
    llm_response = invoke_llm(user_message)  # Your LLM invocation function

    # Step 3: Store the response in cache for future reuse
    store_cache(user_message, llm_response)

    elapsed = (time.time() - start) * 1000
    return {
        "response": llm_response,
        "source": "llm",
        "latency_ms": round(elapsed, 1),
    }

基盤となる Valkey コマンド

次の表は、セマンティックキャッシュの実装に使用される Valkey コマンドを示しています。