363 lines
13 KiB
Python
363 lines
13 KiB
Python
# --- Import librerie ---
|
||
import pandas as pd
|
||
from openai import AzureOpenAI
|
||
import pickle
|
||
from sentence_transformers import SentenceTransformer
|
||
import numpy as np
|
||
import faiss
|
||
import openpyxl
|
||
import re
|
||
import json
|
||
from openpyxl.styles import PatternFill
|
||
from openpyxl import load_workbook
|
||
from collections import Counter
|
||
from prompts.prompt import build_prompt_local
|
||
import warnings
|
||
import logging
|
||
|
||
|
||
# --- Configurazione ---
|
||
endpoint = "https://gpt-sw-central-tap-security.openai.azure.com/"
|
||
deployment = "gpt-4o"
|
||
subscription_key = "8zufUIPs0Dijh0M6NpifkkDvxJHZMFtott7u8V8ySTYNcpYVoRbsJQQJ99BBACfhMk5XJ3w3AAABACOGr6sq"
|
||
|
||
client = AzureOpenAI(
|
||
azure_endpoint=endpoint,
|
||
api_key=subscription_key,
|
||
api_version="2024-05-01-preview",
|
||
)
|
||
|
||
# ----- Step 1: caricare datasets -----
|
||
df_labeled = pd.read_csv("main/datasets/annotated_dataset.csv", encoding="cp1252", sep=';') # colonne: automation, description, category, subcategory, problem_type, gravity
|
||
df_unlabeled = pd.read_csv("main/datasets/unlabeled_dataset.csv", sep='\t', encoding='utf-8')
|
||
print("***STEP 1***\nDataset etichettato caricato. Numero righe:", len(df_labeled), "\nDataset non etichettato caricato. Numero righe:", len(df_unlabeled))
|
||
|
||
def clean_id(x):
|
||
if pd.isna(x):
|
||
return ""
|
||
s = str(x)
|
||
m = re.search(r"\d+", s) # prima sequenza di cifre
|
||
return m.group(0) if m else s.strip()
|
||
|
||
df_labeled["automation_id"] = df_labeled["automation_id"].apply(clean_id)
|
||
df_unlabeled["automation_id"] = df_unlabeled["automation_id"].apply(clean_id)
|
||
df_labeled["folder"] = df_labeled["folder"].astype(str).str.strip()
|
||
df_unlabeled["folder"] = df_unlabeled["folder"].astype(str).str.strip()
|
||
|
||
labeled_pairs = set(zip(df_labeled["automation_id"], df_labeled["folder"])) # Crea set di coppie già etichettate
|
||
|
||
df_unlabeled_filtered = df_unlabeled[
|
||
~df_unlabeled.apply(lambda row: (row["automation_id"], row["folder"]) in labeled_pairs, axis=1) # Filtra il dataset non etichettato
|
||
]
|
||
print("Automazioni non etichettate rimanenti dopo la pulizia:", len(df_unlabeled_filtered))
|
||
|
||
|
||
# --- Step 2: embeddings ---
|
||
# Silenzia warning generici
|
||
warnings.filterwarnings("ignore")
|
||
# Silenzia logging di transformers / sentence-transformers / HF hub
|
||
logging.getLogger("sentence_transformers").setLevel(logging.ERROR)
|
||
logging.getLogger("transformers").setLevel(logging.ERROR)
|
||
logging.getLogger("huggingface_hub").setLevel(logging.ERROR)
|
||
|
||
print("\n***Step 2 ***\nEmbeddings")
|
||
model = SentenceTransformer("all-MiniLM-L6-v2")
|
||
|
||
with open("main/labeled_embeddings.pkl", "rb") as f:
|
||
data = pickle.load(f)
|
||
|
||
embeddings = data['embeddings'].astype("float32")
|
||
print("Shape embeddings:", embeddings.shape)
|
||
|
||
|
||
# ----- Step3: Creazione indice FAISS ---
|
||
dimension = embeddings.shape[1]
|
||
index = faiss.IndexFlatL2(dimension) # indice L2 (distanza Euclidea)
|
||
index.add(embeddings)
|
||
print(f"\n***Step 3: Indice FAISS creato***. \nNumero di vettori nell'indice: {index.ntotal}")
|
||
|
||
|
||
# ----- Step4: Retrieval (similarità) ---
|
||
# Prova con le prime 500 automazioni non annotate
|
||
k = 5
|
||
output_rows = []
|
||
df_sample = df_unlabeled_filtered.head(500)
|
||
llm_rows = []
|
||
|
||
def sim_label(distance: float) -> str:
|
||
if distance <= 0.50:
|
||
return "Match forte"
|
||
elif distance <= 0.75:
|
||
return "Match plausibile"
|
||
elif distance <= 0.90:
|
||
return "Similarità instabile"
|
||
else:
|
||
return "Troppo distante"
|
||
|
||
for i, row in df_sample.iterrows():
|
||
query_text = str(row["human_like"])
|
||
print("numero corrente:", i)
|
||
|
||
# Calcolo embedding della nuova automazione
|
||
query_emb = model.encode([query_text], convert_to_numpy=True).astype("float32")
|
||
# Recupera indici dei k vicini più prossimi
|
||
distances, indices = index.search(query_emb, k)
|
||
|
||
# Metriche globali sui top-k (una volta per automazione)
|
||
topk_cats = []
|
||
top1_distance = float(distances[0][0])
|
||
top1_confidence = 1 / (1 + top1_distance)
|
||
top1_similarity_label = sim_label(top1_distance)
|
||
|
||
for rank in range(k):
|
||
idx = int(indices[0][rank])
|
||
distance = float(distances[0][rank])
|
||
confidence = 1 / (1 + distance)
|
||
label = sim_label(distance)
|
||
|
||
retrieved_row = df_labeled.iloc[idx]
|
||
topk_cats.append(str(retrieved_row["category"]))
|
||
|
||
rank1_category = topk_cats[0] if topk_cats else ""
|
||
majority_category = Counter(topk_cats).most_common(1)[0][0] if topk_cats else ""
|
||
consistency = (sum(c == majority_category for c in topk_cats) / len(topk_cats)) if topk_cats else 0.0
|
||
|
||
for rank in range(k):
|
||
idx = int(indices[0][rank])
|
||
distance = float(distances[0][rank])
|
||
confidence = 1 / (1 + distance)
|
||
label = sim_label(distance)
|
||
|
||
retrieved_row = df_labeled.iloc[idx]
|
||
|
||
output_rows.append({
|
||
# query
|
||
"automazione da etichettare": query_text,
|
||
|
||
# info retrieval per questa riga
|
||
"rank": rank + 1,
|
||
"retrieved_idx": idx,
|
||
"automazione simile": retrieved_row["automation"],
|
||
"categoria automazione simile": retrieved_row["category"],
|
||
"distanza": distance,
|
||
"confidence": round(confidence, 4),
|
||
"similarity": label,
|
||
|
||
# metriche aggregate top-k (ripetute su ogni riga)
|
||
"rank1_distance": top1_distance,
|
||
"rank1_confidence": round(top1_confidence, 4),
|
||
"rank1_similarity_label": top1_similarity_label,
|
||
"rank1_category": rank1_category,
|
||
"majority_category": majority_category,
|
||
"consistency": round(consistency, 3),
|
||
"top5_categories": " | ".join(topk_cats)
|
||
})
|
||
|
||
|
||
# --- Step5: invio dati al LLM ---
|
||
# (1) Costruzione prompt
|
||
retrieved = df_labeled.iloc[indices[0]].copy()
|
||
retrieved["distance"] = distances[0].astype(float)
|
||
retrieved["confidence"] = retrieved["distance"].apply(lambda d: 1 / (1 + float(d)))
|
||
retrieved["similarity"] = retrieved["distance"].apply(sim_label)
|
||
prompt = build_prompt_local(query_text, retrieved, sim_label)
|
||
|
||
# (2) Chiamata al modello: restituisce JSON
|
||
resp = client.chat.completions.create(
|
||
model=deployment,
|
||
messages=[
|
||
{"role": "system", "content": "Return ONLY valid JSON. No extra text."},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
temperature=0
|
||
)
|
||
content = resp.choices[0].message.content.strip()
|
||
|
||
# (3) Parsing della risposta
|
||
try:
|
||
parsed = json.loads(content)
|
||
except Exception:
|
||
parsed = {
|
||
"automation": query_text,
|
||
"category": "",
|
||
"subcategory": "",
|
||
"problem_type": "",
|
||
"gravity": "",
|
||
"scores": {},
|
||
"needs_human_review": True,
|
||
"short_rationale": f"JSON_PARSE_ERROR: {content[:200]}"
|
||
}
|
||
|
||
# (4) Salvataggio di 1 riga per automazione con:
|
||
# - metriche retrieval (rank1/majority/consistency)
|
||
# - output dell'LLM (scores + label finale + review flag)
|
||
llm_category = parsed.get("category", "")
|
||
llm_subcategory = parsed.get("subcategory", "")
|
||
llm_problem_type = parsed.get("problem_type", "")
|
||
llm_gravity = parsed.get("gravity", "")
|
||
final_category = llm_category
|
||
final_subcategory = llm_subcategory
|
||
final_problem_type = llm_problem_type
|
||
final_gravity = llm_gravity
|
||
if llm_category.strip().upper() == "HARMLESS":
|
||
llm_subcategory = ""
|
||
llm_problem_type = "NONE"
|
||
llm_gravity = "NONE"
|
||
|
||
# ================= HUMAN REVIEW LOGIC =================
|
||
needs_human_review = bool(parsed.get("needs_human_review", True))
|
||
OVERRIDE_MAX_DISTANCE = 0.90
|
||
OVERRIDE_MIN_CONSISTENCY = 0.60
|
||
# Allineamento forte: LLM = majority = top1
|
||
aligned_strong = (
|
||
llm_category == majority_category and
|
||
llm_category == rank1_category and
|
||
llm_category != ""
|
||
)
|
||
# distanza non eccessiva e buona consistency
|
||
good_retrieval = (
|
||
top1_distance <= OVERRIDE_MAX_DISTANCE and
|
||
consistency >= OVERRIDE_MIN_CONSISTENCY
|
||
)
|
||
# allora NON richiede revisione anche se il modello aveva messo True
|
||
if aligned_strong and good_retrieval:
|
||
needs_human_review = False
|
||
# =====================================================
|
||
|
||
|
||
llm_rows.append({
|
||
"automation_id": row.get("automation_id", ""),
|
||
"folder": row.get("folder", ""),
|
||
"automation_text": query_text,
|
||
|
||
"rank1_distance": top1_distance,
|
||
"rank1_confidence": round(top1_confidence, 4),
|
||
"rank1_similarity_label": top1_similarity_label,
|
||
"rank1_category": rank1_category,
|
||
"majority_category": majority_category,
|
||
"consistency": round(consistency, 3),
|
||
"top5_categories": " | ".join(topk_cats),
|
||
|
||
"llm_category": llm_category,
|
||
"llm_subcategory": llm_subcategory,
|
||
"llm_problem_type": llm_problem_type,
|
||
"llm_gravity": llm_gravity,
|
||
"llm_needs_human_review": parsed.get("needs_human_review", True),
|
||
"final_needs_human_review": needs_human_review,
|
||
|
||
"final_category": final_category,
|
||
"final_subcategory": final_subcategory,
|
||
"final_problem_type": final_problem_type,
|
||
"final_gravity": final_gravity,
|
||
|
||
"llm_rationale": parsed.get("short_rationale", "")
|
||
})
|
||
|
||
|
||
# --- Step6: integrazione e output ---
|
||
# (5) Esportare l’output finale come dataframe
|
||
df_llm = pd.DataFrame(llm_rows)
|
||
out_path = "main/datasets/labeling_first500.xlsx"
|
||
df_llm.to_excel(out_path, index=False)
|
||
|
||
wb = load_workbook(out_path)
|
||
ws = wb.active
|
||
|
||
# Colori per needs_human_review
|
||
true_fill = PatternFill(start_color="FF6347", end_color="FF6347", fill_type="solid") # rosso
|
||
false_fill = PatternFill(start_color="90EE90", end_color="90EE90", fill_type="solid") # verde
|
||
col_index = {cell.value: idx for idx, cell in enumerate(ws[1], start=1)}
|
||
if "llm_needs_human_review" in col_index:
|
||
c = col_index["llm_needs_human_review"]
|
||
for r in range(2, ws.max_row + 1):
|
||
val = ws.cell(row=r, column=c).value
|
||
if val is True:
|
||
ws.cell(row=r, column=c).fill = true_fill
|
||
elif val is False:
|
||
ws.cell(row=r, column=c).fill = false_fill
|
||
|
||
if "final_needs_human_review" in col_index:
|
||
c = col_index["final_needs_human_review"]
|
||
for r in range(2, ws.max_row + 1):
|
||
val = ws.cell(row=r, column=c).value
|
||
if val is True:
|
||
ws.cell(row=r, column=c).fill = true_fill
|
||
elif val is False:
|
||
ws.cell(row=r, column=c).fill = false_fill
|
||
|
||
wb.save(out_path)
|
||
print(f"\n***Step 6: Retrieval e LLM ***\nExcel LLM salvato in {out_path}")
|
||
|
||
# --- Conteggio needs_human_review ---
|
||
review_counts = df_llm["final_needs_human_review"].value_counts(dropna=False)
|
||
true_count = review_counts.get(True, 0)
|
||
false_count = review_counts.get(False, 0)
|
||
print("\n--- Needs human review summary ---")
|
||
print(f"needs_human_review = True : {true_count}")
|
||
print(f"needs_human_review = False: {false_count}")
|
||
|
||
|
||
# --- Step7: dataset finale su tutte le automazioni (solo testo + etichette) ---
|
||
df_final = df_llm[[
|
||
"automation_text",
|
||
"llm_category",
|
||
"llm_subcategory",
|
||
"llm_gravity",
|
||
"llm_problem_type",
|
||
"final_needs_human_review"
|
||
]].rename(columns={
|
||
"llm_category": "category",
|
||
"llm_subcategory": "subcategory",
|
||
"llm_gravity": "gravity",
|
||
"llm_problem_type": "problem_type"
|
||
})
|
||
|
||
# Normalizza stringhe
|
||
for col in ["category", "subcategory", "gravity", "problem_type"]:
|
||
df_final[col] = df_final[col].fillna("").astype(str).str.strip()
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# Creazione DataFrame risultati
|
||
# df_results = pd.DataFrame(output_rows)
|
||
# output_path = "main/datasets/similarity_analysis.xlsx"
|
||
# df_results.to_excel(output_path, index=False)
|
||
|
||
#wb = load_workbook(output_path)
|
||
#ws = wb.active
|
||
|
||
#distanza_col_idx = None
|
||
#for idx, cell in enumerate(ws[1], start=1):
|
||
#if cell.value == "distanza":
|
||
#distanza_col_idx = idx
|
||
#break
|
||
#if distanza_col_idx is None:
|
||
#raise ValueError("Colonna 'distanza' non trovata!")
|
||
|
||
# Applichiamo i colori in base al valore
|
||
#for row in ws.iter_rows(min_row=2, max_row=ws.max_row, min_col=distanza_col_idx, max_col=distanza_col_idx):
|
||
#cell = row[0]
|
||
#try:
|
||
#val = float(cell.value)
|
||
#if val < 0.5:
|
||
#color = "90EE90" # verde chiaro
|
||
#elif val < 1.0:
|
||
#color = "FFFF00" # giallo
|
||
#else:
|
||
#color = "FF6347" # rosso
|
||
#cell.fill = PatternFill(start_color=color, end_color=color, fill_type="solid")
|
||
#except:
|
||
#continue
|
||
|
||
# Salva il file direttamente con colori applicati
|
||
#wb.save(output_path)
|
||
#print(f"Excel salvato in {output_path}") |