Spaces:
Sleeping
Sleeping
File size: 7,554 Bytes
b38f323 | 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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 | from __future__ import annotations
from pathlib import Path
from typing import Any
def save_training_curves(
*,
history: list[dict[str, float]],
output_path: Path,
title: str = "Training Curves",
) -> Path | None:
if not history:
return None
plt, sns = _load_plot_libs()
sns.set_theme(style="whitegrid")
epochs: list[float] = []
train_loss: list[float] = []
val_loss: list[float] = []
for row in history:
if not isinstance(row, dict):
continue
epoch = row.get("epoch")
tr = row.get("train_loss")
if epoch is None or tr is None:
continue
epochs.append(float(epoch))
train_loss.append(float(tr))
val = row.get("val_loss")
val_loss.append(float(val) if val is not None else float("nan"))
if not epochs:
return None
output_path.parent.mkdir(parents=True, exist_ok=True)
fig, ax = plt.subplots(figsize=(10, 5))
sns.lineplot(x=epochs, y=train_loss, marker="o", label="train_loss", ax=ax)
if any(_is_finite(v) for v in val_loss):
sns.lineplot(x=epochs, y=val_loss, marker="o", label="val_loss", ax=ax)
ax.set_title(title)
ax.set_xlabel("Epoch")
ax.set_ylabel("Loss")
ax.legend()
fig.tight_layout()
fig.savefig(output_path, dpi=140)
plt.close(fig)
return output_path
def save_confusion_matrix_plot(
*,
y_true: list[str],
y_pred: list[str],
labels: list[str],
output_path: Path,
title: str = "Confusion Matrix",
) -> Path | None:
if not y_true or not y_pred or len(y_true) != len(y_pred):
return None
plt, sns = _load_plot_libs()
sns.set_theme(style="white")
matrix = _build_confusion_matrix(y_true=y_true, y_pred=y_pred, labels=labels)
output_path.parent.mkdir(parents=True, exist_ok=True)
fig, ax = plt.subplots(figsize=(8, 6))
sns.heatmap(
matrix,
annot=True,
fmt="d",
cmap="Blues",
xticklabels=labels,
yticklabels=labels,
cbar=True,
ax=ax,
)
ax.set_xlabel("Predicted")
ax.set_ylabel("True")
ax.set_title(title)
fig.tight_layout()
fig.savefig(output_path, dpi=140)
plt.close(fig)
return output_path
def save_retrieval_recall_plot(
*,
recall_at_k: dict[int, float],
hit_at_k: dict[int, float],
output_path: Path,
title: str = "Retrieval Recall@K / Hit@K",
) -> Path | None:
if not recall_at_k and not hit_at_k:
return None
plt, sns = _load_plot_libs()
sns.set_theme(style="whitegrid")
output_path.parent.mkdir(parents=True, exist_ok=True)
ks = sorted(set(recall_at_k.keys()) | set(hit_at_k.keys()))
if not ks:
return None
recall_values = [float(recall_at_k.get(k, float("nan"))) for k in ks]
hit_values = [float(hit_at_k.get(k, float("nan"))) for k in ks]
fig, ax = plt.subplots(figsize=(9, 5))
sns.lineplot(x=ks, y=recall_values, marker="o", label="Recall@K", ax=ax)
sns.lineplot(x=ks, y=hit_values, marker="o", label="Hit@K", ax=ax)
ax.set_ylim(0.0, 1.0)
ax.set_xlabel("K")
ax.set_ylabel("Score")
ax.set_title(title)
ax.legend()
fig.tight_layout()
fig.savefig(output_path, dpi=140)
plt.close(fig)
return output_path
def save_retrieval_mrr_by_label_plot(
*,
mrr_by_label: dict[str, float],
output_path: Path,
title: str = "Retrieval MRR by Decision Label",
) -> Path | None:
if not mrr_by_label:
return None
plt, sns = _load_plot_libs()
sns.set_theme(style="whitegrid")
output_path.parent.mkdir(parents=True, exist_ok=True)
labels = list(mrr_by_label.keys())
values = [float(mrr_by_label[label]) for label in labels]
fig, ax = plt.subplots(figsize=(9, 5))
sns.barplot(x=labels, y=values, ax=ax, palette="Blues_d")
ax.set_ylim(0.0, 1.0)
ax.set_xlabel("Decision label")
ax.set_ylabel("MRR")
ax.set_title(title)
fig.tight_layout()
fig.savefig(output_path, dpi=140)
plt.close(fig)
return output_path
def save_retrieval_user_signal_heatmap(
*,
user_signal_scores: dict[str, dict[str, float]],
output_path: Path,
title: str = "Retrieval Top-K Mean Match Score (User x Signal)",
) -> Path | None:
if not user_signal_scores:
return None
plt, sns = _load_plot_libs()
sns.set_theme(style="white")
output_path.parent.mkdir(parents=True, exist_ok=True)
users = sorted(user_signal_scores.keys())
signals = sorted({signal for row in user_signal_scores.values() for signal in row.keys()})
if not users or not signals:
return None
matrix: list[list[float]] = []
for user in users:
row = user_signal_scores.get(user, {})
matrix.append([float(row.get(signal, 0.0)) for signal in signals])
fig_w = max(10, int(0.45 * len(signals)) + 4)
fig_h = max(4, int(0.6 * len(users)) + 3)
fig, ax = plt.subplots(figsize=(fig_w, fig_h))
sns.heatmap(
matrix,
cmap="YlGnBu",
annot=False,
xticklabels=signals,
yticklabels=users,
cbar=True,
ax=ax,
)
ax.set_xlabel("Signal")
ax.set_ylabel("User")
ax.set_title(title)
fig.tight_layout()
fig.savefig(output_path, dpi=140)
plt.close(fig)
return output_path
def save_ablation_comparison_plot(
*,
with_retrieval: dict[str, float],
without_retrieval: dict[str, float],
output_path: Path,
title: str = "Classification Ablation: With vs Without Retrieval",
) -> Path | None:
keys = ["accuracy", "macro_f1"]
if any(key not in with_retrieval for key in keys) or any(
key not in without_retrieval for key in keys
):
return None
plt, sns = _load_plot_libs()
sns.set_theme(style="whitegrid")
output_path.parent.mkdir(parents=True, exist_ok=True)
labels = ["Accuracy", "Macro-F1"]
x = [0, 1]
with_vals = [float(with_retrieval["accuracy"]), float(with_retrieval["macro_f1"])]
without_vals = [
float(without_retrieval["accuracy"]),
float(without_retrieval["macro_f1"]),
]
fig, ax = plt.subplots(figsize=(8, 5))
width = 0.34
ax.bar([v - width / 2 for v in x], with_vals, width=width, label="with retrieval")
ax.bar([v + width / 2 for v in x], without_vals, width=width, label="without retrieval")
ax.set_xticks(x)
ax.set_xticklabels(labels)
ax.set_ylim(0.0, 1.0)
ax.set_ylabel("Score")
ax.set_title(title)
ax.legend()
fig.tight_layout()
fig.savefig(output_path, dpi=140)
plt.close(fig)
return output_path
def _build_confusion_matrix(
*,
y_true: list[str],
y_pred: list[str],
labels: list[str],
) -> list[list[int]]:
index = {label: i for i, label in enumerate(labels)}
matrix = [[0 for _ in labels] for _ in labels]
for gold, pred in zip(y_true, y_pred):
if gold not in index or pred not in index:
continue
matrix[index[gold]][index[pred]] += 1
return matrix
def _is_finite(value: float) -> bool:
return value == value and value not in (float("inf"), float("-inf"))
def _load_plot_libs() -> tuple[Any, Any]:
try:
import matplotlib.pyplot as plt
import seaborn as sns
except Exception as exc: # pragma: no cover - runtime dependency guard
raise RuntimeError(
"Plotting requires matplotlib and seaborn. "
"Install them with: pip install matplotlib seaborn"
) from exc
return plt, sns
|