Upload 17 files

.env

@@ -0,0 +1,3 @@
+API_BASE_URL=https://api.openai.com/v1
+MODEL_NAME=gpt-4o-mini
+HF_TOKEN=replace_with_your_openai_api_key

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+COPY . /app
+
+EXPOSE 8000
+
+CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,10 +1,158 @@
----
-title: OpenenvScaler
-emoji: 🔥
-colorFrom: pink
-colorTo: gray
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Customer Support OpenEnv Benchmark
+emoji: 🧠
+colorFrom: blue
+colorTo: indigo
+sdk: docker
+pinned: false
+app_port: 8000
+tags:
+  - openenv
+---
+
+# Customer Support OpenEnv Benchmark
+
+This project is a real-world customer support simulation environment for evaluating agent behavior in ticket triage, policy grounding, and escalation handling.
+
+## Motivation
+
+Support teams handle large volumes of incoming tickets where bad automation creates operational and safety risk. This environment evaluates whether an agent can:
+
+- classify issue type correctly,
+- ground responses in the right knowledge base entry,
+- make safe escalation decisions,
+- avoid low-value loops while being polite and policy-consistent.
+
+## OpenEnv compliance
+
+- Full OpenEnv manifest in openenv.yaml
+- Typed models with Pydantic: action, observation, and state
+- Environment API exposed through OpenEnv FastAPI server
+- Standard endpoints available: reset, step, state, schema, health
+- OpenEnv validator passes with openenv validate
+
+## Action and observation spaces
+
+### Action
+
+SupportAction fields:
+
+- action_type: classify | search_kb | respond | escalate
+- content: text payload for that action
+
+### Observation
+
+SupportObservation fields:
+
+- ticket_id, task_id, difficulty
+- query, kb_id, requires_escalation
+- history (trajectory events)
+- reward, done, feedback, metadata
+
+## Tasks and graders
+
+Three deterministic task families are implemented, each with a programmatic grader returning a score in [0.0, 1.0].
+
+1. easy_classify_respond
+Objective: correct category classification and high-fidelity response.
+
+2. medium_kb_grounded_response
+Objective: correct KB retrieval followed by grounded response.
+
+3. hard_escalation_safety
+Objective: correct escalation decision plus safe/polite response behavior.
+
+## Reward design
+
+Reward shaping provides dense trajectory signal:
+
+- positive for correct classify/search_kb decisions
+- response similarity rewards for grounded answers
+- politeness bonus
+- penalties for hallucination risk and unnecessary escalation
+- per-step loop penalty
+
+## Quick start
+
+Install dependencies:
+
+```bash
+pip install -r requirements.txt
+```
+
+Run server locally:
+
+```bash
+uvicorn server.app:app --host 0.0.0.0 --port 8000
+```
+
+Run baseline inference:
+
+```bash
+set API_BASE_URL=https://api.openai.com/v1
+set MODEL_NAME=gpt-4o-mini
+set HF_TOKEN=your_api_token
+python inference.py --csv dataset.csv --limit-per-task 5
+```
+
+Offline smoke run (no API calls):
+
+```bash
+python inference.py --csv dataset.csv --limit-per-task 3 --offline
+```
+
+## Strict inference log format
+
+Inference emits only structured markers:
+
+- [START]
+- [STEP]
+- [END]
+
+## Baseline score reproducibility
+
+- Deterministic decoding (temperature=0)
+- fixed task ordering (easy, medium, hard)
+- deterministic grader criteria
+- one model planning call per episode to reduce runtime variance and cost
+
+## Pre-submission validation
+
+Run all checks in one command:
+
+```bash
+python pre_submission_validate.py --skip-docker
+```
+
+Checks included:
+
+- required env vars: API_BASE_URL, MODEL_NAME, HF_TOKEN
+- openenv validate
+- POST /reset returns HTTP 200
+- optional docker build
+
+For full parity with judge checks, run without skip:
+
+```bash
+python pre_submission_validate.py
+```
+
+## Docker
+
+```bash
+docker build -t customer-support-openenv .
+docker run -p 8000:8000 customer-support-openenv
+```
+
+## Hugging Face Spaces
+
+- Space SDK: Docker
+- Required tag: openenv
+- Health endpoint: /health
+- Reset endpoint: POST /reset
+
+Deploy command (after huggingface login):
+
+```bash
+openenv push
+```

