Embeddings em RAG

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

modelo = SentenceTransformer('all-MiniLM-L6-v2')

# Palavras com significados diferentes
textos = [
    "banco instituição financeira",
    "banco para sentar",
    "gato animal doméstico"
]

embeddings = modelo.encode(textos)

# Similaridade entre os dois "bancos" (significados diferentes)
sim_banco_financeiro_vs_assento = cosine_similarity(
    [embeddings[0]], [embeddings[1]]
)[0][0]  # ~0.15 (baixa)

# Similaridade entre "banco financeiro" e "gato" (nenhuma relação)
sim_banco_vs_gato = cosine_similarity(
    [embeddings[0]], [embeddings[2]]
)[0][0]  # ~0.02 (muito baixa)

from sentence_transformers import SentenceTransformer, util

# 1. Textos de documentos
textos = [
    "A inteligência artificial está transformando a medicina.",
    "Carros elétricos são o futuro do transporte sustentável.",
    "O aquecimento global é causado pela emissão de gases.",
    "Redes neurais profundas aprendem representações hierárquicas."
]

# 2. Query do usuário
query = "Como a IA está mudando a saúde?"

# 3. Carrega modelo e gera embeddings
modelo = SentenceTransformer('all-MiniLM-L6-v2')
embeddings_textos = modelo.encode(textos)
embedding_query = modelo.encode(query)

# 4. Calcula similaridades
similaridades = util.cos_sim(embedding_query, embeddings_textos)[0]

# 5. Recupera os 2 documentos mais relevantes
top_k = 2
indices = similaridades.argsort(descending=True)[:top_k]

print("Resultados da busca semântica:")
for idx in indices:
    print(f"- {textos[idx]} (Score: {similaridades[idx]:.4f})")

import numpy as np
from scipy.spatial.distance import cosine, euclidean

# Exemplo de vetores
v1 = np.array([0.5, 0.3, -0.2])
v2 = np.array([0.4, 0.2, -0.1])
v3 = np.array([-0.9, 0.1, 0.8])

# Similaridade por cosseno (1 = idêntico, 0 = perpendicular, -1 = oposto)
cos_sim = 1 - cosine(v1, v2)  # ≈ 0.95

# Distância Euclidiana (menor = mais similar)
dist_euclidiana = euclidean(v1, v2)  # ≈ 0.17

# Produto Escalar (útil para vetores normalizados)
dot_product = np.dot(v1, v2)  # ≈ 0.29

# Comparando modelos
modelo1 = SentenceTransformer('all-MiniLM-L6-v2')  # 384 dims
modelo2 = SentenceTransformer('all-mpnet-base-v2')  # 768 dims

texto = "Deep learning basics"

emb1 = modelo1.encode(texto)
emb2 = modelo2.encode(texto)

print(f"Dimensões: {len(emb1)} vs {len(emb2)}")
print(f"Norma emb1: {np.linalg.norm(emb1):.4f}")  # ~1.0 (normalizado)
print(f"Norma emb2: {np.linalg.norm(emb2):.4f}")  # ~1.0 (normalizado)

def calcular_similaridade(texto1, texto2, modelo):
    """Calcula similaridade semântica entre dois textos."""
    emb1 = modelo.encode(texto1)
    emb2 = modelo.encode(texto2)
    return cosine_similarity([emb1], [emb2])[0][0]

# Teste com diferentes pares
pares = [
    ("O gato dormiu no sofá.", "O felino descansou no estofado."),
    ("O gato dormiu no sofá.", "O carro vermelho acelerou."),
    ("Python é uma linguagem de programação.", "Java é outra linguagem popular.")
]

for t1, t2 in pares:
    sim = calcular_similaridade(t1, t2, modelo)
    print(f"'{t1[:20]}...' vs '{t2[:20]}...' → {sim:.3f}")

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader

# Carrega documentos
loader = TextLoader("base_conhecimento.txt")
documentos = loader.load()

# Divide em chunks semânticos
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ".", " "],
    length_function=len,
)

chunks = splitter.split_documents(documentos)
print(f"Total de chunks: {len(chunks)}")

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

# Carrega modelo
modelo = SentenceTransformer('all-MiniLM-L6-v2')

# Gera embeddings para todos os chunks
textos_chunks = [chunk.page_content for chunk in chunks]
embeddings_chunks = modelo.encode(textos_chunks)

