Spaces:
Sleeping
Sleeping
| from pathlib import Path | |
| import numpy as np | |
| from src.utils.logger import get_logger | |
| class BackpropFeedbackLoop: | |
| """ | |
| Orchestrates the full backward propagation feedback cycle. | |
| After each forward pass through all 3 models, error signals flow | |
| back to incrementally update BERT, TF-IDF, and Naive Bayes weights, | |
| adjust fuzzy thresholds, and recalibrate ensemble weights. | |
| Full cycle (9 steps): | |
| text_batch -> forward pass -> ground truth -> error signals | |
| -> backward pass -> fuzzy update -> weight recalibration | |
| -> persist -> trend check -> git commit | |
| Part of the MisinformationDetector pipeline. | |
| Args: | |
| detector: MisinformationDetector master instance | |
| config (dict): loaded config.yaml as dict | |
| """ | |
| def __init__(self, detector, config): | |
| self.detector = detector | |
| self.config = config | |
| # Lazy imports to keep module import fast. | |
| from src.feedback.feedback_store import FeedbackStore | |
| from src.feedback.online_trainer import OnlineTrainer | |
| self.store = FeedbackStore() | |
| self.online = OnlineTrainer( | |
| bert_model=detector.bert_model, | |
| bert_tokenizer=detector.bert_tokenizer, | |
| tfidf_model=detector.tfidf_model, | |
| tfidf_vectorizer=detector.tfidf_vectorizer, | |
| nb_model=detector.nb_model, | |
| nb_vectorizer=detector.nb_vectorizer, | |
| device=detector.device, | |
| ) | |
| self.cycle_count = 0 | |
| self.metrics_history = [] | |
| self.logger = get_logger(__name__) | |
| fb_cfg = config.get("feedback", {}) | |
| self.high_error_threshold = float(fb_cfg.get("high_error_threshold", 0.25)) | |
| self.min_f1 = float(fb_cfg.get("min_f1_threshold", 0.75)) | |
| self.fail_limit = int(fb_cfg.get("consecutive_fail_limit", 3)) | |
| def run_cycle(self, text_batch: list, true_labels: list = None) -> dict: | |
| """ | |
| Execute one complete forward + feedback backward cycle. | |
| Args: | |
| text_batch (list): list of text strings to process | |
| true_labels (list, optional): int labels 0/1. If None, the | |
| LLM judge determines ground truth automatically. | |
| Returns: | |
| dict: per-model and ensemble metrics for this cycle | |
| """ | |
| self.logger.info("=== Feedback Cycle #%d START ===", self.cycle_count + 1) | |
| # ── STEP 1: FORWARD PASS ──────────────────────────────────────── | |
| predictions = [] | |
| fuzzy_scores = [] | |
| for text in text_batch: | |
| pred = self.detector.predict(text) | |
| predictions.append(pred) | |
| fb_score = self.store.get_feedback_score(text) | |
| fscore = self.detector.fuzzy_engine.compute( | |
| { | |
| "source_credibility": 0.5, | |
| "bert_confidence": pred["model_breakdown"]["bert"]["confidence"], | |
| "tfidf_confidence": pred["model_breakdown"]["tfidf"]["confidence"], | |
| "nb_confidence": pred["model_breakdown"]["naive_bayes"]["confidence"], | |
| "model_agreement": pred["model_agreement"], | |
| "feedback_score": fb_score, | |
| } | |
| ) | |
| fuzzy_scores.append(fscore) | |
| # ── STEP 2: GROUND TRUTH ──────────────────────────────────────── | |
| judge_confidences = [1.0] * len(text_batch) | |
| if true_labels is None: | |
| try: | |
| judgments = self.detector.llm_judge.evaluate_batch( | |
| list(zip(text_batch, predictions, fuzzy_scores)) | |
| ) | |
| true_labels = [ | |
| 1 if j.get("independent_verdict") == "misinformation" else 0 for j in judgments | |
| ] | |
| judge_confidences = [float(j.get("judge_confidence", 1.0)) for j in judgments] | |
| except Exception as e: | |
| self.logger.warning("LLM judge failed, defaulting labels to 0: %s", e) | |
| true_labels = [0] * len(text_batch) | |
| # ── STEP 3: COMPUTE ERROR SIGNALS ─────────────────────────────── | |
| errors = self._compute_errors(predictions, true_labels, judge_confidences) | |
| high_error_idx = [ | |
| i for i, e in enumerate(errors) if e["total_error"] > self.high_error_threshold | |
| ] | |
| self.logger.info("High-error samples: %d/%d", len(high_error_idx), len(text_batch)) | |
| # ── STEP 4: BACKWARD PASS ─────────────────────────────────────── | |
| if high_error_idx: | |
| err_batch = [(text_batch[i], true_labels[i]) for i in high_error_idx] | |
| self.online.update_bert(err_batch) | |
| self.online.update_tfidf(err_batch) | |
| self.online.update_naive_bayes(err_batch) | |
| # ── STEP 5: UPDATE FUZZY THRESHOLDS ───────────────────────────── | |
| self._update_fuzzy_thresholds(errors, true_labels, fuzzy_scores) | |
| # ── STEP 6: RECALIBRATE ENSEMBLE WEIGHTS ──────────────────────── | |
| cycle_metrics = self._compute_cycle_metrics(predictions, true_labels) | |
| self.detector.ensemble.recalibrate_weights(cycle_metrics) | |
| # ── STEP 7: PERSIST FEEDBACK ───────────────────────────────────── | |
| for i, text in enumerate(text_batch): | |
| self.store.save(text, predictions[i], true_labels[i], fuzzy_scores[i], errors[i]) | |
| # ── STEP 8: TREND CHECK ────────────────────────────────────────── | |
| self.metrics_history.append(cycle_metrics) | |
| self._check_improvement_trend() | |
| # ── STEP 9: GIT COMMIT ─────────────────────────────────────────── | |
| try: | |
| self.detector.git_manager.commit_cycle_results(self.cycle_count + 1, cycle_metrics) | |
| except Exception as e: | |
| self.logger.warning("Git commit failed (non-fatal): %s", e) | |
| self.cycle_count += 1 | |
| ens_f1 = cycle_metrics.get("ensemble", {}).get("f1", 0.0) | |
| self.logger.info( | |
| "=== Feedback Cycle #%d COMPLETE - Ensemble F1: %.4f ===", self.cycle_count, ens_f1 | |
| ) | |
| return cycle_metrics | |
| def _compute_errors(self, predictions, true_labels, judge_confidences) -> list: | |
| """ | |
| Compute per-sample error signals for all 3 models. | |
| Args: | |
| predictions (list): list of predict() result dicts | |
| true_labels (list): list of int ground truth labels | |
| judge_confidences (list): per-sample LLM judge confidence | |
| Returns: | |
| list of dicts: bert_error, tfidf_error, nb_error, | |
| total_error, is_wrong per sample | |
| """ | |
| errors = [] | |
| for pred, label, conf in zip(predictions, true_labels, judge_confidences): | |
| label = float(label) | |
| b = abs(pred["model_breakdown"]["bert"]["confidence"] - label) * conf | |
| t = abs(pred["model_breakdown"]["tfidf"]["confidence"] - label) * conf | |
| n = abs(pred["model_breakdown"]["naive_bayes"]["confidence"] - label) * conf | |
| total = b * 0.5 + t * 0.3 + n * 0.2 | |
| pred_lbl = 1 if pred["crisp_label"] == "misinformation" else 0 | |
| errors.append( | |
| { | |
| "bert_error": b, | |
| "tfidf_error": t, | |
| "nb_error": n, | |
| "total_error": total, | |
| "is_wrong": pred_lbl != int(label), | |
| } | |
| ) | |
| return errors | |
| def _update_fuzzy_thresholds(self, errors, true_labels, fuzzy_scores): | |
| """ | |
| Gradient-free hill climbing update of fuzzy threshold_suspicious. | |
| When the fuzzy engine is confidently wrong, nudge the suspicious | |
| threshold by +/- 0.02 toward reducing that error. | |
| Args: | |
| errors (list): error dicts from _compute_errors | |
| true_labels (list): int ground truth labels | |
| fuzzy_scores (list): float fuzzy scores for each sample | |
| """ | |
| wrong_fuzzy = [ | |
| fuzzy_scores[i] | |
| for i in range(len(errors)) | |
| if errors[i]["is_wrong"] and abs(fuzzy_scores[i] - true_labels[i]) > 0.3 | |
| ] | |
| if not wrong_fuzzy: | |
| return | |
| mean_err = float(np.mean(wrong_fuzzy)) | |
| delta = 0.02 * np.sign(mean_err - 0.5) | |
| old_thr = self.detector.fuzzy_engine.threshold_suspicious | |
| new_thr = float(np.clip(old_thr + delta, 0.35, 0.60)) | |
| self.detector.fuzzy_engine.threshold_suspicious = new_thr | |
| self.logger.info( | |
| "Fuzzy threshold_suspicious: %.3f -> %.3f (delta=%+.3f)", old_thr, new_thr, delta | |
| ) | |
| def _compute_cycle_metrics(self, predictions, true_labels) -> dict: | |
| """ | |
| Compute per-model and ensemble F1 + accuracy for this batch. | |
| Args: | |
| predictions (list): list of predict() result dicts | |
| true_labels (list): int ground truth labels | |
| Returns: | |
| dict: per-model metrics plus ensemble and current weights | |
| """ | |
| from sklearn.metrics import accuracy_score, f1_score | |
| y_true = [int(label) for label in true_labels] | |
| metrics = {} | |
| for name in ["bert", "tfidf", "naive_bayes"]: | |
| y_pred = [p["model_breakdown"][name]["label"] for p in predictions] | |
| metrics[name] = { | |
| "f1": float(f1_score(y_true, y_pred, zero_division=0)), | |
| "accuracy": float(accuracy_score(y_true, y_pred)), | |
| } | |
| ens_pred = [1 if p["crisp_label"] == "misinformation" else 0 for p in predictions] | |
| metrics["ensemble"] = { | |
| "f1": float(f1_score(y_true, ens_pred, zero_division=0)), | |
| "accuracy": float(accuracy_score(y_true, ens_pred)), | |
| } | |
| metrics["weights"] = dict(self.detector.ensemble.weights) | |
| return metrics | |
| def _check_improvement_trend(self): | |
| """ | |
| Alert and write flag file if F1 has been below threshold for | |
| consecutive_fail_limit cycles in a row. | |
| Writes RETRAIN_REQUIRED.flag to project root when triggered. | |
| """ | |
| if len(self.metrics_history) < self.fail_limit: | |
| return | |
| recent_f1 = [ | |
| m.get("ensemble", {}).get("f1", 1.0) for m in self.metrics_history[-self.fail_limit :] | |
| ] | |
| if not all(f < self.min_f1 for f in recent_f1): | |
| return | |
| self.logger.critical( | |
| "ACCURACY ALERT: Ensemble F1 below %.2f for %d consecutive " "cycles. Recent F1s: %s", | |
| self.min_f1, | |
| self.fail_limit, | |
| [("%.4f" % f) for f in recent_f1], | |
| ) | |
| flag = Path("RETRAIN_REQUIRED.flag") | |
| flag.write_text( | |
| "Full retrain required at cycle %d\n" | |
| "Recent F1 scores: %s\n" % (self.cycle_count, [("%.4f" % f) for f in recent_f1]) | |
| ) | |
| self.logger.critical("Wrote %s", flag) | |