client.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from typing import Dict
+
+from openenv.core import EnvClient
+from openenv.core.client_types import StepResult
+
+from models import SupportAction, SupportObservation, SupportState
+
+
+class CustomerSupportEnvClient(EnvClient[SupportAction, SupportObservation, SupportState]):
+    def _step_payload(self, action: SupportAction) -> Dict:
+        return {
+            "action_type": action.action_type,
+            "content": action.content,
+            "metadata": action.metadata,
+        }
+
+    def _parse_result(self, payload: Dict) -> StepResult[SupportObservation]:
+        obs_data = payload.get("observation", {})
+        observation = SupportObservation(
+            ticket_id=obs_data.get("ticket_id", 0),
+            task_id=obs_data.get("task_id", ""),
+            difficulty=obs_data.get("difficulty", "easy"),
+            query=obs_data.get("query", ""),
+            kb_id=obs_data.get("kb_id", ""),
+            requires_escalation=obs_data.get("requires_escalation", False),
+            history=obs_data.get("history", []),
+            feedback=obs_data.get("feedback", ""),
+            done=payload.get("done", False),
+            reward=payload.get("reward"),
+            metadata=obs_data.get("metadata", {}),
+        )
+        return StepResult(
+            observation=observation,
+            reward=payload.get("reward"),
+            done=payload.get("done", False),
+        )
+
+    def _parse_state(self, payload: Dict) -> SupportState:
+        return SupportState(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+            difficulty_filter=payload.get("difficulty_filter"),
+            current_index=payload.get("current_index", 0),
+            task_id=payload.get("task_id", ""),
+            score_so_far=payload.get("score_so_far", 0.0),
+            final_score=payload.get("final_score", 0.0),
+            done=payload.get("done", False),
+        )

daigram.md

@@ -0,0 +1,48 @@
+# Customer Support OpenEnv Benchmark Diagram
+
+```mermaid
+flowchart TD
+    A[dataset.csv] --> B[data_loader.py\nload + normalize + split]
+    B --> C[kb.py\nbuild knowledge base]
+    B --> D[server/customer_support_environment.py\nOpenEnv Environment]
+    C --> D
+    E[tasks.py\n3 tasks + deterministic graders] --> D
+    F[models.py\nTyped Action / Observation / State] --> D
+
+    D --> G[OpenEnv FastAPI App\nserver/app.py]
+    G --> H[POST /reset]
+    G --> I[POST /step]
+    G --> J[GET /state]
+    G --> K[GET /schema]
+
+    L[inference.py\nbaseline agent run] --> G
+    L --> M[[Structured Logs\n[START] [STEP] [END]]]
+    D --> N[[Reward Shaping\npartial progress + penalties]]
+    D --> O[[Episode Final Score\n0.0 to 1.0]]
+
+    P[openenv.yaml] --> G
+    Q[Dockerfile] --> R[HF Space Deployment]
+    G --> R
+```
+
+## Episode Logic (high-level)
+
+```mermaid
+sequenceDiagram
+    participant Agent
+    participant Env as CustomerSupportEnvironment
+    participant Grader as Task Grader
+
+    Agent->>Env: reset(difficulty, index)
+    Env-->>Agent: Observation(ticket, query, kb_id, task_id)
+
+    loop up to max steps
+        Agent->>Env: step(Action)
+        Env->>Env: reward shaping (progress + penalties)
+        Env-->>Agent: Observation + reward + done
+    end
+
+    Env->>Grader: grade_task(task_id, expected, history)
+    Grader-->>Env: final score in [0.0, 1.0]
+    Env-->>Agent: final metadata with score
+```