# Cria índice FAISS para busca eficiente
dimensao = embeddings_chunks.shape[1]
indice = faiss.IndexFlatIP(dimensao)  # Produto interno (similaridade)

# Normaliza embeddings para usar similaridade por cosseno
embeddings_normalizados = embeddings_chunks / np.linalg.norm(embeddings_chunks, axis=1, keepdims=True)
indice.add(embeddings_normalizados.astype('float32'))

print(f"Índice criado com {indice.ntotal} vetores")

def recuperar_contexto(query, indice, modelo, chunks, k=3):
    """Recupera os K chunks mais relevantes para uma query."""
    # Embedding da query
    emb_query = modelo.encode(query)
    emb_query_normalizado = emb_query / np.linalg.norm(emb_query)
    
    # Busca no índice
    similaridades, indices = indice.search(
        emb_query_normalizado.reshape(1, -1).astype('float32'), 
        k
    )
    
    # Recupera os chunks
    contextos = []
    for idx, score in zip(indices[0], similaridades[0]):
        if idx != -1:
            contextos.append({
                'texto': chunks[idx].page_content,
                'score': score
            })
    
    return contextos

# Teste
query = "Quais são as aplicações do aprendizado por transferência?"
contextos = recuperar_contexto(query, indice, modelo, chunks, k=3)

for i, ctx in enumerate(contextos, 1):
    print(f"\nContexto {i} (Score: {ctx['score']:.4f}):")
    print(ctx['texto'][:200] + "...")

from transformers import pipeline

# Carrega um modelo generativo (exemplo com GPT-2)
gerador = pipeline('text-generation', model='gpt2')

def gerar_resposta(query, contextos, gerador):
    """Gera resposta baseada no contexto recuperado."""
    # Constrói prompt
    contexto_texto = "\n\n".join([ctx['texto'] for ctx in contextos])
    
    prompt = f"""
    Contexto:
    {contexto_texto}
    
    Pergunta: {query}
    
    Resposta baseada APENAS no contexto fornecido:
    """
    
    # Gera resposta
    resposta = gerador(
        prompt,
        max_length=300,
        num_return_sequences=1,
        temperature=0.7,
        pad_token_id=50256
    )[0]['generated_text']
    
    return resposta

# Gera resposta
resposta = gerar_resposta(query, contextos, gerador)
print("\nResposta Gerada:")
print(resposta)

def avaliar_recuperacao(queries, relevantes, indice, modelo, chunks, k=5):
    """Avalia a qualidade da recuperação com Precision@K e Recall@K."""
    
    precision_k_total = 0
    recall_k_total = 0
    
    for query, docs_relevantes in zip(queries, relevantes):
        # Recupera documentos
        contextos = recuperar_contexto(query, indice, modelo, chunks, k)
        docs_recuperados = [ctx['texto'] for ctx in contextos]
        
        # Calcula Precision@K
        relevantes_recuperados = sum(1 for doc in docs_recuperados if doc in docs_relevantes)
        precision_k = relevantes_recuperados / k
        precision_k_total += precision_k
        
        # Calcula Recall@K
        recall_k = relevantes_recuperados / len(docs_relevantes)
        recall_k_total += recall_k
    
    return {
        'Precision@K': precision_k_total / len(queries),
        'Recall@K': recall_k_total / len(queries)
    }

# Exemplo de uso
queries = ["O que é transfer learning?", "Explique redes neurais"]
relevantes = [
    ["Documento sobre transfer learning", "Artigo sobre fine-tuning"],
    ["Introdução a redes neurais", "Deep learning book"]
]

metricas = avaliar_recuperacao(queries, relevantes, indice, modelo, chunks)
print(f"Precision@5: {metricas['Precision@K']:.3f}")
print(f"Recall@5: {metricas['Recall@K']:.3f}")

from sentence_transformers import CrossEncoder

# Modelo cross-encoder para re-ranking
re_ranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def re_rank(query, documentos, re_ranker, top_k=3):
    """Re-rank usando cross-encoder."""
    pares = [[query, doc] for doc in documentos]
    scores = re_ranker.predict(pares)
    
    # Ordena por score
    documentos_ranqueados = sorted(
        zip(documentos, scores),
        key=lambda x: x[1],
        reverse=True
    )
    
    return documentos_ranqueados[:top_k]

