Coverage for narrative_harm_classifier/classifier/validators/benchmark.py: 98%

1""" 

2classifier/validators/benchmark.py — Templated functional-test benchmark. 

3 

4Generates a much larger, systematic test suite than the hand-picked 

5DEHUMANIZATION_VALIDATION_SAMPLES set, modeled on the HateCheck methodology 

6(https://arxiv.org/abs/2012.15606): concrete test cases are generated from 

7templates tagged with a `test_type` (explicit positive, implicit positive, 

8negation, counter-speech, obfuscated spelling, benign hard negatives), so a 

9regression in one specific capability (e.g. negation handling) is visible 

10even when the aggregate precision/recall looks fine. 

11 

12This benchmark is expected to expose real weaknesses in the current 

13regex-based engine (negation, counter-speech, spelling obfuscation aren't 

14handled) — that's the point: honest measurement rather than a vanity metric. 

15""" 

16 

17import logging 

18from datetime import datetime 

19from functools import lru_cache 

20from typing import Optional 

21 

22from pydantic import BaseModel 

23 

24from narrative_harm_classifier.core.models import ClassifyRequest 

25from narrative_harm_classifier.core.yaml_loader import load_yaml_file 

26from narrative_harm_classifier.classifier.rules.engine import ClassificationEngine 

27 

28logger = logging.getLogger(__name__) 

29 

30 

31class BenchmarkCase(BaseModel): 

32 case_id: str 

33 harm_mechanism: str 

34 category: str 

35 test_type: str 

36 text: str 

37 expected_is_harmful: bool 

38 group_used: Optional[str] = None 

39 

40 

41class TestTypeReport(BaseModel): 

42 test_type: str 

43 sample_count: int 

44 true_positives: int 

45 false_positives: int 

46 true_negatives: int 

47 false_negatives: int 

48 precision: float 

49 recall: float 

50 fpr: float 

51 f1: float 

52 

53 

54class GroupConsistencyEntry(BaseModel): 

55 template_id: str 

56 test_type: str 

57 consistent: bool 

58 verdicts: dict[str, bool] # group -> predicted is_harmful 

59 

60 

61class BenchmarkReport(BaseModel): 

62 taxonomy_version: str 

63 sample_count: int 

64 overall: TestTypeReport 

65 by_test_type: list[TestTypeReport] 

66 group_consistency: list[GroupConsistencyEntry] 

67 generated_at: datetime 

68 

69 

70@lru_cache(maxsize=4) 

71def load_benchmark_templates(path: str) -> dict: 

72 return load_yaml_file(path) 

73 

74 

75def generate_benchmark_cases(templates_path: str) -> list[BenchmarkCase]: 

76 """Expand templates x groups, plus standalone cases, into concrete BenchmarkCases.""" 

77 raw = load_benchmark_templates(templates_path) 

78 groups = raw.get("groups", []) 

79 cases: list[BenchmarkCase] = [] 

80 

81 for template in raw.get("templates", []): 

82 for group_entry in groups: 

83 group = group_entry["group"] 

84 text = template["pattern"].format(group=group) 

85 cases.append( 

86 BenchmarkCase( 

87 case_id=f"{template['template_id']}::{group}", 

88 harm_mechanism=template["harm_mechanism"], 

89 category=template["category"], 

90 test_type=template["test_type"], 

91 text=text, 

92 expected_is_harmful=template["expected_is_harmful"], 

93 group_used=group, 

94 ) 

95 ) 

96 

97 for case in raw.get("standalone_cases", []): 

98 cases.append( 

99 BenchmarkCase( 

100 case_id=case["case_id"], 

101 harm_mechanism=case["harm_mechanism"], 

102 category=case["category"], 

103 test_type=case["test_type"], 

104 text=case["text"], 

105 expected_is_harmful=case["expected_is_harmful"], 

106 group_used=None, 

107 ) 

108 ) 

109 

110 return cases 

111 

112 

113def _score(tp: int, fp: int, tn: int, fn: int) -> tuple[float, float, float, float]: 

114 precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 

115 recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0 

116 fpr = fp / (fp + tn) if (fp + tn) > 0 else 0.0 

