Upload app.py

app.py

@@ -3,23 +3,24 @@
 #  Backend com RAG Híbrido, CSV, Tavily, NER Entity Graph, FAISS ANN
 # ==============================================================================
 
-import numpy as np
-import pandas as pd
-import torch
-import gc
-import uuid
-import os
-import io
-import json
-import logging
-import time
-import nltk
+import numpy as np
+import pandas as pd
+import torch
+import gc
+import uuid
+import os
+import io
+import json
+import logging
+import time
+import heapq
+import nltk
 from nltk.corpus import stopwords
 from collections import defaultdict
 
 from fastapi import FastAPI, UploadFile, File, Form, HTTPException
 from fastapi.responses import JSONResponse
-from typing import List, Dict, Any, Tuple
+from typing import List, Dict, Any, Tuple, Optional
 from functools import lru_cache
 
 # Ferramentas de Alquimia (ML & NLP)
@@ -115,9 +116,19 @@ logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(
 RETRIEVAL_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"  # Leve e bom PT-BR
 RERANKER_MODEL = "cross-encoder/mmarco-mMiniLMv2-L12-H384-v1"  # Reranker nativo PT-BR
 
-# Parâmetros de Processamento
-BATCH_SIZE = 256
-UMAP_N_NEIGHBORS = 30
+# Parâmetros de Processamento
+BATCH_SIZE = 256
+UMAP_N_NEIGHBORS = 30
+
+# Parâmetros FAISS (HNSW) e Duplicados
+FAISS_HNSW_MIN_SIZE = int(os.environ.get("FAISS_HNSW_MIN_SIZE", 2000))
+FAISS_HNSW_M = int(os.environ.get("FAISS_HNSW_M", 32))
+FAISS_HNSW_EF_CONSTRUCTION = int(os.environ.get("FAISS_HNSW_EF_CONSTRUCTION", 80))
+FAISS_HNSW_EF_SEARCH = int(os.environ.get("FAISS_HNSW_EF_SEARCH", 64))
+
+DUPLICATE_KNN_K = int(os.environ.get("DUPLICATE_KNN_K", 10))
+DUPLICATE_SIM_THRESHOLD = float(os.environ.get("DUPLICATE_SIM_THRESHOLD", 0.98))
+DUPLICATE_TOP_K = int(os.environ.get("DUPLICATE_TOP_K", 5))
 
 # Cache de Sessão (Na memória RAM)
 cache: Dict[str, Any] = {}
@@ -687,9 +698,9 @@ def processar_pipeline(
     del reducer, clusterer, emb_3d; gc.collect()
     return df, embeddings
 
-def calcular_metricas(textos: List[str]) -> Dict[str, Any]:
-    logging.info("Calculando métricas globais...")
-    if not textos: return {}
+def calcular_metricas(textos: List[str]) -> Dict[str, Any]:
+    logging.info("Calculando métricas globais...")
+    if not textos: return {}
     
     # Token pattern: só palavras alfabéticas com 3+ caracteres (ignora números)
     token_pattern = r'\b[a-zA-ZÀ-ÿ]{3,}\b'
@@ -719,33 +730,73 @@ def calcular_metricas(textos: List[str]) -> Dict[str, Any]:
     top_idx_tfidf = np.argsort(soma_tfidf)[-10:][::-1]
     top_tfidf = [{"palavra": vocab_tfidf[i], "score": round(float(soma_tfidf[i]), 4)} for i in top_idx_tfidf]
     
-    return {
-        "riqueza_lexical": len(vocab_count), 
-        "top_tfidf_palavras": top_tfidf, 
-        "entropia": float(entropy(contagens / contagens.sum(), base=2)) if contagens.sum() > 0 else 0.0
-    }
-
-def encontrar_duplicados(df: pd.DataFrame, embeddings: np.ndarray) -> Dict[str, Any]:
-    logging.info("Detectando duplicados...")
-    mask = df["full_text"].duplicated(keep=False)
-    grupos_exatos = {t: [int(i) for i in idxs] for t, idxs in df[mask].groupby("full_text").groups.items()}
-    pares_semanticos = []
-    
-    if 2 < len(embeddings) < 5000:
-        sim = cosine_similarity(embeddings)
-        triu_indices = np.triu_indices_from(sim, k=1)
-        sim_vetor = sim[triu_indices]
-        pares_idx = np.where(sim_vetor > 0.98)[0]
-        top_pares_idx = pares_idx[np.argsort(sim_vetor[pares_idx])[-5:][::-1]]
-        for i in top_pares_idx:
-            idx1, idx2 = triu_indices[0][i], triu_indices[1][i]
-            if df["full_text"].iloc[idx1] != df["full_text"].iloc[idx2]:
-                pares_semanticos.append({
-                    "similaridade": float(sim[idx1, idx2]), 
-                    "texto1": df["full_text"].iloc[idx1], 
-                    "texto2": df["full_text"].iloc[idx2]
-                })
-    return {"grupos_exatos": grupos_exatos, "pares_semanticos": pares_semanticos}
+    return {
+        "riqueza_lexical": len(vocab_count), 
+        "top_tfidf_palavras": top_tfidf, 
+        "entropia": float(entropy(contagens / contagens.sum(), base=2)) if contagens.sum() > 0 else 0.0
+    }
+
+def normalize_embeddings(embeddings: np.ndarray) -> np.ndarray:
+    norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
+    norms[norms == 0] = 1.0
+    return embeddings / norms
+
+def build_faiss_index(embeddings_normalized: np.ndarray) -> faiss.Index:
+    dim = embeddings_normalized.shape[1]
+    if embeddings_normalized.shape[0] >= FAISS_HNSW_MIN_SIZE:
+        index = faiss.IndexHNSWFlat(dim, FAISS_HNSW_M, faiss.METRIC_INNER_PRODUCT)
+        index.hnsw.efConstruction = FAISS_HNSW_EF_CONSTRUCTION
+        index.hnsw.efSearch = FAISS_HNSW_EF_SEARCH
+        index_type = "HNSW"
+    else:
+        index = faiss.IndexFlatIP(dim)
+        index_type = "FlatIP"
+    index.add(embeddings_normalized.astype('float32'))
+    logging.info(f"FAISS index criado ({index_type}) com {index.ntotal} vetores")
+    return index
+
+def encontrar_duplicados(
+    df: pd.DataFrame,
+    embeddings_normalized: np.ndarray,
+    faiss_index: Optional[faiss.Index] = None
+) -> Dict[str, Any]:
+    logging.info("Detectando duplicados...")
+    mask = df["full_text"].duplicated(keep=False)
+    grupos_exatos = {t: [int(i) for i in idxs] for t, idxs in df[mask].groupby("full_text").groups.items()}
+    pares_semanticos = []
+    
+    if len(embeddings_normalized) > 2:
+        if faiss_index is None:
+            faiss_index = build_faiss_index(embeddings_normalized)
+        if hasattr(faiss_index, "hnsw"):
+            faiss_index.hnsw.efSearch = max(faiss_index.hnsw.efSearch, FAISS_HNSW_EF_SEARCH)
+
+        k = min(DUPLICATE_KNN_K + 1, len(embeddings_normalized))
+        scores, indices = faiss_index.search(embeddings_normalized.astype('float32'), k)
+        top_pairs = []
+
+        for i in range(len(embeddings_normalized)):
+            for rank, j in enumerate(indices[i]):
+                if j <= i:
+                    continue
+                score = float(scores[i][rank])
+                if score < DUPLICATE_SIM_THRESHOLD:
+                    continue
+                if df["full_text"].iloc[i] == df["full_text"].iloc[j]:
+                    continue
+
+                if len(top_pairs) < DUPLICATE_TOP_K:
+                    heapq.heappush(top_pairs, (score, i, j))
+                elif score > top_pairs[0][0]:
+                    heapq.heapreplace(top_pairs, (score, i, j))
+
+        for score, idx1, idx2 in sorted(top_pairs, reverse=True):
+            pares_semanticos.append({
+                "similaridade": float(score),
+                "texto1": df["full_text"].iloc[idx1],
+                "texto2": df["full_text"].iloc[idx2]
+            })
+    return {"grupos_exatos": grupos_exatos, "pares_semanticos": pares_semanticos}
 
 def analisar_clusters(df: pd.DataFrame) -> Dict[str, Any]:
     logging.info("Analisando clusters...")
@@ -832,10 +883,9 @@ async def process_api(
             custom_min_samples=min_samples_int
         )
         
-        # Criar índice FAISS para busca rápida (semântica)
-        embeddings_normalized = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
-        faiss_index = faiss.IndexFlatIP(embeddings_normalized.shape[1])  # Inner Product = Cosine sim para vetores normalizados
-        faiss_index.add(embeddings_normalized.astype('float32'))
+        # Criar índice FAISS para busca rápida (semântica)
+        embeddings_normalized = normalize_embeddings(embeddings)
+        faiss_index = build_faiss_index(embeddings_normalized)
         
         # Criar índice BM25 para busca lexical (Hybrid Search)
         corpus_texts = df["full_text"].tolist()
@@ -853,7 +903,7 @@ async def process_api(
         logging.info(f"Job criado: {job_id} (FAISS + BM25 hybrid search)")
 
         metricas_globais = calcular_metricas(df["full_text"].tolist())
-        analise_de_duplicados = encontrar_duplicados(df, embeddings)
+        analise_de_duplicados = encontrar_duplicados(df, embeddings_normalized, faiss_index)
         analise_por_cluster_tfidf = analisar_clusters(df)
 
         n_clusters = len(df["cluster"].unique()) - (1 if "-1" in df["cluster"].unique() else 0)
@@ -1389,4 +1439,4 @@ Use os dados concretos fornecidos, não generalize. Seja específico citando ent
         
     except Exception as e:
         logging.error(f"ERRO EM /analyze_graph/: {e}", exc_info=True)
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))