from sentence_transformers import SentenceTransformer
import time

def benchmark_model(model_name, textos, num_iter=100):
    """Avalia velocidade e qualidade de um modelo."""
    
    # Carrega modelo
    modelo = SentenceTransformer(model_name)
    
    # Mede tempo de encoding
    start = time.time()
    for _ in range(num_iter):
        _ = modelo.encode(textos, show_progress_bar=False)
    tempo_total = time.time() - start
    
    # Gera embeddings
    embeddings = modelo.encode(textos)
    
    # Calcula similaridades
    from sklearn.metrics.pairwise import cosine_similarity
    sim_matrix = cosine_similarity(embeddings)
    
    return {
        'nome': model_name,
        'dimensão': embeddings.shape[1],
        'tempo_médio': tempo_total / num_iter,
        'similaridade_média': sim_matrix.mean()
    }

# Teste com diferentes modelos
textos_teste = [
    "O cachorro está feliz.",
    "O cão está contente.",
    "O carro está rápido.",
    "O veículo é veloz."
]

modelos = ['all-MiniLM-L6-v2', 'all-mpnet-base-v2', 'paraphrase-multilingual-MiniLM-L12-v2']

for modelo in modelos:
    resultado = benchmark_model(modelo, textos_teste)
    print(f"\n{resultado['nome']}:")
    print(f"  Dimensão: {resultado['dimensão']}")
    print(f"  Tempo médio: {resultado['tempo_médio']:.4f}s")
    print(f"  Similaridade média: {resultado['similaridade_média']:.4f}")

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader
import random

# 1. Dados de exemplo (pares similares/dissimilares)
exemplos_treino = []
with open('dados_finetuning.txt', 'r') as f:
    for linha in f:
        texto1, texto2, label = linha.strip().split('\t')
        exemplo = InputExample(texts=[texto1, texto2], label=float(label))
        exemplos_treino.append(exemplo)

# 2. Carrega modelo base
modelo = SentenceTransformer('all-MiniLM-L6-v2')

# 3. Configura treinamento
train_dataloader = DataLoader(exemplos_treino, shuffle=True, batch_size=16)
loss = losses.CosineSimilarityLoss(modelo)

# 4. Fine-tuning
modelo.fit(
    train_objectives=[(train_dataloader, loss)],
    epochs=3,
    warmup_steps=100,
    output_path='modelo_finetuned',
    show_progress_bar=True
)

print("Fine-tuning concluído!")

import pickle
import numpy as np

# Salvar embeddings para uso futuro
def salvar_embeddings(embeddings, caminho):
    with open(caminho, 'wb') as f:
        pickle.dump(embeddings, f)

# Carregar embeddings
def carregar_embeddings(caminho):
    with open(caminho, 'rb') as f:
        return pickle.load(f)

# Exemplo
salvar_embeddings(embeddings_chunks, 'embeddings_base.pkl')
embeddings_carregados = carregar_embeddings('embeddings_base.pkl')
print(f"Embeddings carregados: {embeddings_carregados.shape}")

def cosine_similarity(v1, v2):
    """Implementação manual da similaridade por cosseno."""
    dot = sum(a * b for a, b in zip(v1, v2))
    norm1 = sum(a**2 for a in v1) ** 0.5
    norm2 = sum(b**2 for b in v2) ** 0.5
    return dot / (norm1 * norm2) if norm1 and norm2 else 0.0

v1 = [0.5, 0.3, -0.2]
v2 = [0.4, 0.2, -0.1]
print(f"Similaridade: {cosine_similarity(v1, v2):.4f}")

import faiss
import numpy as np

# 1. Cria dados de exemplo
np.random.seed(42)
dados = np.random.random((10000, 128)).astype('float32')
query = np.random.random((1, 128)).astype('float32')

# 2. Índices FAISS

# Força bruta (exata, lenta)
indice_flat = faiss.IndexFlatL2(128)
indice_flat.add(dados)
distancias, indices = indice_flat.search(query, 5)
print("Força Bruta:", indices[0])

# IVF (Inverted File) - divide o espaço em células
quantizer = faiss.IndexFlatL2(128)
indice_ivf = faiss.IndexIVFFlat(quantizer, 128, 100)  # 100 células
indice_ivf.train(dados)
indice_ivf.add(dados)
indice_ivf.nprobe = 10  # Número de células a explorar
distancias, indices = indice_ivf.search(query, 5)
print("IVF:", indices[0])

