Spaces:
Sleeping
Sleeping
File size: 3,897 Bytes
7b73919 dc65515 7b73919 dc65515 7b73919 92f5069 7b73919 92f5069 | 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """FastAPI server exposing POST /reset, POST /step, GET /state endpoints."""
from __future__ import annotations
from typing import Any, Dict, Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from env.environment import CustomerSupportEnv
from env.models import Action, ActionType
# ── App ───────────────────────────────────────────────────────────────
app = FastAPI(
title="OpenEnv Customer Support",
description="AI Customer Support Ticket Resolution Environment",
version="1.0.0",
)
# Single environment instance
_env = CustomerSupportEnv()
# ── Request / Response schemas ────────────────────────────────────────
class ResetRequest(BaseModel):
task_id: str = "easy_classification"
ticket_index: int = 0
class StepRequest(BaseModel):
action_type: str
classification: Optional[str] = None
response_text: Optional[str] = None
refund_amount: Optional[float] = None
escalation_reason: Optional[str] = None
class ResetResponse(BaseModel):
observation: Dict[str, Any]
class StepResponse(BaseModel):
observation: Dict[str, Any]
reward: Dict[str, Any]
done: bool
info: Dict[str, Any]
class StateResponse(BaseModel):
state: Dict[str, Any]
# ── Endpoints ─────────────────────────────────────────────────────────
@app.post("/reset", response_model=ResetResponse)
def reset_endpoint(req: Optional[ResetRequest] = None) -> ResetResponse:
"""Reset the environment and return the initial observation."""
if req is None:
req = ResetRequest()
try:
obs = _env.reset(task_id=req.task_id, ticket_index=req.ticket_index)
return ResetResponse(observation=obs.model_dump())
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
@app.post("/step", response_model=StepResponse)
def step_endpoint(req: StepRequest) -> StepResponse:
"""Execute one environment step."""
try:
action_type = ActionType(req.action_type)
except ValueError:
raise HTTPException(
status_code=400,
detail=(
f"Invalid action_type '{req.action_type}'. "
f"Valid: {[a.value for a in ActionType]}"
),
)
action = Action(
action_type=action_type,
classification=req.classification,
response_text=req.response_text,
refund_amount=req.refund_amount,
escalation_reason=req.escalation_reason,
)
try:
result = _env.step(action)
except RuntimeError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return StepResponse(
observation=result.observation.model_dump(),
reward=result.reward.model_dump(),
done=result.done,
info=result.info,
)
@app.get("/state", response_model=StateResponse)
def state_endpoint() -> StateResponse:
"""Return the current environment state."""
return StateResponse(state=_env.state())
# ── Health check ──────────────────────────────────────────────────────
@app.get("/health")
def health():
return {"status": "ok"}
# ── Entry point ───────────────────────────────────────────────────────
def main():
import uvicorn
uvicorn.run("server.app:app", host="0.0.0.0", port=7860)
if __name__ == "__main__":
main()
|