AI Engineering Academy · Lekcja

Dokładne buforowanie za pomocą Redis

Buforuj odpowiedzi LLM, haszując kompletny prompt i zapisując wynik w Redis z TTL, aby identyczne żądania obsługiwać natychmiast bez wywołania API.

Lekcja 1 z 413 kroki

Dokładne buforowanie za pomocą Redis to bezpłatna lekcja AI Engineering Academy na CoddyKit. To lekcja 1 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Engineering Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Dlaczego buforować odpowiedzi LLM?

Wywołania API LLM są kosztowne: pojedyncze żądanie GPT-4o może kosztować od 0,005 do 0,15 USD w zależności od liczby tokenów. W wielu aplikacjach znaczna część przychodzących zapytań jest identyczna lub niemal identyczna z wcześniejszymi — dotyczy to na przykład botów FAQ, systemów obsługi klienta czy narzędzi do przeglądu kodu, w których użytkownicy wielokrotnie zadają te same pytania. Buforowanie może wyeliminować od 20 do 50 procent wywołań API w takich zastosowaniach, bezpośrednio obniżając koszty i skracając czas odpowiedzi.

Pamięć podręczna dokładna: projektowanie klucza

Dokładna pamięć podręczna przechowuje odpowiedzi LLM pod kluczem będącym deterministycznym hashem danych wejściowych. Klucz pamięci podręcznej musi uwzględniać każdy parametr wejściowy wpływający na wynik: tablicę wiadomości, nazwę modelu, temperaturę i wszystkie inne parametry zmieniające odpowiedź. Pominięcie któregokolwiek z nich w kluczu powoduje kolizje, w których odpowiedź z pamięci podręcznej zostaje zwrócona dla innego żądania o odmiennych parametrach.

import hashlib
import json

def make_cache_key(messages: list[dict], model: str, temperature: float) -> str:
    # Create a canonical, order-stable representation
    key_data = {
        'model': model,
        'temperature': temperature,
        'messages': messages,  # list order matters
    }
    # Serialize to JSON with sorted keys for determinism
    serialized = json.dumps(key_data, sort_keys=True, ensure_ascii=False)
    # Hash to a fixed-length key safe for Redis
    return 'llm_cache:' + hashlib.sha256(serialized.encode()).hexdigest()

Łączenie z Redisem

Redis jest standardowym wyborem do buforowania odpowiedzi LLM ze względu na opóźnienie odczytu poniżej jednej milisekundy i wbudowaną obsługę TTL. Używaj biblioteki redis-py do dostępu synchronicznego albo aioredis (obecnie scalonej z redis-py jako redis.asyncio) do dostępu asynchronicznego w aplikacjach FastAPI. Przechowuj połączenie z Redisem jako singleton, aby uniknąć wyczerpania puli połączeń.

import redis
import redis.asyncio as aioredis

# Synchronous Redis client
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,  # return str instead of bytes
)

# Async Redis client (for FastAPI)
async_r = aioredis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
)

# Test connection
print(r.ping())  # True if Redis is running

Implementacja wzorca cache-aside

Wzorzec cache-aside jest standardową strategią buforowania dla API LLM. Przy każdym żądaniu: (1) oblicz klucz pamięci podręcznej, (2) sprawdź w Redisie, czy istnieje zapisana odpowiedź, (3) jeśli tak (cache hit), natychmiast ją zwróć, (4) jeśli nie (cache miss), wywołaj API LLM, (5) zapisz odpowiedź w Redisie z TTL, (6) zwróć odpowiedź. Wzorzec ten utrzymuje logikę buforowania poza samym wywołaniem LLM.

import json
from openai import OpenAI

client = OpenAI()

def cached_completion(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
    temperature: float = 0.7,
    ttl_seconds: int = 3600,
) -> str:
    cache_key = make_cache_key(messages, model, temperature)

    # Cache hit?
    cached = r.get(cache_key)
    if cached is not None:
        print('[CACHE HIT]')
        return json.loads(cached)

    # Cache miss: call API
    print('[CACHE MISS]')
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
    )
    result = response.choices[0].message.content

    # Store in cache with TTL
    r.setex(cache_key, ttl_seconds, json.dumps(result))
    return result

Asynchroniczny cache-aside dla FastAPI

W asynchronicznej aplikacji FastAPI używaj asynchronicznego klienta Redis, aby wyszukiwanie w pamięci podręcznej nie blokowało pętli zdarzeń. Wzorzec jest identyczny jak w wersji synchronicznej, ale dla wszystkich operacji Redis używa await. Dzięki temu warstwa buforowania pozostaje w pełni nieblokująca i zgodna z asynchronicznym klientem LLM.

from openai import AsyncOpenAI
import redis.asyncio as aioredis
import json