# HNSW (Hierarchical Navigable Small World) - grafo hierárquico
indice_hnsw = faiss.IndexHNSWFlat(128, 32)  # 32 conexões
indice_hnsw.add(dados)
distancias, indices = indice_hnsw.search(query, 5)
print("HNSW:", indices[0])

# Configurações comuns para HNSW
# efConstruction = 200 (qualidade da construção)
# M = 16-64 (número de conexões)
# efSearch = 100-200 (profundidade da busca)

indice_hnsw_otimizado = faiss.IndexHNSWFlat(128, 64)
indice_hnsw_otimizado.hnsw.efConstruction = 200
indice_hnsw_otimizado.add(dados)
# Para busca, ajustar efSearch
indice_hnsw_otimizado.hnsw.efSearch = 150

# IVF com compressão (PQ - Product Quantization)
# Para economizar memória ainda mais
n_cells = 4096  # Número de células
n_subquantizers = 64  # Subquantizadores para PQ

quantizer = faiss.IndexFlatL2(128)
indice_ivfpq = faiss.IndexIVFPQ(quantizer, 128, n_cells, n_subquantizers, 8)
indice_ivfpq.train(dados)
indice_ivfpq.add(dados)
indice_ivfpq.nprobe = 20

print(f"Memória: {indice_ivfpq.sa_code_size()} bytes por vetor")

import time
import pandas as pd

def benchmark_indices(dados, query, indices, k=10):
    """Compara performance de diferentes índices."""
    resultados = []
    
    for nome, indice in indices:
        # Mede tempo de busca
        start = time.time()
        _, _ = indice.search(query, k)
        tempo = time.time() - start
        
        resultados.append({
            'Índice': nome,
            'Tempo (ms)': tempo * 1000,
            'Memória (MB)': indice.sa_code_size() * indice.ntotal / (1024 * 1024) if hasattr(indice, 'sa_code_size') else None
        })
    
    return pd.DataFrame(resultados)

# Criando índices para benchmark
indices_benchmark = [
    ('Flat (Exato)', faiss.IndexFlatL2(128)),
    ('HNSW (M=32)', faiss.IndexHNSWFlat(128, 32)),
    ('IVF (n_cells=100)', faiss.IndexIVFFlat(faiss.IndexFlatL2(128), 128, 100)),
]

# Configura índices
for nome, idx in indices_benchmark:
    if 'IVF' in nome:
        idx.train(dados)
    idx.add(dados)

# Benchmark
df_resultados = benchmark_indices(dados, query, indices_benchmark)
print(df_resultados)

from pymilvus import Collection, CollectionSchema, FieldSchema, DataType, connections

# 1. Conecta ao Milvus
connections.connect(host='localhost', port='19530')

# 2. Define schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
    FieldSchema(name="texto", dtype=DataType.VARCHAR, max_length=1000)
]
schema = CollectionSchema(fields, "Coleção de embeddings")

# 3. Cria coleção
collection = Collection("documentos_rag", schema)

# 4. Cria índice
index_params = {
    "metric_type": "IP",  # Inner Product (similar ao cosseno)
    "index_type": "HNSW",
    "params": {"M": 16, "efConstruction": 200}
}
collection.create_index("embedding", index_params)

# 5. Insere dados
import random
embeddings_random = np.random.random((100, 384)).tolist()
textos_random = [f"Documento {i}" for i in range(100)]

collection.insert([
    [i for i in range(100)],  # ids
    embeddings_random,        # embeddings
    textos_random             # textos
])

