การแคชเชิงความหมายด้วยเวกเตอร์ฝัง

ข้อจำกัดของการแคชแบบตรงกันทุกประการ

การทำงานของการแคชเชิงความหมาย

# Semantic cache flow
# 1. New query arrives: 'How do I cancel my subscription?'
# 2. Embed it: embed_query = embed('How do I cancel my subscription?')
# 3. Search cache index for nearest cached query embedding
# 4. Find cached: 'What is the process to unsubscribe?' (similarity=0.97)
# 5. 0.97 >= threshold (0.95) → cache HIT, return cached response
# 6. If 0.82 < threshold → cache MISS, call LLM, cache result, add embedding to index

แคชเชิงความหมายในหน่วยความจำด้วย NumPy

import numpy as np
from openai import OpenAI

client = OpenAI()

class InMemorySemanticCache:
    def __init__(self, threshold: float = 0.95):
        self.threshold = threshold
        self.embeddings = []    # list of np.ndarray
        self.responses = []     # list of str
        self.queries = []       # list of str (for inspection)

    def _embed(self, text: str) -> np.ndarray:
        resp = client.embeddings.create(model='text-embedding-3-small', input=text)
        return np.array(resp.data[0].embedding)

    def get(self, query: str) -> str | None:
        if not self.embeddings:
            return None
        q_emb = self._embed(query)
        cache_matrix = np.array(self.embeddings)
        # Cosine similarity: dot product of normalized vectors
        norms = np.linalg.norm(cache_matrix, axis=1)
        q_norm = np.linalg.norm(q_emb)
        sims = (cache_matrix @ q_emb) / (norms * q_norm + 1e-8)
        best_idx = int(np.argmax(sims))
        if sims[best_idx] >= self.threshold:
            print(f'[SEMANTIC HIT] sim={sims[best_idx]:.3f} matched: {self.queries[best_idx]!r}')
            return self.responses[best_idx]
        return None

    def set(self, query: str, response: str):
        emb = self._embed(query)
        self.embeddings.append(emb)
        self.responses.append(response)
        self.queries.append(query)

แคชเชิงความหมายด้วย Redis และไพน์โคน

import redis
from pinecone import Pinecone
import hashlib

r = redis.Redis(decode_responses=True)
pc = Pinecone(api_key='YOUR_KEY')
index = pc.Index('semantic-cache')

SIMILARITY_THRESHOLD = 0.95

def semantic_cache_get(query: str) -> str | None:
    q_emb = embed(query)  # from earlier lesson
    results = index.query(vector=q_emb, top_k=1, include_metadata=True)
    if not results.matches:
        return None
    best = results.matches[0]
    if best.score >= SIMILARITY_THRESHOLD:
        response_key = best.metadata.get('response_key')
        return r.get(response_key)
    return None

def semantic_cache_set(query: str, response: str):
    q_emb = embed(query)
    entry_id = hashlib.sha256(query.encode()).hexdigest()[:16]
    response_key = f'sem_cache_resp:{entry_id}'
    r.setex(response_key, 86400, response)  # 24h TTL
    index.upsert(vectors=[{
        'id': entry_id,
        'values': q_emb,
        'metadata': {'query': query[:200], 'response_key': response_key},
    }])

GPTCache: แคชเชิงความหมายสำเร็จรูป

# pip install gptcache
from gptcache import cache
from gptcache.adapter import openai
from gptcache.embedding import OpenAI as EmbeddingOpenAI
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

# Configure GPTCache with FAISS vector store
cache.init(
    embedding_func=EmbeddingOpenAI().to_embeddings,
    data_manager=get_data_manager(
        CacheBase('sqlite'),
        VectorBase('faiss', dimension=1536),
    ),
    similarity_evaluation=SearchDistanceEvaluation(),
)

# Now use the wrapped openai client — caching is transparent
response = openai.ChatCompletion.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'What is RAG?'}],
)

การเลือกค่าเกณฑ์ความคล้ายคลึง

def validate_threshold(cache, query_pairs_with_labels):
    '''
    query_pairs_with_labels: list of (query1, query2, should_match: bool)
    '''
    true_pos = true_neg = false_pos = false_neg = 0
    for q1, q2, should_match in query_pairs_with_labels:
        e1, e2 = embed(q1), embed(q2)
        sim = cosine_similarity(e1, e2)
        matched = sim >= cache.threshold
        if should_match and matched: true_pos += 1
        elif not should_match and not matched: true_neg += 1
        elif not should_match and matched: false_pos += 1
        else: false_neg += 1
    precision = true_pos / (true_pos + false_pos) if (true_pos + false_pos) else 0
    recall = true_pos / (true_pos + false_neg) if (true_pos + false_neg) else 0
    print(f'Precision: {precision:.3f}, Recall: {recall:.3f}')

