""" EcoGrid-OpenEnv — Unsloth GRPO Training Script Trains an LLM to play the EcoGrid environment using Group Relative Policy Optimization. Leverages unsloth for 4-bit quantised, memory-efficient LoRA training. """ import argparse import json import os import random from typing import List, Dict import numpy as np try: import torch from datasets import Dataset from unsloth import FastLanguageModel, is_bfloat16_supported from trl import GRPOConfig, GRPOTrainer HAS_UNSLOTH = True except ImportError: HAS_UNSLOTH = False try: import wandb HAS_WANDB = True except ImportError: HAS_WANDB = False from env.environment import EcoGridEnv from models.schemas import GridAction # Default to a small model, but allow override DEFAULT_MODEL = "unsloth/Qwen2.5-1.5B-Instruct" MAX_SEQ_LENGTH = 1024 LORA_RANK = 16 def parse_state_from_prompt(prompt) -> dict: """Extract the state JSON from the prompt string or message list.""" try: if isinstance(prompt, list): prompt_str = prompt[-1].get('content', '') else: prompt_str = str(prompt) parts = prompt_str.split("CURRENT STATE:\n") if len(parts) > 1: state_text = parts[1].split("\n\nTASK:")[0] return json.loads(state_text) except Exception: pass return {} def parse_action_from_completion(completion: str) -> GridAction | None: """Extract and validate GridAction JSON from model completion.""" try: start_idx = completion.find('{') end_idx = completion.rfind('}') if start_idx != -1 and end_idx != -1: json_str = completion[start_idx:end_idx+1] data = json.loads(json_str) return GridAction(**data) return None except Exception: return None def format_prompt(state_dict: dict, task_name: str) -> list: """Format the prompt for the model using chat template messages.""" state_json = json.dumps(state_dict, indent=2) system_msg = "You are an expert energy grid operator. Your goal is to balance renewable energy, fossil fuels, and battery storage to meet demand while minimising cost and carbon emissions." user_msg = f"""CURRENT STATE: {state_json} TASK: {task_name} CONSTRAINTS: - renewable_ratio + fossil_ratio <= 1.0 - battery_action must be between -1.0 (discharge) and 1.0 (charge) Output ONLY a valid JSON object: {{ "renewable_ratio": float, "fossil_ratio": float, "battery_action": float }}""" return [ {"role": "system", "content": system_msg}, {"role": "user", "content": user_msg} ] def set_global_seed(seed: int) -> None: """Set all available RNG seeds for reproducible training.""" random.seed(seed) np.random.seed(seed) os.environ["PYTHONHASHSEED"] = str(seed) if HAS_UNSLOTH: torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) torch.use_deterministic_algorithms(True, warn_only=True) def generate_training_data(num_samples: int, task: str, seed: int) -> Dataset: """Generate a dataset of random grid states for training.""" print(f"Generating {num_samples} training states for task '{task}'...") env = EcoGridEnv() rng = random.Random(seed) prompts = [] # We just run the environment randomly to generate a variety of states # Note: We don't need target actions because GRPO learns through trial and error! state = env.reset(task=task, seed=seed) for _ in range(num_samples): state_dict = state.model_dump() prompts.append(format_prompt(state_dict, task)) # Take a random valid action to advance the environment action = GridAction( renewable_ratio=rng.uniform(0, 0.8), fossil_ratio=rng.uniform(0, 0.2), battery_action=rng.uniform(-1, 1), ) try: result = env.step(action) state = result.observation except Exception: # If done or errored, reset state = env.reset(task=task, seed=rng.randint(0, 10000)) return Dataset.from_dict({"prompt": prompts}) def main(): parser = argparse.ArgumentParser(description="Unsloth GRPO Training for EcoGrid") parser.add_argument("--task", type=str, default="hard", choices=["easy", "medium", "hard"]) parser.add_argument("--epochs", type=int, default=1) parser.add_argument("--samples", type=int, default=200) parser.add_argument("--seed", type=int, default=3407) parser.add_argument("--model", type=str, default=DEFAULT_MODEL, help="Model path/name") args = parser.parse_args() if not HAS_UNSLOTH: print("Error: unsloth or trl not installed.") print("Install: pip install unsloth trl datasets") return print(f"Initializing Unsloth GRPO training on {args.model}") set_global_seed(args.seed) if HAS_WANDB: wandb.init( project="ecogrid-openenv", name=f"grpo-{args.task}-{args.model.split('/')[-1]}", config=vars(args) ) # 1. Load Model model, tokenizer = FastLanguageModel.from_pretrained( model_name=args.model, max_seq_length=MAX_SEQ_LENGTH, dtype=None, # Auto detection load_in_4bit=True, ) # Add LoRA adapter model = FastLanguageModel.get_peft_model( model, r=LORA_RANK, target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"], lora_alpha=LORA_RANK, use_gradient_checkpointing="unsloth", random_state=args.seed, ) # 2. Define GRPO Reward Function # We instantiate a fresh environment just for evaluating rewards during training reward_env = EcoGridEnv() def ecogrid_reward_func(prompts: List[str], completions: List[List[Dict[str, str]]], **kwargs) -> List[float]: """Reward function that evaluates model completions using the real environment.""" rewards = [] for prompt, completion_list in zip(prompts, completions): # TRL passes a list of messages for completion. We want the text content. # Depending on format, it might be a list of dicts. We extract the string. if isinstance(completion_list, list) and len(completion_list) > 0: completion_text = completion_list[-1]["content"] else: completion_text = str(completion_list) state_dict = parse_state_from_prompt(prompt) action = parse_action_from_completion(completion_text) if action is None or not state_dict: # Malformed JSON or invalid prompt extraction rewards.append(0.0) continue try: # To accurately calculate reward for THIS specific state and action, # we technically just need to call compute_reward, but it's easier to # forcefully inject the state into a reset environment. # In a true RL loop we'd step through, but GRPO is stateless evaluation. reward_env.reset(task=args.task, seed=42) # Seed doesn't matter here # Hack: inject state directly for evaluation from models.schemas import GridState reward_env._state = GridState(**state_dict) reward_env._done = False result = reward_env.step(action) # The reward is what the environment dictates! rewards.append(result.reward) except Exception as e: # Constraint violation or other error rewards.append(0.0) return rewards def format_reward_func(completions, **kwargs) -> List[float]: """Secondary reward: give a small bonus just for outputting valid JSON.""" rewards = [] for completion_list in completions: text = completion_list[-1]["content"] if isinstance(completion_list, list) else str(completion_list) action = parse_action_from_completion(text) rewards.append(0.1 if action is not None else 0.0) return rewards # 3. Prepare Dataset dataset = generate_training_data(args.samples, args.task, args.seed) # 4. Configure Trainer training_args = GRPOConfig( output_dir="./lora_adapter", learning_rate=2e-5, num_train_epochs=args.epochs, per_device_train_batch_size=2, gradient_accumulation_steps=4, max_prompt_length=1024, max_completion_length=500, num_generations=4, # Number of completions to generate per prompt for relative scoring save_steps=100, logging_steps=10, report_to="wandb" if HAS_WANDB else "none", # W&B tracking ) trainer = GRPOTrainer( model=model, processing_class=tokenizer, reward_funcs=[ecogrid_reward_func, format_reward_func], args=training_args, train_dataset=dataset, ) # 5. Train print("Starting GRPO training...") trainer.train() if HAS_WANDB: wandb.finish() # 6. Save print("Training complete. Saving LoRA adapter...") model.save_pretrained("./lora_adapter") tokenizer.save_pretrained("./lora_adapter") # Extract logs to show improvement log_history = trainer.state.log_history reward_curve = [] for log in log_history: if "eval_ecogrid_reward_func" in log or "reward/ecogrid_reward_func" in log: key = "eval_ecogrid_reward_func" if "eval_ecogrid_reward_func" in log else "reward/ecogrid_reward_func" reward_curve.append({ "step": log.get("step", 0), "reward": log.get(key, 0.0) }) os.makedirs("./logs", exist_ok=True) with open("./logs/reward_curve.json", "w") as f: json.dump(reward_curve, f, indent=2) with open("./logs/training_metrics.json", "w") as f: json.dump( { "task": args.task, "seed": args.seed, "epochs": args.epochs, "samples": args.samples, "model": args.model, "reward_curve": reward_curve, "log_history": log_history, }, f, indent=2, default=str, ) print("Saved reward curve to ./logs/reward_curve.json") print("Saved training metrics to ./logs/training_metrics.json") if __name__ == "__main__": main()