Spaces:
Running
Running
File size: 2,303 Bytes
27d7338 | 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 | """FastAPI server exposing the CodeReviewEnv for evaluation."""
from __future__ import annotations
from typing import Any, Dict, Optional
from fastapi import Body, FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from env.environment import CodeReviewEnv
from env.models import CodeReviewAction, CodeReviewObservation
app = FastAPI()
ENV = CodeReviewEnv()
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
"""Return a JSON error response for unhandled exceptions (never crash server)."""
return JSONResponse(status_code=500, content={"error": str(exc)})
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
"""Return validation errors as JSON without crashing."""
return JSONResponse(status_code=422, content={"error": str(exc)})
@app.get("/")
async def root() -> Dict[str, str]:
"""Root route for HF Spaces UI health."""
return {"status": "ok", "message": "Code Review OpenEnv is running. See /health, /reset, /step, /state."}
@app.post("/reset", response_model=CodeReviewObservation)
async def reset(payload: Optional[Dict[str, Any]] = Body(default=None)) -> CodeReviewObservation:
"""Reset the environment for a given task_id (defaults to easy)."""
task_id = "easy"
if payload and isinstance(payload, dict) and "task_id" in payload:
task_id = str(payload["task_id"])
try:
return ENV.reset(task_id)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
@app.post("/step")
async def step(action: CodeReviewAction) -> Dict[str, Any]:
"""Apply an action to the environment and return the step result."""
observation, reward, done, info = ENV.step(action)
return {"observation": observation.model_dump(), "reward": reward, "done": done, "info": info}
@app.get("/state")
async def state() -> Dict[str, Any]:
"""Return current environment state as JSON."""
return ENV.state()
@app.get("/health")
async def health() -> Dict[str, str]:
"""Health check endpoint."""
return {"status": "ok", "version": "1.0.0"}
|