ขอบเขตของแคชเชิงความหมาย: พรอมป์ตระบบมีความสำคัญ

import hashlib

def make_semantic_cache_namespace(system_prompt: str) -> str:
    return 'sc:' + hashlib.md5(system_prompt.encode()).hexdigest()[:8]

def semantic_cache_get_namespaced(system_prompt: str, user_query: str) -> str | None:
    namespace = make_semantic_cache_namespace(system_prompt)
    q_emb = embed(user_query)
    # Search only within this namespace
    results = index.query(
        vector=q_emb,
        top_k=1,
        filter={'namespace': namespace},
        include_metadata=True,
    )
    if results.matches and results.matches[0].score >= SIMILARITY_THRESHOLD:
        return r.get(results.matches[0].metadata['response_key'])
    return None

การวิเคราะห์อัตราการพบแคชเชิงความหมาย

from sklearn.cluster import KMeans
import numpy as np

def analyze_cache_clusters(cache, n_clusters=10):
    if len(cache.embeddings) < n_clusters:
        print('Not enough cache entries to cluster')
        return

    matrix = np.array(cache.embeddings)
    kmeans = KMeans(n_clusters=n_clusters, n_init=10, random_state=42)
    labels = kmeans.fit_predict(matrix)

    from collections import Counter
    cluster_sizes = Counter(labels)
    print('Query clusters by size:')
    for cluster_id, count in cluster_sizes.most_common():
        representative = cache.queries[labels.tolist().index(cluster_id)]
        print(f'  Cluster {cluster_id}: {count} queries, e.g. {representative!r}')

ข้อควรพิจารณาด้านความปลอดภัยของแคชเชิงความหมาย

import re

PII_PATTERNS = [
    r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',   # phone numbers
    r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b',  # emails
    r'\b\d{9}\b',                           # SSN-like
]

def should_cache(query: str) -> bool:
    for pattern in PII_PATTERNS:
        if re.search(pattern, query, re.IGNORECASE):
            return False  # do not cache queries with PII
    return True

def secure_semantic_completion(user_id: str, query: str) -> str:
    if should_cache(query):
        cached = semantic_cache_get_namespaced(f'user:{user_id}', query)
        if cached:
            return cached
    result = call_llm_api(query)  # actual API call
    if should_cache(query):
        semantic_cache_set_namespaced(f'user:{user_id}', query, result)
    return result

การใช้แคชแบบตรงกันทุกประการร่วมกับแคชเชิงความหมาย

async def two_tier_cached_completion(messages: list[dict], model: str = 'gpt-4o-mini') -> str:
    user_query = messages[-1].get('content', '')
    system_prompt = messages[0].get('content', '') if messages and messages[0]['role'] == 'system' else ''

    # Tier 1: exact cache (instant, free)
    exact_key = make_cache_key(messages, model, temperature=0.0)
    exact_cached = await async_r.get(exact_key)
    if exact_cached:
        return json.loads(exact_cached)

    # Tier 2: semantic cache (one embedding call ~5ms)
    sem_result = semantic_cache_get_namespaced(system_prompt, user_query)
    if sem_result:
        # Backfill exact cache to avoid embedding next time
        await async_r.setex(exact_key, 3600, json.dumps(sem_result))
        return sem_result

    # Tier 3: actual LLM call
    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=0.0
    )
    result = response.choices[0].message.content
    await async_r.setex(exact_key, 3600, json.dumps(result))
    semantic_cache_set_namespaced(system_prompt, user_query, result)
    return result

การอุ่นแคชเพื่อรับมือการเริ่มต้นแบบไม่มีข้อมูล

async def warm_semantic_cache(faq_list: list[dict], system_prompt: str):
    print(f'Warming cache with {len(faq_list)} FAQ entries...')
    for entry in faq_list:
        cached = semantic_cache_get_namespaced(system_prompt, entry['question'])
        if cached:
            print(f'  Already cached: {entry["question"][:50]}')
            continue
        # Generate and cache the response
        response = await async_client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': entry['question']},
            ],
            temperature=0.0,
        )
        answer = response.choices[0].message.content
        semantic_cache_set_namespaced(system_prompt, entry['question'], answer)
        print(f'  Cached: {entry["question"][:50]}')
    print('Cache warming complete')

ตรวจสอบความเข้าใจ

สรุปบทเรียน