Spaces:
Running
Running
File size: 3,331 Bytes
46c2649 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Python Code Action Environment.
This module provides a server-side environment implementation for executing
Python code actions using PyExecutor.
"""
import uuid
from core.env_server import Action, Environment, Observation, Transform
from core.tools import PyExecutor
from ..models import CodeAction, CodeObservation, CodeState
from .transforms import create_safe_coding_transform
class PythonCodeActEnv(Environment):
"""
Python Code Action Environment for executing code and tracking state.
This environment executes Python code submitted as CodeAction during step,
maintains the last exit code in its state, and returns results wrapped
in CodeObservation.
Args:
transform: Optional transform to apply to observations
additional_imports: List of additional module imports to authorize
(e.g., ["numpy", "pandas", "matplotlib"])
Example:
>>> env = PythonCodeActEnv()
>>> obs = env.reset()
>>> action = CodeAction(code="print('Hello, World!')")
>>> obs = env.step(action)
>>> print(obs.stdout) # "Hello, World!\n"
>>> print(obs.exit_code) # 0
>>> print(env.state.last_exit_code) # 0
"""
def __init__(
self,
):
self.transform = create_safe_coding_transform()
self._executor = PyExecutor()
self._state = CodeState()
def reset(self) -> Observation:
"""
Reset environment and start fresh execution session.
Returns:
Initial observation with empty stdout/stderr and exit_code=0
"""
# Initialize fresh state
self._state = CodeState(episode_id=str(uuid.uuid4()), step_count=0)
# Add last_exit_code to state
self._state.last_exit_code = 0
# Return initial observation
observation = CodeObservation(
stdout="",
stderr="",
exit_code=0,
)
return self._apply_transform(observation)
def step(self, action: Action) -> Observation:
"""
Execute code action and return observation.
Args:
action: CodeAction containing the code to execute
Returns:
CodeObservation with execution results (stdout, stderr, exit_code)
Raises:
ValueError: If action is not a CodeAction instance
"""
if not isinstance(action, CodeAction):
raise ValueError(f"Expected CodeAction, got {type(action)}")
# Execute the code using PyExecutor
result = self._executor.run(action.code)
# Update state
self._state.step_count += 1
self._state.last_exit_code = result.exit_code
# Create observation from execution result
observation = CodeObservation(
stdout=result.stdout,
stderr=result.stderr,
exit_code=result.exit_code,
)
return self._apply_transform(observation)
@property
def state(self) -> CodeState:
"""Get current environment state including last exit code."""
return self._state
|