Upload full OpenEnv project structure

Dockerfile

@@ -0,0 +1,25 @@
+FROM python:3.10-slim
+
+# Creates a non-root user with an explicit UID required by Hugging Face Spaces
+RUN useradd -m -u 1000 user
+USER user
+
+# Set environment variables
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+# Create the working directory
+WORKDIR $HOME/app
+
+# Install dependencies first for Docker caching
+COPY --chown=user:user requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the files
+COPY --chown=user:user . .
+
+# Expose the standard Hugging Face Space port
+EXPOSE 7860
+
+# Start the environment API server
+CMD ["uvicorn", "env:app", "--host", "0.0.0.0", "--port", "7860"]

__init__.py

@@ -0,0 +1 @@
+from server import env_instance

client.py

@@ -0,0 +1,40 @@
+import requests
+from typing import Dict, Any, Optional
+
+from server.models import Action, StepResult, ResetResult
+
+class EnvironmentalClient:
+    """
+    Official OpenEnv client wrapper to interface with the containerized environment endpoints.
+    Can be used by baseline inference scripts or remote agent evaluations.
+    """
+    def __init__(self, base_url: str = "http://127.0.0.1:7860"):
+        self.base_url = base_url.rstrip('/')
+        
+    def reset(self, task_id: int = 1) -> ResetResult:
+        """
+        Calls the /reset endpoint to begin a new episode for the specified task.
+        """
+        response = requests.post(f"{self.base_url}/reset", json={"task_id": task_id})
+        response.raise_for_status()
+        return ResetResult(**response.json())
+        
+    def step(self, action: Action) -> StepResult:
+        """
+        Submits a step (action) to the environment and returns the updated state/reward.
+        """
+        # Pydantic dict serialization for requests
+        response = requests.post(f"{self.base_url}/step", json=action.dict())
+        response.raise_for_status()
+        return StepResult(**response.json())
+        
+    def state(self) -> Dict[str, Any]:
+        """
+        Extracts the unstructured metadata state of the running environment.
+        """
+        response = requests.get(f"{self.base_url}/state")
+        response.raise_for_status()
+        return response.json()
+
+# Provide a default singleton for ease of use
+client = EnvironmentalClient()

env.py

@@ -0,0 +1 @@
+from server.app import app, env_instance

inference.py

