| """ |
| scripts/06_build_eval_sets.py |
| Build held-out evaluation sets from the synthetic data: |
| - eval_pictogram_to_sentence.jsonl (500 examples, 100 per persona) |
| - eval_audio_disambig.jsonl (200 examples with audio paths) |
| - eval_visual_ground.jsonl (150 examples with image paths) |
| - eval_adapter_specificity.jsonl (100 fixed prompts, cross-persona) |
| """ |
|
|
| import json |
| import logging |
| import random |
| from pathlib import Path |
|
|
| import pandas as pd |
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") |
| log = logging.getLogger(__name__) |
|
|
| SYNTH_DIR = Path("data/synth") |
| EVAL_DIR = Path("data/eval") |
| PROCESSED_DIR = Path("data/processed") |
| RANDOM_SEED = 42 |
|
|
|
|
| def load_persona_jsonl(persona_id: str) -> list[dict]: |
| path = SYNTH_DIR / f"persona_{persona_id}.jsonl" |
| if not path.exists(): |
| log.warning(f"No synth data for persona: {persona_id}") |
| return [] |
| with open(path, encoding="utf-8") as f: |
| return [json.loads(l) for l in f if l.strip()] |
|
|
|
|
| def build_pictogram_to_sentence_eval(n_per_persona: int = 100) -> list[dict]: |
| """Sample n examples per persona as held-out eval set.""" |
| all_eval = [] |
| personas = ["ananya", "arjun", "priya", "rohan", "zara"] |
|
|
| for pid in personas: |
| records = load_persona_jsonl(pid) |
| if not records: |
| log.warning(f"Skipping {pid} — no data") |
| continue |
| random.shuffle(records) |
| sampled = records[:n_per_persona] |
| for r in sampled: |
| all_eval.append({ |
| "persona_id": pid, |
| "pictogram_sequence": r.get("seq", []), |
| "context": r.get("context", ""), |
| "target_sentence": r.get("target", ""), |
| "alternatives": r.get("alternatives", []), |
| }) |
| log.info(f" {pid}: sampled {len(sampled)} eval examples") |
|
|
| return all_eval |
|
|
|
|
| def build_audio_disambig_eval(n: int = 200) -> list[dict]: |
| """ |
| Build audio disambiguation eval set. |
| Each example: (pictogram_seq + audio_path) → target_sentence. |
| Audio paths reference files in data/processed/audio/. |
| """ |
| audio_manifest = Path("data/processed/audio_manifest.jsonl") |
| if not audio_manifest.exists(): |
| log.warning("Audio manifest not found. Run 03_download_audio_corpora.py first.") |
| |
| return [ |
| { |
| "persona_id": "arjun", |
| "pictogram_sequence": ["water", "want"], |
| "audio_path": "PLACEHOLDER", |
| "audio_transcript_hint": "pani", |
| "target_sentence": "Mujhe pani chahida", |
| "intent_class": "request_food_drink", |
| } |
| ] |
|
|
| with open(audio_manifest) as f: |
| audio_clips = [json.loads(l) for l in f if l.strip()] |
|
|
| random.shuffle(audio_clips) |
| personas = ["ananya", "arjun", "priya", "rohan", "zara"] |
|
|
| records = [] |
| for i, clip in enumerate(audio_clips[:n]): |
| pid = personas[i % len(personas)] |
| persona_data = load_persona_jsonl(pid) |
| if not persona_data: |
| continue |
| base = random.choice(persona_data) |
| records.append({ |
| "persona_id": pid, |
| "pictogram_sequence": base.get("seq", []), |
| "audio_path": clip["path"], |
| "audio_transcript_hint": clip.get("transcript", ""), |
| "target_sentence": base.get("target", ""), |
| "intent_class": _classify_intent(base.get("seq", [])), |
| }) |
|
|
| return records |
|
|
|
|
| def build_visual_ground_eval(n: int = 150) -> list[dict]: |
| """ |
| Build visual grounding eval set. |
| Each example: (home_photo + pictogram_seq) → sentence with visual reference. |
| Uses placeholder image paths since real home photos aren't available. |
| """ |
| personas = ["ananya", "arjun", "priya", "rohan", "zara"] |
| records = [] |
| for i in range(n): |
| pid = personas[i % len(personas)] |
| persona_data = load_persona_jsonl(pid) |
| if not persona_data: |
| continue |
| base = random.choice(persona_data) |
| records.append({ |
| "persona_id": pid, |
| "pictogram_sequence": base.get("seq", []), |
| "image_path": f"data/raw/home_photos/{pid}/sample_{i:03d}.jpg", |
| "target_sentence": base.get("target", ""), |
| "visual_reference": f"object in image #{i % 5 + 1}", |
| }) |
| return records |
|
|
|
|
| def build_adapter_specificity_eval() -> list[dict]: |
| """ |
| Build the fixed prompt set used for adapter-specificity evaluation. |
| 100 prompts, all same prompts applied to each persona adapter. |
| """ |
| |
| UNIVERSAL_PROMPTS = [ |
| {"seq": ["mother", "want"], "context": "morning"}, |
| {"seq": ["water", "give"], "context": "afternoon"}, |
| {"seq": ["pain", "help"], "context": "general"}, |
| {"seq": ["food", "more"], "context": "lunch"}, |
| {"seq": ["sleep", "want"], "context": "evening"}, |
| {"seq": ["play", "want"], "context": "afternoon"}, |
| {"seq": ["toilet", "go"], "context": "general"}, |
| {"seq": ["doctor", "no"], "context": "hospital"}, |
| {"seq": ["happy", "today"], "context": "morning"}, |
| {"seq": ["love", "mother"], "context": "bedtime"}, |
| {"seq": ["school", "go"], "context": "morning"}, |
| {"seq": ["friend", "come"], "context": "afternoon"}, |
| {"seq": ["TV", "watch"], "context": "evening"}, |
| {"seq": ["cold", "milk", "want"], "context": "afternoon"}, |
| {"seq": ["medicine", "no", "want"], "context": "evening"}, |
| {"seq": ["tired", "bed"], "context": "night"}, |
| {"seq": ["hot", "stop"], "context": "bathing"}, |
| {"seq": ["book", "read"], "context": "bedtime"}, |
| {"seq": ["father", "come", "now"], "context": "general"}, |
| {"seq": ["outside", "go", "want"], "context": "afternoon"}, |
| ] * 5 |
|
|
| records = [] |
| for i, p in enumerate(UNIVERSAL_PROMPTS[:100]): |
| records.append({ |
| "prompt_id": f"spec_{i:03d}", |
| "pictogram_sequence": p["seq"], |
| "context": p["context"], |
| "expected_persona_styles": { |
| "ananya": "Tamil/English style, short", |
| "arjun": "Punjabi/Hindi/English mix", |
| "priya": "Bengali/English, very short (4yo)", |
| "rohan": "Hindi/English, older child register", |
| "zara": "Marathi/English, 7yo register", |
| }, |
| }) |
| return records |
|
|
|
|
| def _classify_intent(seq: list[str]) -> str: |
| """Simple rule-based intent classifier for audio eval labels.""" |
| seq_lower = [s.lower() for s in seq] |
| if any(w in seq_lower for w in ["food", "eat", "drink", "water", "milk", "more", "hungry"]): |
| return "request_food_drink" |
| if any(w in seq_lower for w in ["pain", "hurt", "stop", "help"]): |
| return "report_discomfort" |
| if any(w in seq_lower for w in ["play", "fun", "friend", "TV", "outside"]): |
| return "request_activity" |
| if any(w in seq_lower for w in ["love", "happy", "hug", "kiss"]): |
| return "social_emotional" |
| if any(w in seq_lower for w in ["toilet", "bath", "sleep"]): |
| return "personal_care" |
| return "general" |
|
|
|
|
| def save_jsonl(records: list[dict], path: Path): |
| path.parent.mkdir(parents=True, exist_ok=True) |
| with open(path, "w", encoding="utf-8") as f: |
| for r in records: |
| f.write(json.dumps(r, ensure_ascii=False) + "\n") |
| log.info(f"Saved {len(records)} records → {path}") |
|
|
|
|
| def main(): |
| random.seed(RANDOM_SEED) |
| EVAL_DIR.mkdir(parents=True, exist_ok=True) |
|
|
| log.info("Building eval_pictogram_to_sentence...") |
| p2s = build_pictogram_to_sentence_eval(100) |
| save_jsonl(p2s, EVAL_DIR / "eval_pictogram_to_sentence.jsonl") |
|
|
| log.info("Building eval_audio_disambig...") |
| audio = build_audio_disambig_eval(200) |
| save_jsonl(audio, EVAL_DIR / "eval_audio_disambig.jsonl") |
|
|
| log.info("Building eval_visual_ground...") |
| visual = build_visual_ground_eval(150) |
| save_jsonl(visual, EVAL_DIR / "eval_visual_ground.jsonl") |
|
|
| log.info("Building eval_adapter_specificity...") |
| spec = build_adapter_specificity_eval() |
| save_jsonl(spec, EVAL_DIR / "eval_adapter_specificity.jsonl") |
|
|
| log.info("\nEval sets complete:") |
| log.info(f" pictogram_to_sentence: {len(p2s)}") |
| log.info(f" audio_disambig: {len(audio)}") |
| log.info(f" visual_ground: {len(visual)}") |
| log.info(f" adapter_specificity: {len(spec)}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
