| """Hackathon baseline runner for Model Release Env.""" |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| import json |
| import os |
| import re |
| import sys |
| from typing import Any, Dict, List, Optional |
|
|
|
|
| def _sanitize_sys_path() -> None: |
| current_tag = f"python{sys.version_info.major}.{sys.version_info.minor}" |
| sys.path[:] = [ |
| entry |
| for entry in sys.path |
| if entry == "" or "/site-packages" not in entry or current_tag in entry |
| ] |
|
|
|
|
| _sanitize_sys_path() |
|
|
| from openai import OpenAI |
|
|
| try: |
| from model_release_env import ModelReleaseAction, ModelReleaseEnv |
| except ImportError: |
| from client import ModelReleaseEnv |
| from models import ModelReleaseAction |
|
|
|
|
| API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1") |
| MODEL_NAME = os.getenv("MODEL_NAME", "Qwen/Qwen2.5-72B-Instruct") |
| LOCAL_IMAGE_NAME = os.getenv("LOCAL_IMAGE_NAME", "model-release-env:latest") |
| ENV_BASE_URL = os.getenv("ENV_BASE_URL") |
| BENCHMARK = os.getenv("BENCHMARK_NAME", "model_release_env") |
|
|
|
|
| def _get_env_int(name: str, default: str) -> int: |
| try: |
| return int(os.getenv(name, default)) |
| except ValueError as exc: |
| raise SystemExit(f"Invalid integer for {name}: {exc}") from exc |
|
|
|
|
| def _get_env_float(name: str, default: str) -> float: |
| try: |
| return float(os.getenv(name, default)) |
| except ValueError as exc: |
| raise SystemExit(f"Invalid float for {name}: {exc}") from exc |
|
|
|
|
| MAX_STEPS = _get_env_int("MODEL_RELEASE_MAX_STEPS", "8") |
| SUCCESS_THRESHOLD = _get_env_float("MODEL_RELEASE_SUCCESS_THRESHOLD", "0.75") |
|
|
| DEFAULT_TASKS = [ |
| "card_completion_easy", |
| "policy_alignment_medium", |
| "launch_gate_hard", |
| ] |
|
|
| HEURISTIC_PLANS: Dict[str, List[Dict[str, str]]] = { |
| "card_completion_easy": [ |
| {"action_type": "inspect", "target": "release_brief"}, |
| {"action_type": "inspect", "target": "eval_sheet"}, |
| {"action_type": "inspect", "target": "risk_note"}, |
| { |
| "action_type": "set_field", |
| "target": "base_model", |
| "value": "Qwen2.5-7B-Instruct", |
| }, |
| { |
| "action_type": "set_field", |
| "target": "eval_summary", |
| "value": "gsm8k=0.78; math500=0.61; aime24=0.18", |
| }, |
| { |
| "action_type": "set_field", |
| "target": "limitations", |
| "value": "Fails on long derivations and is not approved for medical advice.", |
| }, |
| {"action_type": "set_decision", "value": "beta"}, |
| {"action_type": "submit"}, |
| ], |
| "policy_alignment_medium": [ |
| {"action_type": "inspect", "target": "license_audit"}, |
| {"action_type": "inspect", "target": "safety_report"}, |
| {"action_type": "inspect", "target": "policy_note"}, |
| { |
| "action_type": "set_field", |
| "target": "license", |
| "value": "apache-2.0", |
| }, |
| { |
| "action_type": "set_field", |
| "target": "safety_guardrails", |
| "value": "prompt_injection_filter; pii_redaction", |
| }, |
| { |
| "action_type": "set_field", |
| "target": "blocked_use_cases", |
| "value": "legal advice; medical advice", |
| }, |
| {"action_type": "set_decision", "value": "beta"}, |
| {"action_type": "submit"}, |
| ], |
| "launch_gate_hard": [ |
| {"action_type": "inspect", "target": "regression_report"}, |
| {"action_type": "inspect", "target": "red_team_note"}, |
| {"action_type": "inspect", "target": "launch_policy"}, |
| { |
| "action_type": "set_field", |
| "target": "known_regressions", |
| "value": "latency; throughput", |
| }, |
| { |
| "action_type": "set_field", |
| "target": "launch_blockers", |
| "value": "chain-of-thought extraction bypass", |
| }, |
| { |
| "action_type": "set_field", |
| "target": "public_summary", |
| "value": "release on hold pending latency and safety fixes.", |
| }, |
| {"action_type": "set_decision", "value": "hold"}, |
| {"action_type": "submit"}, |
| ], |
| } |
|
|
|
|
| def _task_names() -> List[str]: |
| raw = os.getenv("MODEL_RELEASE_TASKS") |
| if not raw: |
| return list(DEFAULT_TASKS) |
| return [item.strip() for item in raw.split(",") if item.strip()] |
|
|
|
|
| def _compact_action(action: ModelReleaseAction) -> str: |
| value = action.value.replace(" ", "_") if action.value else "" |
| if action.action_type == "inspect": |
| return f"inspect({action.target})" |
| if action.action_type == "set_field": |
| return f"set_field({action.target}={value})" |
| if action.action_type == "set_decision": |
| return f"set_decision({value})" |
| return "submit()" |
|
|
|
|
| def _stderr(message: str) -> None: |
| print(message, file=sys.stderr) |
|
|
|
|
| def _redact_message(message: str) -> str: |
| redacted = re.sub(r"hf_[A-Za-z0-9]+", "hf_[REDACTED]", message) |
| redacted = re.sub(r"sk_[A-Za-z0-9]+", "sk_[REDACTED]", redacted) |
| redacted = re.sub(r"https://[^\s:@]+:[^\s@]+@", "https://[REDACTED]@", redacted) |
| return redacted |
|
|
|
|
| def _llm_client() -> Optional[OpenAI]: |
| token = os.getenv("HF_TOKEN") |
| if not token: |
| return None |
| return OpenAI(base_url=API_BASE_URL, api_key=token) |
|
|
|
|
| def _extract_json_block(content: str) -> Dict[str, Any]: |
| match = re.search(r"\{.*\}", content, re.DOTALL) |
| if not match: |
| raise ValueError("No JSON object found in model response") |
| return json.loads(match.group(0)) |
|
|
|
|
| def _observation_prompt(observation: Any) -> str: |
| payload = { |
| "task_name": observation.task_name, |
| "goal": observation.goal, |
| "document_index": observation.document_index, |
| "visible_documents": observation.visible_documents, |
| "package_snapshot": observation.package_snapshot, |
| "checklist_status": observation.checklist_status, |
| "critical_gaps": observation.critical_gaps, |
| "available_fields": observation.available_fields, |
| "available_decisions": observation.available_decisions, |
| "inspected_documents": observation.inspected_documents, |
| "remaining_steps": observation.remaining_steps, |
| "score": observation.score, |
| "last_action_error": observation.last_action_error, |
| } |
| return json.dumps(payload, indent=2, sort_keys=True) |
|
|
|
|
| def _heuristic_action(task_name: str, step_index: int) -> ModelReleaseAction: |
| plan = HEURISTIC_PLANS[task_name] |
| if step_index >= len(plan): |
| return ModelReleaseAction(action_type="submit") |
| return ModelReleaseAction(**plan[step_index]) |
|
|
|
|
| def _model_action( |
| client: OpenAI, |
| task_name: str, |
| observation: Any, |
| ) -> ModelReleaseAction: |
| system_prompt = ( |
| "You are operating an OpenEnv release-readiness environment. " |
| "Return exactly one JSON object with keys action_type, target, and value. " |
| "Allowed action_type values: inspect, set_field, set_decision, submit. " |
| "Use inspect before editing. Keep values compact and deterministic." |
| ) |
| user_prompt = ( |
| f"Task: {task_name}\n" |
| "Choose the single best next action given the observation below.\n" |
| "Observation JSON:\n" |
| f"{_observation_prompt(observation)}" |
| ) |
| response = client.chat.completions.create( |
| model=MODEL_NAME, |
| temperature=0.0, |
| max_tokens=220, |
| messages=[ |
| {"role": "system", "content": system_prompt}, |
| {"role": "user", "content": user_prompt}, |
| ], |
| ) |
| if not response.choices: |
| raise ValueError("Model returned no choices") |
| content = response.choices[0].message.content or "" |
| return ModelReleaseAction(**_extract_json_block(content)) |
|
|
|
|
| async def _create_env() -> ModelReleaseEnv: |
| if ENV_BASE_URL: |
| return ModelReleaseEnv(base_url=ENV_BASE_URL) |
| return await ModelReleaseEnv.from_docker_image(LOCAL_IMAGE_NAME) |
|
|
|
|
| async def _run_task(env: ModelReleaseEnv, task_name: str, llm: Optional[OpenAI]) -> float: |
| print(f"[START] benchmark={BENCHMARK} task={task_name}") |
| result = await env.reset(task_name=task_name) |
| step_index = 0 |
|
|
| while not result.done and step_index < MAX_STEPS: |
| try: |
| if llm is None: |
| action = _heuristic_action(task_name, step_index) |
| else: |
| action = _model_action(llm, task_name, result.observation) |
| except Exception as exc: |
| _stderr( |
| f"planner fallback for {task_name}: {_redact_message(str(exc))}" |
| ) |
| action = _heuristic_action(task_name, step_index) |
|
|
| result = await env.step(action) |
| error = result.observation.last_action_error or "null" |
| reward = 0.0 if result.reward is None else float(result.reward) |
| print( |
| f"[STEP] action={_compact_action(action)} reward={reward:.2f} " |
| f"done={str(result.done)} error={error}" |
| ) |
| step_index += 1 |
|
|
| score = float(result.observation.score) |
| success = score >= SUCCESS_THRESHOLD |
| print(f"[END] success={str(success)} score={score:.2f}") |
| return score |
|
|
|
|
| async def main() -> int: |
| llm = _llm_client() |
| env = await _create_env() |
| scores: List[float] = [] |
|
|
| async with env: |
| for task_name in _task_names(): |
| scores.append(await _run_task(env, task_name, llm)) |
|
|
| average_score = sum(scores) / len(scores) if scores else 0.0 |
| _stderr(f"average_score={average_score:.2f}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(asyncio.run(main())) |