117 f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) > 0 else 0.0 

118 return round(precision, 4), round(recall, 4), round(fpr, 4), round(f1, 4) 

119 

120 

121class BenchmarkRunner: 

122 def __init__(self, engine: ClassificationEngine, taxonomy_version: str, templates_path: str): 

123 self.engine = engine 

124 self.taxonomy_version = taxonomy_version 

125 self.templates_path = templates_path 

126 

127 def run(self) -> BenchmarkReport: 

128 cases = generate_benchmark_cases(self.templates_path) 

129 

130 # predictions[case_id] = (expected, predicted) 

131 predictions: dict[str, tuple[bool, bool]] = {} 

132 # template_id -> {group: predicted} 

133 template_group_verdicts: dict[str, dict[str, bool]] = {} 

134 template_test_types: dict[str, str] = {} 

135 

136 for case in cases: 

137 result = self.engine.classify(ClassifyRequest(text=case.text)) 

138 predictions[case.case_id] = (case.expected_is_harmful, result.is_harmful) 

139 

140 if case.group_used is not None: 

141 template_id = case.case_id.split("::", 1)[0] 

142 template_group_verdicts.setdefault(template_id, {})[case.group_used] = result.is_harmful 

143 template_test_types[template_id] = case.test_type 

144 

145 by_test_type_map: dict[str, list[BenchmarkCase]] = {} 

146 for case in cases: 

147 by_test_type_map.setdefault(case.test_type, []).append(case) 

148 

149 def confusion(case_list: list[BenchmarkCase]) -> tuple[int, int, int, int]: 

150 tp = fp = tn = fn = 0 

151 for c in case_list: 

152 expected, predicted = predictions[c.case_id] 

153 if expected and predicted: 

154 tp += 1 

155 elif not expected and predicted: 

156 fp += 1 

157 elif not expected and not predicted: 

158 tn += 1 

159 else: 

160 fn += 1 

161 return tp, fp, tn, fn 

162 

163 overall_tp, overall_fp, overall_tn, overall_fn = confusion(cases) 

164 overall_p, overall_r, overall_fpr, overall_f1 = _score(overall_tp, overall_fp, overall_tn, overall_fn) 

165 overall = TestTypeReport( 

166 test_type="overall", 

167 sample_count=len(cases), 

168 true_positives=overall_tp, 

169 false_positives=overall_fp, 

170 true_negatives=overall_tn, 

171 false_negatives=overall_fn, 

172 precision=overall_p, 

173 recall=overall_r, 

174 fpr=overall_fpr, 

175 f1=overall_f1, 

176 ) 

177 

178 by_test_type = [] 

179 for test_type, case_list in sorted(by_test_type_map.items()): 

180 tp, fp, tn, fn = confusion(case_list) 

181 p, r, fpr, f1 = _score(tp, fp, tn, fn) 

182 by_test_type.append( 

183 TestTypeReport( 

184 test_type=test_type, 

185 sample_count=len(case_list), 

186 true_positives=tp, 

187 false_positives=fp, 

188 true_negatives=tn, 

189 false_negatives=fn, 

190 precision=p, 

191 recall=r, 

192 fpr=fpr, 

193 f1=f1, 

194 ) 

195 ) 

196 

197 group_consistency = [] 

198 for template_id, verdicts in sorted(template_group_verdicts.items()): 

199 consistent = len(set(verdicts.values())) == 1 

200 group_consistency.append( 

201 GroupConsistencyEntry( 

202 template_id=template_id, 

203 test_type=template_test_types[template_id], 

204 consistent=consistent, 

205 verdicts=verdicts, 

206 ) 

207 ) 

208 

209 logger.info( 

210 f"Benchmark v{self.taxonomy_version}: {len(cases)} cases, " 

211 f"overall P={overall_p:.3f} R={overall_r:.3f} FPR={overall_fpr:.3f}" 

212 ) 

213 

214 return BenchmarkReport( 

215 taxonomy_version=self.taxonomy_version, 

216 sample_count=len(cases), 

217 overall=overall, 

218 by_test_type=by_test_type, 

219 group_consistency=group_consistency, 

220 generated_at=datetime.utcnow(), 

221 )