feat: VersionPipeline 反幻觉门控 + AgentScorer 评分系统 + /api/daily/scores 端点

- tests/test_version_gate.py: 3 个测试验证版本状态机（developing→testing→done）和开发 agent 不能直接 mark_done 的门控规则 - src/nmfs_agents/tools/agent_score.py: AgentScorer 类，四维评分（完成率40%+质量30%+ 自主度20%+协作10%），支持 get_score/get_all_scores/get_mode_recommendation - tests/test_agent_score.py: 4 个测试覆盖满分/失败/模式推荐/全量查询场景 - src/nmfs_agents/dashboard/api.py: 新增 GET /api/daily/scores 端点 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-09 23:42:32 +08:00
parent d0bad85afc
commit d9d83b8f1c
4 changed files with 236 additions and 0 deletions
@@ -917,6 +917,11 @@ def make_app(db_path: Path | None = None, config_dir: Path | None = None,
    def api_daily_dates():
        return _daily_db.get_available_dates(limit=30)

+    @app.get("/api/daily/scores")
+    def api_agent_scores(days: int = 7):
+        from nmfs_agents.tools.agent_score import AgentScorer
+        return AgentScorer().get_all_scores(days=days)
+
    # ── Market Insight API ─────────────────────────────────────────────
    from nmfs_agents.tools.project_memory import ProjectMemory as _PM
    _memory = memory if memory is not None else _PM()
@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import logging
+import sqlite3
+from datetime import datetime
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_DB = Path("data/agent_scores.db")
+
+
+class AgentScorer:
+    """Agent 绩效评分系统。
+
+    评分维度：完成率(40%) + 质量(30%) + 自主度(20%) + 协作(10%)
+    """
+
+    def __init__(self, db_path: Path = _DEFAULT_DB) -> None:
+        self._db_path = db_path
+        self._db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init_db()
+
+    def _connect(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(str(self._db_path))
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _init_db(self) -> None:
+        with self._connect() as conn:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS agent_records (
+                    id            INTEGER PRIMARY KEY AUTOINCREMENT,
+                    agent_role    TEXT NOT NULL,
+                    status        TEXT NOT NULL,
+                    quality_pass  INTEGER NOT NULL DEFAULT 1,
+                    mode          TEXT NOT NULL DEFAULT 'auto',
+                    spawned_valid INTEGER NOT NULL DEFAULT 0,
+                    created_at    TEXT NOT NULL
+                )
+            """)
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_as_role ON agent_records(agent_role)"
+            )
+
+    def record(
+        self,
+        agent_role: str,
+        status: str,
+        quality_pass: bool = True,
+        mode: str = "auto",
+        spawned_valid: int = 0,
+    ) -> None:
+        """记录一次 agent 执行结果。"""
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT INTO agent_records"
+                "(agent_role,status,quality_pass,mode,spawned_valid,created_at) "
+                "VALUES(?,?,?,?,?,?)",
+                (
+                    agent_role,
+                    status,
+                    1 if quality_pass else 0,
+                    mode,
+                    spawned_valid,
+                    datetime.now().strftime("%Y-%m-%d %H:%M"),
+                ),
+            )
+
+    def get_score(self, agent_role: str, days: int = 7) -> dict:
+        """计算最近 N 天的综合评分（0-100）。"""
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM agent_records WHERE agent_role=? AND "
+                "created_at >= datetime('now', ?)",
+                (agent_role, f"-{days} days"),
+            ).fetchall()
+        if not rows:
+            return {
+                "total": 50,
+                "completion_rate": 0.5,
+                "quality": 0.5,
+                "autonomy": 0.5,
+                "collab": 0.0,
+                "sample_count": 0,
+            }
+        total = len(rows)
+        done = sum(1 for r in rows if r["status"] == "done")
+        quality = sum(1 for r in rows if r["quality_pass"])
+        auto_done = sum(
+            1 for r in rows if r["mode"] == "auto" and r["status"] == "done"
+        )
+        collab = sum(r["spawned_valid"] for r in rows)
+
+        completion = done / total
+        quality_rate = quality / total
+        autonomy = auto_done / max(done, 1)
+        collab_score = min(1.0, collab / max(total, 1))
+
+        total_score = int(
+            completion * 40 + quality_rate * 30 + autonomy * 20 + collab_score * 10
+        )
+        return {
+            "total": total_score,
+            "completion_rate": round(completion, 2),
+            "quality": round(quality_rate, 2),
+            "autonomy": round(autonomy, 2),
+            "collab": round(collab_score, 2),
+            "sample_count": total,
+        }
+
+    def get_all_scores(self, days: int = 7) -> dict[str, dict]:
+        """返回所有 agent 角色在最近 N 天的评分字典。"""
+        with self._connect() as conn:
+            roles = conn.execute(
+                "SELECT DISTINCT agent_role FROM agent_records WHERE "
+                "created_at >= datetime('now', ?)",
+                (f"-{days} days",),
+            ).fetchall()
+        return {r["agent_role"]: self.get_score(r["agent_role"], days) for r in roles}
+
+    def get_mode_recommendation(self, agent_role: str) -> str:
+        """根据评分推荐运行模式：auto（高分）或 confirm（低分）。"""
+        score = self.get_score(agent_role)
+        if score["sample_count"] < 3:
+            return "auto"
+        if score["total"] >= 80:
+            return "auto"
+        if score["total"] < 50:
+            return "confirm"
+        return "auto"
@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import pytest
+
+from nmfs_agents.tools.agent_score import AgentScorer
+
+
+def test_score_all_done(tmp_path):
+    """全部完成且 auto 模式、有子任务时，评分应为 100。
+
+    公式：完成率(40) + 质量(30) + 自主度(20) + 协作(10) = 100
+    """
+    scorer = AgentScorer(db_path=tmp_path / "s.db")
+    scorer.record("algo-antishake", "done", quality_pass=True, mode="auto", spawned_valid=1)
+    scorer.record("algo-antishake", "done", quality_pass=True, mode="auto", spawned_valid=1)
+    score = scorer.get_score("algo-antishake")
+    assert score["completion_rate"] == 1.0
+    assert score["total"] == 100
+
+
+def test_score_with_failures(tmp_path):
+    """有失败任务时，完成率应为 0.5。"""
+    scorer = AgentScorer(db_path=tmp_path / "s.db")
+    scorer.record("rtp-researcher", "done", quality_pass=True, mode="auto")
+    scorer.record("rtp-researcher", "failed", quality_pass=False, mode="auto")
+    score = scorer.get_score("rtp-researcher")
+    assert score["completion_rate"] == 0.5
+
+
+def test_score_affects_mode_recommendation(tmp_path):
+    """低分 agent 应推荐 confirm 模式。"""
+    scorer = AgentScorer(db_path=tmp_path / "s.db")
+    for _ in range(5):
+        scorer.record("bad-agent", "failed", quality_pass=False, mode="auto")
+    rec = scorer.get_mode_recommendation("bad-agent")
+    assert rec == "confirm"
+
+
+def test_get_all_scores(tmp_path):
+    """get_all_scores 应覆盖所有已记录的 agent 角色。"""
+    scorer = AgentScorer(db_path=tmp_path / "s.db")
+    scorer.record("agent-a", "done", quality_pass=True, mode="auto")
+    scorer.record("agent-b", "failed", quality_pass=False, mode="auto")
+    all_scores = scorer.get_all_scores()
+    assert "agent-a" in all_scores
+    assert "agent-b" in all_scores
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from nmfs_agents.tools.version_pipeline import VersionPipeline
+
+
+def test_version_pipeline_state_machine(tmp_path):
+    """版本状态流转：developing → testing。"""
+    vp = VersionPipeline(db_path=tmp_path / "vp.db")
+    vp.start_version("antishake", "v1", "初始版本")
+
+    status = vp.get_status("antishake")
+    assert status[0]["status"] == "developing"
+
+    vp.start_testing("antishake", "v1")
+    status = vp.get_status("antishake")
+    assert status[0]["status"] == "testing"
+
+
+def test_tester_agent_can_mark_version_done(tmp_path):
+    """只有 TesterAgent 可以将版本标记为 done。"""
+    vp = VersionPipeline(db_path=tmp_path / "vp.db")
+    vp.start_version("antishake", "v1", "test")
+    vp.start_testing("antishake", "v1")
+    vp.mark_done("antishake", "v1")
+
+    status = vp.get_status("antishake")
+    assert status[0]["status"] == "done"
+
+
+def test_dev_agent_cannot_directly_mark_done(tmp_path):
+    """开发 agent 的 [测试通过] 标签应被系统忽略（不能直接调 mark_done）。
+
+    门控点：developer/algo 等 agent 的输出中即使出现 [测试通过] 标签，
+    系统只应调用 vp.start_testing()，不得调用 vp.mark_done()。
+    """
+    vp = VersionPipeline(db_path=tmp_path / "vp.db")
+    vp.start_version("antishake", "v1", "dev version")
+
+    # 模拟开发 agent 输出中出现 [测试通过] 标签
+    report = "[版本] v1: 初始实现\n[测试通过] v1\n[后续] 下轮改进"
+    for m in re.finditer(r"\[测试通过\]\s*(v\d+)", report):
+        version = m.group(1)
+        # 开发 agent 的正确行为：只进入测试状态，不 mark_done
+        vp.start_testing("antishake", version)
+        # 注意：不调用 vp.mark_done("antishake", version)
+
+    status = vp.get_status("antishake")
+    v1 = next(s for s in status if s["version"] == "v1")
+    assert v1["status"] == "testing"   # 进入测试，等待 TesterAgent
+    assert v1["status"] != "done"      # 开发 agent 不能直接 done