Upload 6 files

app.py

@@ -0,0 +1,17 @@
+from openenv.core.env_server import create_app
+from .environment import AccountantEng
+from .schemas import Action, AccountantState
+
+# This creates the API server
+# It automatically handles the WebSocket connections for you!
+app = create_app(
+    env=AccountantEng,
+    action_cls=Action,
+    observation_cls=AccountantState,
+    env_name="mini-accountant",
+)
+
+if __name__ == "__main__":
+    import uvicorn
+    # This runs the server locally on port 8000
+    uvicorn.run(app, host="0.0.0.0", port=8000)

dockerfile

@@ -0,0 +1,21 @@
+# 1. Use an official Python runtime as a parent image
+FROM python:3.10-slim
+
+# 2. Set the working directory in the container
+WORKDIR /app
+
+# 3. Copy the requirements file first (better for Docker caching)
+COPY requirements.txt .
+
+# 4. Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 5. Copy the rest of your application code
+COPY . .
+
+# 6. Expose the port the app runs on
+EXPOSE 8000
+
+# 7. Run the application
+# We use the same command you used locally!
+CMD ["python", "-m", "mini_accountant_env.app"]

environment.py

@@ -0,0 +1,46 @@
+from openenv.core.env_server import Environment as BaseEnv
+from .schemas import CalculateAction, SubmitAction, AccountantState, Action
+
+class AccountantEng(BaseEnv):
+    def __init__(self):
+        self.target_answer = 3000
+        self.max_steps = 5
+        self.messy_text = "Revenue was 5000, costs were 2000. What is the net profit?"
+        self.current_step = 0
+
+    def reset(self) -> AccountantState:
+        self.current_state = 0
+        return AccountantState(
+            observation=self.messy_text,
+            is_done=False
+        )
+    
+    def step(self, action:Action)->tuple[AccountantState, float, bool, dict]:
+        self.current_step+=1
+        reward=-0.1
+        done=False
+        observation=""
+        
+        if self.current_step==self.max_steps:
+            observation="Max steps reached."
+
+        if action.action_type=="calculate":
+            try:
+                result=eval(action.expression)
+                observation = f"Calculation result: {result}"
+            except Exception as e:
+                reward-=0.5
+                
+        elif action.action_type=="submit":
+            done=True
+            if action.net_profit==self.target_answer:
+                reward+=1.0
+                observation = "Correct! Task complete."
+            else:
+                reward-=0.5
+                observation = f"Incorrect. The target was not {action.net_profit}."        
+        if self.current_step==self.max_steps:
+            observation="Max steps reached."
+
+        state = AccountantState(observation=observation, is_done=done)
+        return state, float(reward), done, {}

requirements.txt

@@ -0,0 +1,4 @@
+openenv
+pydantic>=2.0.0
+fastapi
+uvicorn

schemas.py

@@ -0,0 +1,33 @@
+from pydantic import BaseModel, Field
+from typing import Union, Literal
+
+# ==========================================
+# 1. TOOL ARGUMENTS (What the agent can do)
+# ==========================================
+
+class CalculateAction(BaseModel):
+    """Tool to calculate mathematical expressions."""
+    action_type: Literal["calculate"] = "calculate"
+    expression: str = Field(
+        ..., 
+        description="A strict mathematical expression to evaluate. ONLY use numbers and operators (+, -, *, /). Example: '5000 - 1200 - 800'"
+    )
+
+class SubmitAction(BaseModel):
+    """Submits the final answer"""
+    action_type : Literal["submit"]="submit"
+    net_profit: float = Field(
+        ..., 
+        description="The final calculated net profit as a floating point number."
+    )
+
+# The agent's action must be exactly one of these two
+Action = Union[CalculateAction, SubmitAction]
+# ==========================================
+# 2. STATE (What the agent sees)
+# ==========================================
+
+class AccountantState(BaseModel):
+    """The observation space returned to the agent after every step."""
+    observation: str = Field(..., description="The current text prompt, system message, or calculator result.")
+    is_done: bool = Field(default=False, description="True if the episode has finished.")