data_loader.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Dict
+
+import pandas as pd
+
+DIFFICULTIES = ("easy", "medium", "hard")
+DESCRIPTION_COLUMNS = ("Description", "Ticket Description")
+
+REQUIRED_COLUMNS = {
+    "TicketID",
+    "expected_category",
+    "expected_action",
+    "expected_response",
+    "requires_escalation",
+    "knowledge_base_id",
+    "difficulty",
+}
+
+
+def load_dataset(csv_path: str | Path = "dataset.csv") -> pd.DataFrame:
+    path = Path(csv_path)
+    if not path.exists() and not path.is_absolute():
+        candidate = Path(__file__).resolve().parent / path
+        if candidate.exists():
+            path = candidate
+    if not path.exists():
+        raise FileNotFoundError(f"Dataset not found at {path}")
+
+    df = pd.read_csv(path)
+    desc_col = next((c for c in DESCRIPTION_COLUMNS if c in df.columns), None)
+    missing = sorted(REQUIRED_COLUMNS - set(df.columns))
+    if desc_col is None:
+        missing.append("Description or Ticket Description")
+    if missing:
+        raise ValueError(f"Missing required dataset columns: {', '.join(missing)}")
+
+    df = df.copy()
+    if desc_col != "Description":
+        df["Description"] = df[desc_col]
+    df["difficulty"] = df["difficulty"].astype(str).str.lower().str.strip()
+    df["requires_escalation"] = (
+        df["requires_escalation"].astype(str).str.lower().map({"true": True, "false": False}).fillna(False)
+    )
+    df["TicketID"] = df["TicketID"].astype(int)
+    return df
+
+
+def split_difficulty(df: pd.DataFrame) -> Dict[str, pd.DataFrame]:
+    return {d: df[df["difficulty"] == d].reset_index(drop=True).copy() for d in DIFFICULTIES}
+
+
+def build_expected(row: pd.Series) -> dict:
+    return {
+        "ticket_id": int(row["TicketID"]),
+        "query": str(row["Description"]),
+        "expected_category": str(row["expected_category"]),
+        "expected_action": str(row["expected_action"]),
+        "expected_response": str(row["expected_response"]),
+        "requires_escalation": bool(row["requires_escalation"]),
+        "kb_id": str(row["knowledge_base_id"]),
+        "difficulty": str(row["difficulty"]),
+    }

inference.py

