#!/usr/bin/env python3 """Baseline classifiers for human vs. LLM (and model attribution) on 200-token chunks. Splits are done on document numbers (the same document number denotes the same source prompt in every subcorpus; splitting on chunks would leak topics between train and test): doc_num % 5 == 0 -> test, doc_num % 5 == 1 -> dev, rest -> train. Tasks: binary human vs. AI. The human class is ~1.6 % of chunks, so the default decision threshold is useless; the threshold on the SVM decision score is tuned on the dev set to maximize balanced accuracy, and threshold-free ROC-AUC is reported. attribution human + one class per model line (temperatures merged; completion-mode variants kept separate from chat variants, since base-model output differs qualitatively) Feature sets are TF-IDF vectorizers over different token columns. Classifier: LinearSVC (liblinear, internal OVR for multiclass). """ import sys import time from functools import partial from pathlib import Path import numpy as np import pandas as pd from scipy.sparse import hstack from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import SGDClassifier from sklearn.metrics import (accuracy_score, balanced_accuracy_score, classification_report, confusion_matrix, f1_score, roc_auc_score, roc_curve) from sklearn.svm import LinearSVC DATA_DIR = Path(__file__).parent / "data" RESULTS_DIR = Path(__file__).parent / "results_v2" N_FOLDS = 5 # doc_num % 5: 0 -> test, 1 -> dev, 2-4 -> train # Base/completion-mode models mimic the human distribution too well and are # not relevant for detecting assistant-style LLM text in real corpora # (v2 decision, confirmed by Jiří 2026-07-06); they are excluded from both # tasks. EXCLUDE_COMPLETE = True def model_class(row): """Collapse subcorpora into model-line classes for the attribution task.""" if row["model"] == "human": return "human" name = row["model"] if row["mode"] == "complete": name += "-complete" return name def skipgram_analyzer(text): """Skip-bigrams over pre-tokenized text: pairs with 1-3 tokens skipped. Plain bigrams are covered by the word vectorizers; here only the gapped pairs are produced, so the set is complementary. """ toks = text.lower().split() feats = [] for gap in (1, 2, 3): for i in range(len(toks) - gap - 1): feats.append(f"{toks[i]} <{gap}> {toks[i + gap + 1]}") return feats def make_vectorizers(feature_set): """Return list of (column, TfidfVectorizer) for a named feature set.""" # float32 halves matrix memory; the run was OOM-killed with float64 Tfidf = partial(TfidfVectorizer, sublinear_tf=True, dtype=np.float32) word = ("words", Tfidf(ngram_range=(1, 2), min_df=3, lowercase=True)) char = ("words", Tfidf(analyzer="char_wb", ngram_range=(3, 5), min_df=3, max_features=300_000, lowercase=True)) lemma = ("lemmata", Tfidf(ngram_range=(1, 2), min_df=3, lowercase=True)) # long lemma n-grams: LLMs have favourite multi-word phrases (up to ~5 # lemmata) that humans lack; min_df keeps the vocabulary tractable lemma15 = ("lemmata", Tfidf(ngram_range=(1, 5), min_df=5, lowercase=True)) skip = ("words", Tfidf(analyzer=skipgram_analyzer, min_df=5)) pos = ("pos", Tfidf(ngram_range=(1, 3), min_df=3, lowercase=False)) tag = ("TAG", Tfidf(ngram_range=(1, 4), min_df=3, lowercase=False)) fun = ("FUN", Tfidf(ngram_range=(1, 4), min_df=3, lowercase=False)) sets = { "word12": [word], "char35": [char], "lemma12": [lemma], "lemma15": [lemma15], "skipgram": [skip], "pos13": [pos], "tag14": [tag], "fun14": [fun], "morphsyn": [pos, tag, fun], # fully delexicalized "word+pos": [word, pos], "word+skip": [word, skip], # surface-only: needs no lemmata, # usable with a plain tokenizer "lex+skip": [word, lemma15, skip], # lexical phrases + skipgrams "skip+fun": [skip, fun], # skipgrams + deprels "best": [word, lemma15, skip, fun], "full": [word, lemma15, tag, fun], } return sets[feature_set] # liblinear multiclass on multi-million-feature sets was OOM-killed (>52 GB); # for these attribution runs a single-pass SGD one-vs-all is used instead SGD_ATTRIBUTION_SETS = {"lex+skip", "skip+fun", "best", "full"} def make_classifier(task, feature_set): if task == "attribution" and feature_set in SGD_ATTRIBUTION_SETS: return SGDClassifier(loss="log_loss", alpha=1e-6, class_weight="balanced", max_iter=30, tol=1e-4, random_state=0) return LinearSVC(class_weight="balanced", C=1.0) def featurize(df_train, df_evals, feature_set): """Fit vectorizers on train, transform train + each eval frame. Returns [X_train, X_eval1, X_eval2, ...]. """ specs = make_vectorizers(feature_set) parts = [[] for _ in range(1 + len(df_evals))] for col, vec in specs: parts[0].append(vec.fit_transform(df_train[col])) for i, df_eval in enumerate(df_evals, start=1): parts[i].append(vec.transform(df_eval[col])) return [p[0] if len(p) == 1 else hstack(p).tocsr() for p in parts] def tune_threshold(y_dev, scores_dev): """Return the decision-score threshold maximizing balanced accuracy on dev.""" fpr, tpr, thresholds = roc_curve(y_dev, scores_dev) balanced = (tpr + (1 - fpr)) / 2 return thresholds[np.argmax(balanced)] def run_experiment(df, lang, task, feature_set, out_rows): fold = df["doc_num"] % N_FOLDS train = df[fold >= 2] dev = df[fold == 1] test = df[fold == 0] if task == "binary": ytr = (train["model"] != "human").to_numpy() # True = AI ydev = (dev["model"] != "human").to_numpy() yte = (test["model"] != "human").to_numpy() else: ytr = train.apply(model_class, axis=1).to_numpy() yte = test.apply(model_class, axis=1).to_numpy() t0 = time.time() Xtr, Xdev, Xte = featurize(train, [dev, test], feature_set) t_feat = time.time() - t0 clf = make_classifier(task, feature_set) t0 = time.time() clf.fit(Xtr, ytr) t_fit = time.time() - t0 tag = f"{lang}_{task}_{feature_set}" row = {"lang": lang, "task": task, "features": feature_set, "n_features": Xtr.shape[1], "n_train": len(train), "n_test": len(test)} if task == "binary": scores_dev = clf.decision_function(Xdev) scores_te = clf.decision_function(Xte) thr = tune_threshold(ydev, scores_dev) pred = scores_te >= thr row["roc_auc"] = roc_auc_score(yte, scores_te) row["threshold"] = thr else: pred = clf.predict(Xte) row["accuracy"] = accuracy_score(yte, pred) row["balanced_accuracy"] = balanced_accuracy_score(yte, pred) row["macro_f1"] = f1_score(yte, pred, average="macro") auc_str = f" AUC={row['roc_auc']:.4f}" if "roc_auc" in row else "" print(f"[{tag}] acc={row['accuracy']:.4f} " f"bal_acc={row['balanced_accuracy']:.4f} " f"macroF1={row['macro_f1']:.4f}{auc_str} " f"(feat {t_feat:.0f}s, fit {t_fit:.0f}s, {Xtr.shape[1]} features)", flush=True) out_rows.append(row) # per-subcorpus breakdown / confusion matrices on the test set detail = test[["subcorpus"]].copy() if task == "binary": detail["pred_ai"] = pred tab = (detail.groupby("subcorpus")["pred_ai"].agg(["mean", "count"]) .rename(columns={"mean": "frac_predicted_ai", "count": "n_chunks"})) tab.to_csv(RESULTS_DIR / f"persubcorpus_{tag}.tsv", sep="\t") else: labels = sorted(set(yte) | set(pred)) cm = confusion_matrix(yte, pred, labels=labels) pd.DataFrame(cm, index=labels, columns=labels).to_csv( RESULTS_DIR / f"confusion_{tag}.tsv", sep="\t") rep = classification_report(yte, pred, labels=labels, output_dict=True, zero_division=0) pd.DataFrame(rep).T.to_csv(RESULTS_DIR / f"report_{tag}.tsv", sep="\t") def main(): RESULTS_DIR.mkdir(exist_ok=True) binary_features = ["word12", "lemma15", "skipgram", "tag14", "fun14", "morphsyn", "word+pos", "word+skip", "lex+skip", "skip+fun", "best", "full"] attribution_features = ["word12", "morphsyn", "skip+fun", "lex+skip", "best"] langs = sys.argv[1:] if len(sys.argv) > 1 else ["brown", "koditex"] summary_path = RESULTS_DIR / "summary.tsv" out_rows = (pd.read_csv(summary_path, sep="\t").to_dict("records") if summary_path.exists() else []) done = {(r["lang"], r["task"], r["features"]) for r in out_rows} for lang in langs: df = pd.read_pickle(DATA_DIR / f"chunks_{lang}.pkl") if EXCLUDE_COMPLETE: df = df[df["mode"] != "complete"] df = df.drop(columns=["UTAG"]) # unused, frees several GB print(f"=== {lang}: {len(df)} chunks, " f"{(df['model'] == 'human').sum()} human ===", flush=True) for task, feature_sets in [("binary", binary_features), ("attribution", attribution_features)]: for feature_set in feature_sets: if (lang, task, feature_set) in done: print(f"[{lang}_{task}_{feature_set}] already in summary, " f"skipping", flush=True) continue run_experiment(df, lang, task, feature_set, out_rows) pd.DataFrame(out_rows).to_csv(summary_path, sep="\t", index=False) print("done") if __name__ == "__main__": main()