AgentEvalTool/backend/plugins/openclaw/agenteval_skill.py

"""OpenClaw Skill for AgentEvalTool — HTTP API mode.

Triggers an evaluation run via the AgentEvalTool REST API, polls until
completion, and returns a structured Chinese-language summary suitable for
display in an OpenClaw conversation.

Configuration (passed via OpenClaw skill params):
    api_base_url    Base URL of AgentEvalTool (e.g. http://192.168.8.145:8001)
    api_key         X-API-Key header value (if auth is enabled, else omit)
    target_id       ID of the registered evaluation target
    scenario_id     ID of the evaluation scenario
    poll_interval   Seconds between status polls (default: 3)
    timeout         Max seconds to wait for completion (default: 300)
"""

import time
from typing import Any

import httpx


class AgentEvalSkill:
    """OpenClaw skill: trigger evaluation and return a structured summary."""

    def __init__(self, config: dict[str, Any]):
        self.api_base = config["api_base_url"].rstrip("/")
        self.api_key = config.get("api_key")
        self.target_id = config["target_id"]
        self.scenario_id = config["scenario_id"]
        self.poll_interval = float(config.get("poll_interval", 3))
        self.timeout = float(config.get("timeout", 300))

    def _headers(self) -> dict[str, str]:
        h = {"Content-Type": "application/json"}
        if self.api_key:
            h["X-API-Key"] = self.api_key
        return h

    def run(self) -> dict[str, Any]:
        with httpx.Client(base_url=self.api_base, headers=self._headers(), timeout=30) as client:
            # 1. Start the run
            resp = client.post("/api/runs", json={
                "target_id": self.target_id,
                "scenario_id": self.scenario_id,
            })
            if resp.status_code != 200:
                return {"ok": False, "error": f"启动评测失败: HTTP {resp.status_code} {resp.text}"}

            run = resp.json()
            run_id: str = run["id"]

            # 2. Poll until completion or timeout
            deadline = time.time() + self.timeout
            while time.time() < deadline:
                time.sleep(self.poll_interval)
                status_resp = client.get(f"/api/runs/{run_id}")
                if status_resp.status_code != 200:
                    continue
                run = status_resp.json()
                if run.get("status") in ("completed", "failed"):
                    break
            else:
                return {
                    "ok": False,
                    "run_id": run_id,
                    "error": f"评测超时（>{self.timeout}s 未完成）",
                }

            # 3. Fetch structured report
            report_resp = client.get(f"/api/reports/{run_id}")
            report = report_resp.json() if report_resp.status_code == 200 else {}

            return self._format_result(run, report)

    def _format_result(self, run: dict[str, Any], report: dict[str, Any]) -> dict[str, Any]:
        status = run.get("status", "unknown")
        summary = run.get("summary") or {}
        pass_rate = summary.get("pass_rate", 0.0)
        passed_cases = summary.get("passed_cases", 0)
        total_cases = summary.get("total_cases", 0)
        passed_rules = summary.get("passed_rules", 0)
        total_rules = summary.get("total_rules", 0)

        # Build Chinese natural-language summary for display in chat
        if status == "completed":
            verdict = "✅ 评测完成" if pass_rate >= 0.8 else "⚠️ 评测完成（有失败项）"
            summary_text = (
                f"{verdict}\n"
                f"• 评测对象：{report.get('target_name', self.target_id)}\n"
                f"• 评测场景：{report.get('scenario_name', self.scenario_id)}\n"
                f"• 通过率：{pass_rate * 100:.1f}%\n"
                f"• 用例：{passed_cases}/{total_cases} 通过\n"
                f"• 规则：{passed_rules}/{total_rules} 通过\n"
                f"• 报告链接：{self.api_base}/api/reports/{run['id']}"
            )
        else:
            error_info = summary.get("error", {})
            if isinstance(error_info, dict):
                error_msg = error_info.get("message", str(error_info))
            else:
                error_msg = str(error_info)
            summary_text = f"❌ 评测失败：{error_msg}"

        return {
            "ok": status == "completed",
            "run_id": run.get("id"),
            "status": status,
            "pass_rate": pass_rate,
            "summary_text": summary_text,
            "report": report,
        }


def execute(config: dict[str, Any]) -> dict[str, Any]:
    """OpenClaw skill entrypoint."""
    return AgentEvalSkill(config).run()