add frozenlake demo

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)"}}.

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

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']}")

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()

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()