async_client = AsyncOpenAI()
async_r = aioredis.Redis(host='localhost', port=6379, decode_responses=True)

async def async_cached_completion(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
    temperature: float = 0.0,
    ttl: int = 86400,
) -> str:
    key = make_cache_key(messages, model, temperature)

    cached = await async_r.get(key)
    if cached:
        return json.loads(cached)

    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.choices[0].message.content
    await async_r.setex(key, ttl, json.dumps(result))
    return result

Wybór odpowiedniego TTL

TTL (Time To Live) określa, jak długo buforowane odpowiedzi pozostają aktualne. W przypadku pytań i odpowiedzi opartych na stabilnych bazach wiedzy długi TTL (24–72 godziny) maksymalizuje współczynnik trafień pamięci podręcznej. W przypadku odpowiedzi, które powinny uwzględniać najnowsze dane (podsumowania wiadomości, bieżące ceny), właściwy jest krótki TTL (5–15 minut) albo całkowita rezygnacja z buforowania. W przypadku zadań kreatywnych z temperaturą różną od zera buforowane odpowiedzi mogą szybko się dezaktualizować — należy rozważyć buforowanie wyłącznie dla temperature=0.

# TTL strategy by use case
TTL_STRATEGY = {
    'faq_answering':          86400 * 7,   # 7 days — stable facts
    'code_explanation':       86400,        # 1 day — code rarely changes
    'document_summarization': 3600 * 6,    # 6 hours
    'news_analysis':          300,          # 5 minutes — stale quickly
    'creative_writing':       0,            # 0 = don't cache (non-deterministic)
}

def get_ttl_for_use_case(use_case: str) -> int:
    return TTL_STRATEGY.get(use_case, 3600)  # default 1 hour

Metryki i monitorowanie pamięci podręcznej

Śledź współczynnik trafień pamięci podręcznej jako główną metrykę redukcji kosztów. Współczynnik trafień na poziomie 30 procent oznacza, że unika się 30 procent wywołań API. Przechowuj liczbę trafień i chybień w samym Redisie, używając poleceń INCR dla oddzielnych liczników. Udostępnij w aplikacji FastAPI endpoint /metrics, który raportuje bieżący współczynnik trafień, łączną liczbę żądań i szacowane oszczędności, aby określić zwrot z inwestycji w buforowanie.

CACHE_HITS_KEY = 'llm_cache_metrics:hits'
CACHE_MISSES_KEY = 'llm_cache_metrics:misses'

async def async_cached_completion_instrumented(messages, model, temperature=0.0):
    key = make_cache_key(messages, model, temperature)
    cached = await async_r.get(key)

    if cached:
        await async_r.incr(CACHE_HITS_KEY)
        return json.loads(cached)

    await async_r.incr(CACHE_MISSES_KEY)
    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.choices[0].message.content
    await async_r.setex(key, 3600, json.dumps(result))
    return result

async def get_cache_stats():
    hits = int(await async_r.get(CACHE_HITS_KEY) or 0)
    misses = int(await async_r.get(CACHE_MISSES_KEY) or 0)
    total = hits + misses
    return {'hit_rate': hits / total if total > 0 else 0, 'total': total}

Strategie unieważniania pamięci podręcznej

Unieważnianie dokładnej pamięci podręcznej jest proste, ponieważ klucze są deterministyczne. Aby unieważnić konkretny wpis, ponownie oblicz jego klucz i wywołaj r.delete(key). Aby unieważnić wszystkie wpisy dla określonego wzorca promptu, użyj prefiksów kluczy Redis wraz ze skanowaniem z symbolem wieloznacznym. Aby unieważnić całą pamięć podręczną po dużej aktualizacji bazy wiedzy, wywołaj r.flushdb() (zachowaj ostrożność — spowoduje to usunięcie wszystkich kluczy w bazie danych).

async def invalidate_cache_entry(messages, model, temperature):
    key = make_cache_key(messages, model, temperature)
    deleted = await async_r.delete(key)
    print(f'Deleted {deleted} cache entries')

async def invalidate_all_llm_cache():
    # Scan for all keys with prefix 'llm_cache:'
    keys_to_delete = []
    async for key in async_r.scan_iter(match='llm_cache:*'):
        keys_to_delete.append(key)
    if keys_to_delete:
        await async_r.delete(*keys_to_delete)
    print(f'Invalidated {len(keys_to_delete)} cache entries')

Serializowanie złożonych odpowiedzi

Jeśli aplikacja buforuje całe obiekty odpowiedzi API (a nie tylko ich treść tekstową), serializuj je starannie. Pełny obiekt ChatCompletion zawiera informacje o użyciu tokenów, wersji modelu i przyczynie zakończenia — przydatne podczas rejestrowania danych i śledzenia kosztów. Użyj metody .model_dump_json() pakietu SDK, aby serializować obiekty odpowiedzi Pydantic do ciągów JSON, a podczas pobierania z pamięci podręcznej odtwórz je za pomocą ChatCompletion.model_validate_json().

