Capítulo 2: Cache e Performance
Tipos de Cache
| Tipo de Cache | O que é armazenado em cache | Benefício |
|---|---|---|
| Embedding | Embeddings de consulta | Reduz chamadas à API |
| Semântico | Resultados de consultas semelhantes | Pula a recuperação |
| Resposta | Respostas completas do RAG | Mais 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ário | Latência | Custo |
|---|---|---|
| Sem cache | 2-5s | $0,01/consulta |
| Cache de embedding | 1-3s | $0,005/consulta |
| Cache semântico hit | <100ms | ~$0 |
| Cache de resposta hit | <50ms | ~$0 |