File size: 2,718 Bytes
1b50e57 0ee3210 840daec 0ee3210 1b50e57 0ee3210 1b50e57 0ee3210 e8dc38e 1b50e57 840daec 0ee3210 1b50e57 0ee3210 f5607fa 0ee3210 f5607fa 0ee3210 f5607fa | 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 | # server/app.py
from __future__ import annotations
import os
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import PlainTextResponse, RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
try:
from openenv.core.env_server.http_server import create_app
except ImportError as e:
raise ImportError("openenv is required. Install with 'uv sync'") from e
try:
from ..models import ModerationDecision, ModerationObservation, ModerationState
from .env import MetaContentModerationEnv, VALID_TASKS
except ImportError:
from models import ModerationDecision, ModerationObservation, ModerationState
from server.env import MetaContentModerationEnv, VALID_TASKS
# Create the internal openenv interface
app = create_app(
MetaContentModerationEnv,
ModerationDecision,
ModerationObservation,
env_name="meta-content-moderation-env",
max_concurrent_envs=1
)
# Remove the default /state route added by create_app so our custom one below is used
app.router.routes[:] = [r for r in app.router.routes if getattr(r, "path", None) != "/state"]
# Force the state endpoint to return our full ModerationState (including 'score')
@app.get("/state")
def get_state():
"""Returns the full environment state including custom fields like score."""
# We use the singleton instance
env = MetaContentModerationEnv()
return env.state
# Custom extra routes specific to this environment
@app.get("/", include_in_schema=False)
def root():
"""Redirect home page to Swagger documentation."""
return RedirectResponse(url="/docs")
@app.get("/tasks")
def list_tasks() -> dict:
"""List all available task names."""
return {
"tasks": list(VALID_TASKS),
"descriptions": {
"single-label-classify": "Easy \u2014 classify single content item into one violation category",
"multi-label-classify": "Medium \u2014 assign multiple violation labels to content",
"ad-policy-compliance": "Medium-Hard \u2014 review ad copy against ad policies, cite rule IDs",
"thread-moderation-hard": "Hard \u2014 moderate full WhatsApp thread with cultural context + policy conflicts",
}
}
@app.get("/openenv.yaml", response_class=PlainTextResponse)
def serve_openenv_yaml() -> str:
"""Serve the openenv.yaml metadata file."""
yaml_path = Path(__file__).parent.parent / "openenv.yaml"
if not yaml_path.exists():
raise HTTPException(status_code=404, detail="openenv.yaml not found")
return yaml_path.read_text(encoding="utf-8")
def main():
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
if __name__ == "__main__":
main()
|