Buckets:
| """Shared input format, label ids, metrics and JSON helpers; no ML imports required.""" | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| LABELS = ("followed", "violated", "not_applicable") | |
| LABEL2ID = {label: index for index, label in enumerate(LABELS)} | |
| INSTRUCTION = ( | |
| "Classify whether the assistant message follows, violates, or is unrelated to the supplied policy." | |
| ) | |
| def fingerprint(value): | |
| return hashlib.sha256( | |
| json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8") | |
| ).hexdigest() | |
| def read_rows(path): | |
| path = Path(path) | |
| return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] | |
| def write_rows(path, rows): | |
| Path(path).write_text( | |
| "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8" | |
| ) | |
| def input_fields(message, policy, instruction=INSTRUCTION): | |
| """Use the pretrained tokenizer's pair separator; no new embedding needs training.""" | |
| return f"{instruction}\n\nAssistant message:\n{message}", f"Policy:\n{policy}" | |
| def tokenize_pairs(tokenizer, messages, policies, max_length, instruction=INSTRUCTION, **kwargs): | |
| """Keep the full policy and share identical formatting across training and prediction. | |
| Encoders use their pretrained pair separator. Qwen has no pair separator, so | |
| the two fields are separated by ordinary blank-line text; no new token is added. | |
| """ | |
| fields = [ | |
| input_fields(message, policy, instruction) for message, policy in zip(messages, policies, strict=True) | |
| ] | |
| boundary = "" if tokenizer.sep_token else "\n\n" | |
| return tokenizer( | |
| [message + boundary for message, _ in fields], | |
| [policy for _, policy in fields], | |
| truncation="only_first", | |
| max_length=max_length, | |
| return_token_type_ids=False, | |
| **kwargs, | |
| ) | |
| def classification_metrics(truth, predicted): | |
| """Keep all three classes in sklearn's metrics, even when a split lacks one.""" | |
| # Lazy import keeps dataset export and --preflight independent of ML dependencies. | |
| from sklearn.metrics import accuracy_score, f1_score, precision_recall_fscore_support | |
| class_ids = list(range(len(LABELS))) | |
| precision, recall, f1, support = precision_recall_fscore_support( | |
| truth, | |
| predicted, | |
| labels=class_ids, | |
| zero_division=0, | |
| ) | |
| result = {} | |
| for index, name in enumerate(LABELS): | |
| result.update( | |
| { | |
| f"{name}_precision": float(precision[index]), | |
| f"{name}_recall": float(recall[index]), | |
| f"{name}_f1": float(f1[index]), | |
| f"{name}_support": int(support[index]), | |
| } | |
| ) | |
| result["accuracy"] = float(accuracy_score(truth, predicted)) | |
| result["macro_f1"] = float(f1_score(truth, predicted, labels=class_ids, average="macro", zero_division=0)) | |
| return result | |
| def confusion_matrix_counts(truth, predicted): | |
| """sklearn confusion matrix as {actual label: {predicted label: count}}; rows are the truth.""" | |
| from sklearn.metrics import confusion_matrix | |
| matrix = confusion_matrix(truth, predicted, labels=list(range(len(LABELS)))) | |
| return { | |
| actual: {guess: int(matrix[row][column]) for column, guess in enumerate(LABELS)} | |
| for row, actual in enumerate(LABELS) | |
| } | |
Xet Storage Details
- Size:
- 3.37 kB
- Xet hash:
- 6bafd85006c8a02d2b8bdc461eeea670b33937e9a19c17ed656d331ae1fe8eb0
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.