Spaces:
Running
Running
File size: 3,496 Bytes
aabd32c |
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 |
# Questa parte si occupa di segmentare il testo delle sezioni in blocchi più piccoli (chunks)
# adatti alla ricerca vettoriale, utilizzando RecursiveCharacterTextSplitter (RCTS) di LangChain.
import logging
from typing import List, Dict
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
logger = logging.getLogger(__name__)
class Chunker:
def __init__(self, chunk_size: int = 600, chunk_overlap: int = 100):
# Inizializzazione splitter
# Usa il RCTS per suddividere il testo usando una lista di separatori (newline, doppia newline, spazi, etc)
# preservando la coerenza testuale e semantica
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
separators=["\n\n", "\n", ". ", "! ", "? ", " ", ""],
length_function=len,
)
# Questa funzione divide le sezioni logiche del documento in chunk di dimensione fissa con sovrapposizione
# Tra gli args abbiamo:
# sections: Dizionario con titoli di sezione e relativi testi
# filename: Nome del file PDF originale (usato per i metadati)
# Restituisce una lista di oggetti Document di LangChain, ciascuno contenente un chunk e i suoi metadati
def create_chunks(self, sections: Dict[str, str], filename: str) -> List[Document]:
# Gestione input
if not sections:
logger.warning("Nessuna sezione fornita per il chunking")
return []
all_chunks: List[Document] = []
# Iterazione e Chunking
for section_title, section_text in sections.items():
if not section_text.strip():
logger.debug(f"Salto sezione vuota: '{section_title}'")
continue
try:
# Normalizzo il testo in lowercase
section_text = section_text.lower()
# 1. Divide il testo della sezione (lo splitter accetta una lista di testi)
chunks_for_section = self.text_splitter.create_documents([section_text])
# 2. Aggiunge metadati a ciascun chunk
for i, chunk in enumerate(chunks_for_section):
# Metadato 'source' per il nome del documento
chunk.metadata["source"] = filename
# Metadato 'section' per il titolo logico (per il RAG)
chunk.metadata["section"] = section_title
# ID univoco per il chunk (combinazione di filename, sezione e indice)
# Normalizziamo il titolo della sezione per un ID più pulito e sicuro
clean_section_id = section_title.lower().replace(' ', '_').replace('/', '_').replace(':', '_')
chunk.metadata["chunk_id"] = f"{filename}_{clean_section_id}_{i}"
all_chunks.append(chunk)
logger.debug(f"Sezione '{section_title}' divisa in {len(chunks_for_section)} chunk.")
except Exception as e:
logger.error(f"Errore durante il chunking della sezione '{section_title}': {e}")
logger.info(f"Totale {len(all_chunks)} chunk generati.")
return all_chunks |