Add dataset bench and validation agents

backend/app/agents/__init__.py

@@ -0,0 +1,2 @@
+"""Local deterministic agents for app evaluation and validation."""
+

backend/app/agents/evaluation_agent.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from ..catalog import get_catalog
+from ..config import settings
+
+
+REQUIRED_TASKS = {
+    "dataset.upload": "covered_by_api",
+    "dataset.video_frames": "job",
+    "segmodel.train": "job",
+    "segmodel.predict": "job",
+    "yolo.heatmap": "job",
+    "mmseg.flops_fps": "job",
+    "analysis.all": "job",
+}
+
+
+def evaluate_project() -> dict:
+    """Return product/implementation suggestions for the current web platform."""
+    frontend = settings.project_root / "frontend" / "src" / "main.tsx"
+    backend = settings.project_root / "backend" / "app" / "main.py"
+    readme = settings.project_root / "README.md"
+    catalog = get_catalog()
+    checks = []
+    suggestions = []
+
+    frontend_text = frontend.read_text(encoding="utf-8") if frontend.exists() else ""
+    backend_text = backend.read_text(encoding="utf-8") if backend.exists() else ""
+    readme_text = readme.read_text(encoding="utf-8") if readme.exists() else ""
+
+    expectations = {
+        "left_nav_dataset": "数据集" in frontend_text and "#datasets" in frontend_text,
+        "upload_ui": "uploadDatasetFiles" in frontend_text and "labels" in frontend_text and "masks" in frontend_text,
+        "loss_result_ui": "loss" in frontend_text.lower() and "heatmap" in frontend_text.lower(),
+        "dataset_api": "/api/datasets" in backend_text and "api_upload_dataset_files" in backend_text,
+        "no_weight_to_gitea": "Do not push" in readme_text and "check_no_weight_git" in readme_text,
+        "all_core_tasks": all(task in catalog["task_types"] for task in REQUIRED_TASKS if REQUIRED_TASKS[task] == "job"),
+    }
+
+    for name, passed in expectations.items():
+        checks.append({"name": name, "passed": bool(passed)})
+        if not passed:
+            suggestions.append(f"Improve missing capability: {name}")
+
+    if len(catalog["mmseg_algorithms"]) < 31:
+        suggestions.append("MMSeg algorithm catalog should expose all 31 algorithm generators.")
+    if len(catalog["segmodel_architectures"]) < 12:
+        suggestions.append("SegModel catalog should expose all 12 supported architectures.")
+    if not suggestions:
+        suggestions.append("Current platform covers the requested control-plane features; next focus is real dataset/training acceptance tests.")
+
+    score = sum(1 for item in checks if item["passed"]) / max(len(checks), 1)
+    return {
+        "agent": "evaluation_suggestion_agent",
+        "score": round(score, 3),
+        "checks": checks,
+        "suggestions": suggestions,
+    }
+

backend/app/agents/validation_agent.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import tempfile
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+from ..catalog import get_catalog
+from ..config import settings
+from ..modules.system.service import get_conda_envs, get_gpus
+from ..modules.weights.service import load_manifest
+
+
+def _run(command: list[str], cwd: Path | None = None, timeout: int = 60) -> dict:
+    result = subprocess.run(command, cwd=cwd, capture_output=True, text=True, timeout=timeout)
+    return {
+        "command": command,
+        "returncode": result.returncode,
+        "stdout": result.stdout[-4000:],
+        "stderr": result.stderr[-4000:],
+        "passed": result.returncode == 0,
+    }
+
+
+def _fetch(url: str, timeout: int = 5) -> dict:
+    try:
+        with urllib.request.urlopen(url, timeout=timeout) as response:
+            body = response.read(20000).decode("utf-8", errors="replace")
+            return {"url": url, "status": response.status, "body": body, "passed": 200 <= response.status < 300}
+    except urllib.error.HTTPError as exc:
+        body = exc.read(20000).decode("utf-8", errors="replace")
+        return {"url": url, "status": exc.code, "body": body, "passed": False}
+    except Exception as exc:
+        return {"url": url, "error": str(exc), "passed": False}
+
+
+def validate_project(run_build: bool = False) -> dict:
+    """Validate current runtime readiness without launching heavy training."""
+    checks = []
+    catalog = get_catalog()
+    manifest = load_manifest()
+
+    checks.append({"name": "catalog_has_yolo_heatmap", "passed": "yolo.heatmap" in catalog["task_types"]})
+    checks.append({"name": "catalog_has_mmseg_31_algs", "passed": len(catalog["mmseg_algorithms"]) >= 31})
+    checks.append({"name": "weights_manifest_present", "passed": manifest.get("count", 0) >= 1})
+    checks.append({"name": "gpus_query", "passed": bool(get_gpus().get("available"))})
+    env_names = [item["name"] for item in get_conda_envs().get("envs", [])]
+    checks.append({"name": "seg_smp_env_exists", "passed": "seg_smp" in env_names})
+
+    smoke = _run(
+        [
+            "conda",
+            "run",
+            "-n",
+            "seg_smp",
+            "python",
+            "-c",
+            "import fastapi, uvicorn, torch, cv2, segmentation_models_pytorch, ultralytics, albumentations, mmengine, mmseg, mmcv; print(torch.__version__, torch.cuda.is_available())",
+        ],
+        cwd=settings.project_root,
+    )
+    checks.append({"name": "seg_smp_backend_smoke", "passed": smoke["passed"], "detail": smoke})
+
+    no_weight = _run(["bash", "scripts/check_no_weight_git.sh"], cwd=settings.project_root)
+    checks.append({"name": "no_weight_in_git", "passed": no_weight["passed"], "detail": no_weight})
+
+    if os.getenv("SEG_VALIDATE_LIVE", "1") == "1":
+        backend_url = os.getenv("SEG_VALIDATE_BACKEND_URL", "http://127.0.0.1:8010")
+        frontend_url = os.getenv("SEG_VALIDATE_FRONTEND_URL", "http://127.0.0.1:5173")
+        health = _fetch(f"{backend_url}/api/health")
+        datasets = _fetch(f"{backend_url}/api/datasets")
+        frontend = _fetch(frontend_url)
+        checks.append({"name": "live_backend_health", "passed": health["passed"] and '"ok":true' in health.get("body", "").replace(" ", ""), "detail": health})
+        checks.append({"name": "live_dataset_api", "passed": datasets["passed"] and datasets.get("body", "").lstrip().startswith("["), "detail": datasets})
+        checks.append({"name": "live_frontend_index", "passed": frontend["passed"] and "Seg Data Server" in frontend.get("body", ""), "detail": frontend})
+
+    if run_build:
+        tests = _run(["conda", "run", "-n", settings.backend_conda_env, "python", "-m", "pytest", "-q"], cwd=settings.project_root, timeout=120)
+        checks.append({"name": "backend_tests", "passed": tests["passed"], "detail": tests})
+        build = _run(["npm", "run", "build"], cwd=settings.project_root / "frontend", timeout=120)
+        checks.append({"name": "frontend_build", "passed": build["passed"], "detail": build})
+
+    passed = all(item["passed"] for item in checks)
+    return {
+        "agent": "validation_agent",
+        "passed": passed,
+        "checks": checks,
+    }
+
+
+def write_validation_report(path: Path, run_build: bool = False) -> dict:
+    result = validate_project(run_build=run_build)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
+    return result