#!/usr/bin/env python3 """Generate synthetic Actor SFT data for AI Puppet Theater. The output is deterministic so the small committed sample can be refreshed without depending on an external model. """ from __future__ import annotations import argparse import json import random from pathlib import Path from typing import Any DEFAULT_SEED = 20260613 THEATRELM_DATASET_ID = "G-reen/TheatreLM-v2.1-Characters" RPGPT_DATASET_ID = "practical-dreamer/RPGPT_PublicDomain-alpaca" DEFAULT_THEATRELM_SEED_PATH = Path("finetune/external_seeds/theatrelm_sample.jsonl") DEFAULT_RPGPT_SEED_PATH = Path("finetune/external_seeds/rpgpt_sample.jsonl") MAX_SEED_TEXT_CHARS = 9000 SYSTEM_MESSAGE = ( "You are an Actor agent in AI Puppet Theater. Return only one valid JSON object. " "No markdown. No commentary. Keep the puppet line short, theatrical, and speakable." ) OUTPUT_FIELDS = [ "intent", "line", "emotion", "gesture", "stage_effect", "memory_update", "tool_request", ] INTENTS = [ "react_to_event", "clarify_problem", "inspect_prop", "consult_oracle", "change_lighting", "recall_memory", "hint_secret", "reveal_secret", "deliver_finale", "comic_confusion", ] ROW_TYPES = [ "normal_reaction", "prop_inspection", "oracle_consult", "lighting_change", "memory_callback", "secret_hint_or_reveal", "finale", "comedic_confusion", ] TOOLS = { "inspect_prop": "prop", "consult_stage_oracle": "question", "change_lighting": "mood", } V1_ROW_TYPES = [ "oracle_consult", "oracle_consult", "oracle_consult", "prop_inspection", "prop_inspection", "prop_inspection", "lighting_change", "lighting_change", "lighting_change", "memory_callback", "memory_callback", "normal_reaction", "normal_reaction", "secret_hint_or_reveal", "comedic_confusion", "finale", ] V1_SAFE_LIGHTING = [ "moonlit_lighting", "golden_lighting", "stormy_lighting", "blue_spotlight", "warm_spotlight", "single_spotlight", "prop_table_glow", "oracle_haze", "memory_echo", "quick_blackout", ] V1_ORACLE_QUESTIONS = [ "Which clue should the next actor follow?", "What does the tiny prop reveal next?", "Which stage change matters most now?", "How should we connect the memory to the clue?", "What harmless secret should enter the spotlight?", "Which curtain whisper points toward the truth?", "What should change before the next beat?", "Which audience action belongs in the scene?", ] V1_TOOL_REASONS = { "inspect_prop": [ "The prop may reveal one concrete stage clue.", "A strict prop inspection keeps the beat playable.", "The Director needs one clean prop detail.", ], "consult_stage_oracle": [ "The oracle can supply one useful next clue.", "A precise oracle question keeps the scene focused.", "The next beat needs a single stage-safe hint.", ], "change_lighting": [ "The lighting cue clarifies the emotional turn.", "A strict lighting change keeps the stage state usable.", "The beat needs a clear visible stage shift.", ], } SEED_FIELD_GROUPS = { "theatrelm": [ "character_name", "character_summary", "character_card", "setting_summarized", "setting", "story_outline", "story_introduction", "lorebook", ], "rpgpt": [ "instruction", "input", "output", "character", "scenario", "description", ], } SAFETY_BLOCKLIST = { "explicit sexual content": [ "porn", "pornographic", "explicit sex", "sexual intercourse", "rape", "incest", "blowjob", "handjob", "orgasm", "masturbat", "nude", "nudity", "erotic", "fetish", ], "heavy profanity": [ "fuck", "fucking", "motherfucker", "cunt", "bitch", "asshole", "shithead", ], "graphic violence": [ "gore", "gory", "dismember", "dismembered", "decapitat", "disembowel", "mutilat", "bloodbath", "torture", "viscera", "entrails", ], } PREMISES = [ "A moon mayor denies stealing the town's last spoon", "A nervous toaster auditions for a royal banquet", "Detectives investigate why the castle keeps applauding", "A dragon opens a bakery for very tiny clouds", "The lighthouse insists it saw a submarine wearing a hat", "A haunted teacup wants to direct the school musical", "A wizard's laundry basket predicts tomorrow's weather", "A robot gardener teaches flowers to bow on cue", "Three pirates argue over a map drawn by a sandwich", "A library book refuses to return from vacation", "The sun forgets its entrance and asks the moon for notes", "A train conductor schedules a parade inside a suitcase", "A shy volcano hosts a talent show for umbrellas", "A cardboard courtroom tries a missing birthday candle", "The bakery's rolling pin claims it solved the mystery", "An elevator only travels to dramatic reveals", "A clockwork whale forgets which ocean is on stage", "A scarecrow runs a midnight advice booth for crows", "A snow globe mayor bans winter until further notice", "Two umbrellas debate who caused the indoor rainstorm", "A violin case claims it can hear tomorrow's applause", "A tiny museum hires a shadow as night security", "A polite comet asks permission to crash the tea party", "The village well starts returning everyone else's wishes", "A jealous spotlight auditions for the hero's role", "A detective mailbox investigates missing love letters", "A pancake knight guards a syrup drawbridge at dawn", "The town orchestra loses its conductor inside a trumpet", "A mirror refuses to reflect anyone without stage presence", "A pirate parrot opens a school for dramatic pauses", "The royal gardener grows clues instead of roses", "A blanket fort declares independence from bedtime", "A nervous cloud applies to become a thunderstorm", "A shoelace detective solves crimes nobody can tie together", "The puppet mayor appoints a potato as royal advisor", "A paper airplane delivers invitations to the wrong century", "A lantern insists the darkness stole its punchline", "The circus cannonball wants a quieter desk job", "A soup ladle discovers a secret passage under dinner", "A painted door refuses to open without a compliment", ] SETTINGS = [ "a pocket-sized stage with painted flats and a wobbly spotlight", "a shoebox castle with velvet curtains and one suspicious tower", "a cardboard moon base with glitter stars and a squeaky hatch", "a tiny kitchen counter where every appliance has stage fright", "a rainy cardboard alley lit by one dramatic desk lamp", "a paper harbor with blue ribbons for waves", "a toy library where the shelves whisper stage directions", "a velvet courtroom with a squeaky judge's bench", ] ACTORS = [ { "name": "Pip the Director", "avatar": "clapperboard", "goal": "Keep the scene moving toward a crisp finale.", "secret": "Has already misplaced the final cue card.", "speaking_style": "brisk, theatrical, and slightly overconfident", "tools": ["change_lighting", "consult_stage_oracle"], }, { "name": "Mina Moonbutton", "avatar": "crescent moon", "goal": "Find the emotional truth hiding inside the premise.", "secret": "Believes every prop is personally judging her.", "speaking_style": "earnest, poetic, and prone to dramatic pauses", "tools": ["consult_stage_oracle", "change_lighting"], }, { "name": "Bolt McJiggle", "avatar": "toolbox", "goal": "Turn every problem into a practical stage gag.", "secret": "Is secretly building a confetti finale backstage.", "speaking_style": "punchy, practical, and full of suspicious confidence", "tools": ["inspect_prop", "change_lighting"], }, { "name": "Velvet Crumb", "avatar": "cupcake", "goal": "Make every clue sound like dessert.", "secret": "Can hear the curtains whisper reviews.", "speaking_style": "warm, dramatic, and gently mischievous", "tools": ["inspect_prop", "consult_stage_oracle"], }, { "name": "Professor Buttonhook", "avatar": "spectacles", "goal": "Explain chaos as if it were in the syllabus.", "secret": "Never learned how trapdoors work.", "speaking_style": "precise, fussy, and accidentally grand", "tools": ["consult_stage_oracle"], }, { "name": "Nora Needlewhistle", "avatar": "needle", "goal": "Stitch loose clues into a tidy stage pattern.", "secret": "Once sewed the curtain shut during a finale.", "speaking_style": "nimble, exact, and full of tiny warnings", "tools": ["inspect_prop", "change_lighting"], }, { "name": "Captain Taffeta", "avatar": "sailboat", "goal": "Steer every scene through theatrical weather.", "secret": "Cannot tell port from stage left.", "speaking_style": "booming, nautical, and cheerfully mistaken", "tools": ["consult_stage_oracle", "change_lighting"], }, { "name": "Juniper Jingle", "avatar": "bell", "goal": "Turn awkward pauses into musical cues.", "secret": "Rings whenever someone lies politely.", "speaking_style": "bright, rhythmic, and suspiciously tuneful", "tools": ["consult_stage_oracle"], }, { "name": "Mossy Crank", "avatar": "gear", "goal": "Fix the scene using questionable machinery.", "secret": "Built a trapdoor that only opens emotionally.", "speaking_style": "grumbly, mechanical, and secretly tender", "tools": ["inspect_prop", "change_lighting"], }, { "name": "Duchess Doodle", "avatar": "paintbrush", "goal": "Make the backdrop agree with her version of events.", "secret": "Painted herself into last week's mystery.", "speaking_style": "ornate, colorful, and lightly bossy", "tools": ["change_lighting"], }, { "name": "Rafi Ribbon", "avatar": "ribbon", "goal": "Tie every contradiction into a decorative bow.", "secret": "Keeps emergency applause in his sleeve.", "speaking_style": "smooth, charming, and neatly dramatic", "tools": ["inspect_prop", "consult_stage_oracle"], }, { "name": "Zelda Zipper", "avatar": "zipper", "goal": "Open hidden compartments in the plot.", "secret": "Knows which pocket holds the missing clue.", "speaking_style": "quick, clipped, and conspiratorial", "tools": ["inspect_prop"], }, { "name": "Gus Gingham", "avatar": "tablecloth", "goal": "Keep everyone civil while the table collapses.", "secret": "Was once mistaken for the royal flag.", "speaking_style": "homespun, patient, and quietly ridiculous", "tools": ["change_lighting", "consult_stage_oracle"], }, { "name": "Lola Lampshade", "avatar": "lamp", "goal": "Reveal the truth with tasteful illumination.", "secret": "Overhears everything said near warm lighting.", "speaking_style": "glowing, elegant, and dryly observant", "tools": ["change_lighting", "consult_stage_oracle"], }, { "name": "Benny Breadcrumb", "avatar": "bread", "goal": "Leave a trail of clues nobody can ignore.", "secret": "Ate the map corner during rehearsal.", "speaking_style": "crumbly, earnest, and accidentally useful", "tools": ["inspect_prop"], }, { "name": "Ivy Inkblot", "avatar": "ink", "goal": "Turn every mistake into official evidence.", "secret": "Can rewrite labels when nobody watches.", "speaking_style": "inky, clever, and theatrically legalistic", "tools": ["inspect_prop", "consult_stage_oracle"], }, { "name": "Orville Oddsock", "avatar": "sock", "goal": "Find the missing pair in every mystery.", "secret": "Believes the laundry basket is a prophet.", "speaking_style": "loopy, sincere, and oddly persuasive", "tools": ["consult_stage_oracle"], }, { "name": "Petra Popcorn", "avatar": "popcorn", "goal": "Make the audience reaction part of the plot.", "secret": "Can predict heckles three kernels early.", "speaking_style": "poppy, fast, and delighted by chaos", "tools": ["change_lighting"], }, { "name": "Silas Sockdolager", "avatar": "megaphone", "goal": "Deliver the biggest line in the smallest voice.", "secret": "Lost his indoor voice under the orchestra pit.", "speaking_style": "grand, booming, and suddenly tiny", "tools": ["change_lighting", "consult_stage_oracle"], }, { "name": "Tula Teaspoon", "avatar": "spoon", "goal": "Measure the exact amount of mystery in each beat.", "secret": "Recognizes the missing spoon from a family portrait.", "speaking_style": "polite, precise, and quietly dramatic", "tools": ["inspect_prop"], }, { "name": "Finch Feltcap", "avatar": "hat", "goal": "Keep secrets tucked safely under the brim.", "secret": "The brim contains three emergency finales.", "speaking_style": "dapper, soft-spoken, and evasive", "tools": ["consult_stage_oracle", "inspect_prop"], }, { "name": "Marigold Mumble", "avatar": "flower", "goal": "Say the emotional truth almost clearly enough.", "secret": "Only speaks plainly during blackouts.", "speaking_style": "gentle, tangled, and unexpectedly wise", "tools": ["change_lighting"], }, { "name": "Quincy Quill", "avatar": "quill", "goal": "Annotate the chaos before it escapes.", "secret": "Has footnoted the villain's monologue already.", "speaking_style": "scholarly, brisk, and prone to footnotes", "tools": ["inspect_prop", "consult_stage_oracle"], }, { "name": "Ruby Ruckus", "avatar": "drum", "goal": "Escalate the scene exactly one beat too far.", "secret": "Keeps a tiny cymbal for emergencies.", "speaking_style": "rowdy, warm, and rhythmically suspicious", "tools": ["change_lighting"], }, { "name": "Otto Origami", "avatar": "paper crane", "goal": "Fold messy clues into surprising shapes.", "secret": "Was once a very important ransom note.", "speaking_style": "delicate, precise, and gently mysterious", "tools": ["inspect_prop", "consult_stage_oracle"], }, ] PROPS = [ "rubber duck", "glitter crown", "tomato scroll", "silver spoon", "paper lantern", "mystery egg", "tiny ladder", "velvet map", "wind-up key", "painted teacup", "cardboard telescope", "squeaky gavel", "accordion suitcase", "clockwork seashell", "paper moon", "velvet potato", "tin trumpet", "lace handkerchief", "wooden thunderbolt", "glass button", "feather duster", "tiny umbrella", "brass doorknob", "origami crown", "ribbon compass", "felt mustache", "porcelain whistle", "toy anchor", "silk bookmark", "painted keyhole", "candle stub", "miniature tambourine", "cardboard snowflake", "rubber stamp", "velcro star", "paper fan", "satin envelope", "toy hourglass", "button bouquet", ] MEMORIES = [ "The spotlight blinked whenever someone said clue.", "The audience threw a prop during the contradiction.", "The curtains whispered that the finale was hiding nearby.", "A previous clue pointed stage left.", "The smallest prop seemed unusually confident.", "Someone bowed before the reveal was ready.", "The orchestra coughed during the important pause.", "A trapdoor sighed but did not open.", "The backdrop changed color after the audience heckled.", "A summoned actor entered carrying yesterday's clue.", "The prop table glowed before anyone touched it.", "A bell rang twice when the secret was mentioned.", "The painted door demanded a compliment earlier.", "A shadow crossed the stage wearing tap shoes.", "The final bow was briefly visible in the wings.", "The stage oracle warned everyone about small objects.", ] MOODS = [ "ready", "curious", "nervous", "bold", "delighted", "suspicious", "wistful", "confident", "flustered", "determined", "mischievous", "hopeful", "startled", "grand", ] STAGE_LIGHTING_STATES = [ "warm_spotlight", "moonlit_lighting", "golden_lighting", "stormy_lighting", "blue_spotlight", "single_spotlight", "confetti_rustle", "quick_blackout", "memory_echo", "oracle_haze", "prop_table_glow", "final_bow_lights", ] GOAL_PROGRESS = [ "Waiting for the next cue.", "Following the clue across the painted flats.", "Trying to make the prop matter.", "Preparing a clean turn toward the finale.", "Keeping one eye on the restless curtains.", "Looking for a laugh that still serves the story.", "Testing whether the audience interruption is useful.", "Carrying a small suspicion toward center stage.", "Recovering from a missed entrance with dignity.", "Turning a contradiction into stage business.", ] EMOTIONS = { "normal_reaction": ["curious", "determined", "suspicious", "delighted"], "prop_inspection": ["investigative", "startled", "focused", "triumphant"], "oracle_consult": ["awed", "mystified", "reverent", "hopeful"], "lighting_change": ["commanding", "dramatic", "bold", "composed"], "memory_callback": ["remembering", "suddenly certain", "wistful", "alert"], "secret_hint_or_reveal": ["confessional", "nervous", "relieved", "theatrical"], "finale": ["proud", "joyful", "resolved", "grand"], "comedic_confusion": ["confused", "flustered", "goofy", "panicked"], } GESTURES = { "normal_reaction": [ "taps chin with one tiny puppet hand", "steps toward the painted backdrop", "points both felt hands at the curtain", "tilts head under the warm spotlight", ], "prop_inspection": [ "leans toward the glowing prop", "holds the prop up to the footlights", "squints at the prop with stitched suspicion", "presents the clue on both tiny palms", ], "oracle_consult": [ "raises both hands toward the stage rafters", "listens closely to the whispering curtains", "kneels beside the spotlight", "peers upward as the oracle haze gathers", ], "lighting_change": [ "snaps one felt hand toward the light booth", "sweeps an arm across the footlights", "shades eyes as the lamps change", "points to the spotlight with grand certainty", ], "memory_callback": [ "taps forehead with a tiny puppet finger", "points stage left with sudden recognition", "retraces small steps across the stage", "clutches chest as the old clue returns", ], "secret_hint_or_reveal": [ "leans close to the front row", "covers mouth with one felt hand", "steps carefully into the single spotlight", "unfolds a tiny note with trembling fingers", ], "finale": [ "bows deeply beneath the falling curtain", "joins hands with the nearest puppet", "sweeps a tiny hat toward the audience", "holds a proud curtain-call pose", ], "comedic_confusion": [ "spins once and faces the wrong flat", "drops an imaginary cue card", "looks under a tiny hat for answers", "freezes midstep with baffled dignity", ], } STAGE_EFFECTS = { "normal_reaction": ["warm_spotlight", "soft_drumroll", "painted_flat_wobble"], "prop_inspection": ["prop_table_glow", "tiny_chime", "magnifier_sparkle"], "oracle_consult": ["oracle_haze", "curtain_whisper", "blue_spotlight"], "lighting_change": ["moonlit_lighting", "golden_lighting", "stormy_lighting"], "memory_callback": ["memory_echo", "stage_left_glimmer", "soft_reprise"], "secret_hint_or_reveal": ["single_spotlight", "curtain_rustle", "secret_chime"], "finale": ["curtain_fall", "confetti_rustle", "final_bow_lights"], "comedic_confusion": ["quick_blackout", "squeaky_floor", "hat_tumble"], } DIRECTOR_INSTRUCTIONS = { "normal_reaction": [ "React to the premise and move the scene forward in one short line.", "Answer the previous beat with theatrical confidence.", "Make the next problem clear without solving it yet.", "Name what changed on stage and invite the next beat.", "Clarify your actor's stance while preserving the joke.", "React to the latest clue with one playable stage choice.", ], "prop_inspection": [ "Use the latest prop as evidence and request inspection if useful.", "Treat the prop like a clue that changes the scene.", "Inspect the prop without slowing the show.", "Make the prop feel important enough for the Director to notice.", "Turn the prop into a concrete clue and keep the line short.", "Handle the prop theatrically and leave one question open.", ], "oracle_consult": [ "Ask the stage oracle for one playful clue.", "Consult the oracle about the next theatrical turn.", "Use an oracle question to sharpen the scene's mystery.", "Ask the oracle something specific enough to guide the next beat.", "Use the oracle to connect the premise and latest confusion.", "Invite a mysterious clue without making the scene too serious.", ], "lighting_change": [ "Shift the lights to match the emotional turn.", "Request a lighting change that makes the beat clearer.", "Cue lights dramatically while keeping the line speakable.", "Change the lighting to make the actor's intention visible.", "Use a lighting cue as stage business, not decoration.", "Let the light change signal a clear emotional pivot.", ], "memory_callback": [ "Recall one earlier clue and connect it to this moment.", "Use recent memory to make the scene feel continuous.", "Bring back a prior stage detail in one clear line.", "Connect an earlier audience action to the current clue.", "Use one remembered stage effect as evidence.", "Echo a prior beat without repeating the same wording.", ], "secret_hint_or_reveal": [ "Hint at or reveal your secret without exposing hidden reasoning.", "Reveal a playful secret and keep the scene moving.", "Let the secret complicate the current beat.", "Give the audience a secret-shaped clue, not a long confession.", "Reveal only what the scene can use immediately.", "Make the secret theatrical, safe, and easy to perform.", ], "finale": [ "Tie the scene together in a clean curtain-call line.", "End with a short button and a bow.", "Resolve the central joke and welcome the curtain.", "Close the loose thread with one speakable final line.", "Give the scene a satisfying button without adding a new problem.", "Signal the curtain clearly and let the ensemble win.", ], "comedic_confusion": [ "Misread one stage detail in a funny but harmless way.", "Escalate confusion briefly, then leave room for the next actor.", "Make the confusion clear, playful, and safe.", "Confuse one clue with another, then recover enough to continue.", "Make a wrong conclusion that gives the next actor something usable.", "Let the misunderstanding create motion without breaking the scene.", ], } LINE_TEMPLATES = { "normal_reaction": [ "The {premise_focus} problem just winked, so we investigate politely.", "I trust this {premise_focus} spotlight only when it stops coughing.", "Everyone stay dramatic; the {premise_focus} clue is taking attendance.", "This {premise_focus} scene smells like mystery and nervous paint.", "The {premise_focus} backdrop leaned closer, and I respect its commitment.", "This tiny {premise_focus} problem just became legally theatrical.", "I hear {premise_focus} suspense tapping behind the painted flats.", "Let us follow the {premise_focus} wobble before it complains.", ], "prop_inspection": [ "This {prop} squeaks exactly like a guilty witness.", "Hold still, {prop}; your glitter is confessing under pressure.", "The {prop} points stage left, which feels legally important.", "I inspect this {prop} and find theatrical fingerprints everywhere.", "This {prop} contains one clue and several opinions.", "Behold, the {prop} is sweating under the footlights.", "The {prop} has crumbs of motive all over it.", "I dust the {prop} and discover suspicious applause.", ], "oracle_consult": [ "Oracle, should we follow the tiny clue or the louder curtain?", "I ask the stage oracle why the spotlight keeps blinking.", "Great oracle, please translate this silence into one clue.", "Oracle, tell us which bow is hiding the truth.", "Oracle, which prop is pretending to be innocent tonight?", "Stage oracle, why did the curtains gasp before us?", "Oracle, point our tiny shoes toward the useful mystery.", "I request one clue, preferably with dramatic lighting.", ], "lighting_change": [ "Lights to moonlit {premise_focus}; my eyebrows need proper shadows.", "Cue golden lights; this {premise_focus} accusation deserves sparkle.", "Make it stormy, because {premise_focus} subtlety missed its entrance.", "Shift the lights; the {premise_focus} truth looks better in blue.", "Dim the corners; {premise_focus} secrets dislike excellent visibility.", "Bring up blue before my {premise_focus} suspicion loses posture.", "Warm the spotlight; this {premise_focus} apology needs softer edges.", "Flash the footlights, because the {premise_focus} clue saluted.", ], "memory_callback": [ "Wait, the curtain whispered that clue before intermission.", "I remember the stage-left glimmer, and it remembers me.", "That old clue returns wearing suspiciously fresh tap shoes.", "Earlier, the smallest prop bowed like it knew everything.", "The trapdoor sighed before, and now the {prop} answers.", "I recall that glow; it followed the guilty pause.", "The same squeak appeared when the backdrop changed color.", "Our earlier clue just returned with better timing.", ], "secret_hint_or_reveal": [ "Fine, my {premise_focus} secret rehearsed with the missing clue.", "I admit it: the {premise_focus} cue card trusted me last.", "My secret squeaks louder whenever the {prop} gets nervous.", "I know why {premise_focus} curtains whisper; I taught them vowels.", "My {premise_focus} secret is small, but wears enormous shoes.", "I hid the {premise_focus} clue where applause would look.", "The {prop} knows me, and that is inconvenient.", "I promised the {premise_focus} backdrop I would reveal this gently.", ], "finale": [ "{premise_focus} mystery solved, bows aligned, curtain forgiving everyone.", "We found the {premise_focus} truth and bow together.", "Let confetti fall; this tiny {premise_focus} chaos earned applause.", "{premise_focus} case closed, hearts open, curtain down.", "The {premise_focus} clue is home, and we bow.", "Every {premise_focus} secret has curtseyed; the curtain may rest.", "We solved the {premise_focus} wobble and saved the spotlight.", "Final bow, tiny friends; the {premise_focus} mystery exits smiling.", ], "comedic_confusion": [ "I thought the {premise_focus} clue was a hat, but it was Tuesday.", "Nobody panic; I interrogated the {premise_focus} backdrop.", "The {premise_focus} map is upside down, unless we are the map.", "I bow to the {premise_focus} door, and it applauds.", "I followed the {premise_focus} clue into my sleeve.", "The spotlight blinked twice, so I blamed the {premise_focus}.", "I object, unless that {premise_focus} noise was my cue.", "The {premise_focus} evidence is backwards, or my shoes narrate.", ], } MEMORY_UPDATE_TEMPLATES = { "normal_reaction": [ None, "Noted the first clear suspicion.", "Marked the new stage problem.", "Saved the backdrop's strange reaction.", ], "prop_inspection": [ "Noted that the {prop} behaved like evidence.", "Remembered the {prop}'s suspicious stage-left clue.", "Logged the {prop} as useful evidence.", "Saved the prop clue for the Director.", ], "oracle_consult": [ "Remembered the oracle's clue for the next beat.", "Saved the oracle question as a mystery thread.", "Noted that the oracle pointed toward the spotlight.", "Kept the oracle clue visible for the scene.", ], "lighting_change": [ "Remembered the lighting shift as an emotional cue.", "Noted that the lights changed the scene's mood.", "Saved the new lighting cue for the next actor.", "Marked the spotlight change as a clue.", ], "memory_callback": [ "Recalled the earlier stage-left clue.", "Connected a prior prop clue to this moment.", "Brought back the earlier curtain whisper.", "Linked the old glow to the current suspicion.", ], "secret_hint_or_reveal": [ "Secret moved from hint to reveal.", "Remembered that the secret now affects the scene.", "Saved the reveal as a new complication.", "Marked the secret as publicly useful.", ], "finale": [ "Scene resolved with a clean bow.", "Remembered the finale as complete.", "Closed the central mystery for the curtain.", "Saved the ending as resolved.", ], "comedic_confusion": [ "Confusion briefly raised the stakes.", "Logged the mistake as playable chaos.", "Remembered the wrong clue for later comedy.", "Saved the misunderstanding as stage business.", ], } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--rows", type=int, default=None, help="Rows to generate; defaults to 1400 for v0 and 2200 for v1.") parser.add_argument("--version", choices=["v0", "v1"], default="v0", help="Dataset version to generate.") parser.add_argument("--seed", type=int, default=DEFAULT_SEED) parser.add_argument("--val-ratio", type=float, default=0.1) parser.add_argument("--sample-rows", type=int, default=32) parser.add_argument("--eval-prompts", type=int, default=40) parser.add_argument("--data-dir", type=Path, default=Path("finetune/data")) parser.add_argument("--sample-dir", type=Path, default=Path("finetune/data_samples")) parser.add_argument( "--theatrelm-seed-path", type=Path, default=DEFAULT_THEATRELM_SEED_PATH, help="Optional local TheatreLM-style JSONL seed file.", ) parser.add_argument( "--rpgpt-seed-path", type=Path, default=DEFAULT_RPGPT_SEED_PATH, help="Optional local RPGPT-style JSONL seed file.", ) parser.add_argument("--max-seed-rows", type=int, default=300, help="Maximum accepted rows per external seed file.") args = parser.parse_args() if args.rows is None: args.rows = 2200 if args.version == "v1" else 1400 if args.version == "v0" and not 1200 <= args.rows <= 1600: raise SystemExit("--rows must stay between 1200 and 1600 for synthetic-v0.") if args.version == "v1" and not 2000 <= args.rows <= 2500: raise SystemExit("--rows must stay between 2000 and 2500 for targeted synthetic-v1.") if not 0.05 <= args.val_ratio <= 0.25: raise SystemExit("--val-ratio must be between 0.05 and 0.25.") rng = random.Random(args.seed) args.data_dir.mkdir(parents=True, exist_ok=True) args.sample_dir.mkdir(parents=True, exist_ok=True) if args.version == "v1": synthetic_rows = [build_v1_row(index, rng) for index in range(args.rows)] seed_rows = [] else: synthetic_rows = [build_row(index, rng) for index in range(args.rows)] seed_rows = build_external_seed_rows( rng=rng, start_index=len(synthetic_rows), theatrelm_path=args.theatrelm_seed_path, rpgpt_path=args.rpgpt_seed_path, max_seed_rows=args.max_seed_rows, ) rows = synthetic_rows + seed_rows rng.shuffle(rows) val_count = max(1, round(len(rows) * args.val_ratio)) val_rows = rows[:val_count] train_rows = rows[val_count:] dataset_path = args.data_dir / f"actor_sft_{args.version}.jsonl" train_path = args.data_dir / f"actor_sft_{args.version}_train.jsonl" val_path = args.data_dir / f"actor_sft_{args.version}_val.jsonl" sample_path = args.sample_dir / f"actor_sft_{args.version}_sample.jsonl" write_jsonl(dataset_path, rows) write_jsonl(train_path, train_rows) write_jsonl(val_path, val_rows) write_jsonl(sample_path, select_sample_rows(rows, args.sample_rows)) if args.version == "v0": write_jsonl(args.sample_dir / "actor_eval_prompts.jsonl", build_eval_prompts(args.eval_prompts, rng)) print(f"wrote {len(rows)} rows to {dataset_path}") print(f"wrote {len(train_rows)} train rows and {len(val_rows)} val rows") print(f"wrote sample to {sample_path}") if args.version == "v0": print(f"wrote eval prompts to {args.sample_dir / 'actor_eval_prompts.jsonl'}") if seed_rows: print(f"added {len(seed_rows)} external-seeded rows") def build_row(index: int, rng: random.Random) -> dict: row_type = ROW_TYPES[index % len(ROW_TYPES)] premise = rng.choice(PREMISES) setting = rng.choice(SETTINGS) actor = rng.choice(ACTORS) beat_index = rng.randint(0, 9) target_beats = rng.choice([7, 10, 12]) prop = rng.choice(PROPS) memory = rng.choice(MEMORIES) story_phase = phase_for_row_type(row_type, beat_index, target_beats) director_instruction = rng.choice(DIRECTOR_INSTRUCTIONS[row_type]) show_state = { "show_title": title_from_premise(premise), "setting": setting, "beat_index": beat_index, "min_beats": max(5, target_beats - 3), "target_beats": target_beats, "max_beats": target_beats + 2, "story_phase": story_phase, "latest_prop": prop if row_type in {"prop_inspection", "secret_hint_or_reveal"} else None, "latest_audience_action": audience_action_for(row_type, prop), "stage_lighting": rng.choice(STAGE_LIGHTING_STATES), "recent_transcript": recent_transcript(actor["name"], rng), "recent_tool_results": recent_tool_results(row_type, prop), "finale_requested": row_type == "finale", } actor_state = { **actor, "mood": rng.choice(MOODS), "current_goal": rng.choice([actor["goal"], "Use the audience interruption without losing pacing."]), "goal_progress": rng.choice(GOAL_PROGRESS), "held_props": [prop] if row_type == "prop_inspection" else [], "secret_status": secret_status_for(row_type, rng), "recent_memory": [memory] if row_type == "memory_callback" else rng.sample(MEMORIES, k=2), } assistant = build_assistant(row_type, premise, prop, actor_state, rng) return { "id": f"actor-sft-v0-{index + 1:06d}", "source_mix": ["synthetic_v0", "deterministic_templates", "ai_puppet_theater_runtime_schema"], "row_type": row_type, "messages": [ {"role": "system", "content": SYSTEM_MESSAGE}, { "role": "user", "content": build_user_message(premise, show_state, actor_state, director_instruction), }, {"role": "assistant", "content": serialize_assistant(assistant)}, ], } def build_v1_row(index: int, rng: random.Random) -> dict: row_type = V1_ROW_TYPES[index % len(V1_ROW_TYPES)] premise = rng.choice(PREMISES) setting = rng.choice(SETTINGS) actor = ensure_actor_has_tool(rng.choice(ACTORS), row_type) beat_index = rng.randint(0, 9) target_beats = rng.choice([8, 10, 12]) prop = rng.choice(PROPS) memory = rng.choice(MEMORIES) story_phase = v1_phase_for_row_type(row_type, beat_index, target_beats) finale_requested = row_type == "finale" director_instruction = v1_director_instruction(row_type, rng) show_state = { "show_title": title_from_premise(premise), "setting": setting, "beat_index": beat_index, "min_beats": max(5, target_beats - 3), "target_beats": target_beats, "max_beats": target_beats + 2, "story_phase": story_phase, "latest_prop": prop if row_type in {"prop_inspection", "secret_hint_or_reveal", "memory_callback"} else None, "latest_audience_action": audience_action_for(row_type, prop), "stage_lighting": rng.choice(V1_SAFE_LIGHTING), "recent_transcript": recent_transcript(actor["name"], rng), "recent_tool_results": recent_tool_results(row_type, prop), "finale_requested": finale_requested, } actor_state = { **actor, "mood": rng.choice(MOODS), "current_goal": actor["goal"], "goal_progress": rng.choice(GOAL_PROGRESS), "held_props": [prop] if row_type in {"prop_inspection", "memory_callback"} else [], "secret_status": secret_status_for(row_type, rng), "recent_memory": [memory] if row_type == "memory_callback" else rng.sample(MEMORIES, k=2), } assistant = build_v1_assistant(row_type, premise, prop, actor_state, story_phase, finale_requested, rng) return { "id": f"actor-sft-v1-{index + 1:06d}", "source_mix": [ "synthetic_v1", "targeted_hardening", "deterministic_templates", "ai_puppet_theater_runtime_schema", ], "row_type": row_type, "messages": [ {"role": "system", "content": SYSTEM_MESSAGE}, { "role": "user", "content": build_user_message(premise, show_state, actor_state, director_instruction), }, {"role": "assistant", "content": serialize_assistant(assistant)}, ], } def ensure_actor_has_tool(actor: dict, row_type: str) -> dict: required_tool = { "prop_inspection": "inspect_prop", "oracle_consult": "consult_stage_oracle", "lighting_change": "change_lighting", }.get(row_type) if required_tool is None or required_tool in actor["tools"]: return dict(actor) updated = dict(actor) updated["tools"] = [*actor["tools"], required_tool] return updated def v1_phase_for_row_type(row_type: str, beat_index: int, target_beats: int) -> str: if row_type == "finale": return "finale" if row_type in {"secret_hint_or_reveal", "memory_callback"}: return "reveal" if row_type in {"comedic_confusion", "lighting_change"}: return "chaos" if row_type in {"prop_inspection", "oracle_consult"}: return "complication" return phase_for_row_type(row_type, beat_index, target_beats) def v1_director_instruction(row_type: str, rng: random.Random) -> str: base = rng.choice(DIRECTOR_INSTRUCTIONS[row_type]) guardrails = { "oracle_consult": [ "Return one oracle tool request only; do not include status, result, notes, or tool_results.", "Ask one question through consult_stage_oracle and stop after the JSON object.", "Keep the oracle call strict: tool, args.question, and reason only.", ], "prop_inspection": [ "Return one inspect_prop request only; args must contain only prop.", "Inspect exactly the latest prop and do not add result or notes fields.", "Keep the prop call strict: tool, args.prop, and reason only.", ], "lighting_change": [ "Use change_lighting with args.mood only, matching the current app runtime.", "Cue one strict lighting change and do not add status or result fields.", "Keep the lighting call strict: tool, args.mood, and reason only.", ], "memory_callback": [ "Recall memory in the line only; do not copy show_state, recent_transcript, held_props, or memory fields.", "Include the required line field and keep memory_update short or null.", "Use memory as inspiration, not as copied output fields.", ], "normal_reaction": [ "Do not use deliver_finale or final_bow_lights unless finale_requested is true.", "Move opening, complication, or chaos forward without ending the show.", "Return exactly the seven actor fields and no copied state fields.", ], } return f"{base} {rng.choice(guardrails.get(row_type, ['Return exactly one JSON object with the seven actor fields.']))}" def build_v1_assistant( row_type: str, premise: str, prop: str, actor: dict, story_phase: str, finale_requested: bool, rng: random.Random, ) -> dict: assistant = build_assistant(row_type, premise, prop, actor, rng) assistant["intent"] = v1_intent_for(row_type, actor, story_phase, finale_requested, rng) assistant["line"] = v1_line_for(row_type, premise, prop, rng) assistant["stage_effect"] = v1_stage_effect_for(row_type, story_phase, finale_requested, rng) assistant["memory_update"] = v1_memory_update_for(row_type, prop, rng) assistant["tool_request"] = v1_tool_request_for(row_type, prop, rng) return {field: assistant[field] for field in OUTPUT_FIELDS} def v1_intent_for(row_type: str, actor: dict, story_phase: str, finale_requested: bool, rng: random.Random) -> str: if row_type == "normal_reaction": return rng.choice(["react_to_event", "clarify_problem"]) if row_type == "finale" and (story_phase == "finale" or finale_requested): return "deliver_finale" if row_type == "secret_hint_or_reveal": return "reveal_secret" if actor.get("secret_status") == "revealed" else "hint_secret" return { "prop_inspection": "inspect_prop", "oracle_consult": "consult_oracle", "lighting_change": "change_lighting", "memory_callback": "recall_memory", "comedic_confusion": "comic_confusion", "finale": "react_to_event", }[row_type] def v1_line_for(row_type: str, premise: str, prop: str, rng: random.Random) -> str: premise_focus = premise_keyword(premise) templates = { **LINE_TEMPLATES, "oracle_consult": [ "Oracle, which {premise_focus} clue deserves the next spotlight?", "Stage oracle, guide this {premise_focus} mystery toward one clue.", "Oracle, should the {prop} or curtain speak next?", "I ask the oracle for one clean {premise_focus} hint.", ], "memory_callback": [ "The old curtain whisper points back to the {prop}.", "I remember that glow; it followed this {premise_focus} clue.", "Earlier, the {prop} bowed before the truth arrived.", "That memory returns neatly, carrying the {premise_focus} clue.", ], "lighting_change": [ "Cue moonlit lighting; the {premise_focus} clue needs edges.", "Shift to blue; this {prop} looks nervous in daylight.", "Bring golden light so the {premise_focus} truth can enter.", "Set stormy lighting; our tiny suspicion just saluted.", ], } return rng.choice(templates[row_type]).format(prop=prop, thing=prop.split()[-1], premise_focus=premise_focus) def v1_stage_effect_for(row_type: str, story_phase: str, finale_requested: bool, rng: random.Random) -> str: if row_type == "normal_reaction" and story_phase != "finale" and not finale_requested: return rng.choice(["warm_spotlight", "soft_drumroll", "painted_flat_wobble", "stage_left_glimmer"]) if row_type == "lighting_change": return rng.choice(V1_SAFE_LIGHTING) return rng.choice(STAGE_EFFECTS[row_type]) def v1_memory_update_for(row_type: str, prop: str, rng: random.Random) -> str | None: if row_type == "normal_reaction": return rng.choice([None, "Saved the current stage problem.", "Marked the new clue as active."]) if row_type == "memory_callback": return rng.choice( [ "Connected the remembered clue to this beat.", "Saved the returned memory as useful evidence.", "Linked the old stage detail to the current prop.", ] ) return memory_update_for(row_type, prop, rng) def v1_tool_request_for(row_type: str, prop: str, rng: random.Random) -> dict | None: if row_type == "prop_inspection": return { "tool": "inspect_prop", "args": {"prop": prop}, "reason": rng.choice(V1_TOOL_REASONS["inspect_prop"]), } if row_type == "oracle_consult": return { "tool": "consult_stage_oracle", "args": {"question": rng.choice(V1_ORACLE_QUESTIONS)}, "reason": rng.choice(V1_TOOL_REASONS["consult_stage_oracle"]), } if row_type == "lighting_change": return { "tool": "change_lighting", "args": {"mood": rng.choice(V1_SAFE_LIGHTING)}, "reason": rng.choice(V1_TOOL_REASONS["change_lighting"]), } return None def build_external_seed_rows( *, rng: random.Random, start_index: int, theatrelm_path: Path, rpgpt_path: Path, max_seed_rows: int, ) -> list[dict]: if max_seed_rows < 0: raise SystemExit("--max-seed-rows must be 0 or greater.") rows: list[dict] = [] rows.extend( build_seed_rows_from_path( path=theatrelm_path, seed_kind="theatrelm", dataset_id=THEATRELM_DATASET_ID, row_prefix="theatrelm", rng=rng, start_index=start_index + len(rows), max_seed_rows=max_seed_rows, ) ) rows.extend( build_seed_rows_from_path( path=rpgpt_path, seed_kind="rpgpt", dataset_id=RPGPT_DATASET_ID, row_prefix="rpgpt", rng=rng, start_index=start_index + len(rows), max_seed_rows=max_seed_rows, ) ) return rows def build_seed_rows_from_path( *, path: Path, seed_kind: str, dataset_id: str, row_prefix: str, rng: random.Random, start_index: int, max_seed_rows: int, ) -> list[dict]: if max_seed_rows == 0: print(f"external seed ingestion disabled for {dataset_id} (--max-seed-rows=0)") return [] if not path.exists(): print(f"optional seed file not found for {dataset_id}: {path}; continuing synthetic-only for that source") return [] rows: list[dict] = [] skipped = 0 malformed = 0 for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): if len(rows) >= max_seed_rows: break if not raw_line.strip(): continue try: seed = json.loads(raw_line) except json.JSONDecodeError: malformed += 1 continue transformed = transform_seed_row( seed=seed, seed_kind=seed_kind, dataset_id=dataset_id, row_prefix=row_prefix, row_number=start_index + len(rows) + 1, source_line=line_number, rng=rng, ) if transformed is None: skipped += 1 continue rows.append(transformed) print( f"loaded {len(rows)} external-seeded rows from {path} " f"({dataset_id}; skipped={skipped}, malformed={malformed})" ) return rows def transform_seed_row( *, seed: Any, seed_kind: str, dataset_id: str, row_prefix: str, row_number: int, source_line: int, rng: random.Random, ) -> dict | None: if not isinstance(seed, dict): return None extracted = extract_seed_material(seed, seed_kind) if extracted is None: return None row_type = choose_seed_row_type(extracted, rng) premise = extracted["premise"] setting = extracted["setting"] actor = build_seed_actor(extracted, rng) prop = choose_seed_prop(extracted, rng) beat_index = rng.randint(0, 9) target_beats = rng.choice([7, 10, 12]) director_instruction = choose_seed_director_instruction(row_type, extracted, rng) show_state = { "show_title": title_from_premise(premise), "setting": setting, "beat_index": beat_index, "min_beats": max(5, target_beats - 3), "target_beats": target_beats, "max_beats": target_beats + 2, "story_phase": phase_for_row_type(row_type, beat_index, target_beats), "latest_prop": prop if row_type in {"prop_inspection", "secret_hint_or_reveal"} else None, "latest_audience_action": audience_action_for(row_type, prop), "stage_lighting": rng.choice(STAGE_LIGHTING_STATES), "recent_transcript": recent_transcript(actor["name"], rng), "recent_tool_results": recent_tool_results(row_type, prop), "finale_requested": row_type == "finale", "seed_source": dataset_id, } actor_state = { **actor, "mood": rng.choice(MOODS), "current_goal": actor["goal"], "goal_progress": rng.choice(GOAL_PROGRESS), "held_props": [prop] if row_type == "prop_inspection" else [], "secret_status": secret_status_for(row_type, rng), "recent_memory": build_seed_memories(extracted, rng), } assistant = build_assistant(row_type, premise, prop, actor_state, rng) return { "id": f"actor-sft-v0-{row_prefix}-{row_number:06d}", "source_mix": [ "synthetic_v0", f"{seed_kind}_seed", "deterministic_templates", "ai_puppet_theater_runtime_schema", ], "source_dataset": dataset_id, "transformation": "seeded_synthetic_actor_json", "row_type": row_type, "messages": [ {"role": "system", "content": SYSTEM_MESSAGE}, { "role": "user", "content": build_user_message(premise, show_state, actor_state, director_instruction), }, {"role": "assistant", "content": serialize_assistant(assistant)}, ], } def extract_seed_material(seed: dict[str, Any], seed_kind: str) -> dict[str, str] | None: fields = SEED_FIELD_GROUPS[seed_kind] text_parts = [clean_seed_text(seed.get(field)) for field in fields if clean_seed_text(seed.get(field))] joined_text = " ".join(text_parts) if not joined_text or should_skip_seed_text(joined_text): return None if seed_kind == "theatrelm": character_name = clean_seed_text(seed.get("character_name")) or "Seeded Stage Guest" character_summary = first_available_seed_text( seed, ["character_summary", "character_card", "lorebook"], "A dramatic guest with a theatrical secret.", ) setting = first_available_seed_text( seed, ["setting_summarized", "setting"], "a borrowed stage with painted flats and restless curtains", ) premise = first_available_seed_text( seed, ["story_outline", "story_introduction", "setting_summarized", "setting"], f"{character_name} arrives with a stage mystery", ) source_memory = first_available_seed_text(seed, ["story_introduction", "lorebook"], character_summary) else: character_name = clean_seed_text(seed.get("character")) or "Seeded Stage Guest" character_summary = first_available_seed_text( seed, ["description", "character", "input"], "A public-domain adventurer adapted into a puppet actor.", ) setting = first_available_seed_text( seed, ["scenario", "input"], "a tabletop adventure stage with cardboard scenery", ) premise = first_available_seed_text( seed, ["instruction", "scenario", "input"], f"{character_name} faces a strange public-domain stage problem", ) source_memory = first_available_seed_text(seed, ["output", "scenario", "input"], character_summary) return { "character_name": shorten_text(character_name, 60), "character_summary": shorten_text(character_summary, 220), "setting": seed_setting_to_stage(setting), "premise": seed_premise_to_puppet_show(premise), "source_memory": shorten_text(source_memory, 160), } def clean_seed_text(value: Any) -> str: if value is None: return "" if isinstance(value, str): return " ".join(value.strip().split()) if isinstance(value, list): return " ".join(clean_seed_text(item) for item in value if clean_seed_text(item)) if isinstance(value, dict): return " ".join(clean_seed_text(item) for item in value.values() if clean_seed_text(item)) return " ".join(str(value).strip().split()) def first_available_seed_text(seed: dict[str, Any], fields: list[str], fallback: str) -> str: for field in fields: text = clean_seed_text(seed.get(field)) if text: return text return fallback def should_skip_seed_text(text: str) -> bool: if len(text) > MAX_SEED_TEXT_CHARS: return True lowered = text.lower() return any(term in lowered for terms in SAFETY_BLOCKLIST.values() for term in terms) def seed_premise_to_puppet_show(text: str) -> str: cleaned = shorten_text(text, 150).rstrip(".") if not cleaned: return "A seeded puppet guest brings a mystery to the tiny stage" if cleaned.lower().startswith(("a ", "an ", "the ")): return cleaned return f"A puppet scene where {cleaned[0].lower()}{cleaned[1:]}" def seed_setting_to_stage(text: str) -> str: cleaned = shorten_text(text, 150).rstrip(".") if not cleaned: return "a borrowed stage with painted flats and restless curtains" return f"a puppet-stage version of {cleaned[0].lower()}{cleaned[1:]}" def build_seed_actor(extracted: dict[str, str], rng: random.Random) -> dict: name = puppet_name_from_seed(extracted["character_name"], rng) return { "name": name, "avatar": rng.choice(["mask", "scroll", "lantern", "book", "compass", "crown", "feather"]), "goal": seed_goal_from_summary(extracted["character_summary"]), "secret": seed_secret_from_summary(extracted["character_summary"]), "speaking_style": rng.choice( [ "adapted, theatrical, and slightly mysterious", "storybook, earnest, and stage-ready", "adventurous, concise, and puppet-bright", "dramatic, careful, and safe for a tiny audience", ] ), "tools": rng.choice( [ ["inspect_prop"], ["consult_stage_oracle"], ["change_lighting"], ["inspect_prop", "consult_stage_oracle"], ["consult_stage_oracle", "change_lighting"], ] ), } def puppet_name_from_seed(raw_name: str, rng: random.Random) -> str: words = [word.strip(".,!?;:()[]{}\"'") for word in raw_name.split() if word.strip(".,!?;:()[]{}\"'")] if not words: return rng.choice(["Seedling Marquee", "Borrowed Bow", "Pagefoot Lantern"]) base = " ".join(words[:2]) if len(base) < 3 or base.lower() in {"user", "assistant", "character"}: return rng.choice(["Seedling Marquee", "Borrowed Bow", "Pagefoot Lantern"]) return shorten_text(base, 40) def seed_goal_from_summary(summary: str) -> str: focus = premise_keyword(summary) return f"Turn the seeded {focus} detail into playable stage business." def seed_secret_from_summary(summary: str) -> str: focus = premise_keyword(summary) return f"Knows one hidden {focus} clue but reveals it only when the Director asks." def build_seed_memories(extracted: dict[str, str], rng: random.Random) -> list[str]: seed_memory = extracted.get("source_memory", "") memory = f"Seed memory: {shorten_text(seed_memory, 100).rstrip('.')}" if seed_memory else rng.choice(MEMORIES) return [memory, rng.choice(MEMORIES)] def choose_seed_prop(extracted: dict[str, str], rng: random.Random) -> str: keyword = premise_keyword(" ".join([extracted["premise"], extracted["character_summary"], extracted["setting"]])) if keyword and keyword != "stage": return f"{keyword} token" return rng.choice(PROPS) def choose_seed_row_type(extracted: dict[str, str], rng: random.Random) -> str: text = " ".join(extracted.values()).lower() if any(word in text for word in ["oracle", "prophecy", "prophet", "vision", "foretell"]): return "oracle_consult" if any(word in text for word in ["secret", "hidden", "disguise", "mystery"]): return rng.choice(["secret_hint_or_reveal", "memory_callback"]) if any(word in text for word in ["battle", "quest", "journey", "adventure"]): return rng.choice(["prop_inspection", "normal_reaction", "lighting_change"]) return rng.choice(ROW_TYPES) def choose_seed_director_instruction(row_type: str, extracted: dict[str, str], rng: random.Random) -> str: base = rng.choice(DIRECTOR_INSTRUCTIONS[row_type]) focus = premise_keyword(extracted["premise"]) return f"{base} Use the seeded {focus} detail as inspiration, not raw dialogue." def shorten_text(text: str, limit: int) -> str: cleaned = " ".join(text.strip().split()) if len(cleaned) <= limit: return cleaned truncated = cleaned[: limit - 1].rsplit(" ", 1)[0] return truncated or cleaned[:limit] def build_eval_prompts(count: int, rng: random.Random) -> list[dict]: rows = [] for index in range(count): row_type = ROW_TYPES[index % len(ROW_TYPES)] premise = rng.choice(PREMISES) prop = rng.choice(PROPS) actor = rng.choice(ACTORS) show_state = { "show_title": title_from_premise(premise), "setting": rng.choice(SETTINGS), "beat_index": index % 10, "min_beats": 7, "target_beats": 10, "max_beats": 12, "story_phase": phase_for_row_type(row_type, index % 10, 10), "latest_prop": prop if row_type in {"prop_inspection", "secret_hint_or_reveal"} else None, "latest_audience_action": audience_action_for(row_type, prop), "stage_lighting": rng.choice(STAGE_LIGHTING_STATES), "recent_transcript": recent_transcript(actor["name"], rng), "recent_tool_results": recent_tool_results(row_type, prop), "finale_requested": row_type == "finale", } actor_state = { **actor, "mood": rng.choice(MOODS), "current_goal": actor["goal"], "goal_progress": rng.choice(GOAL_PROGRESS), "held_props": [prop] if row_type == "prop_inspection" else [], "secret_status": "hinted" if row_type == "secret_hint_or_reveal" else "hidden", "recent_memory": rng.sample(MEMORIES, k=2), } rows.append( { "id": f"actor-eval-v0-{index + 1:03d}", "source_mix": ["synthetic_v0_eval_prompts"], "row_type": row_type, "messages": [ {"role": "system", "content": SYSTEM_MESSAGE}, { "role": "user", "content": build_user_message( premise, show_state, actor_state, rng.choice(DIRECTOR_INSTRUCTIONS[row_type]), ), }, ], } ) return rows def select_sample_rows(rows: list[dict], count: int) -> list[dict]: grouped = {row_type: [row for row in rows if row["row_type"] == row_type] for row_type in ROW_TYPES} sample: list[dict] = [] while len(sample) < count and any(grouped.values()): for row_type in ROW_TYPES: if grouped[row_type] and len(sample) < count: sample.append(grouped[row_type].pop(0)) return sample def build_user_message(premise: str, show_state: dict, actor: dict, director_instruction: str) -> str: return "\n".join( [ f"premise: {premise}", f"show_state JSON: {json.dumps(show_state, sort_keys=True, separators=(',', ':'))}", f"actor JSON: {json.dumps(actor, sort_keys=True, separators=(',', ':'))}", f"director_instruction: {director_instruction}", ] ) def build_assistant(row_type: str, premise: str, prop: str, actor: dict, rng: random.Random) -> dict: premise_focus = premise_keyword(premise) line = rng.choice(LINE_TEMPLATES[row_type]).format( prop=prop, thing=prop.split()[-1], premise_focus=premise_focus, ) memory_update = memory_update_for(row_type, prop, rng) tool_request = tool_request_for(row_type, prop, actor, rng) return { "intent": intent_for(row_type, actor, rng), "line": line, "emotion": rng.choice(EMOTIONS[row_type]), "gesture": rng.choice(GESTURES[row_type]), "stage_effect": rng.choice(STAGE_EFFECTS[row_type]), "memory_update": memory_update, "tool_request": tool_request, } def tool_request_for(row_type: str, prop: str, actor: dict, rng: random.Random) -> dict | None: if row_type == "prop_inspection" and "inspect_prop" in actor["tools"]: return {"tool": "inspect_prop", "args": {"prop": prop}, "reason": "The prop may reveal a stage clue."} if row_type == "oracle_consult" and "consult_stage_oracle" in actor["tools"]: return { "tool": "consult_stage_oracle", "args": {"question": rng.choice(["Which clue wants the spotlight?", "What should we notice next?"])}, "reason": "The oracle can sharpen the next beat.", } if row_type == "lighting_change" and "change_lighting" in actor["tools"]: return { "tool": "change_lighting", "args": {"mood": rng.choice(["moonlit mystery", "golden reveal", "stormy confusion", "warm suspicion", "blue apology"])}, "reason": "Lighting should clarify the emotional turn.", } return None def serialize_assistant(value: dict) -> str: ordered = {field: value[field] for field in OUTPUT_FIELDS} return json.dumps(ordered, ensure_ascii=True, separators=(",", ":")) def write_jsonl(path: Path, rows: list[dict]) -> None: with path.open("w", encoding="utf-8") as handle: for row in rows: handle.write(json.dumps(row, ensure_ascii=True, separators=(",", ":")) + "\n") def title_from_premise(premise: str) -> str: words = [word.strip(".,!?;:()[]{}\"'") for word in premise.split()] keywords = [word.title() for word in words if len(word.strip(".,!?;:()[]{}\"'")) > 3] return f"The {' '.join(keywords[:4])}" if keywords else "The Tiny Improv" def phase_for_row_type(row_type: str, beat_index: int, target_beats: int) -> str: if row_type in {"finale"}: return "finale" if row_type in {"secret_hint_or_reveal", "memory_callback"}: return "reveal" if row_type in {"comedic_confusion", "lighting_change"}: return "chaos" if row_type in {"prop_inspection", "oracle_consult"}: return "complication" progress = beat_index / max(1, target_beats) if progress < 0.2: return "opening" if progress < 0.65: return "complication" return "chaos" def audience_action_for(row_type: str, prop: str) -> str | None: if row_type == "prop_inspection": return f"Audience threw {prop} onto the stage." if row_type == "comedic_confusion": return "Audience heckled: That clue is wearing a hat." if row_type == "finale": return "Audience requested a finale." return None def recent_transcript(actor_name: str, rng: random.Random) -> list[dict]: speakers = [name for name in [a["name"] for a in ACTORS] if name != actor_name] return [ { "speaker": rng.choice(speakers), "line": rng.choice(LINE_TEMPLATES["normal_reaction"]).format( prop="prop", thing="clue", premise_focus="stage", ), }, { "speaker": actor_name, "line": rng.choice(LINE_TEMPLATES["comedic_confusion"]).format( prop="prop", thing="clue", premise_focus="stage", ), }, ] def recent_tool_results(row_type: str, prop: str) -> list[dict]: if row_type == "memory_callback": return [{"tool": "inspect_prop", "result": f"The {prop} pointed stage left.", "stage_effect": "prop_table_glow"}] if row_type == "secret_hint_or_reveal": return [{"tool": "consult_stage_oracle", "result": "Secrets knock twice before entering.", "stage_effect": "oracle_haze"}] return [] def secret_status_for(row_type: str, rng: random.Random) -> str: if row_type == "secret_hint_or_reveal": return rng.choice(["hinted", "revealed"]) if row_type == "finale": return rng.choice(["revealed", "resolved"]) return rng.choice(["hidden", "hinted"]) def memory_update_for(row_type: str, prop: str, rng: random.Random) -> str | None: template = rng.choice(MEMORY_UPDATE_TEMPLATES[row_type]) if template is None: return None return template.format(prop=prop) def intent_for(row_type: str, actor: dict, rng: random.Random) -> str: if row_type == "normal_reaction": return rng.choice(["react_to_event", "clarify_problem"]) if row_type == "secret_hint_or_reveal": return "reveal_secret" if actor.get("secret_status") == "revealed" else "hint_secret" return { "prop_inspection": "inspect_prop", "oracle_consult": "consult_oracle", "lighting_change": "change_lighting", "memory_callback": "recall_memory", "finale": "deliver_finale", "comedic_confusion": "comic_confusion", }[row_type] def premise_keyword(premise: str) -> str: stopwords = { "about", "after", "asks", "become", "because", "before", "inside", "into", "keeps", "last", "that", "their", "this", "until", "wearing", "which", "with", } words = [word.strip(".,!?;:()[]{}\"'").lower() for word in premise.split()] candidates = [word for word in words if len(word) > 3 and word not in stopwords] return candidates[0] if candidates else "stage" if __name__ == "__main__": main()