@@ -0,0 +1,144 @@
+import os
+import json
+from openai import OpenAI
+from server.models import Action, BrowserGymAction # using our local Action model alias
+from server.app import env_instance as env
+
+# Environment Configuration
+API_BASE_URL = os.environ.get("API_BASE_URL", "https://api.openai.com/v1")
+MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o")
+API_KEY = os.environ.get("OPENAI_API_KEY") or os.environ.get("HF_TOKEN", "")
+
+MAX_STEPS = 15
+TEMPERATURE = 0.2
+MAX_TOKENS = 512
+
+SYSTEM_PROMPT = """
+You are an expert Data Engineer interacting with a simulated SQLite database.
+You will be given a task goal, the current database schema, and the most recent step's SQL output or error.
+Your goal is to complete the task by executing SQL commands. 
+
+CRITICAL RULES:
+1. You may only execute ONE SQL statement at a time. Do not chain statements with semicolons.
+2. If you need to review data, use short SELECT queries.
+3. If your previous action resulted in an SQL error, fix the error and try again.
+4. If you need multiple steps to achieve the goal (e.g. create tables, then insert data), execute them one by one.
+5. You MUST output ONLY a valid JSON object matching this schema:
+{
+  "action_str": "YOUR SQL QUERY HERE"
+}
+Do not wrap your response in markdown code blocks. Just valid JSON.
+"""
+
+def build_user_prompt(step: int, observation, history: list) -> str:
+    prompt = f"--- Step {step} ---\n"
+    prompt += f"Goal: {observation.goal}\n\n"
+    if observation.schema_dump:
+        prompt += f"Current DB Schema:\n{observation.schema_dump}\n\n"
+    
+    prompt += f"Last Result (or Error):\n{observation.result}\n\n"
+    
+    if history:
+        prompt += "Action History (Last 3 steps):\n"
+        for h in history[-3:]:
+            prompt += h + "\n"
+            
+    prompt += "\nProvide the JSON with your next `action_str`:"
+    return prompt
+
+def parse_model_action(response_text: str) -> str:
+    # Try to parse JSON
+    text = response_text.strip()
+    if text.startswith("```json"): text = text[7:]
+    if text.startswith("```"): text = text[3:]
+    if text.endswith("```"): text = text[:-3]
+    text = text.strip()
+    
+    try:
+        data = json.loads(text)
+        return data.get("action_str", "SELECT 1;")
+    except json.JSONDecodeError:
+        # Fallback if model doesn't follow json format correctly
+        return text
+
+def run_task(task_id: int):
+    print(f"\n{'='*50}\nStarting Task {task_id}\n{'='*50}")
+    
+    client = OpenAI(
+        base_url=API_BASE_URL,
+        api_key=API_KEY
+    )
+    
+    history = []
+    
+    # Using the local env object wrapper
+    result = env.reset(task_id=task_id)
+    observation = result.observation
+    print(f"Episode goal: {observation.goal}\n")
+
+    for step in range(1, MAX_STEPS + 1):
+        # We handle done from the step result, but for initial step we check just in case
+        user_prompt = build_user_prompt(step, observation, history)
+        
+        # print("PROMPT:", user_prompt)
+        
+        messages = [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": user_prompt},
+        ]
+
+        try:
+            completion = client.chat.completions.create(
+                model=MODEL_NAME,
+                messages=messages,
+                temperature=TEMPERATURE,
+                max_tokens=MAX_TOKENS,
+                stream=False,
+                response_format={"type": "json_object"} # enforce json output
+            )
+            response_text = completion.choices[0].message.content or ""
+            action_str = parse_model_action(response_text)
+        except Exception as exc: 
+            failure_msg = f"Model request failed ({exc}). Using fallback action."
+            print(failure_msg)
+            action_str = "SELECT 1;"
+
+        print(f"Step {step}: model suggested -> {action_str[:100]}...")
+
+        # Step the environment
+        step_result = env.step(BrowserGymAction(action_str=action_str))
+        observation = step_result.observation
+        reward = step_result.reward
+        done = step_result.done
+
+        error_flag = " ERROR" if observation.last_action_error else ""
+        history_line = f"Step {step}: {action_str[:50]}... -> reward {reward:+.2f}{error_flag}"
+        history.append(history_line)
+        
+        print(f"  Reward: {reward:+.2f} | Done: {done} | Last action error: {observation.last_action_error}")
+
+        if done:
+            final_score = step_result.info.get("current_score", 0.0)
+            print(f"\nEpisode complete! Final Score: {final_score}/1.0")
+            break
+    else:
+        final_score = env.state().get("current_score", 0.0)
+        print(f"\nReached max steps ({MAX_STEPS}). Final Score: {final_score}/1.0")
+        
+    return final_score
+
+def main():
+    print("Testing OpenEnv Data Engineer Inference Baseline")
+    
+    if not API_KEY:
+        print("Warning: API_KEY/HF_TOKEN not set. Will likely fail unless local LLM.")
+
+    scores = {}
+    for task_id in [1, 2, 3]:
+        score = run_task(task_id)
+        scores[f"Task_{task_id}"] = score
+        
+    print(f"\n{'*'*50}\nEVALUATION COMPLETE\n{scores}\n{'*'*50}")
+
+if __name__ == "__main__":
+    main()

models.py

@@ -0,0 +1,8 @@
+from server.models import (
+    Action, 
+    BrowserGymAction, 
+    Observation, 
+    Reward, 
+    StepResult, 
+    ResetResult
+)

