Spaces:
Running
Running
| # 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 |