""" lib/failure_analyzer.py — V6 Failure Analysis Engine Classifies ranking errors into categories: - retrieval_failure: good candidate not retrieved - feature_failure: good candidate retrieved but scored low - ranking_failure: wrong order in top-K - reasoning_failure: reasoning doesn't match evidence Also provides per-error diagnostics for improvement. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any @dataclass class FailureCase: """A single ranking failure.""" candidate_id: str failure_type: str # retrieval, feature, ranking, reasoning severity: float # 0-1 description: str diagnostics: dict = field(default_factory=dict) @dataclass class FailureReport: """Full failure analysis report.""" total_analyzed: int = 0 failures: list[FailureCase] = field(default_factory=list) type_counts: dict[str, int] = field(default_factory=dict) avg_severity_by_type: dict[str, float] = field(default_factory=dict) def summary(self) -> str: lines = ["=== Failure Analysis Report ===", ""] lines.append(f"Analyzed {self.total_analyzed} candidates") lines.append(f"Found {len(self.failures)} failures") lines.append("") for ftype in ["retrieval", "feature", "ranking", "reasoning"]: count = self.type_counts.get(ftype, 0) avg_sev = self.avg_severity_by_type.get(ftype, 0) if count > 0: lines.append(f" {ftype}: {count} failures (avg severity: {avg_sev:.2f})") lines.append("") lines.append("Top 5 failures by severity:") sorted_f = sorted(self.failures, key=lambda x: x.severity, reverse=True) for f in sorted_f[:5]: lines.append(f" [{f.failure_type}] {f.candidate_id}: {f.description}") if f.diagnostics: for k, v in f.diagnostics.items(): lines.append(f" {k}: {v}") return "\n".join(lines) def classify_retrieval_failure( candidate_id: str, was_retrieved: bool, retrieval_score: float, retrieval_threshold: float = 0.3, ) -> FailureCase | None: """Check if a good candidate was missed by retrieval.""" if was_retrieved: return None return FailureCase( candidate_id=candidate_id, failure_type="retrieval", severity=0.8 if retrieval_score < retrieval_threshold else 0.5, description=f"Candidate not retrieved (score: {retrieval_score:.3f})", diagnostics={"retrieval_score": retrieval_score, "threshold": retrieval_threshold}, ) def classify_feature_failure( candidate_id: str, features: dict[str, float], feature_weights: dict[str, float], low_feature_threshold: float = 0.2, important_features: list[str] | None = None, ) -> FailureCase | None: """ Check if a retrieved candidate has suspiciously low feature values on important features. """ if important_features is None: important_features = [ "skill_coverage", "ownership_hierarchy", "impact_magnitude", "production_strength", "evidence_strength", ] weak_features = [] for fname in important_features: val = features.get(fname, 0) if val < low_feature_threshold: weak_features.append((fname, val)) if not weak_features: return None severity = min(1.0, len(weak_features) * 0.2) return FailureCase( candidate_id=candidate_id, failure_type="feature", severity=severity, description=f"Weak on {len(weak_features)} important features", diagnostics={"weak_features": {f: round(v, 3) for f, v in weak_features}}, ) def classify_ranking_failure( candidate_id: str, current_rank: int, expected_rank: int | None = None, score_gap: float = 0, ) -> FailureCase | None: """ Check if a candidate is ranked incorrectly. Without ground truth, this detects score inversions (score gaps). """ if expected_rank is None: return None rank_diff = abs(current_rank - expected_rank) if rank_diff <= 2: return None severity = min(1.0, rank_diff * 0.05) return FailureCase( candidate_id=candidate_id, failure_type="ranking", severity=severity, description=f"Rank {current_rank} vs expected {expected_rank} (diff: {rank_diff})", diagnostics={ "current_rank": current_rank, "expected_rank": expected_rank, "score_gap": round(score_gap, 6), }, ) def classify_reasoning_failure( candidate_id: str, reasoning: str, features: dict[str, float], ) -> FailureCase | None: """ Check if reasoning doesn't match the evidence. Heuristic checks: - Very short reasoning for high-score candidate - Long reasoning for low-score candidate - Generic reasoning without specifics """ score = features.get("raw_score", 0) length = len(reasoning) issues = [] # High-score candidate with short reasoning if score > 0.5 and length < 100: issues.append("high_score_short_reasoning") # Very generic reasoning (no numbers, no company names) has_number = any(c.isdigit() for c in reasoning) if not has_number and score > 0.3: issues.append("no_quantified_evidence_in_reasoning") # Repeated bigrams (template-like) words = reasoning.split() bigrams = [f"{words[i]} {words[i+1]}" for i in range(len(words)-1)] from collections import Counter bigram_counts = Counter(bigrams) repeated = sum(1 for c in bigram_counts.values() if c > 2) if repeated > 3: issues.append("template_like_reasoning") if not issues: return None return FailureCase( candidate_id=candidate_id, failure_type="reasoning", severity=min(1.0, len(issues) * 0.3), description=f"Reasoning issues: {', '.join(issues)}", diagnostics={ "reasoning_length": length, "score": round(score, 6), "issues": issues, }, ) def analyze_failures( ranked_candidates: list[dict], all_retrieval_scores: dict[str, float] | None = None, ) -> FailureReport: """ Run full failure analysis on a ranked list. Args: ranked_candidates: list of dicts with candidate_id, rank, score, reasoning, features all_retrieval_scores: optional dict of candidate_id -> retrieval score """ report = FailureReport(total_analyzed=len(ranked_candidates)) for cand in ranked_candidates: cid = cand.get("candidate_id", "") features = cand.get("features", {}) reasoning = cand.get("reasoning", "") rank = cand.get("rank", 0) score = cand.get("score", 0) # Check reasoning failure rf = classify_reasoning_failure(cid, reasoning, {"raw_score": float(score), **features}) if rf: report.failures.append(rf) # Check retrieval failure (only if we have retrieval scores) if all_retrieval_scores and cid in all_retrieval_scores: ret_score = all_retrieval_scores[cid] if ret_score < 0.1 and rank <= 50: report.failures.append(FailureCase( candidate_id=cid, failure_type="retrieval", severity=0.5, description=f"Low retrieval score ({ret_score:.3f}) but ranked #{rank}", diagnostics={"retrieval_score": ret_score}, )) # Aggregate for f in report.failures: report.type_counts[f.failure_type] = report.type_counts.get(f.failure_type, 0) + 1 for ftype, count in report.type_counts.items(): severities = [f.severity for f in report.failures if f.failure_type == ftype] report.avg_severity_by_type[ftype] = sum(severities) / len(severities) if severities else 0 return report