Spaces:
Running
Running
File size: 24,648 Bytes
e3f85da | 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 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 | """The public page of a governed system: the signed model, its monitoring page, and the dossier.
Built for the AI Governance Practicum (AI Governance Collective). One app.py serves all three organizations; space_config.json says which."""
import json, os, pickle, io, warnings
import numpy as np, pandas as pd
import gradio as gr
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
warnings.filterwarnings("ignore")
HERE = os.path.dirname(os.path.abspath(__file__))
# Hugging Face's free tier runs Gradio Spaces on ZeroGPU, which refuses to start an app that declares no GPU function.
# This app needs no GPU: the declaration below satisfies the check and is never called. On a plain CPU Space it is skipped.
try:
import spaces
@spaces.GPU
def _gpu_not_used():
return "This app runs on the CPU."
except Exception:
pass
CFG = json.load(open(os.path.join(HERE, "space_config.json")))
REF = json.load(open(os.path.join(HERE, "reference.json")))
MODEL = pickle.load(open(os.path.join(HERE, CFG["model_file"]), "rb"))
RECAL = pickle.load(open(os.path.join(HERE, CFG["recal_file"]), "rb")) if CFG.get("recal_file") else None
REASONS = pd.read_csv(os.path.join(HERE, CFG["reasons_file"])) if CFG.get("reasons_file") else None
EXAMPLES = pd.read_csv(os.path.join(HERE, "examples.csv"), keep_default_na=False)
INPUTS, KEPT, SHARE = CFG["inputs"], CFG["kept"], CFG["share"]
CAL_FLOOR = 25 # a calibration band is read only when it expects at least this many outcomes; a smaller band says nothing either way
SPACE_ID = os.environ.get("SPACE_ID", "")
WEEK6 = CFG["week6"]
# ----------------------------------------------------------------------------- scoring
def score_frame(X):
"""The production score. Where the Week 6 memo ratified fitted steps (the recalibration and the odds correction), they run here."""
X = X[INPUTS]
if RECAL is not None:
pf = RECAL.predict_proba(X)[:, 1]; f = CFG["odds_factor"]
return pf * f / (pf * f + 1 - pf)
return MODEL.predict_proba(X)[:, 1]
def contributions(X):
"""Each input's fixed weight times the record's value on the scaled, remover-adjusted inputs. The record's own reasons."""
Z = X[INPUTS]
for name, step in MODEL.steps[:-1]:
Z = step.transform(Z)
coef = MODEL.named_steps["model"].coef_[0]
return pd.DataFrame(np.asarray(Z) * coef, columns=KEPT, index=X.index)
def allowed_reasons(con):
"""Drop the columns a reason may never cite: a medication the record marks No (Alderwyn, Week 6 memo, ruling 5)."""
drugs = CFG.get("drug_fields", [])
if not drugs: return con
keep = [c for c in con.index if not any(c == d + "_No" for d in drugs)]
return con[keep]
def allowed_reasons_frame(frame):
drugs = CFG.get("drug_fields", [])
return frame[[c for c in frame.columns if not any(c == d + "_No" for d in drugs)]] if drugs else frame
def readable(col):
if REASONS is not None:
r = REASONS.loc[(REASONS.input == col) | ((REASONS.input + "_" + REASONS.value.astype(str)) == col), "reason"]
if len(r): return r.iloc[0]
for c in CFG["categorical"]:
if col.startswith(c + "_"): return f"{c} is {col[len(c) + 1:]}"
return col
def coerce(df):
"""Read a file the way the working file is read: numeric inputs as numbers, categorical inputs as text."""
df = df.copy()
for c in CFG["numeric"]:
if c in df: df[c] = pd.to_numeric(df[c], errors="coerce")
for c in CFG["categorical"]:
if c in df: df[c] = df[c].astype(str)
return df
def groups_of(frame):
g = {}
for gname, d in CFG["groups"].items():
col = d["col"]
if col not in frame: continue
if "bands" in d:
v = pd.to_numeric(frame[col], errors="coerce"); lab = pd.Series(["unrecorded"] * len(frame), index=frame.index, dtype=object)
for lo, hi, name in d["bands"]: lab[(v >= lo) & (v <= hi)] = name
g[gname] = lab
else:
s = frame[col].astype(str); g[gname] = s.map(d["map"]).fillna(s) if d.get("map") else s
return pd.DataFrame(g, index=frame.index)
def pct(x, d=1): return f"{x * 100:.{d}f} percent"
# ----------------------------------------------------------------------------- the signed version (the collapsed panel at the top)
def model_header():
a = CFG["artifacts"][CFG["model_file"]]
L = [f"**Organization:** {CFG['org']}. **Signed by:** {CFG['caio']} ({CFG['memo_ref']}). **Kept by:** {CFG['lead']}.",
f"**File:** `{CFG['model_file']}`, {a['bytes']:,} bytes, SHA-256 `{a['sha256']}`."]
if RECAL is not None:
r = CFG["artifacts"][CFG["recal_file"]]
L.append(f"**Fitted steps the memo ratified:** the isotonic recalibration (`{CFG['recal_file']}`, SHA-256 `{r['sha256']}`), then the odds correction to the {int(CFG['calibration']['book_rate']*100)} percent book rate (factor {CFG['odds_factor']}). Every score on this page has both applied.")
L.append(f"**What it is:** a logistic regression with {len(KEPT)} fitted columns from {len(INPUTS)} input fields. The weights are fixed. The same record gets the same score every time.")
L.append(f"**Operating point:** the top {int(SHARE*100)} percent of the ranked file gets {CFG['action']}. On the Week 6 test set ({WEEK6['test_rows']:,} {CFG['units']}) that was {WEEK6['flagged']:,} {CFG['units']}, of which {WEEK6['caught']:,} were {CFG['positive_label']}. Test-set ROC-AUC {WEEK6['roc_auc']:.3f}.")
L.append(f"**Libraries:** scikit-learn {CFG['versions']['sklearn']}, fairlearn {CFG['versions']['fairlearn']}. The model does not load on another version of scikit-learn, by design.")
return "\n\n".join(L)
def weights_table(n=25):
return pd.DataFrame([dict(column=c, weight=round(w, 4), meaning=readable(c)) for c, w in CFG["weights"][:n]])
# ----------------------------------------------------------------------------- tab 1: one record
CAT = [c for c in INPUTS if c in CFG["categorical"]]
NUM = [c for c in INPUTS if c in CFG["numeric"]]
FORM = CAT + NUM # the order of the form's controls
def _typical():
d = {c: REF["numeric"][c]["median"] for c in NUM}
for c in CAT:
r = REF["categorical"][c]; d[c] = max(r, key=r.get)
return d
def _case_row(i):
row = EXAMPLES.iloc[i]
return {c: (float(row[c]) if c in NUM else str(row[c])) for c in FORM}
PRESETS = {f"Typical {CFG['unit']} (training-file medians and most common values)": _typical()}
for i in range(len(EXAMPLES)):
PRESETS[f"{EXAMPLES['case'].iloc[i].capitalize()} (Week 6 case, rank {EXAMPLES['week6_rank'].iloc[i]} of {WEEK6['test_rows']:,})"] = _case_row(i)
FIRST = list(PRESETS)[0]
def apply_preset(name):
d = PRESETS[name]
return [d[c] for c in FORM]
def field_label(c):
"""The input's name, with the meaning of its codes where the program's group definitions give one (SEX: 1 male, 2 female)."""
for d in CFG["groups"].values():
if d["col"] == c and d.get("map"):
return c + " (" + ", ".join(f"{k} {v}" for k, v in d["map"].items()) + ")"
return c
def choices_for(c):
ch = sorted(REF["categorical"][c], key=str)
for v in EXAMPLES[c].astype(str).unique():
if v not in ch: ch.append(v)
return ch
def score_one(*vals):
row = {c: v for c, v in zip(FORM, vals)}
X = coerce(pd.DataFrame([row]))
p = float(score_frame(X)[0]); cut = REF["cut_score"]; inside = p >= cut
verdict = (f"## Score: **{p:.1%}**\n\n"
f"### {'馃敶 Above the cut' if inside else '馃煝 Below the cut'} (Week 6 cut = {cut:.1%}, the score of the last {CFG['unit']} inside the top {int(SHARE*100)} percent)\n\n"
f"At the operating point the {CFG['memo_ref']} set, this {CFG['unit']} {'would' if inside else 'would not'} be {CFG['flagged_word']}: {CFG['action']}. "
f"The score says where the record ranks. It is not a decision about the {CFG['person']}.")
con = allowed_reasons(contributions(X).iloc[0])
up = con[con > 0].sort_values(ascending=False).head(6); down = con[con < 0].sort_values().head(4)
items = list(up.items()) + list(down.items())
labs = [readable(c) for c, _ in items][::-1]; v = [float(x) for _, x in items][::-1]
fig, ax = plt.subplots(figsize=(11, 0.45 * len(v) + 1.8))
ax.barh(range(len(v)), v, color=["#d62728" if x > 0 else "#2ca02c" for x in v])
ax.set_yticks(range(len(v))); ax.set_yticklabels(labs, fontsize=9); ax.axvline(0, color="black", lw=0.8)
ax.set_xlabel(f"Toward {CFG['positive_label']} (red) or away from it (green): the fixed weight times this record's value", fontsize=9)
ax.set_title(f"Why {CFG['system']} scored this {CFG['unit']} {p:.1%}: the record's own reasons", fontsize=11, loc="left")
fig.tight_layout()
return verdict, fig
# ----------------------------------------------------------------------------- tab 2: batch upload
def score_batch(file, top_n):
if file is None: return None, f"Upload a CSV of {CFG['units']} to score (the model's input columns, one row per {CFG['unit']}).", None
try: df = coerce(pd.read_csv(file.name if hasattr(file, "name") else file, low_memory=False, keep_default_na=False, na_values=[]))
except Exception as e: return None, f"Could not read the file: {e}", None
missing = [c for c in INPUTS if c not in df.columns]
if missing: return None, "The file is missing input columns: " + ", ".join(missing) + ". Nothing was scored.", None
s = score_frame(df); order = np.argsort(-s, kind="stable"); k = int(round(SHARE * len(df)))
idc = CFG.get("id_col"); ids = df[idc].values if idc and idc in df.columns else np.arange(1, len(df) + 1)
con = allowed_reasons_frame(contributions(df)); top = [readable(con.columns[i]) for i in np.argmax(con.values, axis=1)]
out = pd.DataFrame({(idc if idc and idc in df.columns else "row"): ids, "score": np.round(s, 4), "ranked_in": np.where(np.arange(len(df)) >= 0, "", ""), "top_reason": top})
out = out.iloc[order].reset_index(drop=True); out.insert(0, "rank", np.arange(1, len(df) + 1)); out["ranked_in"] = np.where(out["rank"] <= k, "yes", "no")
path = os.path.join("/tmp", "ranked_list.csv"); out.to_csv(path, index=False)
msg = (f"Scored **{len(df):,}** {CFG['units']}. The top {int(SHARE*100)} percent, the operating point, is **{k:,}** {CFG['units']}. "
f"Showing the top **{int(top_n)}** by score (the team's capacity). The download carries every record with its score, rank, whether it is ranked in, and its own top reason.")
return out.head(int(top_n)), msg, path
# ----------------------------------------------------------------------------- tab 2: monitoring
def psi(ref_shares, new_shares):
keys = set(ref_shares) | set(new_shares); eps = 1e-4; v = 0.0
for k in keys:
a, b = max(ref_shares.get(k, 0.0), eps), max(new_shares.get(k, 0.0), eps); v += (b - a) * np.log(b / a)
return float(v)
def drift_table(df):
rows = []
for c in INPUTS:
if c not in df.columns: continue
watch = c in CFG["watch_fields"]
if c in REF["numeric"]:
r = REF["numeric"][c]; v = pd.to_numeric(df[c], errors="coerce"); edges = r["edges"]
if len(edges) < 3: continue
def shares(x):
b = np.clip(np.searchsorted(edges, x, side="right") - 1, 0, len(edges) - 2); return {str(i): float((b == i).mean()) for i in range(len(edges) - 1)}
ref_sh = r["shares"]; new_sh = shares(v.dropna().values)
idx = psi(ref_sh, new_sh)
rows.append(dict(input=c, kind="number", training_file=f"mean {r['mean']:,.2f}, median {r['median']:,.2f}", this_file=f"mean {v.mean():,.2f}, median {v.median():,.2f}", missing=int(v.isna().sum()), stability_index=round(idx, 3), reading=reading(idx), watched="yes" if watch else ""))
elif c in REF["categorical"]:
r = REF["categorical"][c]; vc = df[c].astype(str).value_counts(normalize=True); new_sh = {str(k): float(x) for k, x in vc.items()}
idx = psi(r, new_sh); top_r = max(r, key=r.get); top_n = max(new_sh, key=new_sh.get) if new_sh else ""
new_levels = [k for k in new_sh if k not in r]
rows.append(dict(input=c, kind="category", training_file=f"most common {top_r} ({pct(r[top_r])})", this_file=f"most common {top_n} ({pct(new_sh.get(top_n, 0))})" + (f"; new levels: {', '.join(new_levels)}" if new_levels else ""),
missing=int((df[c].astype(str) == "nan").sum()), stability_index=round(idx, 3), reading=reading(idx), watched="yes" if watch else ""))
t = pd.DataFrame(rows)
return t.sort_values(["watched", "stability_index"], ascending=[False, False]).reset_index(drop=True) if len(t) else t
def reading(idx):
return "stable" if idx < 0.10 else ("watch" if idx < 0.25 else "SHIFTED")
def group_table(df, s, flag, y):
"""Each group's rate at the cut beside the Week 6 test-set reading and the tripwire. With outcomes: the ruled check (false alarm or miss). Without: the share ranked in."""
G = groups_of(df); rows = []; check = CFG["group_check"]; line = CFG["group_tripwire"]
if y is not None:
pos, neg = y == 1, y == 0
port = (((flag == 1) & neg).sum() / max(neg.sum(), 1)) if check == "false_alarm" else (((flag == 0) & pos).sum() / max(pos.sum(), 1))
for gname in G.columns:
d = CFG["groups"][gname]; order = [b[2] for b in d["bands"]] if "bands" in d else d.get("order", [])
for lvl in sorted(G[gname].unique(), key=lambda s: (order.index(s) if s in order else len(order), str(s))):
m = (G[gname] == lvl).values; n = int(m.sum())
wk6 = REF["group_rates"].get(gname, {}).get(str(lvl), {})
covered = n >= CFG["min_group"] # the memo's group floor, applied to the month's file
if y is not None:
rate = (((flag == 1) & neg & m).sum() / max((neg & m).sum(), 1)) if check == "false_alarm" else (((flag == 0) & pos & m).sum() / max((pos & m).sum(), 1))
base = wk6.get("false_alarm" if check == "false_alarm" else "miss", np.nan)
over = rate > port * (1 + line)
result = (f"not judged (under the group floor of {CFG['min_group']:,})" if not covered else ("TRIP" if over else "inside the line"))
rows.append({"group": gname, "level": lvl, "records this file": n, ("false alarm rate" if check == "false_alarm" else "miss rate"): pct(rate), "whole file": pct(port), "line": pct(port * (1 + line)), "Week 6 test set": pct(base) if base == base else "", "watched": "yes" if str(lvl) in CFG["watch_groups"] else "", "result": result})
else:
rate = float(flag[m].mean()); base = wk6.get("flagged_share", np.nan)
rows.append({"group": gname, "level": lvl, "records this file": n, "share ranked in": pct(rate), "whole file": pct(SHARE), "Week 6 test set": pct(base) if base == base else "", "watched": "yes" if str(lvl) in CFG["watch_groups"] else "", "result": "no outcomes yet: share ranked in only" if covered else f"not judged (under the group floor of {CFG['min_group']:,})"})
return pd.DataFrame(rows)
def calibration_table(s, y):
cal = CFG["calibration"]; rows = []
for lo, hi, name in cal["bands"]:
m = (s >= lo) & (s < hi)
if m.sum() == 0: rows.append(dict(band=name, records=0, predicted="", observed="", gap_share_of_observed="", result="empty band")); continue
mp, ob = float(s[m].mean()), float(y[m].mean()); gap = abs(mp - ob); expected = mp * m.sum()
if expected < CAL_FLOOR: res = f"not read (fewer than {CAL_FLOOR} expected outcomes in the band)"
else: res = "no outcomes in band" if ob == 0 else ("pass" if gap / ob <= cal["tripwire"] else "TRIP")
rows.append(dict(band=name, records=int(m.sum()), predicted=pct(mp, 2), observed=pct(ob, 2), gap_share_of_observed=(pct(gap / ob, 1) if ob else ""), result=res))
return pd.DataFrame(rows)
def monitor(file):
empty = (None, None, None)
if file is None: return "Upload a monthly file: one row per " + CFG["unit"] + ", the model's input columns, the measurement fields, and the outcome column once it is known.", *empty
try: df = coerce(pd.read_csv(file.name if hasattr(file, "name") else file, low_memory=False, keep_default_na=False, na_values=[]))
except Exception as e: return f"Could not read the file: {e}", *empty
missing = [c for c in INPUTS if c not in df.columns]
if missing: return "The file is missing input columns: " + ", ".join(missing) + ". The job would hold this file out of the cycle (the input contract).", *empty
s = score_frame(df); order = np.argsort(-s, kind="stable"); k = int(round(SHARE * len(df))); flag = np.zeros(len(df), int); flag[order[:k]] = 1
y = (df[CFG["outcome"]].astype(str) == str(CFG["positive"])).astype(int).values if CFG["outcome"] in df.columns else None
above_cut = float((s >= REF["cut_score"]).mean())
L = [f"**File:** {len(df):,} {CFG['units']}, {df.shape[1]} columns. Outcome column `{CFG['outcome']}`: " + ("present, so the ruled checks run." if y is not None else "absent, so only the input and score readings run. Upload again when the outcomes are known."),
f"**Scores:** mean {s.mean():.4f} (training-file positive rate {pct(REF['positive_rate'])}). {pct(above_cut)} of this file scores above the Week 6 cut of {REF['cut_score']:.4f}, against {pct(SHARE, 0)} by construction on the test set. The top {int(SHARE*100)} percent of this file is {k:,} {CFG['units']}.",
f"**Inputs:** the stability index compares each input's distribution with the training file's ({REF['rows']:,} rows). Under 0.10 is stable, 0.10 to 0.25 is watch, over 0.25 is shifted. These lines are the usual working lines, provisional until the Week 8 monitoring plan sets the program's own. Watched fields: {CFG['watch_fields_label']}."]
if y is not None:
pos = y == 1
L.append(f"**At the cut:** {int((flag & pos).sum()):,} of the {int(pos.sum()):,} {CFG['positive_label']} {CFG['units']} are ranked in ({pct((flag & pos).sum() / max(pos.sum(), 1))}); the whole-file {'false alarm' if CFG['group_check'] == 'false_alarm' else 'miss'} rate is the line every group is read against, plus {int(CFG['group_tripwire']*100)} percent ({CFG['memo_ref']}).")
cal = calibration_table(s, y) if (CFG.get("calibration") and y is not None) else None
if cal is not None:
worst = [r for r in cal.result if r == "TRIP"]; read = [r for r in cal.result if r in ("TRIP", "pass")]
L.append(f"**Calibration:** predicted against observed by score band, line {int(CFG['calibration']['tripwire']*100)} percent of observed. Bands that expect fewer than {CAL_FLOOR} outcomes are not read. " + (f"{len(read)} band(s) read, every one inside the line." if not worst else f"{len(worst)} of {len(read)} band(s) read are over the line: the calibration condition (ruling 1) is not met for this file."))
return "\n\n".join(L), drift_table(df), group_table(df, s, flag, y), cal
# ----------------------------------------------------------------------------- tab 4: the dossier
def dossier_markdown():
L = [f"## The dossier: {CFG['system']} at {CFG['org']}", "", "The six deliverables of the practicum, in course order. Each opens in the browser.", ""]
for f, label in CFG["docs"]:
p = os.path.join(HERE, "docs", f)
if SPACE_ID: url = f"https://huggingface.co/spaces/{SPACE_ID}/resolve/main/docs/{f}"
else: url = f"docs/{f}"
L.append(f"- [{label}]({url})" + ("" if os.path.exists(p) else " (not uploaded yet)"))
L += ["", f"**Monitoring targets from the {CFG['memo_ref']}:**"] + [f"- {m}" for m in CFG["monitoring_lines"]]
L += ["", f"**Data:** [{CFG['data_file']}]({CFG['data_url']}) in the program's dataset repository (MD5 `{CFG['md5']}`). The data is not copied into this page."]
return "\n".join(L)
# ----------------------------------------------------------------------------- the page
INTRO = f"""
# {CFG['system']} {CFG['version']}: {CFG['org']}
**Governance analyst: [YOUR NAME]** 路 a teaching deployment for the *AI Governance Practicum*, Track 1
---
{CFG['system']} estimates, for each {CFG['unit']}, the likelihood of {CFG['positive_label']}, ranks the file, and sends the top {int(SHARE*100)} percent to a person for {CFG['action']}. This page runs the signed version ({CFG['memo_ref']}). It is for **education**, not real decisions. A score says where a {CFG['person']}'s record ranks on one list. It is not a decision about the {CFG['person']}.
"""
with gr.Blocks(title=f"{CFG['org']} {CFG['system']}") as demo:
gr.Markdown(INTRO)
with gr.Accordion("About the signed version: the file, its hash, the fitted steps, the weights", open=False):
gr.Markdown(model_header())
gr.Markdown("**The fixed weights.** The largest by size, with what each column means. A positive weight pushes the score up.")
gr.Dataframe(weights_table(), interactive=False, wrap=True)
with gr.Tab(f"1 路 Single {CFG['unit']}"):
gr.Markdown(f"Score one {CFG['unit']}, then read the **reasons**. Load a Week 6 case to fill the form, or edit any field and score again.")
with gr.Row():
preset = gr.Dropdown(list(PRESETS), label="Load an example", value=FIRST)
d0 = PRESETS[FIRST]
widgets = {}
with gr.Accordion("Categorical inputs", open=True):
for i in range(0, len(CAT), 4):
with gr.Row():
for c in CAT[i:i + 4]:
widgets[c] = gr.Dropdown(choices_for(c), value=d0[c], label=field_label(c), allow_custom_value=True)
with gr.Accordion("Numeric inputs", open=(len(CAT) == 0)):
for i in range(0, len(NUM), 4):
with gr.Row():
for c in NUM[i:i + 4]:
widgets[c] = gr.Number(value=d0[c], label=c)
ordered = [widgets[c] for c in FORM]
score_btn = gr.Button(f"Score this {CFG['unit']}", variant="primary")
out_md = gr.Markdown(); out_plot = gr.Plot()
preset.change(apply_preset, preset, ordered)
score_btn.click(score_one, ordered, [out_md, out_plot])
with gr.Tab("2 路 Batch upload"):
gr.Markdown(f"Upload a CSV of {CFG['units']} (the same columns as the working file). {CFG['system']} scores every row, ranks the file, and marks the top {int(SHARE*100)} percent, the operating point the {CFG['memo_ref']} set. Top-N is how many of the ranked list the team can work. `sample_month.csv` in this repository is a file to try.")
with gr.Row():
up = gr.File(label=f"{CFG['units'].capitalize()} CSV", file_types=[".csv"])
with gr.Column():
topn = gr.Slider(5, 100, value=20, step=5, label="Top-N by capacity (how many the team can work)")
batch_btn = gr.Button("Score & rank", variant="primary")
batch_msg = gr.Markdown(); batch_tbl = gr.Dataframe(interactive=False, wrap=True); batch_dl = gr.File(label="Download the full ranked list")
batch_btn.click(score_batch, [up, topn], [batch_tbl, batch_msg, batch_dl])
with gr.Tab("3 路 Monitoring"):
gr.Markdown(f"**One month at a time.** Upload the month's file. The page reads it against the training file and the tripwires the {CFG['memo_ref']} set: the inputs' distributions, the scores, the groups at the cut, " + ("and the calibration table." if CFG.get("calibration") else "and the ruled group check.") + f" `sample_month.csv` in this repository is a stand-in drawn from the Week 6 test set, so the page can be tried before a monthly file exists.")
mf = gr.File(label="Monthly file (CSV)", file_types=[".csv"])
mon_btn = gr.Button("Read this month", variant="primary")
m_msg = gr.Markdown()
m_drift = gr.Dataframe(interactive=False, label="Inputs against the training file", wrap=True)
m_groups = gr.Dataframe(interactive=False, label="Groups at the cut, against the whole file and the line", wrap=True)
m_cal = gr.Dataframe(interactive=False, label="Calibration by score band", visible=bool(CFG.get("calibration")))
mon_btn.click(monitor, mf, [m_msg, m_drift, m_groups, m_cal])
with gr.Tab("4 路 The dossier"):
gr.Markdown(dossier_markdown())
gr.Markdown(f"---\n*{CFG['system']} is a fictional model built on a public dataset for the AI Governance Practicum (AI Governance Collective). The organization, the people, and the program are fictional. Data: [{CFG['data_file']}]({CFG['data_url']}).*")
if __name__ == "__main__":
demo.launch()
|