# 6. Busca
search_params = {"metric_type": "IP", "params": {"ef": 64"}}
result = collection.search(
    [query_embedding.tolist()],
    "embedding",
    search_params,
    limit=5,
    output_fields=["texto"]
)

for hits in result:
    for hit in hits:
        print(f"Score: {hit.score:.4f} - Texto: {hit.entity.get('texto')}")

# Reduz precisão para economizar memória
from sklearn.preprocessing import QuantileTransformer

def quantizar_embeddings(embeddings, n_bits=8):
    """Quantiza embeddings para n_bits de precisão."""
    transformer = QuantileTransformer(n_quantiles=2**n_bits)
    quantizados = transformer.fit_transform(embeddings)
    return quantizados.astype(np.uint8)

from functools import lru_cache

@lru_cache(maxsize=1000)
def buscar_query_cache(query_text, k=5):
    """Cache para queries frequentes."""
    embedding = modelo.encode(query_text)
    _, indices = indice.search(embedding.reshape(1, -1), k)
    return indices.tolist()

# Primeira busca (lenta)
resultados = buscar_query_cache("O que é machine learning?")

# Busca seguinte (rápida - do cache)
resultados = buscar_query_cache("O que é machine learning?")

# Pseudocódigo para sharding
class ShardedVectorIndex:
    def __init__(self, num_shards):
        self.shards = [create_index() for _ in range(num_shards)]
        self.num_shards = num_shards
    
    def add(self, vectors, ids):
        for vec, id in zip(vectors, ids):
            shard_id = hash(id) % self.num_shards
            self.shards[shard_id].add(vec)
    
    def search(self, query, k):
        results = []
        for shard in self.shards:
            results.extend(shard.search(query, k))
        return sorted(results, key=lambda x: x.score)[:k]

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    PythonCodeTextSplitter
)

# Para documentos em Markdown
markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
)
chunks_markdown = markdown_splitter.split_text(documento_markdown)

# Para código Python
python_splitter = PythonCodeTextSplitter(chunk_size=200, chunk_overlap=20)
chunks_python = python_splitter.split_text(codigo_python)

# Chunking avançado com análise semântica
from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")

def chunk_semantico(texto, max_tokens=500, overlap=50):
    """Divide texto em chunks baseados em frases e parágrafos."""
    sentences = texto.split('. ')
    chunks = []
    current_chunk = []
    current_length = 0
    
    for sent in sentences:
        sent_tokens = len(tokenizer.tokenize(sent))
        
        if current_length + sent_tokens > max_tokens and current_chunk:
            # Finaliza chunk atual
            chunks.append('. '.join(current_chunk))
            
            # Overlap: mantém últimas frases
            overlap_tokens = 0
            overlap_chunk = []
            for s in reversed(current_chunk):
                s_tokens = len(tokenizer.tokenize(s))
                if overlap_tokens + s_tokens <= overlap:
                    overlap_chunk.insert(0, s)
                    overlap_tokens += s_tokens
                else:
                    break
            current_chunk = overlap_chunk
            current_length = overlap_tokens
        
        current_chunk.append(sent)
        current_length += sent_tokens
    
    if current_chunk:
        chunks.append('. '.join(current_chunk))
    
    return chunks

# Mapeamento de modelo por domínio
MODELOS_POR_DOMINIO = {
    'geral': 'all-MiniLM-L6-v2',
    'alta_precisao': 'all-mpnet-base-v2',
    'multilingue': 'paraphrase-multilingual-MiniLM-L12-v2',
    'medicina': 'microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext',
    'juridico': 'nlpaueb/legal-bert-base-uncased',
    'codigo': 'microsoft/codebert-base',
    'financas': 'ProsusAI/finbert',
}

def escolher_modelo(dominio):
    """Retorna o modelo recomendado para um domínio."""
    return SentenceTransformer(MODELOS_POR_DOMINIO.get(dominio, 'all-MiniLM-L6-v2'))

# Exemplo
modelo_medicina = escolher_modelo('medicina')

import numpy as np

def normalizar_batch(embeddings):
    """Normaliza um batch de embeddings (L2 norm)."""
    norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
    return embeddings / (norms + 1e-8)  # Evita divisão por zero

# Antes da indexação
embeddings_normalizados = normalizar_batch(embeddings_chunks)

# Verifica
print(f"Norma antes: {np.linalg.norm(embeddings_chunks[0]):.4f}")
print(f"Norma depois: {np.linalg.norm(embeddings_normalizados[0]):.4f}")

# Para FAISS com similaridade por cosseno
if 'IndexFlatIP' in str(type(indice)):  # Inner Product
    embeddings_para_index = embeddings_normalizados
else:
    embeddings_para_index = embeddings_chunks

from sklearn.metrics import ndcg_score
import numpy as np

