diff --git a/cookbook/frozenlake/__init__.py b/cookbook/frozenlake/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/cookbook/frozenlake/frozenlake_prompts.yaml b/cookbook/frozenlake/frozenlake_prompts.yaml new file mode 100644 index 00000000..4c1f5a70 --- /dev/null +++ b/cookbook/frozenlake/frozenlake_prompts.yaml @@ -0,0 +1,47 @@ +frozenlake_sys_prompt_no_slippery: | + You are an AI agent playing FrozenLake game. Your goal is to navigate from Start (S) to Goal (G) while avoiding Holes (H). + + Game Rules: + - S: Starting position (safe) + - F: Frozen surface (safe to walk on) + - H: Hole (you fall in and lose) + - G: Goal (you win!) + - []: Your current position + + Actions: + - 0: Move LEFT + - 1: Move DOWN + - 2: Move RIGHT + - 3: Move UP + + Your task: Analyze the current state and choose the best action (0-3) to reach the Goal while avoiding Holes. + While ensuring a safe arrival at the goal, you should aim to complete the task in as few steps as possible. + Think step by step, and respond with your thoughts and then clearly state your action as a number (0-3) in format {"action":"(0-3)"}. + +frozenlake_sys_prompt_slippery: | + You are an AI agent playing FrozenLake game. Your goal is to navigate from Start (S) to Goal (G) while avoiding Holes (H). + + Game Rules: + - S: Starting position (safe) + - F: Frozen surface (safe to walk on) + - H: Hole (you fall in and lose) + - G: Goal (you win!) + - []: Your current position + + Actions: + - 0: Move LEFT + - 1: Move DOWN + - 2: Move RIGHT + - 3: Move UP + + The ice is slippery, so you might not always move in the intended direction! + you will move in intended direction with probability of 1/3 else will move in either perpendicular direction with equal probability of 1/3 in both directions. + + For example, if action is left, then: + - P(move left)=1/3 + - P(move up)=1/3 + - P(move down)=1/3 + + Your task: Analyze the current state and choose the best action (0-3) to reach the Goal while avoiding Holes. + While ensuring a safe arrival at the goal, you should aim to complete the task in as few steps as possible. + Think step by step, and respond with your thoughts and then clearly state your action as a number (0-3) in format {{"action":"(0-3)"}}. diff --git a/cookbook/frozenlake/frozenlake_react_agent.py b/cookbook/frozenlake/frozenlake_react_agent.py new file mode 100644 index 00000000..2e215db6 --- /dev/null +++ b/cookbook/frozenlake/frozenlake_react_agent.py @@ -0,0 +1,379 @@ +import os +import re +import time +import json + +import ray +import requests +import random +from typing import List, Dict, Any, Optional +from dataclasses import dataclass +import numpy as np +import gymnasium as gym +from gymnasium.envs.toy_text.frozen_lake import generate_random_map +from openai import OpenAI +from loguru import logger +import yaml +from dotenv import load_dotenv +from tqdm import tqdm + +load_dotenv("../../.env") + +@dataclass +class GameResult: + task_id: str + run_id: int + experiment_name: str + success: bool + steps: int + reward: float + trajectory: List[Dict] + map_config: Dict[str, Any] + +@ray.remote +class FrozenLakeReactAgent: + """A ReAct Agent for FrozenLake game with experience learning.""" + + def __init__(self, + index: int, + task_configs: List[Dict], + experiment_name: str, + model_name: str = "qwen3-8b", + temperature: float = 0.7, + max_steps: int = 50, + num_runs: int = 1, + use_experience: bool = False, + make_experience: bool = False): + + self.index = index + self.task_configs = task_configs + self.experiment_name = experiment_name + self.model_name = model_name + self.temperature = temperature + self.max_steps = max_steps + self.num_runs = num_runs + self.use_experience = use_experience + self.make_experience = make_experience + + self.llm_client = OpenAI() + self.action_map = {0: "LEFT", 1: "DOWN", 2: "RIGHT", 3: "UP"} + + # Load prompts + self.prompts = self._load_prompts() + + def _load_prompts(self) -> Dict[str, str]: + """Load prompts from yaml file""" + try: + with open("frozenlake_prompts.yaml", 'r', encoding='utf-8') as f: + return yaml.safe_load(f) + except FileNotFoundError: + logger.warning("Prompt file not found, using default prompts") + return { + "frozenlake_sys_prompt_no_slippery": "You are playing FrozenLake. Navigate from S to G avoiding H.", + "frozenlake_sys_prompt_slippery": "You are playing FrozenLake. Navigate from S to G avoiding H. Ice is slippery!" + } + + def call_llm(self, messages: List[Dict]) -> str: + """Call LLM with retry logic""" + for i in range(5): + try: + response = self.llm_client.chat.completions.create( + model=self.model_name, + messages=messages, + temperature=self.temperature, + extra_body={"enable_thinking": False}, + seed=0 + ) + return response.choices[0].message.content + except Exception as e: + logger.warning(f"LLM call failed (attempt {i + 1}): {e}") + time.sleep(1 + i * 2) + return "LLM call failed" + + def observe_state(self, env, observation: int) -> str: + """Convert environment observation to text description""" + desc = env.unwrapped.desc + nrow, ncol = desc.shape + + # Convert to string grid + grid = [[cell.decode('utf-8') for cell in row] for row in desc] + + # Get current position + row, col = observation // ncol, observation % ncol + + # Create visual representation + state_text = "Current State:\n" + for i in range(nrow): + for j in range(ncol): + if i == row and j == col: + state_text += f"[{grid[i][j]}]" + else: + state_text += f" {grid[i][j]} " + state_text += "\n" + + state_text += "\nLegend: S=Start, F=Frozen, H=Hole, G=Goal, []=Your Position" + return state_text + + def build_system_prompt(self, is_slippery: bool) -> str: + """Build system prompt based on game configuration""" + if is_slippery: + return self.prompts["frozenlake_sys_prompt_slippery"] + else: + return self.prompts["frozenlake_sys_prompt_no_slippery"] + + def get_experience(self, map_desc: str, is_slippery: bool) -> str: + """Retrieve relevant experience from experience service""" + if not self.use_experience: + return "" + + try: + query = f"FrozenLake game map: {map_desc}, slippery: {is_slippery}" + base_url = "http://0.0.0.0:8001/" + workspace_id = self.experiment_name + + response = requests.post( + url=base_url + "retriever", + json={ + "workspace_id": workspace_id, + "query": query, + "top_k": 3 + }, + timeout=60 + ) + + if response.status_code == 200: + data = response.json() + return data.get("experience_merged", "") + else: + logger.warning(f"Experience retrieval failed: {response.status_code}") + return "" + + except Exception as e: + logger.warning(f"Failed to get experience: {e}") + return "" + + def action_parser(self, response: str) -> int: + """Parse action from LLM response""" + # Look for {"action":"X"} pattern + patterns = [ + r'["\']action["\']\s*:\s*["\']([0-3])["\']', + r'"action"\s*:\s*"([0-3])"', + r"'action'\s*:\s*'([0-3])'", + r'\baction["\']?\s*[:=]\s*["\']?([0-3])', + r'\b([0-3])\b(?=\s*$)', # Single digit at end + ] + + for pattern in patterns: + match = re.search(pattern, response) + if match: + action = int(match.group(1)) + if 0 <= action <= 3: + return action + + # Random fallback + action = random.randint(0, 3) + logger.warning(f"Could not parse action from response, using random: {action}") + return action + + def run_single_episode(self, task_config: Dict, run_id: int) -> GameResult: + """Run a single episode of the game""" + map_size = task_config.get("map_size", 4) + is_slippery = task_config.get("is_slippery", True) + map_desc = task_config.get("map_desc", None) + + # Create environment + env_kwargs = { + "render_mode": None, + "is_slippery": is_slippery, + } + + if map_desc is not None: + env_kwargs["desc"] = map_desc + else: + env_kwargs["desc"] = generate_random_map(size=map_size) + + env = gym.make("FrozenLake-v1", **env_kwargs) + + # Get map description for experience + map_str = '\n'.join([''.join([cell.decode('utf-8') for cell in row]) + for row in env.unwrapped.desc]) + + # Build messages + system_prompt = self.build_system_prompt(is_slippery) + experience = self.get_experience(map_str, is_slippery) + + messages = [{"role": "system", "content": system_prompt}] + + if experience: + exp_content = f"Here are some relevant tips from previous successful games:\n\n{experience}\n\nUse these tips to help you succeed." + messages.append({"role": "user", "content": exp_content}) + messages.append( + {"role": "assistant", "content": "I'll use these tips to navigate the frozen lake successfully."}) + + # Initialize game + observation, info = env.reset() + trajectory = [] + + # Add initial state + initial_state = self.observe_state(env, observation) + messages.append({"role": "user", "content": initial_state}) + + success = False + total_reward = 0 + + for step in range(self.max_steps): + # Get action from LLM + response = self.call_llm(messages) + action = self.action_parser(response) + + messages.append({"role": "assistant", "content": response}) + + # Take action + next_observation, reward, terminated, truncated, info = env.step(action) + total_reward += reward + done = terminated or truncated + + # Record trajectory step + trajectory.append({ + "step": step, + "state": observation, + "action": action, + "action_name": self.action_map[action], + "reward": reward, + "next_state": next_observation, + "done": done, + "llm_response": response + }) + + if done: + if terminated and reward > 0: + success = True + result_msg = f"Success! You reached the goal in {step + 1} steps!" + else: + result_msg = f"Game over! You fell into a hole or ran out of time." + + messages.append({"role": "user", "content": result_msg}) + break + else: + # Continue game + next_state = self.observe_state(env, next_observation) + step_msg = f"Step {step + 1}: You moved {self.action_map[action]}. Reward: {reward}\n{next_state}" + messages.append({"role": "user", "content": step_msg}) + observation = next_observation + + env.close() + + # Create result + map_id = task_config.get("map_id", f"unknown_{self.index}_{run_id}") + task_id = f"{task_config.get('task_type', 'test')}_map{map_id}_{run_id}" + result = GameResult( + task_id=task_id, + run_id=run_id, + experiment_name=self.experiment_name, + success=success, + steps=len(trajectory), + reward=total_reward, + trajectory=trajectory, + map_config={ + "map_desc": map_str, + "map_id": map_id, + "is_slippery": is_slippery, + "map_size": map_size, + "use_experience": self.use_experience + } + ) + + return result, messages + + def save_experience(self, results: List[GameResult], messages_list: List[List[Dict]]): + """Save successful trajectories as experience""" + if not self.make_experience: + return + + trajs = [] + for result, messages in zip(results, messages_list): + if result.success: + # Create trajectory for experience service + traj = { + "messages": messages, + "query" : result.map_config["map_desc"], + "score": 1.0, # Success + } + trajs.append(traj) + else: + traj = { + "messages": messages, + "query" : result.map_config["map_desc"], + "score": 0.0, # Success + } + trajs.append(traj) + + if trajs: + try: + base_url = "http://0.0.0.0:8001/" + workspace_id = self.experiment_name + + response = requests.post( + url=base_url + "summarizer", + json={ + "workspace_id": workspace_id, + "traj_list": trajs + }, + timeout=300 + ) + + if response.status_code == 200: + logger.info(f"Saved {len(trajs)} trajectories as experience") + else: + logger.warning(f"Failed to save experience: {response.status_code}") + + except Exception as e: + logger.error(f"Error saving experience: {e}") + + def execute(self) -> List[Dict]: + """Execute all tasks""" + all_results = [] + all_messages = [] + + for task_index, task_config in tqdm(enumerate(self.task_configs), desc="Processing tasks:"): + for run_id in range(self.num_runs): + logger.info(f"Ray {self.index}, Task {task_index}, Run {run_id}") + + result, messages = self.run_single_episode(task_config, run_id) + all_results.append(result) + all_messages.append(messages) + + # Convert result to dict for JSON serialization + result_dict = { + "task_id": result.task_id, + "run_id": result.run_id, + "experiment_name": result.experiment_name, + "task_completed": result.success, + "success": result.success, + "steps": result.steps, + "reward": result.reward, + "map_config": result.map_config, + "trajectory": result.trajectory + } + all_results[-1] = result_dict + + # Save experience if needed + if self.make_experience: + # Convert back to GameResult objects for experience saving + game_results = [] + for i, result_dict in enumerate(all_results): + game_result = GameResult( + task_id=result_dict["task_id"], + run_id=result_dict["run_id"], + experiment_name=result_dict["experiment_name"], + success=result_dict["success"], + steps=result_dict["steps"], + reward=result_dict["reward"], + trajectory=result_dict["trajectory"], + map_config=result_dict["map_config"] + ) + game_results.append(game_result) + + self.save_experience(game_results, all_messages) + + return all_results \ No newline at end of file diff --git a/cookbook/frozenlake/map_manager.py b/cookbook/frozenlake/map_manager.py new file mode 100644 index 00000000..3184c9a7 --- /dev/null +++ b/cookbook/frozenlake/map_manager.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +""" +Map Management Tool - Pre-generate and manage test maps +""" + +import json +import numpy as np +from pathlib import Path +from typing import List, Optional, Dict, Any +from loguru import logger +from gymnasium.envs.toy_text.frozen_lake import generate_random_map + + +class MapManager: + """Map Manager - pre-generating, storing and loading test maps""" + + def __init__(self, data_dir: str = "./map/"): + self.data_dir = Path(data_dir) + self.data_dir.mkdir(parents=True, exist_ok=True) + + def generate_test_maps(self, num_maps: int, map_size: int = 4, + base_seed: int = 10000) -> str: + """ + Generate test map collection and save + + Args: + num_maps: Number of maps to generate + map_size: Map size + base_seed: Base random seed + + Returns: + Path of saved file + """ + logger.info(f"πŸ—ΊοΈ Generating {num_maps} test maps (size={map_size})") + + maps_data = [] + for i in range(num_maps): + seed = base_seed + i + np.random.seed(seed) + map_desc = generate_random_map(size=map_size) + + maps_data.append({ + "map_id": i, + "seed": seed, + "map_size": map_size, + "map_desc": map_desc # Convert to list for JSON serialization + }) + + # Save to file + filename = f"test_maps_{num_maps}_{map_size}x{map_size}.jsonl" + filepath = self.data_dir / filename + + with open(filepath, "w", encoding="utf-8") as f: + for map_data in maps_data: + f.write(json.dumps(map_data, ensure_ascii=False) + "\n") + + logger.info(f"βœ… Test maps saved to {filepath}") + return str(filepath) + + def load_test_maps(self, filepath: str) -> List[Dict[str, Any]]: + """ + Load test maps + + Args: + filepath: Map file path + + Returns: + Map data list + """ + if not Path(filepath).exists(): + raise FileNotFoundError(f"Map file not found: {filepath}") + + maps_data = [] + with open(filepath, "r", encoding="utf-8") as f: + for line in f: + if line.strip(): + map_data = json.loads(line) + # Convert list back to numpy array + maps_data.append(map_data) + + logger.info(f"πŸ“– Loaded {len(maps_data)} test maps from {filepath}") + return maps_data + + def get_map_by_index(self, maps_data: List[Dict], index: int) -> Optional[list]: + """Get map by index""" + if 0 <= index < len(maps_data): + return maps_data[index]["map_desc"] + return None + + def get_or_create_test_maps(self, num_maps: int, map_size: int = 4) -> List[Dict[str, Any]]: + """ + Get or create test maps + If file exists and has sufficient quantity, load directly; otherwise regenerate + """ + filename = f"test_maps_{num_maps}_{map_size}x{map_size}.jsonl" + filepath = self.data_dir / filename + + if filepath.exists(): + try: + maps_data = self.load_test_maps(str(filepath)) + if len(maps_data) >= num_maps: + logger.info(f"βœ… Using existing test maps: {filepath}") + return maps_data[:num_maps] # Return required number of maps + except Exception as e: + logger.warning(f"⚠️ Failed to load existing maps: {e}, regenerating...") + + # File doesn't exist or insufficient quantity, regenerate + self.generate_test_maps(num_maps, map_size) + return self.load_test_maps(str(filepath)) + + +if __name__ == "__main__": + # Usage example + manager = MapManager() + + # Generate 100 4x4 test maps + manager.generate_test_maps(num_maps=100, map_size=4) + + # Load and view the first map + maps = manager.load_test_maps("./map/test_maps_100_4x4.jsonl") + print(f"First map:\n{maps[0]['map_desc']}") \ No newline at end of file diff --git a/cookbook/frozenlake/run_exp_statistic.py b/cookbook/frozenlake/run_exp_statistic.py new file mode 100644 index 00000000..3646b896 --- /dev/null +++ b/cookbook/frozenlake/run_exp_statistic.py @@ -0,0 +1,370 @@ +import json +import pandas as pd +from pathlib import Path +from collections import defaultdict +from typing import Dict, List, Tuple +from loguru import logger + + +def calculate_best_at_k(scores: List[float], k: int) -> float: + """ + Calculate best@k metric. + Divide scores into groups of size k, take the maximum value in each group, + then average these maximum values. + + Args: + scores: List of success scores (0 or 1) for all runs of a task + k: Group size + + Returns: + best@k value + """ + if len(scores) % k != 0: + raise ValueError(f"Length of scores ({len(scores)}) must be divisible by k ({k})") + + group_maxs = [] + for i in range(0, len(scores), k): + group = scores[i:i + k] + group_maxs.append(max(group)) + + return sum(group_maxs) / len(group_maxs) + + +def get_possible_k_values(total_runs: int) -> List[int]: + """Get all possible k values (divisors of total_runs)""" + k_values = [] + for k in range(1, total_runs + 1): + if total_runs % k == 0: + k_values.append(k) + return sorted(k_values, reverse=True) + + +def parse_task_config(task_id: str, map_config: Dict) -> Tuple[str, bool, bool]: + """ + Parse task configuration from task_id and map_config. + + Returns: + (condition, is_slippery, use_experience) + """ + is_slippery = map_config.get("is_slippery", True) + use_experience = map_config.get("use_experience", False) + + # Create condition string + slip_str = "slippery" if is_slippery else "no_slip" + exp_str = "with_exp" if use_experience else "no_exp" + condition = f"{slip_str}_{exp_str}" + + return condition, is_slippery, use_experience + + +def analyze_frozenlake_results(): + """Analyze FrozenLake experiment results""" + path = Path("./exp_result") + + if not path.exists(): + logger.error("Experiment results directory not found!") + return + + all_results = {} + + # Process all result files + for file in path.glob("*test*.jsonl"): + logger.info(f"Processing {file.name}") + + # Group results by condition and map + condition_results = defaultdict(lambda: defaultdict(list)) + + with open(file, "r") as f: + for line in f: + if not line.strip(): + continue + + try: + data = json.loads(line) + + if isinstance(data, list): + for item in data: + process_single_result(item, condition_results) + else: + process_single_result(data, condition_results) + + except json.JSONDecodeError as e: + logger.warning(f"Invalid JSON in {file.name}: {e}") + continue + + if not condition_results: + logger.warning(f"No valid data found in {file.name}") + continue + + # Calculate metrics for this file + file_metrics = calculate_file_metrics(condition_results, file.name) + all_results[file.name] = file_metrics + + # Generate comprehensive report + if all_results: + generate_analysis_report(all_results) + else: + logger.warning("No valid results found!") + + +def process_single_result(data: Dict, condition_results: Dict): + """Process a single result entry""" + map_config = data.get("map_config", {}) + task_id = data.get("task_id", "unknown") + success = data.get("success", False) + + # Parse condition + condition, is_slippery, use_experience = parse_task_config(task_id, map_config) + + # Extract map identifier - prefer map_id from map_config + map_id = map_config.get("map_id", "unknown") + if map_id == "unknown" and "test_map" in task_id: + # Fallback to parsing from task_id + parts = task_id.split("_") + for part in parts: + if part.startswith("map"): + try: + # Extract number from "mapXX" + map_num = ''.join(filter(str.isdigit, part)) + if map_num: + map_id = int(map_num) + break + except: + pass + + # Store result + success_score = 1.0 if success else 0.0 + condition_results[condition][f"map_{map_id}"].append(success_score) + + +def calculate_file_metrics(condition_results: Dict, filename: str) -> Dict: + """Calculate metrics for a single file""" + file_metrics = {"file": filename} + + for condition, map_results in condition_results.items(): + condition_scores = [] + + # Collect all scores for this condition + for map_id, scores in map_results.items(): + condition_scores.extend(scores) + + if not condition_scores: + continue + + # Check if all maps have the same number of runs + run_counts = [len(scores) for scores in map_results.values()] + if len(set(run_counts)) > 1: + logger.warning(f"Inconsistent runs for {condition}: {set(run_counts)}") + continue + + num_runs = run_counts[0] if run_counts else 0 + if num_runs == 0: + continue + + # Calculate overall success rate + overall_success = sum(condition_scores) / len(condition_scores) + file_metrics[f"{condition}_success_rate"] = overall_success + + # Calculate best@k metrics + k_values = get_possible_k_values(num_runs) + for k in k_values: + try: + # Calculate best@k for each map, then average + map_best_k_scores = [] + for map_id, scores in map_results.items(): + map_best_k = calculate_best_at_k(scores, k) + map_best_k_scores.append(map_best_k) + + avg_best_k = sum(map_best_k_scores) / len(map_best_k_scores) + file_metrics[f"{condition}_best@{k}"] = avg_best_k + + except ValueError as e: + logger.warning(f"Error calculating best@{k} for {condition}: {e}") + + # Map-level analysis + map_success_rates = {} + for map_id, scores in map_results.items(): + map_success_rate = sum(scores) / len(scores) + map_success_rates[map_id] = map_success_rate + + file_metrics[f"{condition}_map_details"] = map_success_rates + + logger.info(f"{filename} - {condition}: {overall_success:.3f} success rate, " + f"{len(map_results)} maps, {num_runs} runs each") + + return file_metrics + + +def generate_analysis_report(all_results: Dict): + """Generate comprehensive analysis report""" + logger.info("Generating comprehensive analysis report...") + + # 1. Create summary table + summary_data = [] + for file_name, metrics in all_results.items(): + row = {"file": file_name} + + # Extract success rates and best@k metrics + for key, value in metrics.items(): + if key != "file" and not key.endswith("_map_details"): + row[key] = value + + summary_data.append(row) + + if summary_data: + df_summary = pd.DataFrame(summary_data) + df_summary = df_summary.set_index('file') + + print("\n" + "=" * 100) + print("FROZENLAKE EXPERIMENT RESULTS SUMMARY") + print("=" * 100) + print(df_summary.round(4)) + print("=" * 100) + + # Save summary table + output_path = Path("./exp_result") / "frozenlake_summary.csv" + df_summary.to_csv(output_path) + logger.info(f"Summary table saved to: {output_path}") + + # 2. Condition comparison + print("\n" + "=" * 80) + print("CONDITION COMPARISON") + print("=" * 80) + + condition_comparison = defaultdict(list) + + for file_name, metrics in all_results.items(): + for key, value in metrics.items(): + if "_success_rate" in key: + condition = key.replace("_success_rate", "") + condition_comparison[condition].append(value) + + # Calculate average performance per condition + condition_avg = {} + for condition, scores in condition_comparison.items(): + if scores: + avg_score = sum(scores) / len(scores) + condition_avg[condition] = avg_score + print(f"{condition:20s}: {avg_score:.4f} (Β±{pd.Series(scores).std():.4f})") + + # 3. Experience effect analysis + print("\n" + "=" * 80) + print("EXPERIENCE EFFECT ANALYSIS") + print("=" * 80) + + experience_analysis = analyze_experience_effect(condition_avg) + for analysis_line in experience_analysis: + print(analysis_line) + + # 4. Map difficulty analysis + print("\n" + "=" * 80) + print("MAP DIFFICULTY ANALYSIS") + print("=" * 80) + + map_analysis = analyze_map_difficulty(all_results) + for map_id, difficulty in map_analysis.items(): + print(f"{map_id:10s}: {difficulty:.4f} average success rate") + + # 5. Detailed statistics + print("\n" + "=" * 80) + print("DETAILED STATISTICS") + print("=" * 80) + + generate_detailed_stats(all_results) + + +def analyze_experience_effect(condition_avg: Dict[str, float]) -> List[str]: + """Analyze the effect of experience on performance""" + analysis = [] + + # Compare with/without experience for each slippery condition + slippery_no_exp = condition_avg.get("slippery_no_exp", 0) + slippery_with_exp = condition_avg.get("slippery_with_exp", 0) + no_slip_no_exp = condition_avg.get("no_slip_no_exp", 0) + no_slip_with_exp = condition_avg.get("no_slip_with_exp", 0) + + if slippery_no_exp > 0 and slippery_with_exp > 0: + improvement_slippery = (slippery_with_exp - slippery_no_exp) / slippery_no_exp * 100 + analysis.append(f"Slippery condition - Experience effect: {improvement_slippery:+.1f}%") + analysis.append(f" Without exp: {slippery_no_exp:.4f}") + analysis.append(f" With exp: {slippery_with_exp:.4f}") + + if no_slip_no_exp > 0 and no_slip_with_exp > 0: + improvement_no_slip = (no_slip_with_exp - no_slip_no_exp) / no_slip_no_exp * 100 + analysis.append(f"No-slip condition - Experience effect: {improvement_no_slip:+.1f}%") + analysis.append(f" Without exp: {no_slip_no_exp:.4f}") + analysis.append(f" With exp: {no_slip_with_exp:.4f}") + + # Overall experience effect + exp_conditions = [v for k, v in condition_avg.items() if "with_exp" in k] + no_exp_conditions = [v for k, v in condition_avg.items() if "no_exp" in k] + + if exp_conditions and no_exp_conditions: + avg_with_exp = sum(exp_conditions) / len(exp_conditions) + avg_without_exp = sum(no_exp_conditions) / len(no_exp_conditions) + overall_improvement = (avg_with_exp - avg_without_exp) / avg_without_exp * 100 + analysis.append(f"Overall experience effect: {overall_improvement:+.1f}%") + + return analysis + + +def analyze_map_difficulty(all_results: Dict) -> Dict[str, float]: + """Analyze difficulty of different maps""" + map_scores = defaultdict(list) + + for file_name, metrics in all_results.items(): + for key, value in metrics.items(): + if key.endswith("_map_details") and isinstance(value, dict): + for map_id, success_rate in value.items(): + map_scores[map_id].append(success_rate) + + # Calculate average difficulty per map + map_difficulty = {} + for map_id, scores in map_scores.items(): + if scores: + avg_success = sum(scores) / len(scores) + map_difficulty[map_id] = avg_success + + # Sort by difficulty (hardest first) + return dict(sorted(map_difficulty.items(), key=lambda x: x[1])) + + +def generate_detailed_stats(all_results: Dict): + """Generate detailed statistics""" + total_experiments = len(all_results) + total_conditions = set() + + for metrics in all_results.values(): + for key in metrics.keys(): + if "_success_rate" in key: + condition = key.replace("_success_rate", "") + total_conditions.add(condition) + + print(f"Total experiment files: {total_experiments}") + print(f"Total conditions tested: {len(total_conditions)}") + print(f"Conditions: {', '.join(sorted(total_conditions))}") + + # Best performing conditions + all_success_rates = [] + for metrics in all_results.values(): + for key, value in metrics.items(): + if "_success_rate" in key and isinstance(value, (int, float)): + all_success_rates.append((key.replace("_success_rate", ""), value)) + + if all_success_rates: + best_condition = max(all_success_rates, key=lambda x: x[1]) + worst_condition = min(all_success_rates, key=lambda x: x[1]) + + print(f"Best performance: {best_condition[0]} ({best_condition[1]:.4f})") + print(f"Worst performance: {worst_condition[0]} ({worst_condition[1]:.4f})") + + +def main(): + """Main function for statistics analysis""" + logger.info("πŸ” Starting FrozenLake Results Analysis") + analyze_frozenlake_results() + logger.info("πŸ“Š Analysis completed!") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/cookbook/frozenlake/run_frozenlake.py b/cookbook/frozenlake/run_frozenlake.py new file mode 100644 index 00000000..7be7117c --- /dev/null +++ b/cookbook/frozenlake/run_frozenlake.py @@ -0,0 +1,296 @@ +import os +import time +import json +import ray +from pathlib import Path +from typing import List, Dict +import numpy as np +from loguru import logger +from gymnasium.envs.toy_text.frozen_lake import generate_random_map + +from frozenlake_react_agent import FrozenLakeReactAgent +from map_manager import MapManager + + +def generate_training_configs(num_maps: int = 20, map_size: int = 4, is_slippery: bool=False) -> List[Dict]: + """Generate random maps for training/experience generation""" + configs = [] + + for i in range(num_maps): + # Generate both slippery and non-slippery versions + random_map = generate_random_map(size=map_size) + + config = { + "task_type": "training", + "map_desc": random_map, + "map_size": map_size, + "is_slippery": is_slippery, + "task_id": f"train_{i}_{is_slippery}" + } + configs.append(config) + + return configs + + +def generate_test_configs(num_test_maps: int = 100, is_slippery: bool = False) -> List[Dict]: + """Generate test configurations using MapManager""" + logger.info(f"πŸ“‹ Generating test configurations for {num_test_maps} maps") + + # Initialize MapManager and get test maps + map_manager = MapManager() + maps_data = map_manager.get_or_create_test_maps(num_maps=num_test_maps, map_size=4) + + configs = [] + + for map_data in maps_data: + map_desc = np.array([list(row) for row in map_data["map_desc"]], dtype='c') + map_id = map_data["map_id"] + + for use_exp in [True, False]: + config = { + "task_type": "test", + "map_desc": map_desc, + "map_size": 4, + "is_slippery": is_slippery, + "use_experience": use_exp, + "map_id": map_id, + "task_id": f"test_map{map_id}_slip{is_slippery}_exp{use_exp}" + } + configs.append(config) + + logger.info(f"βœ… Generated {len(configs)} test configurations") + return configs + + +def train(experiment_name: str, max_workers: int = 2, num_runs: int = 3, num_training_maps= 15, is_slippery: bool= False) -> None: + """Phase 1: Generate experience from random maps""" + logger.info("🎯 Starting Training Phase - Generating Experience") + logger.info("=" * 60) + + training_configs = generate_training_configs(num_maps=num_training_maps, map_size=4, is_slippery=is_slippery) + path = Path("./exp_result") + path.mkdir(parents=True, exist_ok=True) + + results = [] + + def dump_results(): + output_file = path / f"{experiment_name}_training.jsonl" + with open(output_file, "w") as f: + for result in results: + f.write(json.dumps(result) + "\n") + logger.info(f"Training results saved to {output_file}") + + if max_workers > 1: + # Distributed training + future_list = [] + for i in range(max_workers): + worker_configs = training_configs[i::max_workers] + if worker_configs: # Only create worker if it has tasks + agent = FrozenLakeReactAgent.remote( + index=i, + task_configs=worker_configs, + experiment_name=experiment_name, + num_runs=num_runs, + use_experience=False, # No experience in training phase + make_experience=True,# Generate experience + ) + future = agent.execute.remote() + future_list.append(future) + time.sleep(1) + + logger.info(f"Started {len(future_list)} training workers") + + for i, future in enumerate(future_list): + worker_results = ray.get(future) + if worker_results: + results.extend(worker_results) + logger.info(f"results: {results[0]}") + logger.info(f"Training worker {i + 1}/{len(future_list)} completed") + dump_results() + + else: + # Single process training + agent = FrozenLakeReactAgent( + index=0, + task_configs=training_configs, + experiment_name=experiment_name, + num_runs=num_runs, + use_experience=False, + make_experience=True + ) + results = agent.execute() + dump_results() + + # Calculate training statistics + successful_runs = [r for r in results if r["success"]] + total_runs = len(results) + success_rate = len(successful_runs) / total_runs if total_runs > 0 else 0 + + logger.info(f"Training completed: {len(successful_runs)}/{total_runs} successful ({success_rate:.2%})") + return results + + +def test(experiment_name: str, max_workers: int = 2, num_runs: int = 5, num_test_maps: int = 100, is_slippery: bool=False) -> None: + """Phase 2: Test on fixed maps with/without experience""" + logger.info("πŸ§ͺ Starting Test Phase - Evaluating Performance") + logger.info(f"πŸ“Š Testing on {num_test_maps} maps with {num_runs} runs each") + logger.info("=" * 60) + + test_configs = generate_test_configs(num_test_maps=num_test_maps, is_slippery=is_slippery) + path = Path("./exp_result") + path.mkdir(parents=True, exist_ok=True) + + # Group configs by experience usage for separate experiments + exp_configs = [c for c in test_configs if c.get("use_experience", False)] + no_exp_configs = [c for c in test_configs if not c.get("use_experience", False)] + + logger.info(f"πŸ“ Configs without experience: {len(no_exp_configs)}") + logger.info(f"πŸ“ Configs with experience: {len(exp_configs)}") + + + + def dump_results(suffix: str): + output_file = path / f"{experiment_name}_test_{suffix}.jsonl" + with open(output_file, "w") as f: + for result in all_results: + f.write(json.dumps(result) + "\n") + logger.info(f"πŸ’Ύ Test results saved to {output_file}") + + # Test without experience first + logger.info("🚫 Testing WITHOUT experience...") + all_results = [] + results_no_exp = run_test_configs( + configs=no_exp_configs, + experiment_name=experiment_name, + max_workers=max_workers, + num_runs=num_runs, + use_experience=False + ) + all_results.extend(results_no_exp) + dump_results("no_exp") + + # Test with experience + logger.info("βœ… Testing WITH experience...") + all_results = [] + results_with_exp = run_test_configs( + configs=exp_configs, + experiment_name=experiment_name, + max_workers=max_workers, + num_runs=num_runs, + use_experience=True + ) + all_results.extend(results_with_exp) + dump_results("with_exp") + + return all_results + + +def run_test_configs(configs: List[Dict], experiment_name: str, max_workers: int, + num_runs: int, use_experience: bool) -> List[Dict]: + """Run a set of test configurations""" + results = [] + + if max_workers > 1: + future_list = [] + for i in range(max_workers): + worker_configs = configs[i::max_workers] + if worker_configs: + agent = FrozenLakeReactAgent.remote( + index=i, + task_configs=worker_configs, + experiment_name=experiment_name, + num_runs=num_runs, + use_experience=use_experience, + make_experience=False + ) + future = agent.execute.remote() + future_list.append(future) + time.sleep(1) + + for i, future in enumerate(future_list): + worker_results = ray.get(future) + if worker_results: + results.extend(worker_results) + logger.info(f"Test worker {i + 1}/{len(future_list)} completed") + + else: + agent = FrozenLakeReactAgent( + index=0, + task_configs=configs, + experiment_name=experiment_name, + num_runs=num_runs, + use_experience=use_experience, + make_experience=False + ) + results = agent.execute() + + return results + + +def main(): + """Main execution function""" + experiment_name = "frozenlake_no_slippery" + max_workers = 4 + training_runs = 4 # Runs per training map + num_training_maps = 50 + test_runs = 1 # Runs per test configuration + num_test_maps = 100 # Number of test maps to use + is_slippery = False + # model_name = "qwen-max-latest" + + # Initialize Ray if using multiple workers + if max_workers > 1: + ray.init(num_cpus=max_workers) + + try: + # Phase 1: Training (Experience Generation) + logger.info("πŸš€ Starting FrozenLake Experiment") + logger.info(f"🎯 Experiment: {experiment_name}") + logger.info(f"πŸƒ Workers: {max_workers}") + logger.info(f"πŸ“Š Test maps: {num_test_maps}") + logger.info(f"πŸ”„ Test runs per map: {test_runs}") + + training_results = train( + experiment_name=experiment_name, + max_workers=max_workers, + num_runs=training_runs, + num_training_maps=num_training_maps, + is_slippery=is_slippery + ) + + # Wait a bit for experience service to process + logger.info("⏰ Waiting for experience service to process data...") + time.sleep(10) + + + # Phase 2: Testing (Performance Evaluation) + test_results = test( + experiment_name=experiment_name, + max_workers=max_workers, + num_runs=test_runs, + num_test_maps=num_test_maps, + is_slippery=is_slippery + ) + + # Summary + logger.info("πŸŽ‰ Experiment completed!") + logger.info(f"πŸ“ˆ Training results: {len(training_results)} episodes") + logger.info(f"πŸ“ˆ Test results: {len(test_results)} episodes") + + # Quick statistics + successful_training = sum(1 for r in training_results if r.get("success", False)) + training_success_rate = successful_training / len(training_results) if training_results else 0 + + successful_test = sum(1 for r in test_results if r.get("success", False)) + test_success_rate = successful_test / len(test_results) if test_results else 0 + + logger.info(f"πŸ“Š Training success rate: {training_success_rate:.2%}") + logger.info(f"πŸ“Š Test success rate: {test_success_rate:.2%}") + + finally: + if max_workers > 1: + ray.shutdown() + + +if __name__ == "__main__": + main() \ No newline at end of file