Реалізація кешування LLM-відповідей Exact Semantic Cache

Реалізація кешування LLM-відповідей: Exact та Semantic Cache

LLM-запити дорогі та повільні. Кешування — найдешевший спосіб знизити обидва показники. Два підходи: Exact Cache (хеш промпту) — для строго однакових запитів; Semantic Cache (еmbeddings + векторний пошук) — для семантично подібних запитів. У типовому додатку 30–40% запитів можна закрити кешем.

Exact Cache

import hashlib
import json
import redis
from typing import Optional
from functools import wraps

class ExactLLMCache:

    def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl

    def _make_key(self, messages: list[dict], model: str, temperature: float) -> str:
        """Створює ключ кешу з параметрів запиту"""
        cache_input = {
            "messages": messages,
            "model": model,
            "temperature": temperature,
        }
        content = json.dumps(cache_input, sort_keys=True, ensure_ascii=False)
        return f"llm:exact:{hashlib.sha256(content.encode()).hexdigest()}"

    def get(self, messages: list[dict], model: str, temperature: float = 0) -> Optional[str]:
        key = self._make_key(messages, model, temperature)
        cached = self.redis.get(key)
        if cached:
            return cached.decode()
        return None

    def set(self, messages: list[dict], model: str, temperature: float, response: str):
        key = self._make_key(messages, model, temperature)
        self.redis.setex(key, self.ttl, response.encode())

    def cached_complete(self, complete_fn):
        """Декоратор для кешування функцій"""
        @wraps(complete_fn)
        def wrapper(messages, model="gpt-4o", temperature=0, **kwargs):
            cached = self.get(messages, model, temperature)
            if cached:
                return cached

            result = complete_fn(messages, model=model, temperature=temperature, **kwargs)
            self.set(messages, model, temperature, result)
            return result
        return wrapper

Semantic Cache з векторним пошуком

from openai import OpenAI
import numpy as np
from dataclasses import dataclass

@dataclass
class CachedEntry:
    query_embedding: list[float]
    question: str
    answer: str
    model: str
    created_at: float

class SemanticLLMCache:
    """Кеш на основі семантичної схожості запитань"""

    def __init__(
        self,
        similarity_threshold: float = 0.92,
        max_entries: int = 10000,
    ):
        self.openai = OpenAI()
        self.threshold = similarity_threshold
        self.entries: list[CachedEntry] = []

    def _get_embedding(self, text: str) -> list[float]:
        response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=text,
        )
        return response.data[0].embedding

    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        a_arr = np.array(a)
        b_arr = np.array(b)
        return np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr))

    def get(self, question: str, model: str = None) -> Optional[str]:
        """Шукає подібне запитання в кешу"""
        if not self.entries:
            return None

        query_embedding = self._get_embedding(question)

        best_similarity = 0
        best_answer = None

        for entry in self.entries:
            if model and entry.model != model:
                continue

            similarity = self._cosine_similarity(query_embedding, entry.query_embedding)
            if similarity > best_similarity:
                best_similarity = similarity
                best_answer = entry.answer

        if best_similarity >= self.threshold:
            return best_answer
        return None

    def set(self, question: str, answer: str, model: str):
        """Додає запис до кешу"""
        import time
        embedding = self._get_embedding(question)
        entry = CachedEntry(
            query_embedding=embedding,
            question=question,
            answer=answer,
            model=model,
            created_at=time.time(),
        )
        self.entries.append(entry)

        # Обмежуємо розмір кешу
        if len(self.entries) > 10000:
            self.entries = sorted(self.entries, key=lambda e: e.created_at)[-10000:]

Комбінований кеш з Redis + векторним сховищем

import chromadb
import time

class ProductionSemanticCache:
    """Production-ready кеш: Redis для exact, Chroma для semantic"""

    def __init__(self):
        self.redis = redis.from_url("redis://localhost:6379")
        self.chroma = chromadb.HttpClient(host="localhost", port=8000)
        self.collection = self.chroma.get_or_create_collection("llm_cache")
        self.openai = OpenAI()
        self.similarity_threshold = 0.93
        self.exact_ttl = 3600
        self.semantic_ttl = 86400  # 24 години

    def get(self, question: str, model: str) -> Optional[dict]:
        # 1. Exact match спочатку (швидко)
        exact_key = f"llm:exact:{hashlib.md5(f'{question}:{model}'.encode()).hexdigest()}"
        exact_hit = self.redis.get(exact_key)
        if exact_hit:
            return {"answer": exact_hit.decode(), "cache_type": "exact"}

        # 2. Semantic match
        embedding = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=question,
        ).data[0].embedding

        results = self.collection.query(
            query_embeddings=[embedding],
            n_results=1,
            where={"model": model},
        )

        if results["distances"] and results["distances"][0]:
            distance = results["distances"][0][0]
            similarity = 1 - distance  # Chroma використовує косинусну відстань

            if similarity >= self.similarity_threshold:
                answer = results["documents"][0][0]
                return {"answer": answer, "cache_type": "semantic", "similarity": similarity}

        return None

    def set(self, question: str, answer: str, model: str):
        # Exact cache в Redis
        exact_key = f"llm:exact:{hashlib.md5(f'{question}:{model}'.encode()).hexdigest()}"
        self.redis.setex(exact_key, self.exact_ttl, answer.encode())

        # Semantic cache в Chroma
        embedding = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=question,
        ).data[0].embedding

        self.collection.add(
            ids=[f"{int(time.time())}_{hash(question)}"],
            embeddings=[embedding],
            documents=[answer],
            metadatas=[{"model": model, "question": question, "created_at": time.time()}],
        )

Практичний кейс: FAQ-бот

Профіль: 5000 запитань/день, 70% повторюваних (FAQ про продукт).

До кешу: всі запити → GPT-4o = $180/міс, p95 latency 2.3 сек.

Після кешу:

• Exact cache hit: 35% запитів, latency < 5 мс
• Semantic cache hit: 28% запитів, latency ~50 мс (embedding + пошук)
• LLM запити: 37% від первинного обсягу = $67/міс (-63%)
• Середня latency: 2.3 сек → 0.4 сек

Строки реалізації

• Exact cache (Redis): 0.5–1 день
• Semantic cache (Chroma + embeddings): 2–3 дні
• Production з метриками hit rate: 1 тиждень