Семантический кэш (semantic cache) — это техника кэширования ответов AI-моделей, основанная на смысловом сходстве запросов, а не на точном совпадении строк. Если пользователь задаёт вопрос, семантически идентичный ранее обработанному, система возвращает кэшированный ответ без обращения к LLM. Это экономит до 70% расходов на токены и снижает латентность с секунд до миллисекунд.
Почему обычный кэш не работает для LLM
Классический кэш (Redis, Memcached) работает по точному совпадению ключа. Для LLM это неэффективно, потому что одну и ту же мысль можно выразить десятками способов:
- «Какая столица Франции?»
- «Назови столицу Франции»
- «Столица Франции — это...?»
- «What is the capital of France?» (если поддерживается мультиязычность)
Все эти запросы имеют одинаковый смысл и один ответ, но точный кэш обработает каждый как уникальный и отправит 4 запроса в LLM.
Как работает semantic cache
Алгоритм semantic cache:
- Преобразовать запрос пользователя в embedding (вектор)
- Найти ближайший вектор в кэше (cosine similarity)
- Если сходство выше порога (например, 0.95) — вернуть кэшированный ответ
- Иначе — отправить запрос в LLM, сохранить ответ и embedding в кэш
import numpy as np
from openai import OpenAI
client = OpenAI(
base_url="https://api.modelswitch.ru/v1",
api_key="msk_ваш_ключ",
)
class SemanticCache:
def __init__(self, similarity_threshold=0.95):
self.threshold = similarity_threshold
self.cache = [] # [{embedding, query, response, timestamp}]
def get_embedding(self, text: str) -> list[float]:
resp = client.embeddings.create(
model="text-embedding-3-small",
input=text,
)
return resp.data[0].embedding
def cosine_similarity(self, a: list[float], b: list[float]) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
def get(self, query: str) -> str | None:
query_emb = self.get_embedding(query)
best_score, best_response = 0.0, None
for entry in self.cache:
score = self.cosine_similarity(query_emb, entry["embedding"])
if score > best_score:
best_score = score
best_response = entry["response"]
if best_score >= self.threshold:
return best_response
return None
def put(self, query: str, response: str):
embedding = self.get_embedding(query)
self.cache.append({
"embedding": embedding,
"query": query,
"response": response,
"timestamp": time.time(),
})
# Использование
cache = SemanticCache(similarity_threshold=0.95)
def ask(question: str) -> str:
cached = cache.get(question)
if cached:
print("Cache HIT")
return cached
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": question}],
)
answer = response.choices[0].message.content
cache.put(question, answer)
print("Cache MISS")
return answerProduction-реализация с Redis и векторным поиском
Для production используйте векторную базу данных (Qdrant, Pinecone, Weaviate) или Redis с модулем RediSearch:
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid
import time
qdrant = QdrantClient(host="localhost", port=6333)
# Создаём коллекцию для кэша
qdrant.recreate_collection(
collection_name="prompt_cache",
vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)
class ProductionSemanticCache:
def __init__(self, threshold=0.95, ttl_seconds=3600):
self.threshold = threshold
self.ttl = ttl_seconds
def get(self, query_embedding: list[float]) -> str | None:
results = qdrant.search(
collection_name="prompt_cache",
query_vector=query_embedding,
limit=1,
score_threshold=self.threshold,
)
if results:
payload = results[0].payload
if time.time() - payload["timestamp"] < self.ttl:
return payload["response"]
return None
def put(self, query_embedding: list[float], query: str, response: str):
qdrant.upsert(
collection_name="prompt_cache",
points=[PointStruct(
id=str(uuid.uuid4()),
vector=query_embedding,
payload={"query": query, "response": response,
"timestamp": time.time()},
)],
)Стратегии инвалидации кэша
Кэшированные ответы устаревают. Стратегии инвалидации:
- TTL (Time-to-Live) — автоматическое удаление через N секунд. Для фактологических вопросов — 1-24 часа, для творческих — не кэшировать
- При смене модели — при переключении на новую модель сбрасывайте кэш
- При изменении системного промпта — новый промпт = новые ответы
- По объёму — LRU-вытеснение при достижении лимита записей
ModelSwitch передаёт запросы к моделям прозрачно, поэтому semantic cache можно реализовать на стороне вашего приложения, а экономию отслеживать через дашборд расходов.
Заключение
Semantic cache — один из самых эффективных способов снизить расходы на AI API. В приложениях с повторяющимися запросами (FAQ-боты, поддержка, образовательные платформы) экономия может достигать 70%. Начните с простой in-memory реализации, затем перейдите на векторную базу данных для production. Стоимость embedding-запроса ($0.02 за 1M токенов) ничтожна по сравнению со стоимостью запроса к GPT-4o ($10 за 1M output-токенов).