|
| 1 | +# Agentic Engineering Practice Documentation |
| 2 | + |
| 3 | +This document introduces the development practices of the Agentic component in the ROLL framework, including environment manager development protocols, GlobalDataset usage, validation mode configuration, and trajectory synthesis functionality. |
| 4 | + |
| 5 | +## 1. EnvManager Development Protocol |
| 6 | + |
| 7 | +EnvManager is the core component of the Agentic framework, responsible for environment management and trajectory generation. Developing new EnvManagers requires following the following protocol: |
| 8 | + |
| 9 | +### 1.1 Core Loop Mechanism |
| 10 | + |
| 11 | +EnvManager must implement the `run_rollout_loop` method, which follows the following protocol: |
| 12 | + |
| 13 | +```python |
| 14 | +def run_rollout_loop(self, data: DataProto): |
| 15 | + """ |
| 16 | + 1. Each time run_rollout_loop is called, it continuously executes episodes |
| 17 | + until receiving a command that data collection is complete |
| 18 | + 2. Need to reset seed to ensure consistency across all groups |
| 19 | + 3. episode_id is obtained from the scheduler |
| 20 | +
|
| 21 | + Seed update logic: |
| 22 | + group_seed = base_seed + group_id |
| 23 | + episode_seed = group_seed + episode_id |
| 24 | +
|
| 25 | + trajectory_id: f"{group_id}_{episode_id}_{episode_seed}" |
| 26 | + """ |
| 27 | + |
| 28 | + # Minimal call example |
| 29 | + while self.running: |
| 30 | + # Get episode_id from scheduler |
| 31 | + self.episode_id = ray.get(self.output_queue.get_episode_id.remote(self.env_config["group_id"])) |
| 32 | + if self.episode_id is None: |
| 33 | + break |
| 34 | + |
| 35 | + # Reset environment |
| 36 | + rollout_cache = self.reset() |
| 37 | + |
| 38 | + while rollout_cache is not None and not rollout_cache.terminated and not rollout_cache.truncated: |
| 39 | + # Make decision |
| 40 | + lm_output = self.make_decision(rollout_cache) |
| 41 | + # Execute environment step |
| 42 | + rollout_cache = self.step(lm_output) |
| 43 | + |
| 44 | + # Submit trajectory |
| 45 | + rollout = self.formulate_rollouts(rollout_cache) |
| 46 | + ray.get(self.output_queue.put.remote(self.env_config['group_id'], self.episode_id, start_step, rollout)) |
| 47 | +``` |
| 48 | + |
| 49 | +### 1.2 EnvManager Development Constraints |
| 50 | + |
| 51 | +- **While loop infinite loop**: EnvManager continuously executes episodes through a while loop |
| 52 | +- **Exit only when dataset traversal is complete**: When dataset traversal is complete, the `reset()` method returns None, triggering loop exit |
| 53 | +- **Each episode must have corresponding trajectory put**: Each completed episode must submit trajectory data through `output_queue.put` |
| 54 | + |
| 55 | +## 2. GlobalDataset Usage |
| 56 | + |
| 57 | +### 2.1 Design Purpose |
| 58 | + |
| 59 | +To avoid memory access/memory bottlenecks caused by each env reading data independently, the framework provides the GlobalDataset component at the framework level to implement unified management and distribution of datasets. |
| 60 | + |
| 61 | +### 2.2 Class Definition and Location |
| 62 | + |
| 63 | +```python |
| 64 | +# Location: roll.datasets.global_dataset.GlobalDataset |
| 65 | +@ray.remote |
| 66 | +class GlobalDataset: |
| 67 | + def __init__(self, dataset_name, split: str = "train", mode="sample", dataset_kwargs: Dict = None): |
| 68 | + # mode: "sample" or "traversal" |
| 69 | +``` |
| 70 | + |
| 71 | +### 2.3 Two Working Modes |
| 72 | + |
| 73 | +#### Sample Mode |
| 74 | +- **Purpose**: Random sampling of datasets in training mode |
| 75 | +- **Features**: Randomly select data items each time |
| 76 | +- **Configuration**: `mode="sample"` |
| 77 | + |
| 78 | +#### Traversal Mode |
| 79 | +- **Purpose**: Need to traverse the entire dataset in validation mode |
| 80 | +- **Features**: Traverse dataset sequentially, ensuring each data item is accessed |
| 81 | +- **Configuration**: `mode="traversal"` |
| 82 | + |
| 83 | +### 2.4 Core Features |
| 84 | + |
| 85 | +- **Deterministic sampling**: The `get_data_item` method ensures the same seed returns the same data |
| 86 | +- **State management**: Internally maintains index state, supporting dataset reset and traversal |
| 87 | + |
| 88 | +### 2.5 Usage Example |
| 89 | + |
| 90 | +Refer to the implementation of MathEnv, SWEEnv, TerminalBenchEnv: |
| 91 | + |
| 92 | +```python |
| 93 | +class MathEnv(GEMMathEnv): |
| 94 | + def __init__(self, dataset_name: str = "", mode: str = "train", **kwargs): |
| 95 | + # Convert train/val mode to sample/traversal |
| 96 | + global_dataset_mode = "sample" if self.mode == "train" else "traversal" |
| 97 | + |
| 98 | + self.dataset = GlobalDataset.options( |
| 99 | + name=f"{self.mode}_{dataset_name}", |
| 100 | + get_if_exists=True, |
| 101 | + namespace=RAY_NAMESPACE |
| 102 | + ).remote( |
| 103 | + dataset_name=dataset_name, |
| 104 | + split=split, |
| 105 | + mode=global_dataset_mode |
| 106 | + ) |
| 107 | + |
| 108 | + # Create and register dataset_manager, this is necessary for implementing multiple val |
| 109 | + self.dataset_manager = GlobalDatasetManager.options( |
| 110 | + name=f"{self.mode}_dataset_manager", |
| 111 | + get_if_exists=True, |
| 112 | + namespace=RAY_NAMESPACE |
| 113 | + ).remote() |
| 114 | + ray.get(self.dataset_manager.register.remote( |
| 115 | + dataset_name=dataset_name, |
| 116 | + dataset_ref=self.dataset |
| 117 | + )) |
| 118 | +``` |
| 119 | + |
| 120 | +## 3. Validation Dataset Traversal Configuration During Training |
| 121 | + |
| 122 | +### 3.1 Configuration Principles |
| 123 | + |
| 124 | +For scenarios that require dataset traversal (such as math/code/swe validation scenarios), special configuration is required: |
| 125 | + |
| 126 | +- **Set mode parameter when defining env**: In val mode, need to set `mode=val` |
| 127 | +- **Set val_batch_size=-1**: This allows traversal of the entire val dataset |
| 128 | +- **Exit when env.reset returns None**: When dataset traversal is complete, env.reset will return None |
| 129 | + |
| 130 | +### 3.2 MathEnv Implementation Reference |
| 131 | + |
| 132 | +Location: `roll.pipeline.agentic.env.gem.math_env.MathEnv` |
| 133 | + |
| 134 | +```python |
| 135 | +class MathEnv(GEMMathEnv): |
| 136 | + def __init__(self, mode: str = "train", **kwargs): |
| 137 | + # Convert mode |
| 138 | + global_dataset_mode = "sample" if self.mode == "train" else "traversal" |
| 139 | + self.dataset = GlobalDataset.remote( |
| 140 | + dataset_name=dataset_name, |
| 141 | + split=split, |
| 142 | + mode=global_dataset_mode |
| 143 | + ) |
| 144 | + |
| 145 | + def reset(self, seed: Optional[None] = None) -> Tuple[str, dict[str, Any]]: |
| 146 | + data = ray.get(self.dataset.get_data_item.remote(seed=seed)) |
| 147 | + if data is None: |
| 148 | + return None, None # Dataset traversal complete |
| 149 | + # Process data... |
| 150 | +``` |
| 151 | + |
| 152 | +### 3.3 YAML Configuration Example |
| 153 | + |
| 154 | +```yaml |
| 155 | +rollout_batch_size: 128 |
| 156 | +val_batch_size: -1 # Traverse entire dataset |
| 157 | + |
| 158 | +deep_math: |
| 159 | + env_type: "roll_math" |
| 160 | + env_config: |
| 161 | + mode: val # Set to validation mode |
| 162 | + dataset_name: data/math_deepmath_deal.jsonl |
| 163 | + split: train |
| 164 | + question_key: prompt |
| 165 | + answer_key: ground_truth |
| 166 | +``` |
| 167 | +
|
| 168 | +### 3.4 Random Sampling Evaluation Scenarios |
| 169 | +
|
| 170 | +For random sampling evaluation scenarios such as games, simply configure in the conventional way, ensuring the same seed returns the same data. Random sampling is the default implementation, no special configuration required. |
| 171 | +
|
| 172 | +## 4. Trajectory Synthesis Dataset Traversal Configuration |
| 173 | +
|
| 174 | +### 4.1 AgenticRolloutPipeline Implementation |
| 175 | +
|
| 176 | +Location: `roll/pipeline/agentic/agentic_rollout_pipeline.py` |
| 177 | + |
| 178 | +### 4.2 Startup Method |
| 179 | + |
| 180 | +```shell |
| 181 | +python examples/start_agentic_rollout_pipeline.py --config_path $CONFIG_PATH --config_name $CONFIG_NAME |
| 182 | +``` |
| 183 | + |
| 184 | +### 4.3 Core Configuration Reference |
| 185 | + |
| 186 | +```yaml |
| 187 | +# Trajectory storage directory |
| 188 | +rollout_dump_dir: /data/oss_bucket_0/lixing/log/swe/${model_name}/rollout_trajectories |
| 189 | +
|
| 190 | +# Support ODPS storage |
| 191 | +# rollout_dump_dir: odps://odps_project/tables/table_name/ds=${model_name} |
| 192 | +
|
| 193 | +# Environment manager configuration |
| 194 | +train_env_manager: |
| 195 | + max_env_num_per_worker: 16 |
| 196 | + num_env_groups: 32 |
| 197 | + group_size: 1 # Support multiple trajectories for the same prompt rollout |
| 198 | + tags: [SWEEnvVal] |
| 199 | + num_groups_partition: [32] |
| 200 | +
|
| 201 | +# Custom environment configuration |
| 202 | +custom_envs: |
| 203 | + SWEEnvVal: |
| 204 | + env_type: "swe_env" |
| 205 | + env_config: |
| 206 | + mode: val # Validation mode |
| 207 | +``` |
| 208 | + |
| 209 | +### 4.4 Trajectory Dump Configuration |
| 210 | + |
| 211 | +In the `formulate_rollouts` method of EnvManager, need to register dump fields and types: |
| 212 | + |
| 213 | +```python |
| 214 | +def formulate_rollouts(self, rollout_cache: RolloutCache): |
| 215 | + # Prepare data |
| 216 | + save = { |
| 217 | + "task_idx": task_idx, |
| 218 | + "episode_score": episode_score, |
| 219 | + "traj_messages": traj_messages, |
| 220 | + "metrics": metrics, |
| 221 | + # ... other fields |
| 222 | + } |
| 223 | + |
| 224 | + # Register dump fields |
| 225 | + lm_input.non_tensor_batch["model_name"] = np.array( |
| 226 | + [os.path.basename(self.pipeline_config.base_dir)], dtype=object |
| 227 | + ) |
| 228 | + lm_input.non_tensor_batch["save_content"] = np.array([json.dumps(save)], dtype=object) |
| 229 | + lm_input.non_tensor_batch["step"] = np.array([self.current_step], dtype=object) |
| 230 | + lm_input.non_tensor_batch["task_idx"] = np.array([task_idx], dtype=object) |
| 231 | + lm_input.non_tensor_batch["stop_reason"] = np.array([stop_reason], dtype=object) |
| 232 | + lm_input.non_tensor_batch["mode"] = np.array([self.mode], dtype=object) |
| 233 | + lm_input.non_tensor_batch["episode_score"] = np.array([episode_score], dtype=object) |
| 234 | + |
| 235 | + # Configure database field types |
| 236 | + colummns_config = [ |
| 237 | + ["task_idx", "bigint"], |
| 238 | + ["model_name", "string"], |
| 239 | + ["stop_reason", "string"], |
| 240 | + ["episode_score", "double"], |
| 241 | + ["mode", "string"], |
| 242 | + ["save_content", "string"], |
| 243 | + ] |
| 244 | + lm_input.meta_info["COLUMMNS_CONFIG"] = colummns_config |
| 245 | + |
| 246 | + return lm_input |
| 247 | +``` |
| 248 | + |
| 249 | +### 4.5 Important Notes |
| 250 | + |
| 251 | +- **Keys in columns_config will be removed from data_proto after dump** |
| 252 | +- **save_content field contains complete trajectory information, stored in JSON format** |
| 253 | +- **Support local file system and ODPS table storage** |
| 254 | +- **Each trajectory has a unique trajectory_id for tracking** |
| 255 | + |
| 256 | +## 5. Trajectory Filtering |
| 257 | + |
| 258 | +### 5.1 Usage Method |
| 259 | + |
| 260 | +The trajectory filtering function is implemented by configuring the filter class through `roll.pipeline.agentic.agentic_config.EnvManagerConfig.group_filter_cls`. `roll.pipeline.agentic.agentic_pipeline.GroupFilter` is the default implementation. |
| 261 | + |
| 262 | +### 5.2 Custom Filtering Logic |
| 263 | + |
| 264 | +Custom complex trajectory filtering logic can be implemented, for example: |
| 265 | + |
| 266 | +```python |
| 267 | +class GroupFilter: |
| 268 | + def __init__(self, config: AgenticConfig, env_manager_config: EnvManagerConfig, mode: str): |
| 269 | + pass |
| 270 | +
|
| 271 | + def filter(self, group_id: int, episode_id: int, group: list[DataProto]): |
| 272 | + for data in group: |
| 273 | + if data.meta_info["drop_flag"]: |
| 274 | + return True |
| 275 | +``` |
| 276 | + |
| 277 | +Through custom filter functions, flexible filtering strategies can be implemented based on various trajectory attributes (such as score, length, stop reason, etc.). |
| 278 | + |
| 279 | +## 6. Frequently Asked Questions |
| 280 | + |
| 281 | +### Q1: How to handle dataset traversal completion? |
| 282 | + |
| 283 | +A: Check the `get_data_item` return value in the `reset` method. If it returns None, it means dataset traversal is complete, and you should return None to exit the loop. |
| 284 | + |
| 285 | +### Q2: How to ensure experiment reproducibility? |
| 286 | + |
| 287 | +A: Through a unified seed management mechanism, ensure the same seed returns the same data. The `get_data_item` method of GlobalDataset guarantees this. |
| 288 | + |
| 289 | +### Q3: How to handle large-scale trajectory data storage? |
| 290 | + |
| 291 | +A: You can use ODPS table storage by configuring `rollout_dump_dir` as an `odps://` format URL. For example: |
| 292 | + |
| 293 | +```yaml |
| 294 | +rollout_dump_dir: odps://odps_project/tables/table_name/ds=${model_name} |
| 295 | +``` |
| 296 | + |
| 297 | +### Q4: How to debug the trajectory generation process? |
| 298 | + |
| 299 | +A: You can debug the trajectory generation process by configuring log levels and adding custom logs. Trajectory data will be completely saved in JSON format for easy analysis. |
| 300 | + |
| 301 | +For multi-round interaction local debugging, refer to the documentation: [Debug Guide](../../../简体中文/快速开始/debug_guide.md) |
0 commit comments