Spaces:
Sleeping
Sleeping
File size: 4,262 Bytes
c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 19f02f2 b6835b7 598f6bf c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 b6835b7 c7b3de4 | 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 | # 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.
"""
Lint Coding Agent Environment Implementation.
A simple test environment that echoes back messages sent to it.
Perfect for testing HTTP server infrastructure.
"""
import json
import os
from uuid import uuid4
from openenv.core.env_server.interfaces import Environment
from openenv.core.env_server.types import State
try:
from ..models import LintCodingAgentAction, LintCodingAgentObservation
except ImportError:
from models import LintCodingAgentAction, LintCodingAgentObservation
class LintCodingAgentEnvironment(Environment):
"""
A data-driven environment that loads a 50-level curriculum from a JSON file.
Designed for professional code-linting agent evaluation.
"""
SUPPORTS_CONCURRENT_SESSIONS: bool = True
def __init__(self):
# Initialize internal state using OpenEnv's State object
self._state = State(episode_id=str(uuid4()), step_count=0)
self.level = 1
# --- DATA LOADING BLOCK ---
# Get the path to QUESTIONS.json relative to this file to avoid pathing errors in Docker
current_dir = os.path.dirname(os.path.abspath(__file__))
json_path = os.path.join(current_dir, "QUESTIONS.json")
try:
with open(json_path, "r") as f:
self.curriculum = json.load(f)
except Exception as e:
print(f"CRITICAL ERROR: Could not load QUESTIONS.json: {e}")
# Minimal fallback to prevent total server crash
self.curriculum = {"1": {"lang": "Python", "task": "Fallback", "context": "print 'Hi'", "ans": "print("}}
self.max_levels = len(self.curriculum)
def reset(self) -> LintCodingAgentObservation:
"""Reset the environment to the beginning of the curriculum."""
self._state = State(episode_id=str(uuid4()), step_count=0)
self.level = 1
return self._get_observation("Curriculum Loaded. Ready for Level 1.", 0.0, False)
def step(self, action: LintCodingAgentAction) -> LintCodingAgentObservation:
"""Process the agent's solution against the current JSON task."""
self._state.step_count += 1
# Get current level data (JSON keys are strings)
task_data = self.curriculum.get(str(self.level), self.curriculum.get("1"))
# Logic: Extract fields from the strict Action model
solution = action.code_solution
explanation = action.explanation
# Case-insensitive grading logic
is_correct = task_data["ans"].lower() in solution.lower()
if is_correct:
reward = 1.0
self.level += 1
feedback = f"Correct! Level {self.level - 1} cleared. Reason: {explanation}"
else:
reward = -0.1
feedback = f"Incorrect fix. Retry Level {self.level}. Hint: {task_data['task']}"
# Done if all levels complete or agent gives up
done = self.level > self.max_levels or "QUIT" in solution.upper()
return self._get_observation(feedback, reward, done)
def _get_observation(self, feedback: str, reward: float, done: bool) -> LintCodingAgentObservation:
"""Helper to package internal data into the strict Observation Pydantic model."""
# Use string indexing for JSON curriculum access
task_data = self.curriculum.get(str(self.level), self.curriculum.get("1"))
return LintCodingAgentObservation(
level=self.level,
language=task_data["lang"],
problem_statement=f"{feedback} | Goal: {task_data['task']}",
code_context=task_data["context"],
last_test_results=feedback,
reward=reward,
done=done,
metadata={
"step": self._state.step_count,
"episode_id": self._state.episode_id
}
)
@property
def state(self) -> State:
return self._state |