Pular para o conteúdo principal

Capítulo 2: Cache e Performance

Tipos de Cache​

Tipo de CacheO que é armazenado em cacheBenefício
EmbeddingEmbeddings de consultaReduz chamadas à API
SemânticoResultados de consultas semelhantesPula a recuperação
RespostaRespostas completas do RAGMais rápido

Cache de Embedding​

from functools import lru_cache
import hashlib

class CachedEmbeddings:
def __init__(self, embeddings, cache_size=10000):
self.embeddings = embeddings
self._cache = {}

def embed_query(self, text):
key = hashlib.md5(text.encode()).hexdigest()
if key not in self._cache:
self._cache[key] = self.embeddings.embed_query(text)
return self._cache[key]

Cache Semântico​

Cache baseado na similaridade da consulta:

from langchain.cache import GPTCache
import gptcache

def get_similar_query(query):
"""Encontrar resultado em cache para consulta semelhante"""
query_embedding = embed(query)

# Buscar no armazenamento vetorial do cache
results = cache_store.similarity_search(
query_embedding,
threshold=0.95 # Consultas muito semelhantes
)

if results:
return results[0].metadata["response"]
return None

Cache de Resposta com Redis​

import redis
import json

r = redis.Redis()
CACHE_TTL = 3600 # 1 hora

def cached_rag(query, rag_chain):
# Verificar cache
cached = r.get(f"rag:{query}")
if cached:
return json.loads(cached)

# Gerar e armazenar em cache
response = rag_chain.invoke(query)
r.setex(f"rag:{query}", CACHE_TTL, json.dumps(response))

return response

Invalidação de Cache​

def invalidate_on_update(document_id):
"""Invalidar caches quando documentos mudam"""
# Limpar caches de resposta relacionados
r.delete_pattern(f"rag:*doc:{document_id}*")

# Remover do cache semântico
semantic_cache.delete(document_id)

Comparação de Performance​

CenárioLatênciaCusto
Sem cache2-5s$0,01/consulta
Cache de embedding1-3s$0,005/consulta
Cache semântico hit<100ms~$0
Cache de resposta hit<50ms~$0