Guida RAG: Realizzazione di un Sistema Retrieval-Augmented Generation
1. Introduzione
Questa guida descrive passo-passo come realizzare un sistema di Retrieval-Augmented Generation (RAG) che:
- Estrae e pulisce il testo da un PDF
- Divide il testo in "chunk" per mantenere il contesto
- Genera embeddings tramite Sentence-Transformers
- Indicizza gli embeddings con FAISS (usando un indice HNSW per elevate prestazioni)
- Salva i dati in MongoDB
- Espone API tramite un'app Flask per interrogare il sistema
- Distribuisce l'applicazione in produzione con Gunicorn, systemd e Nginx
2. Estrazione e Pre-elaborazione del Testo dal PDF
2.1 Estrazione del Testo con pdfminer.six
from pdfminer.high_level import extract_text
def extract_text_from_pdf(pdf_path):
try:
text = extract_text(pdf_path)
return text
except Exception as e:
print(f"Errore durante l'estrazione da {pdf_path}: {e}")
return ""
2.2 Pulizia del Testo
import re
def clean_text(text):
lines = text.splitlines()
cleaned_lines = []
for line in lines:
line = line.strip()
if not line:
continue
if "www." in line.lower():
continue
if "copyright" in line.lower():
continue
if line.isdigit():
continue
if re.search(r"^\d+(\s|$)", line):
continue
cleaned_lines.append(line)
return "\n".join(cleaned_lines)
3. Chunking e Generazione degli Embeddings
3.1 Chunking del Testo
from nltk.tokenize import sent_tokenize, word_tokenize
def chunk_text(text, max_words=200, overlap=50):
sentences = sent_tokenize(text)
chunks = []
current_chunk = []
current_count = 0
for sentence in sentences:
sentence_words = word_tokenize(sentence)
sentence_word_count = len(sentence_words)
if current_count + sentence_word_count > max_words:
chunk = " ".join(current_chunk)
chunks.append(chunk)
current_chunk_words = word_tokenize(chunk)
new_chunk = " ".join(current_chunk_words[-overlap:])
current_chunk = [new_chunk] if new_chunk else []
current_count = len(word_tokenize(new_chunk))
current_chunk.append(sentence)
current_count += sentence_word_count
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
3.2 Generazione degli Embeddings con Sentence-Transformers
from sentence_transformers import SentenceTransformer
import numpy as np
def generate_embeddings(chunks):
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
embeddings = model.encode(chunks, convert_to_numpy=True)
return embeddings
4. Costruzione dell'Indice FAISS
import faiss
import numpy as np
def build_faiss_index(embeddings):
embedding_matrix = np.array(embeddings).astype('float32')
dimension = embedding_matrix.shape[1]
M = 32 # Numero di collegamenti per nodo (HNSW)
index = faiss.IndexHNSWFlat(dimension, M)
index.hnsw.efConstruction = 40
index.hnsw.efSearch = 50
index.add(embedding_matrix)
return index
5. Salvataggio degli Embeddings in MongoDB
import pymongo
MONGO_URI = "mongodb+srv://..."
client = pymongo.MongoClient(MONGO_URI)
db = client["gptbuild"]
def process_and_save_pdf(pdf_path, tag):
text = extract_text_from_pdf(pdf_path)
cleaned = clean_text(text)
chunks = chunk_text(cleaned, max_words=200, overlap=50)
embeddings = generate_embeddings(chunks)
collection = db[tag]
for chunk, embedding in zip(chunks, embeddings):
document = {
"chunk": chunk,
"embedding": embedding.tolist()
}
collection.insert_one(document)
return {"status": "success", "num_chunks": len(chunks)}
6. Creazione delle API con Flask
Il file serverTransformer.py contiene l'app Flask che espone gli endpoint:
/be/api/upload: Caricamento del PDF e salvataggio degli embeddings/be/api/list_embeddings: Elenco delle collection disponibili/be/api/chat: Interrogazione della collection per ottenere il contesto
@app.route('/be/api/chat', methods=['POST'])
def chat():
data = request.get_json()
query = data.get("query", "")
collection_name = data.get("collection", "")
retrieved_chunks = search_collection(collection_name, query, k=3)
context = "\n".join(retrieved_chunks)
prompt = f"Contesto:\n{context}\n\nDomanda: {query}\nRisposta:"
# Chiamata all'LLM locale
payload = {"prompt": prompt, "model": "deepseek-r1:32b"}
llm_response = requests.post(llm_api_url, json=payload)
return jsonify({"response": llm_response})
7. Avvio in Produzione
7.1 Gunicorn
gunicorn --workers 3 --timeout 620 --bind 0.0.0.0:5000 serverTransformer:app
7.2 Servizio systemd
[Unit]
Description=Gunicorn instance to serve Flask API (RAG)
After=network.target
[Service]
User=domenicosimone
WorkingDirectory=/home/domenicosimone/dist
ExecStart=/opt/venv/llm/bin/gunicorn --workers 3 --timeout 620 --bind 0.0.0.0:5000 serverTransformer:app
Restart=always
[Install]
WantedBy=multi-user.target
7.3 Configurazione Nginx
server {
listen 443 ssl;
server_name llm.domenico-simonemarsella.org;
location /be/api/ {
proxy_pass http://127.0.0.1:5000/be/api/;
proxy_set_header Host $host;
proxy_read_timeout 600s;
}
}
7.4 Riepilogo
Con questa guida hai realizzato un sistema RAG completo che:
- Estrae e pulisce il testo da PDF
- Divide il testo in chunk e genera embeddings
- Indicizza i vettori con FAISS
- Salva i dati in MongoDB
- Espone API Flask per caricare e interrogare i dati
- Distribuisce l'app con Gunicorn e Nginx