def avaliar_recuperacao_completa(
    queries, 
    documentos_relevantes, 
    funcao_busca, 
    k_values=[1, 3, 5, 10]
):
    """Avaliação abrangente da recuperação."""
    resultados = {}
    
    for k in k_values:
        precision_sum = 0
        recall_sum = 0
        ndcg_sum = 0
        
        for query, relevantes in zip(queries, documentos_relevantes):
            # Recupera top-k
            docs_recuperados = funcao_busca(query, k)
            
            # Precision@K
            relevantes_recuperados = sum(1 for doc in docs_recuperados if doc in relevantes)
            precision = relevantes_recuperados / k
            precision_sum += precision
            
            # Recall@K
            recall = relevantes_recuperados / len(relevantes)
            recall_sum += recall
            
            # NDCG@K
            # Cria relevância binária (1 se relevante, 0 senão)
            relevancia_recuperados = [1 if doc in relevantes else 0 for doc in docs_recuperados]
            if len(relevancia_recuperados) < k:
                relevancia_recuperados += [0] * (k - len(relevancia_recuperados))
            
            # NDCG requer relevância ideal
            relevancia_ideal = [1] * min(len(relevantes), k) + [0] * max(0, k - len(relevantes))
            ndcg = ndcg_score([relevancia_ideal], [relevancia_recuperados])
            ndcg_sum += ndcg
        
        resultados[f'Precision@{k}'] = precision_sum / len(queries)
        resultados[f'Recall@{k}'] = recall_sum / len(queries)
        resultados[f'NDCG@{k}'] = ndcg_sum / len(queries)
    
    return resultados

# Exemplo de uso
def busca_exemplo(query, k):
    # Simula busca (substituir pela real)
    return [f"Doc {i}" for i in range(k)]

resultados_avaliacao = avaliar_recuperacao_completa(
    queries=["O que é RAG?", "Explique embeddings"],
    documentos_relevantes=[["Doc 1", "Doc 2"], ["Doc 0", "Doc 3"]],
    funcao_busca=busca_exemplo
)

for metrica, valor in resultados_avaliacao.items():
    print(f"{metrica}: {valor:.3f}")

class IndexAtualizavel:
    """Índice com suporte a atualizações incrementais."""
    
    def __init__(self, modelo, dimensao):
        self.modelo = modelo
        self.dimensao = dimensao
        self.indice = faiss.IndexFlatIP(dimensao)
        self.documentos = []
        self.embeddings = []
    
    def adicionar_documentos(self, novos_documentos):
        """Adiciona novos documentos ao índice."""
        novos_embeddings = self.modelo.encode(novos_documentos)
        novos_embeddings_normalizados = normalizar_batch(novos_embeddings)
        
        # Adiciona ao índice
        self.indice.add(novos_embeddings_normalizados.astype('float32'))
        
        # Armazena documentos
        self.documentos.extend(novos_documentos)
        self.embeddings.append(novos_embeddings)
        
        print(f"Índice atualizado: {self.indice.ntotal} documentos")
    
    def remover_documentos(self, indices_remover):
        """Remove documentos (reconstroi índice)."""
        # Mantém apenas documentos não removidos
        manter_indices = [i for i in range(len(self.documentos)) if i not in indices_remover]
        
        self.documentos = [self.documentos[i] for i in manter_indices]
        embeddings_manter = [self.embeddings[i] for i in manter_indices]
        
        # Reconstrói índice
        self.indice = faiss.IndexFlatIP(self.dimensao)
        if embeddings_manter:
            embeddings_array = np.vstack(embeddings_manter)
            embeddings_normalizados = normalizar_batch(embeddings_array)
            self.indice.add(embeddings_normalizados.astype('float32'))
    
    def buscar(self, query, k=5):
        """Busca documentos similares."""
        emb_query = self.modelo.encode([query])
        emb_query_normalizado = normalizar_batch(emb_query)
        
        distancias, indices = self.indice.search(
            emb_query_normalizado.astype('float32'), 
            k
        )
        
        return [(self.documentos[idx], dist) for idx, dist in zip(indices[0], distancias[0])]

# Uso
index_atualizavel = IndexAtualizavel(modelo, 384)
index_atualizavel.adicionar_documentos(["Doc1", "Doc2", "Doc3"])
index_atualizavel.adicionar_documentos(["Doc4", "Doc5"])

# Busca
resultados = index_atualizavel.buscar("Pergunta de exemplo", k=3)

import logging
from datetime import datetime
import json

