Как сократить расходы на API нейросетей на 80%: Семантическое кэширование в Redis

import numpy as np
from openai import OpenAI
import redis

client = OpenAI()
r = redis.Redis(host="localhost", port=6379, decode_responses=False)

def get_embedding(text: str) -> list:
    resp = client.embeddings.create(input=text, model="text-embedding-3-small")
    return resp.data[0].embedding

def query_with_semantic_cache(prompt: str, threshold: float = 0.94) -> str:
    query_vector = get_embedding(prompt)
    
    # 1. Поиск в Redis (имитация векторного совпадения)
    cached_response = check_redis_vector_similarity(query_vector, threshold)
    if cached_response:
        print("[CACHE HIT] Ответ получен из Redis за 4 мс! Стоимость $0.00")
        return cached_response

    # 2. Промах кэша: отправляем запрос в дорогую модель
    print("[CACHE MISS] Обращение к frontier LLM...")
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    answer = response.choices[0].message.content
    
    # 3. Запись в кэш
    save_to_redis_vector(query_vector, answer)
    return answer