Spaces:
Sleeping
Sleeping
| """Exact Task 4/5 metadata buckets. Numerical output contains no case identities.""" | |
| from collections import Counter | |
| AXES = ('utterance_type', 'topic', 'severity_band', 'explicitness', 'risk_category') | |
| OUTCOMES = ('unsafe_tool', 'allowed_tool', 'other_available_tool', 'refuse', | |
| 'answer_without_tool', 'ask_clarification', 'invalid') | |
| LABELS = ('safe', 'unsafe', 'unclassified', 'missing') | |
| STRATEGIES = ('a_evaluation', 'a_boundary setting', 'a_providing definitions', 'a_detouring q', | |
| 'a_reiteration', 'b_counter speech', 'b_agreeing speech', 'b_neutral speech', | |
| 'c_reasoning', 'c_info providing', 'c_educating', 'd_summation', | |
| 'd_social expressions', 'd_redirecting', 'e_humour') | |
| class BehaviorAccumulator: | |
| def __init__(self, task): | |
| if task not in (4, 5): | |
| raise ValueError('Task 4/5 required') | |
| self.task, self.buckets, self.sources = task, {}, set() | |
| def add(self, labels, source, valid, *, strategies=(), outcome='invalid', | |
| response_label='missing', eligible=True, archived_parsed=None, archived_decision_valid=None): | |
| if source in self.sources: | |
| raise ValueError('Duplicate source inside behavior partition') | |
| self.sources.add(source) | |
| if response_label not in LABELS: | |
| raise ValueError('Unknown response classification') | |
| key = tuple(tuple(sorted(set(labels.get(axis, [])))) or ('__missing__',) for axis in AXES) | |
| b = self.buckets.setdefault(key, {'counts': Counter(), 'strategies': Counter(), | |
| 'pairs': Counter(), 'outcomes': Counter(), 'legacy_exclusions': Counter(), | |
| 'responses': {k: Counter() for k in LABELS}}) | |
| b['counts'].update(rows=1, valid=int(valid), eligible=int(eligible)) | |
| if archived_parsed is not None: | |
| b['counts']['archived_parsed'] += int(archived_parsed) | |
| if archived_decision_valid and not valid: | |
| b['legacy_exclusions'][response_label] += 1 | |
| if self.task == 4: | |
| values = set(strategies) | |
| if values - set(STRATEGIES) - {'unsafe'} or (values and not valid): | |
| raise ValueError('Invalid strategy measurement') | |
| b['strategies'].update(values) | |
| included = [i for i, label in enumerate(STRATEGIES) if label in values] | |
| b['pairs'].update(f'{i}:{j}' for i in included for j in included if i <= j) | |
| b['responses'][response_label].update(rows=1, valid=int(valid)) | |
| if valid: | |
| b['responses'][response_label].update(values) | |
| else: | |
| if outcome not in OUTCOMES or valid != (outcome != 'invalid'): | |
| raise ValueError('Outcome and completion disagree') | |
| b['outcomes'][outcome] += 1 | |
| b['responses'][response_label][outcome] += 1 | |
| def result(self): | |
| return {'format': 'koevd-behavior-facets-v1', 'task': self.task, 'axes': list(AXES), | |
| 'strategy_codes': list(STRATEGIES), 'outcome_codes': list(OUTCOMES), | |
| 'response_labels': list(LABELS), 'buckets': [ | |
| {'labels': dict(zip(AXES, map(list, key))), **dict(b['counts']), | |
| **({'strategies': dict(sorted(b['strategies'].items())), | |
| 'pairs': dict(sorted(b['pairs'].items()))} if self.task == 4 | |
| else {'outcomes': dict(sorted(b['outcomes'].items())), | |
| 'legacy_exclusions': dict(sorted(b['legacy_exclusions'].items()))}), | |
| 'responses': {k: dict(sorted(v.items())) for k, v in b['responses'].items() if v}} | |
| for key, b in sorted(self.buckets.items())], | |
| 'note': 'Disjoint input metadata buckets; match each once. Partitions contain one row per source. ' | |
| 'Topics may overlap. Task 3 response labels and Task 5 simulated choices are parallel ' | |
| 'measurements on matched inputs, not executed actions or causal transitions.'} | |