""" CloudSRE v2 — Formal Evaluation Framework Evaluates a trained model against ALL tiers with proper metrics: - Resolution rate per tier - Average steps to resolution - Reward distribution - Diagnostic accuracy (did model check correct service?) - Fix accuracy (did model use correct fix type?) Usage: python evaluate.py --model-dir ./cloudsre-agent --episodes-per-tier 10 """ import argparse import json import os import sys import time from collections import defaultdict def main(): parser = argparse.ArgumentParser(description="CloudSRE v2 Evaluation") parser.add_argument("--env-url", required=True, help="Environment URL") parser.add_argument("--model-dir", required=True, help="Path to trained model") parser.add_argument("--episodes-per-tier", type=int, default=10) parser.add_argument("--output", default="eval_results.json") args = parser.parse_args() import httpx TIERS = ["warmup", "single_fault", "cascade", "multi_cascade", "adversarial"] client = httpx.Client(base_url=args.env_url, timeout=120) # Load model try: from unsloth import FastLanguageModel model, tokenizer = FastLanguageModel.from_pretrained( model_name=args.model_dir, max_seq_length=2048, load_in_4bit=True, ) FastLanguageModel.for_inference(model) print(f"Model loaded from {args.model_dir}") except Exception as e: print(f"Could not load model: {e}") print("Running in mock mode (random actions) for framework validation...") model = None tokenizer = None SYSTEM_PROMPT = """You are an expert Cloud SRE. Output ONLY the next command to run. No explanations. COMMANDS: status, restart_service , queue drain , curl http://..internal/healthz, cat /var/log//error.log REGIONS: us-east-1(payment,auth,billing,gateway,loadbalancer,config) eu-west-1(worker,scheduler,search,storage,metrics_collector) ap-south-1(frontend,cache,notification,email,dns)""" def generate_action(obs, history): """Generate next action from model (or fallback heuristic).""" health = obs.get("service_health", {}) broken = [n for n, h in health.items() if h.get("status") != "healthy"] if model is None: # Heuristic fallback for framework validation if not history: return "status" elif len(history) == 1 and broken: return f"cat /var/log/{broken[0]}/error.log" elif broken: err = health.get(broken[0], {}).get("error", "") if "queue" in err.lower(): return "queue drain 200" return f"restart_service {broken[0]}" return "status" # Real model inference alert = obs.get("alert", "") cmd_output = obs.get("command_output", "") health_text = "\n".join(f" {n}: {h.get('status','?')}" for n, h in health.items()) history_text = "\n".join(f" $ {h}" for h in history[-5:]) prompt = f"""{SYSTEM_PROMPT} ALERT: {alert} OUTPUT: {cmd_output[:300]} HEALTH: {health_text} PREVIOUS: {history_text} Next command:""" messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ] inputs = tokenizer.apply_chat_template(messages, return_tensors="pt", add_generation_prompt=True) inputs = inputs.to(model.device) outputs = model.generate(inputs, max_new_tokens=64, temperature=0.7, do_sample=True) response = tokenizer.decode(outputs[0][inputs.shape[1]:], skip_special_tokens=True).strip() # Extract first line as command cmd = response.split("\n")[0].strip() return cmd # ═══════════════════════════════════════════════════════ # Evaluate each tier # ═══════════════════════════════════════════════════════ results = {} for tier in TIERS: print(f"\n{'='*60}") print(f"Evaluating: {tier} ({args.episodes_per_tier} episodes)") print(f"{'='*60}") tier_data = { "resolved": 0, "failed": 0, "total_steps": [], "total_rewards": [], "fix_types_used": defaultdict(int), "episodes": [], } for ep in range(args.episodes_per_tier): r = client.post("/reset", json={"task_id": tier}) data = r.json() obs = data.get("observation", data) max_steps = obs.get("max_steps", 15) scenario = obs.get("scenario_id", "") history = [] total_reward = 0 resolved = False for step in range(max_steps): cmd = generate_action(obs, history) history.append(cmd) # Track fix types if "restart" in cmd: tier_data["fix_types_used"]["restart"] += 1 elif "drain" in cmd: tier_data["fix_types_used"]["drain"] += 1 elif "status" in cmd or "healthz" in cmd: tier_data["fix_types_used"]["diagnostic"] += 1 elif "cat" in cmd or "log" in cmd: tier_data["fix_types_used"]["log_check"] += 1 r2 = client.post("/step", json={"action": {"command": cmd}}) d2 = r2.json() obs = d2.get("observation", d2) reward = float(d2.get("reward", obs.get("reward", 0))) total_reward += reward done = d2.get("done", obs.get("done", False)) if done: resolved = True break if resolved: tier_data["resolved"] += 1 else: tier_data["failed"] += 1 tier_data["total_steps"].append(len(history)) tier_data["total_rewards"].append(total_reward) tier_data["episodes"].append({ "scenario": scenario, "resolved": resolved, "steps": len(history), "reward": total_reward, }) status = "✅" if resolved else "❌" print(f" {status} Ep {ep+1:2d} | {scenario:40s} | {len(history):2d} steps | reward={total_reward:+.2f}") # Tier summary n = args.episodes_per_tier rate = tier_data["resolved"] / n * 100 avg_steps = sum(tier_data["total_steps"]) / n avg_reward = sum(tier_data["total_rewards"]) / n print(f"\n Resolution: {tier_data['resolved']}/{n} ({rate:.0f}%)") print(f" Avg steps: {avg_steps:.1f}") print(f" Avg reward: {avg_reward:+.2f}") results[tier] = { "resolution_rate": rate, "avg_steps": avg_steps, "avg_reward": avg_reward, "resolved": tier_data["resolved"], "failed": tier_data["failed"], "fix_types": dict(tier_data["fix_types_used"]), "episodes": tier_data["episodes"], } # Save results with open(args.output, "w") as f: json.dump(results, f, indent=2) # Final summary print(f"\n{'='*60}") print(f"EVALUATION COMPLETE") print(f"{'='*60}") for tier in TIERS: r = results[tier] print(f" {tier:15s}: {r['resolution_rate']:5.1f}% resolved | {r['avg_steps']:.1f} avg steps | {r['avg_reward']:+.2f} avg reward") overall_resolved = sum(results[t]["resolved"] for t in TIERS) overall_total = sum(results[t]["resolved"] + results[t]["failed"] for t in TIERS) print(f"\n Overall: {overall_resolved}/{overall_total} ({overall_resolved/overall_total*100:.0f}%)") print(f" Results saved to: {args.output}") client.close() if __name__ == "__main__": main()