Esempio in python di RAG per lettura di PDF
Dopo esserci immersi nello studio di alcuni progetti pionieristici che impiegano la tecnologia di Retrieval Augmented Generation, averne esaminato i principi fondamentali e il framework open-source LangChain, proseguiamo nel nostro viaggio alla scoperta di ulteriori iniziative legate a questo ambito: Esempio in python di RAG per lettura di PDF.
In NetAI, ci siamo spesso trovati nella necessità di reperire definizioni specifiche su vari argomenti. Ci ritrovavamo costantemente a consultare numerosi siti web alla ricerca delle informazioni desiderate, ignorando spesso che la risposta poteva trovarsi semplicemente all’interno di un libro scaricato mesi prima e mai sfogliato in modo approfondito.
Grazie alle nostre competenze acquisite attraverso l’uso di LangChain, abbiamo quindi optato per creare un nostro database vettoriale locale contenente tutti i libri in formato PDF che potrebbero rivelarsi utili per il nostro lavoro.
Questo ci permette di rivolgere al nostro Bot-PDF qualsiasi domanda riguardante terminologie specifiche, ricevendo in risposta le informazioni accurate e precise che necessitiamo.
Tecnologie Utlilizzate
ChromaDB: un database vettoriale che si basa sulle rappresentazioni vettoriali dei documenti testuali. Questo permette di immagazzinare e interrogare grandi quantità di testo in modo efficiente, consentendo di effettuare ricerche avanzate e ottenere risposte pertinenti alle domande degli utenti.
LangChain: un framework open source per la creazione di applicazioni basate su modelli linguistici di grandi dimensioni (LLM). Gli LLM sono grandi modelli di deep learning pre-addestrati su grandi quantità di dati in grado di generare risposte alle query degli utenti, ad esempio rispondere a domande o creare immagini da prompt basati su testo.
Spiegazione dell’ Esempio in python di RAG per lettura di PDF
Lo script si propone di creare un database vettoriale utilizzando ChromaDB, popolandolo con le informazioni estratte da una serie di documenti PDF presenti in una specifica cartella e nelle sue sottocartelle. Vediamo i passaggi principali:
Lettura dei PDF: Utilizzando la libreria PyPDFLoader, vengono caricati i documenti PDF presenti nella cartella specificata.
Divisione dei Documenti e Preparazione: Per superare i limiti di elaborazione degli LLM, i documenti selezionati vengono suddivisi in segmenti minori o “porzioni” utilizzando CharacterTextSplitter, rendendo più agevole l’elaborazione.
Vettorizzazione delle Porzioni di Documento: Le porzioni di documento vengono trasformate in rappresentazioni numeriche attraverso l’uso di modelli di embedding.
Generazione degli Hash: Viene generato un hash MD5 per il contenuto di ciascun PDF, al fine di identificarlo univocamente.
Controllo dell’Unicità: Gli hash generati vengono salvati in un file di testo per verificare l’unicità dei documenti nel database. Questo garantisce che i documenti non vengano duplicati all’interno del database.
Creazione del Database Chroma: Utilizzando ChromaDB, i documenti vengono memorizzati nel database vettoriale, consentendo di effettuare ricerche avanzate basate sul contenuto testuale.
Interrogazione del Database: Attraverso il metodo similarity_search() fornito da LangChain, il database può essere interrogato con una specifica domanda. Il sistema restituirà quindi i documenti più pertinenti in base al contenuto della domanda.
Generazione della Risposta: Infine, l’LLM elabora il prompt arricchito con le informazioni recuperate e genera una risposta. Grazie al contesto aggiuntivo fornito dalle porzioni di documento pertinenti, la risposta del modello sarà più informata, accurata e contestualmente rilevante.
Il codice che abbiamo usato
import os
from dotenv import load_dotenv
load_dotenv()
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
import hashlib
# Path delle cartelle di destinazione
pdf_folder_path = "<your-pdf-folder>"
database_directory = "<your-db-folder>"
existing_hashes = set()
embedding_function = OpenAIEmbeddings()
def generate_pdf_hash(pdf_path):
"""Genera un hash MD5 per il contenuto di un PDF."""
hasher = hashlib.md5()
with open(pdf_path, "rb") as f:
buf = f.read()
hasher.update(buf)
return hasher.hexdigest()
# salvare in un file txt un elenco di stringhe passate come variabile 'hash'
def salva_hash_in_file(hash, database_directory):
"""
Salva una stringa hash in un file di testo se non è già presente.
:param hash: La stringa hash da salvare.
:param percorso_file: Percorso del file di testo in cui salvare l'hash.
:return: True se l'hash è stato salvato, False se l'hash è già presente nel file.
"""
try:
percorso_file = os.path.join(database_directory, "tabella_pdfhash.txt")
# Legge gli hash esistenti nel file, se il file esiste.
try:
with open(percorso_file, "r", encoding="utf-8") as file:
hash_esistenti = set(file.read().splitlines())
except FileNotFoundError:
hash_esistenti = set()
# Controlla se l'hash è già presente e lo salva se non presente.
if hash not in hash_esistenti:
with open(percorso_file, "a", encoding="utf-8") as file:
file.write(hash + "\n")
return True
else:
print(f"L'hash {hash} è già presente nel file.")
return False
except Exception as e:
print(f"Si è verificato un errore durante il salvataggio dell'hash: {e}")
return False
documents_all = []
# Usa os.walk() per iterare su tutte le cartelle e sottocartelle
for root, dirs, filenames in os.walk(pdf_folder_path):
for filename in filenames:
if filename.endswith(".pdf"):
pdf_path = os.path.join(root, filename)
pdf_hash = generate_pdf_hash(pdf_path)
# Controlla se il PDF è già nel database usando l'hash
if not salva_hash_in_file(pdf_hash, database_directory):
print(f"Il PDF {filename} è già presente nell'elenco hash.")
continue
else:
# Salva il pdf sul db vettoriale
# Carica e prepara il documento
loader = PyPDFLoader(pdf_path)
documents = loader.load()
text_splitter = CharacterTextSplitter(
chunk_size=1500, chunk_overlap=500
)
chunked_documents = text_splitter.split_documents(documents)
# Crea semplici ID per i documenti
ids = [
pdf_hash + "_" + str(i)
for i in range(1, len(chunked_documents) + 1)
]
print(ids)
vectordb = Chroma.from_documents(
documents=chunked_documents,
embedding=embedding_function,
persist_directory=database_directory,
)
vectordb = Chroma(
persist_directory="<your-db-folder>",
embedding_function=embedding_function,
)
# Definisci la tua query
query = "Explain how to Create subsystem bulkheads?"
# Usa il metodo similarity_search() per fare la query
docs = vectordb.similarity_search(query)
print(docs)Implementazione tecnica
Queste righe importano le librerie necessarie per eseguire lo script. os viene utilizzato per operazioni di sistema, dotenv per caricare variabili d’ambiente, e le librerie di LangChain e Chroma per l’elaborazione del testo e la gestione del database vettoriale.
import os
from dotenv import load_dotenv
load_dotenv()
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
import hashlibQueste variabili definiscono il percorso della cartella contenente i PDF e la directory in cui verrà memorizzato il database. existing_hashes è un set che conterrà gli hash dei PDF già presenti nel database. embedding_function è un’istanza dell’oggetto OpenAIEmbeddings utilizzato per ottenere rappresentazioni vettoriali del testo.
# Path delle cartelle di destinazione
pdf_folder_path = "<your-pdf-folder>"
database_directory = "<your-db-folder>"
existing_hashes = set()
embedding_function = OpenAIEmbeddings()La funzione generate_pdf_hash(pdf_path) riceve il percorso di un file PDF e genera un hash MD5 basato sul suo contenuto. Questo hash verrà utilizzato per identificare univocamente ciascun PDF.
def generate_pdf_hash(pdf_path):
"""Genera un hash MD5 per il contenuto di un PDF."""
hasher = hashlib.md5()
with open(pdf_path, "rb") as f:
buf = f.read()
hasher.update(buf)
return hasher.hexdigest()
La funzione salva_hash_in_file(hash, database_directory) prende in input un hash e lo salva in un file di testo se non è già presente. Questo è utile per mantenere traccia degli hash dei PDF già presenti nel database.
# salvare in un file txt un elenco di stringhe passate come variabile 'hash'
def salva_hash_in_file(hash, database_directory):
"""
Salva una stringa hash in un file di testo se non è già presente.
:param hash: La stringa hash da salvare.
:param percorso_file: Percorso del file di testo in cui salvare l'hash.
:return: True se l'hash è stato salvato, False se l'hash è già presente nel file.
"""
try:
percorso_file = os.path.join(database_directory, "tabella_pdfhash.txt")
# Legge gli hash esistenti nel file, se il file esiste.
try:
with open(percorso_file, "r", encoding="utf-8") as file:
hash_esistenti = set(file.read().splitlines())
except FileNotFoundError:
hash_esistenti = set()
# Controlla se l'hash è già presente e lo salva se non presente.
if hash not in hash_esistenti:
with open(percorso_file, "a", encoding="utf-8") as file:
file.write(hash + "\n")
return True
else:
print(f"L'hash {hash} è già presente nel file.")
return False
except Exception as e:
print(f"Si è verificato un errore durante il salvataggio dell'hash: {e}")
return FalseIl seguente blocco di codice itera attraverso tutti i file PDF nella cartella specificata. Per ciascun file, viene generato un hash e verificato se è già presente nel database. Se non lo è, il PDF viene elaborato, suddiviso in segmenti più piccoli e memorizzato nel database vettoriale ChromaDB.
documents_all = []
# Usa os.walk() per iterare su tutte le cartelle e sottocartelle
for root, dirs, filenames in os.walk(pdf_folder_path):
for filename in filenames:
if filename.endswith(".pdf"):
pdf_path = os.path.join(root, filename)
pdf_hash = generate_pdf_hash(pdf_path)
# Controlla se il PDF è già nel database usando l'hash
if not salva_hash_in_file(pdf_hash, database_directory):
print(f"Il PDF {filename} è già presente nell'elenco hash.")
continue
else:
# Salva il pdf sul db vettoriale
# Carica e prepara il documento
loader = PyPDFLoader(pdf_path)
documents = loader.load()
text_splitter = CharacterTextSplitter(
chunk_size=1500, chunk_overlap=500
)
chunked_documents = text_splitter.split_documents(documents)
# Crea semplici ID per i documenti
ids = [
pdf_hash + "_" + str(i)
for i in range(1, len(chunked_documents) + 1)
]
print(ids)
vectordb = Chroma.from_documents(
documents=chunked_documents,
embedding=embedding_function,
persist_directory=database_directory,
)
Infine, viene creato un’istanza di ChromaDB con i pdf agguinti pronto per essere interrogato da una query specifica definita dalla variabile query. Viene quindi utilizzato il metodo similarity_search() di LangChain per trovare i documenti più pertinenti rispetto alla domanda e stamparli a schermo.
vectordb = Chroma(
persist_directory="<your-db-folder>",
embedding_function=embedding_function,
)
# Definisci la tua query
query = "Explain how to Create subsystem bulkheads?"
# Usa il metodo similarity_search() per fare la query
docs = vectordb.similarity_search(query)
print(docs)
chromadb database vettoriale esempio codice langchain pdf python