from openai.types.chat import ChatCompletion

async def cached_completion_full_response(
    messages, model='gpt-4o-mini', temperature=0.0
):
    key = make_cache_key(messages, model, temperature) + ':full'
    cached = await async_r.get(key)

    if cached:
        return ChatCompletion.model_validate_json(cached)  # reconstruct object

    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    # Serialize Pydantic model to JSON
    await async_r.setex(key, 3600, response.model_dump_json())
    return response

Buforowanie a niedeterminizm

Dokładne buforowanie ma sens tylko w przypadku żądań deterministycznych lub niemal deterministycznych. Przy temperature=0 i top_p=1.0 większość LLM generuje ten sam wynik dla tych samych danych wejściowych (choć nie jest to gwarantowane z powodu niedeterminizmu obliczeń zmiennoprzecinkowych). Przy wyższych wartościach temperatury buforowane odpowiedzi szybko się dezaktualizują, ponieważ model wygenerowałby inne wyniki. Zawsze buforuj odpowiedzi przy temperature=0 albo wyraźnie zaznacz w kluczu pamięci podręcznej, że odpowiedzi mogą się różnić.

Klaster Redis i konfiguracja produkcyjna

W środowiskach produkcyjnych z dużą ilością danych w pamięci podręcznej używaj Redis Cluster do poziomego fragmentowania danych między wiele węzłów albo zarządzanej usługi Redis, takiej jak AWS ElastiCache lub Redis Cloud. Ustaw politykę maxmemory (zwykle allkeys-lru, aby usuwać najdawniej używane wpisy po zapełnieniu pamięci), aby zapobiec wyczerpaniu pamięci Redis i automatycznie zarządzać rozmiarem pamięci podręcznej.

# Redis configuration for production LLM caching
# In redis.conf:
# maxmemory 2gb
# maxmemory-policy allkeys-lru

# Connection with retry and connection pool
import redis
from redis.retry import Retry
from redis.backoff import ExponentialBackoff

retry = Retry(ExponentialBackoff(base=0.1), 3)
production_redis = redis.Redis(
    host='your-redis-host.cache.amazonaws.com',
    port=6379,
    ssl=True,
    decode_responses=True,
    max_connections=50,
    retry=retry,
    retry_on_error=[redis.ConnectionError, redis.TimeoutError],
)

Szybkie sprawdzenie

Sprawdź swoją wiedzę na temat dokładnego buforowania odpowiedzi LLM za pomocą Redisa z tej lekcji.

Odpowiedz w interaktywnej lekcji

Podsumowanie lekcji

W tej lekcji dowiedział się Pan / dowiedziała się Pani, że dokładne buforowanie tworzy hash ze wszystkich danych wejściowych LLM, aby uzyskać deterministyczny klucz pamięci podręcznej, wzorzec cache-aside sprawdza Redis przed wywołaniem API i zapisuje wyniki po chybieniu, a wybór TTL powinien odzwierciedlać częstotliwość zmian treści — dłuższy TTL sprawdza się w przypadku stabilnej wiedzy, a krótszy w przypadku dynamicznych danych. Monitoruj współczynnik trafień pamięci podręcznej jako główną metrykę redukcji kosztów. W następnej części zbudujemy pamięć podręczną semantyczną dla podobnych, lecz nieidentycznych zapytań.

Bezpłatny start

Ucz się Python dzięki korepetycjom AI — za darmo

Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.

Kursy: 30
Lekcje: 120

Zacznij naukę bezpłatnie Wyświetl pełny kurs

Często zadawane pytania

Czy lekcja „Dokładne buforowanie za pomocą Redis” jest bezpłatna?

Tak — pełny tekst „Dokładne buforowanie za pomocą Redis” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Engineering Academy, przejdź na CoddyKit PRO. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Co nauczysz się w „Dokładne buforowanie za pomocą Redis”?

Buforuj odpowiedzi LLM, haszując kompletny prompt i zapisując wynik w Redis z TTL, aby identyczne żądania obsługiwać natychmiast bez wywołania API. Ćwiczysz AI Engineering Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Engineering Academy?

Nie wymagamy żadnego doświadczenia. AI Engineering Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 1 z 4.

Ile czasu zajmuje lekcja „Dokładne buforowanie za pomocą Redis”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Engineering Academy?

Tak. Każda lekcja AI Engineering Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

Dokładne buforowanie za pomocą Redis
Buforowanie semantyczne z embeddingami
Buforowanie prefiksów promptów OpenAI
Batchowanie, routing modeli i pulpity kosztów

← Powrót do AI Engineering Academy