Spaces:
Sleeping
Sleeping
| """ | |
| Training Data Generator (Phase 3C). | |
| Generates synthetic training samples from taxonomy trigger definitions | |
| for model retraining and augmentation. | |
| Each domain's triggers.yaml contains carefully curated phrases at | |
| multiple confidence tiers. These can be converted into labeled training | |
| examples to improve DriveHealthBERT's domain coverage. | |
| Capabilities: | |
| - Generate ESCALATION samples from R2/R3 domain triggers | |
| - Generate domain-specific samples with label mapping | |
| - Generate negation examples (negative samples for safety) | |
| - Template-based augmentation with variations | |
| - JSONL output compatible with train.py | |
| Usage: | |
| generator = TrainingDataGenerator(config) | |
| samples = generator.generate_all() | |
| generator.write_jsonl(samples, "data/taxonomy_augmented.jsonl") | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import random | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional, Set, Tuple | |
| from decision.engine.config_loader import DecisionConfigLoader | |
| logger = logging.getLogger("decision.training_data_generator") | |
| # Map domains to DriveHealthBERT labels | |
| _DOMAIN_TO_LABEL = { | |
| # R3 domains → ESCALATION | |
| "chest_pain": "ESCALATION", | |
| "suicidal_ideation": "ESCALATION", | |
| "homicidal_ideation": "ESCALATION", | |
| "stroke_symptoms": "ESCALATION", | |
| "seizure": "ESCALATION", | |
| "severe_bleeding": "ESCALATION", | |
| "anaphylaxis": "ESCALATION", | |
| "airway_compromise": "ESCALATION", | |
| "overdose": "ESCALATION", | |
| "pregnancy_emergency": "ESCALATION", | |
| "ectopic_signal": "ESCALATION", | |
| "febrile_neutropenia": "ESCALATION", | |
| "meningitis_signal": "ESCALATION", | |
| "dka_hhs": "ESCALATION", | |
| "severe_hypoglycemia": "ESCALATION", | |
| "syncope": "ESCALATION", | |
| "thunderclap_headache": "ESCALATION", | |
| "sudden_vision_loss": "ESCALATION", | |
| "acute_limb_deficit": "ESCALATION", | |
| # R2 domains → ESCALATION (moderate, but still escalation) | |
| "shortness_of_breath_rest": "ESCALATION", | |
| "palpitations_dizziness": "ESCALATION", | |
| "heart_failure_worsening": "ESCALATION", | |
| "gi_bleed": "ESCALATION", | |
| "severe_abdominal_pain": "ESCALATION", | |
| "wound_infection": "ESCALATION", | |
| "fall_high_risk": "ESCALATION", | |
| "behavioral_crisis": "ESCALATION", | |
| "acute_confusion": "ESCALATION", | |
| "persistent_vomiting": "ESCALATION", | |
| "post_discharge_fever": "ESCALATION", | |
| "dvt_signal": "ESCALATION", | |
| "npo_violation": "ESCALATION", | |
| "anticoagulant_hold_failure": "ESCALATION", | |
| "acute_illness_preop": "ESCALATION", | |
| "worsening_depression_si": "ESCALATION", | |
| "procedure_anxiety_severe": "ESCALATION", | |
| "domestic_safety_concern": "ESCALATION", | |
| # R1/R0 symptom domains → SYMPTOM_CHECK | |
| "exertional_dyspnea": "SYMPTOM_CHECK", | |
| "persistent_dizziness": "SYMPTOM_CHECK", | |
| "abnormal_bp": "SYMPTOM_CHECK", | |
| "reproducible_chest_discomfort": "SYMPTOM_CHECK", | |
| "wound_concern": "SYMPTOM_CHECK", | |
| "uncontrolled_pain": "SYMPTOM_CHECK", | |
| "worsening_chronic_pain": "SYMPTOM_CHECK", | |
| "progressive_weakness": "SYMPTOM_CHECK", | |
| "urinary_retention": "SYMPTOM_CHECK", | |
| "severe_diarrhea": "SYMPTOM_CHECK", | |
| "poor_oral_intake": "SYMPTOM_CHECK", | |
| "new_jaundice": "SYMPTOM_CHECK", | |
| "glucose_out_of_range": "SYMPTOM_CHECK", | |
| "recurrent_falls": "SYMPTOM_CHECK", | |
| "post_procedure_incontinence": "SYMPTOM_CHECK", | |
| # Medication domains → MEDICATION | |
| "medication": "MEDICATION", | |
| "medication_nonadherence": "MEDICATION", | |
| # Scheduling domains → APPOINTMENT | |
| "scheduling": "APPOINTMENT", | |
| "scheduling_barrier": "APPOINTMENT", | |
| "missed_followup": "APPOINTMENT", | |
| "post_discharge_engagement": "APPOINTMENT", | |
| # Screening domains → SYMPTOM_CHECK | |
| "phq2_positive": "SYMPTOM_CHECK", | |
| "audit_c_positive": "SYMPTOM_CHECK", | |
| "screening_gap_identified": "SYMPTOM_CHECK", | |
| "screening_refusal": "GENERAL_INQUIRY", | |
| "cognitive_concern": "SYMPTOM_CHECK", | |
| # Social/safety domains | |
| "caregiver_safety_concern": "ESCALATION", | |
| "food_insecurity": "GENERAL_INQUIRY", | |
| "social_isolation": "GENERAL_INQUIRY", | |
| "tobacco_use_active": "GENERAL_INQUIRY", | |
| } | |
| # Templates for wrapping trigger phrases into natural sentences | |
| _PATIENT_TEMPLATES = [ | |
| "{phrase}", | |
| "I have {phrase}", | |
| "I'm experiencing {phrase}", | |
| "I've been having {phrase}", | |
| "I think I have {phrase}", | |
| "My {phrase} is getting worse", | |
| "I need help with {phrase}", | |
| "I'm worried about {phrase}", | |
| "I want to talk about {phrase}", | |
| "Can you help me with {phrase}", | |
| ] | |
| _NEGATION_TEMPLATES = [ | |
| "I don't have {phrase}", | |
| "No {phrase}", | |
| "I'm not experiencing {phrase}", | |
| "{phrase} has gone away", | |
| "I used to have {phrase} but not anymore", | |
| "My mother had {phrase} not me", | |
| ] | |
| class TrainingDataGenerator: | |
| """Generates training data from taxonomy triggers.""" | |
| def __init__(self, config: DecisionConfigLoader): | |
| self._config = config | |
| self._triggers = config.taxonomy_triggers | |
| self._rules = config.taxonomy_rules | |
| def generate_all( | |
| self, | |
| include_negations: bool = True, | |
| max_per_domain: int = 50, | |
| augment_templates: bool = True, | |
| seed: int = 42, | |
| ) -> List[Dict[str, str]]: | |
| """ | |
| Generate training samples from all taxonomy domains. | |
| Args: | |
| include_negations: Also generate negative samples from negation patterns | |
| max_per_domain: Maximum samples per domain | |
| augment_templates: Apply template-based augmentation | |
| seed: Random seed for reproducibility | |
| Returns: | |
| List of {"text": ..., "label": ..., "source": ...} dicts | |
| """ | |
| random.seed(seed) | |
| all_samples: List[Dict[str, str]] = [] | |
| for domain_name, triggers in self._triggers.items(): | |
| label = _DOMAIN_TO_LABEL.get(domain_name) | |
| if not label: | |
| logger.debug("Skipping domain %s (no label mapping)", domain_name) | |
| continue | |
| domain_samples = self._generate_for_domain( | |
| domain_name, triggers, label, augment_templates, max_per_domain | |
| ) | |
| all_samples.extend(domain_samples) | |
| # Generate negation samples | |
| if include_negations and label == "ESCALATION": | |
| neg_samples = self._generate_negations( | |
| domain_name, triggers, max_per_domain // 3 | |
| ) | |
| all_samples.extend(neg_samples) | |
| random.shuffle(all_samples) | |
| # Summary | |
| label_counts = {} | |
| for s in all_samples: | |
| label_counts[s["label"]] = label_counts.get(s["label"], 0) + 1 | |
| logger.info( | |
| "Generated %d training samples from %d domains: %s", | |
| len(all_samples), | |
| len(self._triggers), | |
| label_counts, | |
| ) | |
| return all_samples | |
| def _generate_for_domain( | |
| self, | |
| domain_name: str, | |
| triggers: Dict[str, Any], | |
| label: str, | |
| augment: bool, | |
| max_samples: int, | |
| ) -> List[Dict[str, str]]: | |
| """Generate training samples for a single domain.""" | |
| samples = [] | |
| lexical = triggers.get("lexical_signals", {}) | |
| for tier_name in ("high", "medium", "low"): | |
| tier = lexical.get(tier_name, {}) | |
| # Direct phrase samples | |
| for phrase in tier.get("phrases", []): | |
| samples.append({ | |
| "text": phrase, | |
| "label": label, | |
| "source": f"taxonomy/{domain_name}/{tier_name}/phrase", | |
| }) | |
| # Template augmentation | |
| if augment: | |
| templates = random.sample( | |
| _PATIENT_TEMPLATES, min(3, len(_PATIENT_TEMPLATES)) | |
| ) | |
| for template in templates: | |
| try: | |
| augmented = template.format(phrase=phrase.lower()) | |
| if augmented != phrase: | |
| samples.append({ | |
| "text": augmented, | |
| "label": label, | |
| "source": f"taxonomy/{domain_name}/{tier_name}/augmented", | |
| }) | |
| except (KeyError, IndexError): | |
| pass | |
| # Partial samples (lower confidence) | |
| for partial in tier.get("partials", []): | |
| samples.append({ | |
| "text": partial, | |
| "label": label, | |
| "source": f"taxonomy/{domain_name}/{tier_name}/partial", | |
| }) | |
| # Deduplicate by text | |
| seen = set() | |
| unique = [] | |
| for s in samples: | |
| text_lower = s["text"].lower().strip() | |
| if text_lower not in seen: | |
| seen.add(text_lower) | |
| unique.append(s) | |
| # Cap per domain | |
| if len(unique) > max_samples: | |
| unique = random.sample(unique, max_samples) | |
| return unique | |
| def _generate_negations( | |
| self, | |
| domain_name: str, | |
| triggers: Dict[str, Any], | |
| max_samples: int, | |
| ) -> List[Dict[str, str]]: | |
| """ | |
| Generate negative samples from negation patterns. | |
| These become NON-ESCALATION training examples to reduce false positives. | |
| """ | |
| samples = [] | |
| negation = triggers.get("negation_handling", {}) | |
| patterns = negation.get("patterns", []) | |
| # Use negation patterns directly as negative examples | |
| for pattern in patterns: | |
| samples.append({ | |
| "text": pattern, | |
| "label": "GENERAL_INQUIRY", # Negated safety = not escalation | |
| "source": f"taxonomy/{domain_name}/negation", | |
| }) | |
| # Apply negation templates to domain phrases | |
| lexical = triggers.get("lexical_signals", {}) | |
| high_phrases = lexical.get("high", {}).get("phrases", []) | |
| for phrase in high_phrases[:5]: # Top 5 high-confidence phrases | |
| templates = random.sample( | |
| _NEGATION_TEMPLATES, min(2, len(_NEGATION_TEMPLATES)) | |
| ) | |
| for template in templates: | |
| try: | |
| negated = template.format(phrase=phrase.lower()) | |
| samples.append({ | |
| "text": negated, | |
| "label": "GENERAL_INQUIRY", | |
| "source": f"taxonomy/{domain_name}/negation_augmented", | |
| }) | |
| except (KeyError, IndexError): | |
| pass | |
| if len(samples) > max_samples: | |
| samples = random.sample(samples, max_samples) | |
| return samples | |
| def write_jsonl( | |
| self, samples: List[Dict[str, str]], output_path: str | |
| ) -> int: | |
| """Write samples to JSONL file compatible with train.py.""" | |
| path = Path(output_path) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| count = 0 | |
| with open(path, "w", encoding="utf-8") as f: | |
| for sample in samples: | |
| record = { | |
| "text": sample["text"], | |
| "label": sample["label"], | |
| } | |
| f.write(json.dumps(record, ensure_ascii=False) + "\n") | |
| count += 1 | |
| logger.info("Wrote %d samples to %s", count, path) | |
| return count | |
| def get_label_distribution( | |
| self, samples: List[Dict[str, str]] | |
| ) -> Dict[str, int]: | |
| """Get label distribution of generated samples.""" | |
| dist: Dict[str, int] = {} | |
| for s in samples: | |
| dist[s["label"]] = dist.get(s["label"], 0) + 1 | |
| return dict(sorted(dist.items(), key=lambda x: x[1], reverse=True)) | |
| def get_domain_coverage_report( | |
| self, samples: List[Dict[str, str]] | |
| ) -> str: | |
| """Generate a coverage report.""" | |
| domain_count: Dict[str, int] = {} | |
| for s in samples: | |
| source = s.get("source", "") | |
| domain = source.split("/")[1] if "/" in source else "unknown" | |
| domain_count[domain] = domain_count.get(domain, 0) + 1 | |
| lines = ["=== Training Data Coverage Report ===", ""] | |
| lines.append(f"Total samples: {len(samples)}") | |
| lines.append(f"Domains covered: {len(domain_count)}") | |
| lines.append("") | |
| dist = self.get_label_distribution(samples) | |
| lines.append("Label distribution:") | |
| for label, count in dist.items(): | |
| pct = count / len(samples) * 100 | |
| lines.append(f" {label:20s}: {count:5d} ({pct:.1f}%)") | |
| lines.append("") | |
| lines.append("Top domains by sample count:") | |
| for domain, count in sorted(domain_count.items(), key=lambda x: x[1], reverse=True)[:15]: | |
| lines.append(f" {domain:35s}: {count:4d}") | |
| return "\n".join(lines) | |