openenv.yaml

@@ -0,0 +1,20 @@
+name: OpenEnv-SQL-Data-Engineer
+version: 1.0.0
+description: >
+  A real-world OpenEnv environment simulating a Database Administrator/Data Engineer. 
+  The agent must interact with a live SQLite mock database using SQL commands to perform data extraction, data cleaning, and schema normalization.
+entrypoint: env:app
+models:
+  action: models:Action
+  observation: models:Observation
+  reward: models:Reward
+tags:
+  - database
+  - sql
+  - data-engineering
+  - structured-data
+  - real-world
+port: 7860
+dockerfile: Dockerfile
+environment_vars:
+  - HF_TOKEN

pyproject.toml

@@ -0,0 +1,24 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "sql-data-engineer-env"
+version = "0.1.0"
+description = "A real-world SQL data engineering environment for agent evaluation."
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "openenv-core",
+    "fastapi",
+    "uvicorn",
+    "pydantic",
+    "openai",
+    "pandas"
+]
+
+[tool.setuptools]
+packages = ["."]
+
+[project.scripts]
+server = "server.app:main"

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+pydantic
+openai
+requests

server/__init__.py

@@ -0,0 +1,2 @@
+from .app import env_instance
+from .models import Observation, Action

server/app.py

@@ -0,0 +1,178 @@
+import sqlite3
+import os
+from typing import Dict, Any, Optional
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+import uvicorn
+
+from .models import Action, Observation, Reward, StepResult, ResetResult
+from .tasks import TASKS
+
+app = FastAPI(title="OpenEnv SQL Data Engineer")
+
+class SQLEnvironment:
+    def __init__(self):
+        self.conn: Optional[sqlite3.Connection] = None
+        self.task_id = 1
+        self.step_count = 0
+        self.current_score = 0.0
+        self.db_path = "tmp_env.db"
+
+    def get_schema_dump(self) -> str:
+        if not self.conn:
+            return ""
+        try:
+            c = self.conn.cursor()
+            c.execute("SELECT type, name, sql FROM sqlite_master WHERE type='table' OR type='view'")
+            rows = c.fetchall()
+            dump = []
+            for t, name, sql in rows:
+                if name.startswith('sqlite_'): continue
+                dump.append(f"[{t.upper()}] {name}:\n  {sql}")
+                
+            return "\n".join(dump) if dump else "Database is empty."
+        except Exception as e:
+            return f"Error extracting schema: {e}"
+
+    def reset(self, task_id: int = 1) -> ResetResult:
+        if self.conn:
+            self.conn.close()
+        
+        # Clean up existing temp db
+        if os.path.exists(self.db_path):
+            os.remove(self.db_path)
+            
+        self.task_id = task_id
+        if self.task_id not in TASKS:
+            raise ValueError(f"Task ID {self.task_id} not found.")
+            
+        self.conn = sqlite3.connect(self.db_path)
+        self.step_count = 0
+        self.current_score = 0.0
+        
+        # Initialize standard SQLite settings
+        self.conn.execute("PRAGMA foreign_keys = ON")
+        
+        # Setup specific task data
+        task = TASKS[self.task_id]
+        task.setup_db(self.conn)
+        self.current_score = task.grade(self.conn) # Base score
+        
+        goal_text = task.get_goal()
+        # Add basic info about actual task goal
+        instructions = f"Task Goal: {goal_text}\n"
+        
+        obs = Observation(
+            goal=instructions,
+            result="Environment initialized. Schema ready.",
+            step=self.step_count,
+            last_action_error=False,
+            schema_dump=self.get_schema_dump()
+        )
+        return ResetResult(observation=obs, info={"task_id": self.task_id, "initial_score": self.current_score})
+
+    def step(self, action: Action) -> StepResult:
+        if not self.conn:
+            raise ValueError("Environment not initialized. Call reset() first.")
+            
+        self.step_count += 1
+        last_action_error = False
+        query_result = ""
+        
+        try:
+            c = self.conn.cursor()
+            query = action.action_str.strip()
+            # Basic mitigation of forbidden queries just in case (though we're in mock)
+            if query.upper().startswith("DROP TABLE sqlite_"):
+                raise Exception("Cannot modify system tables.")
+                
+            c.execute(query)
+            
+            if query.upper().startswith("SELECT") or query.upper().startswith("PRAGMA"):
+                rows = c.fetchmany(10) # limit output size for LLM observation
+                col_names = [description[0] for description in c.description] if c.description else []
+                # Format tabular output
+                result_str = " | ".join(col_names) + "\n"
+                result_str += "-" * len(result_str) + "\n"
+                for r in rows:
+                    result_str += " | ".join([str(val) for val in r]) + "\n"
+                if len(rows) == 10:
+                    result_str += "... (output truncated)"
+                query_result = result_str
+            else:
+                self.conn.commit()
+                query_result = f"Command executed successfully. Rowcount: {c.rowcount}"
+                
+        except Exception as e:
+            last_action_error = True
+            query_result = f"SQL Error: {str(e)}"
+            
+        # Run grader
+        task = TASKS[self.task_id]
+        new_score = task.grade(self.conn)
+        
+        # Reward is dense: change in score + small penalty for errors
+        reward_value = new_score - self.current_score
+        
+        if last_action_error:
+            # minor penalty for syntax errors
+            reward_value -= 0.05
+            
+        self.current_score = new_score
+        
+        # Episode terminates when score is 1.0
+        done = (self.current_score >= 1.0) or (self.step_count > 30)
+        
+        obs = Observation(
+            goal=task.get_goal(),
+            result=query_result,
+            step=self.step_count,
+            last_action_error=last_action_error,
+            schema_dump=self.get_schema_dump() if not last_action_error else None # only dump if no error to save tokens
+        )
+        
+        return StepResult(
+            observation=obs,
+            reward=reward_value,
+            done=done,
+            info={"current_score": self.current_score}
+        )
+
+    def state(self) -> Any:
+        # Return state as unstructured dict per standard API
+        return {
+            "task_id": self.task_id,
+            "step": self.step_count,
+            "current_score": self.current_score,
+            "schema_dump": self.get_schema_dump()
+        }
+
+# Global instance
+env_instance = SQLEnvironment()
+
+class ResetRequest(BaseModel):
+    task_id: int = 1
+
+@app.post("/reset", response_model=ResetResult)
+def reset(req: ResetRequest):
+    try:
+        return env_instance.reset(task_id=req.task_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+@app.post("/step", response_model=StepResult)
+def step(action: Action):
+    try:
+        return env_instance.step(action)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+@app.get("/state")
+def state():
+    return env_instance.state()
+
+def main():
+    uvicorn.run("server.app:app", host="0.0.0.0", port=7860, reload=False)
+
+if __name__ == "__main__":
+    main()

server/models.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel
+from typing import Optional, Any, Dict, List
+
+class Action(BaseModel):
+    action_str: str
+    
+# Alias for compatibility with the sample inference script
+BrowserGymAction = Action
+
+class Observation(BaseModel):
+    goal: str
+    result: str
+    step: int
+    last_action_error: bool
+    schema_dump: Optional[str] = None
+    
+class Reward(BaseModel):
+    value: float
+    reason: Optional[str] = None
+
+class StepResult(BaseModel):
+    observation: Observation
+    reward: float
+    done: bool
+    info: Dict[str, Any]
+
+class ResetResult(BaseModel):
+    observation: Observation
+    info: Dict[str, Any]

server/tasks.py

@@ -0,0 +1,196 @@
+import sqlite3
+import re
+
+class Task:
+    def __init__(self, task_id: int):
+        self.task_id = task_id
+        
+    def setup_db(self, conn: sqlite3.Connection):
+        raise NotImplementedError
+        
+    def get_goal(self) -> str:
+        raise NotImplementedError
+        
+    def grade(self, conn: sqlite3.Connection) -> float:
+        raise NotImplementedError
+
+class EasyTask(Task):
+    def __init__(self):
+        super().__init__(1)
+
+    def setup_db(self, conn: sqlite3.Connection):
+        c = conn.cursor()
+        c.execute("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, total_spent REAL)")
+        c.executemany("INSERT INTO customers (name, total_spent) VALUES (?, ?)", [
+            ("Alice", 500.0),
+            ("Bob", 1200.0),
+            ("Charlie", 50.0),
+            ("Diana", 3000.0),
+            ("Eve", 1000.01) # over 1000
+        ])
+        conn.commit()
+
+    def get_goal(self) -> str:
+        return "Create a view named 'high_value_customers' containing all customers who have a 'total_spent' greater than 1000.0. The view should contain the exact same columns as the customers table."
+
+    def grade(self, conn: sqlite3.Connection) -> float:
+        c = conn.cursor()
+        try:
+            # Check if view exists
+            c.execute("SELECT name FROM sqlite_master WHERE type='view' AND name='high_value_customers'")
+            if not c.fetchone():
+                return 0.0
+            
+            # Check rows
+            c.execute("SELECT name, total_spent FROM high_value_customers ORDER BY name")
+            rows = c.fetchall()
+            
+            if len(rows) != 3:
+                return 0.5 # partially correct, exists but wrong rows
+                
+            expected = [("Bob", 1200.0), ("Diana", 3000.0), ("Eve", 1000.01)]
+            if rows == expected:
+                return 1.0
+            return 0.5
+        except Exception:
+            return 0.0
+
+
+class MediumTask(Task):
+    def __init__(self):
+        super().__init__(2)
+
+    def setup_db(self, conn: sqlite3.Connection):
+        c = conn.cursor()
+        c.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT, category TEXT, price TEXT)")
+        c.executemany("INSERT INTO products (name, category, price) VALUES (?, ?, ?)", [
+            ("Laptop", "Electronics", "$999.99"),
+            ("Mouse", "electronics", "25.50 USD"),
+            ("Desk", "FURNITURE", "150.0"),
+            ("Chair", "furniture", "$85.00"),
+            ("Headphones", "ELEC", "€45.00") # We'll just ask them to remove letters/symbols
+        ])
+        conn.commit()
+
+    def get_goal(self) -> str:
+        return (
+            "The 'products' table is messy. "
+            "1. Standardize the 'category' column to be fully UPPERCASE. (Hint: treat 'ELEC' as 'ELECTRONICS'). "
+            "2. Create a new column 'price_usd' of type REAL. Extract the numeric value from the 'price' string and populate 'price_usd'. "
+            "Do not drop any original columns."
+        )
+
+    def grade(self, conn: sqlite3.Connection) -> float:
+        score = 0.0
+        c = conn.cursor()
+        try:
+            # Check column exists
+            c.execute("PRAGMA table_info(products)")
+            columns = [row[1] for row in c.fetchall()]
+            if 'price_usd' in columns:
+                score += 0.3
+                
+                # Check data accuracy for price
+                c.execute("SELECT price_usd FROM products ORDER BY id")
+                prices = [row[0] for row in c.fetchall()]
+                expected_prices = [999.99, 25.50, 150.0, 85.0, 45.0]
+                
+                # allow small float diffs
+                correct_prices = sum(1 for p, e in zip(prices, expected_prices) if p is not None and abs(p - e) < 0.01)
+                score += (correct_prices / 5.0) * 0.4 # up to 0.4 for correct prices
+
+            # Check category uppercase
+            c.execute("SELECT category FROM products ORDER BY id")
+            categories = [row[0] for row in c.fetchall()]
+            expected_cats = ["ELECTRONICS", "ELECTRONICS", "FURNITURE", "FURNITURE", "ELECTRONICS"]
+            
+            correct_cats = sum(1 for c, e in zip(categories, expected_cats) if c == e)
+            score += (correct_cats / 5.0) * 0.3 # up to 0.3 for correct categories
+            
+            return min(1.0, score)
+        except Exception:
+            return score
+
+
+class HardTask(Task):
+    def __init__(self):
+        super().__init__(3)
+
+    def setup_db(self, conn: sqlite3.Connection):
+        c = conn.cursor()
+        c.execute("""
+            CREATE TABLE hospital_records (
+                patient_name TEXT, 
+                patient_dob TEXT, 
+                doctor_name TEXT, 
+                doctor_specialty TEXT, 
+                appointment_date TEXT, 
+                diagnosis TEXT
+            )
+        """)
+        records = [
+            ("John Doe", "1980-01-01", "Dr. Smith", "Cardiology", "2023-10-01", "Hypertension"),
+            ("Jane Roe", "1992-05-15", "Dr. Jones", "Neurology", "2023-10-02", "Migraine"),
+            ("John Doe", "1980-01-01", "Dr. Smith", "Cardiology", "2023-11-01", "Follow-up"),
+            ("Bob Guy", "1975-11-20", "Dr. Smith", "Cardiology", "2023-10-05", "Checkup")
+        ]
+        c.executemany("INSERT INTO hospital_records VALUES (?, ?, ?, ?, ?, ?)", records)
+        conn.commit()
+
+    def get_goal(self) -> str:
+        return (
+            "Normalize the flat 'hospital_records' table into 3 tables: "
+            "'patients' (id INTEGER PRIMARY KEY, name TEXT, dob TEXT), "
+            "'doctors' (id INTEGER PRIMARY KEY, name TEXT, specialty TEXT), and "
+            "'appointments' (id INTEGER PRIMARY KEY, patient_id INTEGER, doctor_id INTEGER, date TEXT, diagnosis TEXT). "
+            "Migrate all data from 'hospital_records' correctly without duplication. "
+            "Ensure foreign keys are correctly pointing to the new IDs."
+        )
+
+    def grade(self, conn: sqlite3.Connection) -> float:
+        score = 0.0
+        c = conn.cursor()
+        try:
+            # Check tables exist
+            c.execute("SELECT name FROM sqlite_master WHERE type='table'")
+            tables = [row[0] for row in c.fetchall()]
+            
+            if 'patients' in tables: score += 0.1
+            if 'doctors' in tables: score += 0.1
+            if 'appointments' in tables: score += 0.2
+            
+            if score < 0.4:
+                return score
+                
+            # Check data counts (3 unique patients, 2 unique doctors, 4 appointments)
+            c.execute("SELECT COUNT(*) FROM patients")
+            if c.fetchone()[0] == 3: score += 0.1
+            
+            c.execute("SELECT COUNT(*) FROM doctors")
+            if c.fetchone()[0] == 2: score += 0.1
+            
+            c.execute("SELECT COUNT(*) FROM appointments")
+            if c.fetchone()[0] == 4: score += 0.1
+            
+            # Check referential integrity (can we reconstruct the original view?)
+            query = """
+                SELECT p.name, p.dob, d.name, d.specialty, a.date, a.diagnosis
+                FROM appointments a
+                JOIN patients p ON a.patient_id = p.id
+                JOIN doctors d ON a.doctor_id = d.id
+                ORDER BY p.name, a.date
+            """
+            c.execute(query)
+            reconstructed = c.fetchall()
+            if len(reconstructed) == 4:
+                score += 0.3
+                
+            return min(1.0, score)
+        except Exception:
+            return score
+
+TASKS = {
+    1: EasyTask(),
+    2: MediumTask(),
+    3: HardTask()
+}