openenv/repl

1# Copyright (c) Meta Platforms, Inc. and affiliates.2# All rights reserved.3#4# This source code is licensed under the BSD-style license found in the5# LICENSE file in the root directory of this source tree.6 7"""8Data models for the REPL Environment.9 10The REPL environment provides a Python REPL for training language models11on code execution tasks, based on the Recursive Language Models (RLM) paradigm.12 13Supports two finalization patterns:141. RLM-style: print('FINAL(answer)') or print('FINAL_VAR(var_name)')152. Prime Intellect style: answer = {"content": "...", "ready": True}16"""17 18from typing import Dict, List, Optional19 20from pydantic import BaseModel, Field21 22try:23    from openenv.core.env_server.types import Action, Observation, State24except ImportError:25    from openenv.core.env_server.types import Action, Observation, State26 27 28class REPLAction(Action):29    """Action containing Python code to execute in the REPL.30 31    Supports multiple finalization patterns:32    1. RLM-style: print('FINAL(answer)') or print('FINAL_VAR(var_name)') in code33    2. Prime Intellect style: answer = {"content": "...", "ready": True} in namespace34    3. Explicit: Set is_final=True with final_answer35    """36 37    code: str = Field(default="", description="Python code to execute")38    is_final: bool = Field(39        default=False,40        description="Whether this action signals the final answer",41    )42    final_answer: Optional[str] = Field(43        default=None, description="Final answer if is_final=True"44    )45 46 47class CodeBlockResult(BaseModel):48    """Result of executing a single code block."""49 50    stdout: str = Field(default="", description="Standard output from execution")51    stderr: str = Field(default="", description="Standard error from execution")52    locals_snapshot: Dict[str, str] = Field(53        default_factory=dict,54        description="String representations of new/modified variables",55    )56    execution_time: float = Field(57        default=0.0, ge=0, description="Execution time in seconds"58    )59    success: bool = Field(default=True, description="Whether execution succeeded")60    exception: Optional[str] = Field(61        default=None, description="Exception message if execution failed"62    )63 64 65class REPLObservation(Observation):66    """Observation returned after code execution in the REPL."""67 68    result: CodeBlockResult = Field(69        default_factory=CodeBlockResult, description="Result of code execution"70    )71    context_preview: Optional[str] = Field(72        default=None,73        description="Preview of the context (first N chars) if context is loaded",74    )75    context_length: int = Field(76        default=0, ge=0, description="Total length of context in characters"77    )78    available_variables: List[str] = Field(79        default_factory=list,80        description="List of variable names available in the namespace",81    )82    iteration: int = Field(default=0, ge=0, description="Current iteration number")83    max_iterations: int = Field(84        default=30, ge=1, description="Maximum allowed iterations"85    )86 87 88class REPLState(State):89    """Extended state for REPL environment."""90 91    context: Optional[str] = Field(92        default=None, description="The context/problem to work with"93    )94    task_prompt: Optional[str] = Field(95        default=None, description="The task description to solve"96    )97    iteration: int = Field(default=0, ge=0, description="Current iteration number")98    max_iterations: int = Field(99        default=30, ge=1, description="Max iterations before termination"100    )101    namespace_keys: List[str] = Field(102        default_factory=list, description="Variables currently in namespace"103    )104    final_answer: Optional[str] = Field(105        default=None, description="Final answer if episode is complete"106    )107    total_execution_time: float = Field(108        default=0.0, ge=0, description="Total code execution time in seconds"109    )110