@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from typing import Dict, List
+
+from openai import OpenAI
+
+from models import SupportAction
+from server.customer_support_environment import CustomerSupportEnvironment
+from tasks import grade_task
+
+
+def log_start(task: str, env: str, model: str) -> None:
+    print(f"[START] task={task} env={env} model={model}", flush=True)
+
+
+def log_step(step: int, action: str, reward: float, done: bool, error: str | None = None) -> None:
+    error_value = "null" if error is None else error.replace("\n", " ")
+    print(
+        f"[STEP] step={step} action={action!r} reward={reward:.4f} done={done} error={error_value}",
+        flush=True,
+    )
+
+
+def log_end(success: bool, steps: int, score: float, rewards: List[float]) -> None:
+    reward_text = ",".join(f"{r:.4f}" for r in rewards)
+    print(
+        f"[END] success={str(success).lower()} steps={steps} score={score:.4f} rewards=[{reward_text}]",
+        flush=True,
+    )
+
+
+def get_model_plan(client: OpenAI, model_name: str, task_prompt: str) -> Dict[str, str | bool]:
+    completion = client.chat.completions.create(
+        model=model_name,
+        temperature=0,
+        messages=[
+            {
+                "role": "system",
+                "content": (
+                    "You are a customer support policy agent. Reply with strict JSON only and no markdown. "
+                    "Schema: {\"category\": string, \"search_kb\": bool, \"escalate\": bool, \"response\": string}."
+                ),
+            },
+            {"role": "user", "content": task_prompt},
+        ],
+        response_format={"type": "json_object"},
+    )
+    raw = completion.choices[0].message.content or "{}"
+    data = json.loads(raw)
+    return {
+        "category": str(data.get("category", "technical_issue")),
+        "search_kb": bool(data.get("search_kb", True)),
+        "escalate": bool(data.get("escalate", False)),
+        "response": str(data.get("response", "Thank you for contacting support. We will assist shortly.")),
+    }
+
+
+def heuristic_plan(observation: object) -> Dict[str, str | bool]:
+    query = str(getattr(observation, "query")).lower()
+    requires_escalation = bool(getattr(observation, "requires_escalation"))
+
+    category = "technical_issue"
+    if any(t in query for t in ("refund", "charge", "invoice", "billing")):
+        category = "billing"
+    elif any(t in query for t in ("cancel", "termination", "close account", "cancellation")):
+        category = "cancellation"
+    elif any(t in query for t in ("warranty", "buy", "purchase", "availability", "product")):
+        category = "product_inquiry"
+    elif any(t in query for t in ("refund not received", "return", "reimburse")):
+        category = "refund"
+
+    return {
+        "category": category,
+        "search_kb": True,
+        "escalate": requires_escalation,
+        "response": "Thank you for contacting us. We will review your case and assist you shortly.",
+    }
+
+
+def build_actions(observation: object, plan: Dict[str, str | bool]) -> List[SupportAction]:
+    actions = [
+        SupportAction(action_type="classify", content=str(plan["category"])),
+    ]
+
+    if bool(plan.get("search_kb", True)):
+        actions.append(SupportAction(action_type="search_kb", content=str(getattr(observation, "kb_id"))))
+
+    if bool(plan.get("escalate", False)):
+        actions.append(SupportAction(action_type="escalate", content="Escalating to specialist support for manual review."))
+    else:
+        actions.append(SupportAction(action_type="respond", content=str(plan.get("response", "Thank you for contacting support."))))
+    return actions
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="OpenEnv baseline inference for customer support benchmark")
+    parser.add_argument("--csv", default="dataset.csv")
+    parser.add_argument("--limit-per-task", type=int, default=5)
+    parser.add_argument("--max-steps", type=int, default=4)
+    parser.add_argument("--offline", action="store_true", help="Use heuristic policy without API calls")
+    args = parser.parse_args()
+
+    api_base_url = os.getenv("API_BASE_URL")
+    model_name = os.getenv("MODEL_NAME")
+    hf_token = os.getenv("HF_TOKEN")
+
+    if not args.offline:
+        missing = [name for name, value in (("API_BASE_URL", api_base_url), ("MODEL_NAME", model_name), ("HF_TOKEN", hf_token)) if not value]
+        if missing:
+            raise RuntimeError(f"Missing required env vars: {', '.join(missing)}")
+
+    client = OpenAI(base_url=api_base_url, api_key=hf_token) if not args.offline else None
+    env = CustomerSupportEnvironment(csv_path=args.csv)
+
+    all_rewards: List[float] = []
+    task_scores: List[float] = []
+    task_labels: List[str] = []
+    total_steps = 0
+
+    log_start(task="all", env="customer_support_benchmark", model=model_name or "offline-heuristic")
+    difficulties = ["easy", "medium", "hard"]
+    for difficulty in difficulties:
+        for idx in range(args.limit_per_task):
+            obs = env.reset(difficulty=difficulty, index=idx)
+            rewards: List[float] = []
+            done = False
+
+            prompt = (
+                "Return the best support plan for this ticket. "
+                "Prefer policy-safe responses and escalate only when needed.\n"
+                f"TicketID={obs.ticket_id}\n"
+                f"Difficulty={obs.difficulty}\n"
+                f"Task={obs.task_id}\n"
+                f"Query={obs.query}\n"
+                f"KB={obs.kb_id}\n"
+                f"RequiresEscalation={obs.requires_escalation}\n"
+            )
+
+            error = None
+            try:
+                plan = heuristic_plan(obs)
+                if client is not None and model_name is not None:
+                    plan = get_model_plan(client, model_name, prompt)
+            except Exception as exc:
+                error = str(exc)
+                plan = heuristic_plan(obs)
+
+            planned_actions = build_actions(obs, plan)
+
+            for step, action in enumerate(planned_actions[: args.max_steps], start=1):
+                if done:
+                    break
+                obs = env.step(action)
+
+                reward = float(obs.reward or 0.0)
+                rewards.append(reward)
+                all_rewards.append(reward)
+                total_steps += 1
+                done = bool(obs.done)
+                step_error = error if step == 1 else None
+                log_step(step=step, action=f"{action.action_type}|{action.content}", reward=reward, done=done, error=step_error)
+
+            final_score = grade_task(obs.task_id, obs.metadata, obs.history)
+            task_scores.append(final_score)
+            task_labels.append(difficulty)
+            log_step(
+                step=total_steps,
+                action=f"episode_score|difficulty={difficulty}|index={idx}",
+                reward=final_score,
+                done=True,
+                error=None,
+            )
+
+    benchmark_score = sum(task_scores) / len(task_scores) if task_scores else 0.0
+    by_task: dict[str, list[float]] = {"easy": [], "medium": [], "hard": []}
+    for label, score in zip(task_labels, task_scores):
+        by_task[label].append(score)
+    for label in ("easy", "medium", "hard"):
+        values = by_task[label]
+        avg = sum(values) / len(values) if values else 0.0
+        log_step(
+            step=total_steps,
+            action=f"task_average|difficulty={label}",
+            reward=avg,
+            done=False,
+            error=None,
+        )
+
+    print(
+        f"[END] success={str(benchmark_score >= 0.7).lower()} steps={total_steps} score={benchmark_score:.4f} rewards_count={len(all_rewards)}",
+        flush=True,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

kb.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+
+import pandas as pd
+
+
+def build_knowledge_base(df: pd.DataFrame) -> dict[str, dict]:
+    grouped: dict[str, list] = defaultdict(list)
+    for _, row in df.iterrows():
+        grouped[str(row["knowledge_base_id"])].append(row)
+
+    kb: dict[str, dict] = {}
+    for kb_id, rows in grouped.items():
+        kb[kb_id] = {
+            "title": _most_common([str(row.get("Subject", kb_id)) for row in rows]),
+            "category": _most_common([str(row["expected_category"]) for row in rows]),
+            "canonical_action": _most_common([str(row["expected_action"]) for row in rows]),
+            "canonical_response": _most_common([str(row["expected_response"]) for row in rows]),
+            "summary": "Ground responses in policy and avoid unsupported claims.",
+        }
+    return kb
+
+
+def _most_common(items: list[str]) -> str:
+    if not items:
+        return ""
+    return Counter(items).most_common(1)[0][0]

models.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+
+from openenv.core.env_server.types import Action, Observation, State
+from pydantic import Field
+
+
+class SupportAction(Action):
+    action_type: Literal["classify", "search_kb", "respond", "escalate"] = Field(
+        ..., description="Action category taken by the agent"
+    )
+    content: str = Field(default="", description="Action payload text")
+
+
+class SupportObservation(Observation):
+    ticket_id: int = Field(..., description="Current ticket identifier")
+    task_id: str = Field(..., description="Current task id")
+    difficulty: Literal["easy", "medium", "hard"] = Field(...)
+    query: str = Field(..., description="Customer support query")
+    kb_id: str = Field(..., description="Expected knowledge-base identifier")
+    requires_escalation: bool = Field(..., description="Whether escalation is required")
+    history: List[Dict[str, Any]] = Field(default_factory=list)
+    feedback: str = Field(default="", description="Reward-shaping feedback")
+
+
+class SupportState(State):
+    difficulty_filter: Optional[str] = Field(default=None)
+    current_index: int = Field(default=0)
+    task_id: str = Field(default="")
+    score_so_far: float = Field(default=0.0)
+    final_score: float = Field(default=0.0)
+    done: bool = Field(default=False)

openenv.yaml

@@ -0,0 +1,6 @@
+spec_version: 1
+name: customer_support_benchmark
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000

pre_submission_validate.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parent
+
+
+def _run(cmd: list[str], timeout: int = 600, cwd: Path | None = None) -> tuple[int, str]:
+    process = subprocess.run(
+        cmd,
+        cwd=str(cwd or ROOT),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        timeout=timeout,
+        text=True,
+        check=False,
+    )
+    return process.returncode, process.stdout
+
+
+def _check_env_vars() -> bool:
+    required = ["API_BASE_URL", "MODEL_NAME", "HF_TOKEN"]
+    missing = [name for name in required if not os.getenv(name)]
+    if missing:
+        print(f"[FAIL] missing env vars: {', '.join(missing)}")
+        return False
+    print("[PASS] required env vars are set")
+    return True
+
+
+def _check_openenv_validate() -> bool:
+    openenv_path = shutil.which("openenv")
+    if openenv_path is None:
+        scripts_dir = Path(sys.executable).resolve().parent
+        candidate = scripts_dir / ("openenv.exe" if os.name == "nt" else "openenv")
+        if candidate.exists():
+            openenv_path = str(candidate)
+    if openenv_path is None:
+        print("[FAIL] openenv CLI not found in PATH")
+        return False
+    code, output = _run([openenv_path, "validate"], timeout=300)
+    if code != 0:
+        print("[FAIL] openenv validate failed")
+        print(output)
+        return False
+    print("[PASS] openenv validate")
+    return True
+
+
+def _check_reset_ping(port: int = 8000) -> bool:
+    python_bin = sys.executable
+    cmd = [python_bin, "-m", "uvicorn", "server.app:app", "--host", "127.0.0.1", "--port", str(port)]
+    proc = subprocess.Popen(
+        cmd,
+        cwd=str(ROOT),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        deadline = time.time() + 30
+        url = f"http://127.0.0.1:{port}/reset"
+        payload = json.dumps({}).encode("utf-8")
+        request = urllib.request.Request(url, data=payload, method="POST")
+        request.add_header("Content-Type", "application/json")
+
+        while time.time() < deadline:
+            try:
+                with urllib.request.urlopen(request, timeout=5) as response:
+                    if response.status == 200:
+                        print("[PASS] POST /reset returned 200")
+                        return True
+            except urllib.error.URLError:
+                time.sleep(1)
+                continue
+        print("[FAIL] POST /reset did not return 200 in time")
+        return False
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+
+
+def _check_docker(skip: bool) -> bool:
+    if skip:
+        print("[SKIP] docker build check")
+        return True
+    docker_path = shutil.which("docker")
+    if docker_path is None:
+        print("[FAIL] docker not found")
+        return False
+    code, output = _run([docker_path, "build", "-t", "customer-support-openenv", "."], timeout=1200)
+    if code != 0:
+        print("[FAIL] docker build failed")
+        print(output)
+        return False
+    print("[PASS] docker build")
+    return True
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Local pre-submission checks for OpenEnv benchmark")
+    parser.add_argument("--skip-docker", action="store_true", help="Skip docker build check")
+    args = parser.parse_args()
+
+    checks = [
+        _check_env_vars(),
+        _check_openenv_validate(),
+        _check_reset_ping(),
+        _check_docker(args.skip_docker),
+    ]
+    passed = all(checks)
+    print("[DONE] pre-submission checks passed" if passed else "[DONE] pre-submission checks failed")
+    return 0 if passed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pyproject.toml

@@ -0,0 +1,23 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "customer-support-openenv-benchmark"
+version = "0.1.0"
+description = "OpenEnv customer support benchmark environment with graded tasks"
+requires-python = ">=3.10"
+dependencies = [
+    "pandas>=2.2.0",
+    "pydantic>=2.7.0",
+    "openenv-core[core]>=0.2.2",
+    "fastapi>=0.115.0",
+    "uvicorn>=0.34.0",
+    "openai>=1.76.0",
+]
+
+[project.scripts]
+server = "server.app:main"
+
+[tool.setuptools]
+py-modules = ["models", "tasks", "data_loader", "kb", "client", "inference"]

requirements.txt

@@ -0,0 +1,6 @@
+pandas>=2.2.0
+pydantic>=2.7.0
+openenv-core[core]>=0.2.2
+fastapi>=0.115.0
+uvicorn>=0.34.0
+openai>=1.76.0

tasks.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from difflib import SequenceMatcher
+from typing import Callable, Dict, List
+
+
+@dataclass(frozen=True)
+class TaskSpec:
+    task_id: str
+    difficulty: str
+    objective: str
+    grader_name: str
+
+
+TASKS: Dict[str, TaskSpec] = {
+    "easy_classify_respond": TaskSpec(
+        task_id="easy_classify_respond",
+        difficulty="easy",
+        objective="Classify ticket correctly and provide the expected response.",
+        grader_name="grade_easy",
+    ),
+    "medium_kb_grounded_response": TaskSpec(
+        task_id="medium_kb_grounded_response",
+        difficulty="medium",
+        objective="Use the correct KB entry and provide a grounded response.",
+        grader_name="grade_medium",
+    ),
+    "hard_escalation_safety": TaskSpec(
+        task_id="hard_escalation_safety",
+        difficulty="hard",
+        objective="Make the right escalation decision and keep the response safe/polite.",
+        grader_name="grade_hard",
+    ),
+}
+
+
+def task_for_difficulty(difficulty: str) -> TaskSpec:
+    normalized = difficulty.strip().lower()
+    for task in TASKS.values():
+        if task.difficulty == normalized:
+            return task
+    return TASKS["easy_classify_respond"]
+
+
+def grade_easy(expected: dict, history: List[dict]) -> float:
+    classify_ok = any(
+        h["action_type"] == "classify" and _norm(h["content"]) == _norm(expected["expected_category"])
+        for h in history
+    )
+    response_score = _best_response_similarity(expected["expected_response"], history)
+    score = (0.5 if classify_ok else 0.0) + 0.5 * response_score
+    return _clamp01(score)
+
+
+def grade_medium(expected: dict, history: List[dict]) -> float:
+    kb_ok = any(
+        h["action_type"] == "search_kb" and _norm(h["content"]) == _norm(expected["kb_id"])
+        for h in history
+    )
+    response_score = _best_response_similarity(expected["expected_response"], history)
+    score = (0.4 if kb_ok else 0.0) + 0.6 * response_score
+    return _clamp01(score)
+
+
+def grade_hard(expected: dict, history: List[dict]) -> float:
+    escalated = any(h["action_type"] == "escalate" for h in history)
+    escalation_ok = escalated == bool(expected["requires_escalation"])
+    response_score = _best_response_similarity(expected["expected_response"], history)
+    safe = any(h["action_type"] == "respond" and _is_safe(h["content"]) for h in history)
+    score = (0.5 if escalation_ok else 0.0) + 0.25 * response_score + (0.25 if safe else 0.0)
+    return _clamp01(score)
+
+
+GRADERS: Dict[str, Callable[[dict, List[dict]], float]] = {
+    "grade_easy": grade_easy,
+    "grade_medium": grade_medium,
+    "grade_hard": grade_hard,
+}
+
+
+def grade_task(task_id: str, expected: dict, history: List[dict]) -> float:
+    task = TASKS[task_id]
+    grader = GRADERS[task.grader_name]
+    return grader(expected, history)
+
+
+def _best_response_similarity(expected_response: str, history: List[dict]) -> float:
+    expected = _norm(expected_response)
+    if not expected:
+        return 0.0
+    best = 0.0
+    for h in history:
+        if h["action_type"] != "respond":
+            continue
+        candidate = _norm(h["content"])
+        if not candidate:
+            continue
+        best = max(best, SequenceMatcher(None, expected, candidate).ratio())
+    return _clamp01(best)
+
+
+def _is_safe(content: str) -> bool:
+    lowered = _norm(content)
+    unsafe_markers = ("guaranteed fix", "ignore policy", "skip verification", "hack")
+    if any(marker in lowered for marker in unsafe_markers):
+        return False
+    safe_markers = ("thank", "please", "assist", "support", "review", "escalat")
+    return any(marker in lowered for marker in safe_markers)
+
+
+def _norm(value: str) -> str:
+    return " ".join(str(value).strip().lower().split())
+
+
+def _clamp01(value: float) -> float:
+    return max(0.0, min(1.0, round(float(value), 4)))

test_api_smoke.py

@@ -0,0 +1,52 @@
+import asyncio
+
+from client import CustomerSupportEnvClient
+from models import SupportAction
+
+
+async def run_episode(difficulty: str, index: int) -> dict:
+    env = CustomerSupportEnvClient(base_url="http://127.0.0.1:8000")
+    await env.connect()
+    reset_result = await env.reset(difficulty=difficulty, index=index)
+
+    rewards: list[float] = []
+    done = bool(reset_result.done)
+    observation = reset_result.observation
+
+    actions = [
+        SupportAction(action_type="classify", content="technical_issue"),
+        SupportAction(action_type="search_kb", content=observation.kb_id),
+        SupportAction(
+            action_type="respond",
+            content="Thank you for contacting us. We will assist you shortly.",
+        ),
+    ]
+
+    for action in actions:
+        if done:
+            break
+        step_result = await env.step(action)
+        rewards.append(float(step_result.reward or 0.0))
+        done = bool(step_result.done)
+        observation = step_result.observation
+
+    await env.close()
+    return {
+        "difficulty": difficulty,
+        "index": index,
+        "ticket_id": observation.ticket_id,
+        "done": done,
+        "steps": len(rewards),
+        "reward_sum": round(sum(rewards), 4),
+    }
+
+
+async def main() -> None:
+    for difficulty in ("easy", "medium", "hard"):
+        for index in (0, 1):
+            result = await run_episode(difficulty, index)
+            print(result)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())