Spaces:
Sleeping
Sleeping
File size: 2,207 Bytes
6b42632 4ded5ed 6b42632 d145b94 6b42632 4ded5ed 6b42632 d145b94 4ded5ed cf4ce1e 4ded5ed 6b42632 54bff2c 4ded5ed 6b42632 4ded5ed 6b42632 54bff2c 4ded5ed 54bff2c | 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 | # 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.
"""
FastAPI application for the Coding Environment.
This module creates an HTTP server that exposes the PythonCodeActEnv
over HTTP and WebSocket endpoints, compatible with EnvClient.
Usage:
# Development (with auto-reload):
uvicorn envs.coding_env.server.app:app --reload --host 0.0.0.0 --port 8000
# Production:
uvicorn envs.coding_env.server.app:app --host 0.0.0.0 --port 8000 --workers 4
# Or run directly:
python -m envs.coding_env.server.app
"""
from fastapi import Query
try:
from coding_env.models import CodeAction, CodeObservation
from coding_env.server.python_codeact_env import PythonCodeActEnv
from coding_env.server.task_bank import get_episode_score, list_tasks
except ImportError:
from ..models import CodeAction, CodeObservation
from .python_codeact_env import PythonCodeActEnv
from .task_bank import get_episode_score, list_tasks
from openenv.core.env_server import create_app
# Create the app with web interface and README integration
# Pass the class (factory) instead of an instance for WebSocket session support
app = create_app(PythonCodeActEnv, CodeAction, CodeObservation, env_name="coding_env")
@app.get("/tasks", tags=["Environment Info"])
def tasks():
"""Return available benchmark tasks and their difficulty."""
return {"tasks": list_tasks()}
@app.get("/grader", tags=["Environment Info"])
def grader(
task_id: str = Query(..., description="Task identifier"),
episode_id: str = Query(..., description="Episode identifier"),
):
"""Return normalized score in [0.0, 1.0] for task/episode."""
score = get_episode_score(task_id, episode_id)
return {"task_id": task_id, "episode_id": episode_id, "score": float(score)}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
def main():
"""Main entry point for running the server."""
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
main()
|