File size: 5,966 Bytes
a968971 c1b1880 a968971 c1b1880 a968971 06c5cc4 a968971 cc3f780 a968971 c1b1880 a968971 cc3f780 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 a968971 c1b1880 cc3f780 a968971 c1b1880 a968971 c1b1880 cc3f780 c1b1880 cc3f780 a968971 c1b1880 a968971 c1b1880 a968971 4aa29d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | import re
import numpy as np
import nltk
from sklearn.metrics.pairwise import cosine_similarity
from dotenv import load_dotenv
from langchain_huggingface import HuggingFaceEmbeddings
# Carico l'ambiente. Su HF Spaces andrà a pescare dai secrets, in locale dal .env
load_dotenv()
class ActivaSemanticSplitter:
def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2", batch_size=32):
self.batch_size = batch_size
print("🔄 Inizializzazione HuggingFace Embedding Engine...")
# Scelto MiniLM-L6: per questo prototipo ci serve un modello veloce e leggero in RAM
# che non faccia da collo di bottiglia durante l'ingestion massiva di documenti.
try:
self.embedding_model = HuggingFaceEmbeddings(model_name=model_name)
print("✅ Modello caricato correttamente.")
except Exception as e:
print(f"❌ Errore caricamento modello: {e}")
raise e
# Check preventivo sui tokenizer.
try:
nltk.data.find('tokenizers/punkt')
nltk.data.find('tokenizers/punkt_tab')
except LookupError:
print("⬇️ Download risorse NLTK...")
nltk.download('punkt', quiet=True)
nltk.download('punkt_tab', quiet=True)
def _split_sentences(self, text):
# La pulizia base. Fondamentale per i testi estratti da vecchi OCR o documenti sporchi.
text = text.strip()
try:
# Recupero il tokenizer dell'italiano. Evito sent_tokenize() puro perché è una black box
# e mi serve poter iniettare eccezioni custom per la punteggiatura.
try:
tokenizer = nltk.data.load('tokenizers/punkt/italian.pickle')
except:
# Fallback di sicurezza se il path del pickle salta
from nltk.tokenize.punkt import PunktSentenceTokenizer
tokenizer = PunktSentenceTokenizer()
# --- LISTA ECCEZIONI ABBREVIAZIONI ---
# Evito che il chunker mi spezzi la frase a metà quando incontra "pag." o "art."
# cosa che distruggerebbe il senso semantico prima ancora di passare all'LLM.
custom_abbrevs = ['sec', 's', 'prof', 'dott', 'avv', 'pag', 'fig', 'nr', 'art']
for abbr in custom_abbrevs:
tokenizer._params.abbrev_types.add(abbr)
sentences = tokenizer.tokenize(text)
except ImportError:
print("⚠️ NLTK non installato. Fallback su Regex semplice.")
sentences = re.split(r'(?<=[.?!])\s+', text)
except Exception as e:
print(f"⚠️ Errore NLTK ({e}). Fallback su Regex.")
sentences = re.split(r'(?<=[.?!])\s+', text)
# Filtro via il rumore di fondo (stringhe troppo corte o spazi rimasti appesi)
return [s.strip() for s in sentences if len(s.strip()) > 5]
def combine_sentences(self, sentences, buffer_size=1):
# Sliding window per dare contesto: embeddare una frase singola tipo "Di conseguenza." non ha senso vettoriale.
# Le affianco la frase prima e quella dopo per "spalmare" il significato
# ed evitare che una frase breve sballi il calcolo del coseno.
combined = []
for i in range(len(sentences)):
start = max(0, i - buffer_size)
end = min(len(sentences), i + 1 + buffer_size)
combined_context = " ".join(sentences[start:end])
combined.append(combined_context)
return combined
def calculate_cosine_distances(self, sentences):
# Embeddo tutto in batch. Se arrivano malloppi enormi da estrarre non voglio saturare la memoria.
embeddings = []
total = len(sentences)
for i in range(0, total, self.batch_size):
batch = sentences[i : i + self.batch_size]
batch_embeddings = self.embedding_model.embed_documents(batch)
embeddings.extend(batch_embeddings)
# Calcolo le distanze sequenziali tra la frase N e la frase N+1
distances = []
for i in range(len(embeddings) - 1):
similarity = cosine_similarity([embeddings[i]], [embeddings[i+1]])[0][0]
# Inverto la similarità in distanza (0 = concetti identici, 1 = cambio totale di argomento)
distance = 1.0 - similarity
distances.append(distance)
return distances, embeddings
def create_chunks(self, text, percentile_threshold=95):
single_sentences = self._split_sentences(text)
if not single_sentences:
return [], [], 0
combined_sentences = self.combine_sentences(single_sentences)
distances, _ = self.calculate_cosine_distances(combined_sentences)
if not distances:
# Testo troppo breve per essere splittato, lo tengo intero
return [text], [], 0
# Calcolo la soglia di taglio dinamicamente in base alle variazioni semantiche del documento stesso.
threshold = np.percentile(distances, percentile_threshold)
# Individuo i "punti di rottura" dove l'argomento cambia radicalmente
indices_above_thresh = [i for i, x in enumerate(distances) if x > threshold]
chunks = []
start_index = 0
breakpoints = indices_above_thresh + [len(single_sentences)]
# Ricostruisco i paragrafi unendo le frasi originali (non quelle col buffer)
# delimitandole dai punti di rottura che abbiamo appena trovato.
for i in breakpoints:
end_index = i + 1
chunk_text = " ".join(single_sentences[start_index:end_index])
if len(chunk_text) > 20: # Salto micro-frammenti spazzatura (es. singole parole o punteggiatura)
chunks.append(chunk_text)
start_index = end_index
return chunks, distances, threshold |