class RAGMonitor:
    """Monitora performance do sistema RAG."""
    
    def __init__(self, log_file='rag_logs.json'):
        self.log_file = log_file
        self.logs = []
    
    def log_query(self, query, num_resultados, tempo_busca):
        """Registra uma consulta."""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'query': query[:100],  # Trunca
            'num_resultados': num_resultados,
            'tempo_busca_ms': tempo_busca * 1000
        }
        self.logs.append(log_entry)
        
        # Salva em arquivo
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(log_entry) + '\n')
    
    def relatorio(self):
        """Gera relatório de estatísticas."""
        if not self.logs:
            return "Sem logs"
        
        tempos = [log['tempo_busca_ms'] for log in self.logs]
        
        return {
            'total_queries': len(self.logs),
            'tempo_medio_ms': sum(tempos) / len(tempos),
            'tempo_max_ms': max(tempos),
            'tempo_min_ms': min(tempos),
            'queries_ultima_hora': sum(
                1 for log in self.logs 
                if (datetime.now() - datetime.fromisoformat(log['timestamp'])).seconds < 3600
            )
        }

# Uso
monitor = RAGMonitor()
monitor.log_query("O que é RAG?", 5, 0.032)
monitor.log_query("Explique embeddings", 3, 0.028)
print(monitor.relatorio())

# rag_pipeline.py

import os
from typing import List, Dict, Any
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
from langchain.text_splitter import RecursiveCharacterTextSplitter

class SistemaRAG:
    def __init__(
        self, 
        modelo_nome: str = 'all-MiniLM-L6-v2',
        chunk_size: int = 500,
        chunk_overlap: int = 50,
        top_k: int = 5
    ):
        self.modelo = SentenceTransformer(modelo_nome)
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.top_k = top_k
        self.indice = None
        self.chunks = []
        self.metadados = []
        
    def preparar_documentos(self, documentos: List[Dict[str, str]]):
        """Prepara e indexa documentos."""
        # Divide em chunks
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap
        )
        
        for doc in documentos:
            chunks = splitter.split_text(doc['texto'])
            for chunk in chunks:
                self.chunks.append(chunk)
                self.metadados.append({
                    'fonte': doc.get('fonte', 'Desconhecido'),
                    'data': doc.get('data', ''),
                    'titulo': doc.get('titulo', '')
                })
        
        # Gera embeddings e indexa
        embeddings = self.modelo.encode(self.chunks)
        embeddings_normalizados = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
        
        self.indice = faiss.IndexFlatIP(embeddings.shape[1])
        self.indice.add(embeddings_normalizados.astype('float32'))
        
        print(f"✓ {len(self.chunks)} chunks indexados")
    
    def consultar(self, query: str) -> Dict[str, Any]:
        """Executa uma consulta e retorna contexto e resposta."""
        # Embedding da query
        emb_query = self.modelo.encode([query])
        emb_query_normalizado = emb_query / np.linalg.norm(emb_query)
        
        # Busca
        scores, indices = self.indice.search(
            emb_query_normalizado.astype('float32'), 
            self.top_k
        )
        
        # Recupera contexto
        contextos = []
        for idx, score in zip(indices[0], scores[0]):
            if idx != -1:
                contextos.append({
                    'texto': self.chunks[idx],
                    'score': float(score),
                    'metadados': self.metadados[idx]
                })
        
        # Prepara para LLM (aqui você integraria com GPT, LLaMA, etc.)
        contexto_texto = "\n\n".join([c['texto'] for c in contextos])
        
        return {
            'query': query,
            'contextos': contextos,
            'contexto_completo': contexto_texto,
            'num_chunks_recuperados': len(contextos)
        }

# Uso
if __name__ == "__main__":
    # Inicializa sistema
    rag = SistemaRAG()
    
    # Carrega documentos
    docs = [
        {'texto': 'Aprendizado de máquina é uma subárea da IA...', 'fonte': 'Wiki'},
        {'texto': 'Redes neurais são inspiradas no cérebro...', 'fonte': 'Livro'}
    ]
    rag.preparar_documentos(docs)
    
    # Consulta
    resultado = rag.consultar("O que é aprendizado de máquina?")
    print(f"Query: {resultado['query']}")
    print(f"Contextos recuperados: {len(resultado['contextos'])}")
    for ctx in resultado['contextos']:
        print(f"- {ctx['texto'][:100]}... (score: {ctx['score']:.3f})")