739d586aec
- EvalRun.triggered_by 全链路（manual/ai_assistant/cli）+ 迁移 b7d4e6f81c22 - 标准 agenteval-run SKILL.md 纳入版本管理，deploy 脚本同步 + API Key 注入 - 简单登录：AGENTEVAL_ADMIN_PASSWORD + HMAC 会话 token，require_auth 双凭据 - 对比报告限同场景（400）+ 空 results 误判修复 - /api/stats/dashboard 扩展聚合；/api/runs 返回场景/对象名 - 测试 218 → 232
121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
"""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,
|
|
"triggered_by": "ai_assistant",
|
|
},
|
|
)
|
|
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()
|