Files

T

wangweiandCopilot 8ebf554989 docs: add advisor comparison implementation plan

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

2026-07-02 16:37:38 +08:00

44 KiB

Raw Blame History

优化建议历史对比 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 报告详情页「优化建议」区域自动显示"相比同场景上一次运行"的顾问诊断差异摘要（落地架构设计 §11 回归复测方法论）。

Architecture: 新增 webapp 层只读服务 webapp/services/advisor_comparison.py，现场调用已公开的 rag_eval.advisor.diagnose() 分别对当前 run 和"同 scenario_name 最近一次前序 run"的 scores.csv 重新计算诊断，按指标名分类差异状态（resolved/regressed/still_triggered/new_metric），挂载到 ReportData.advisor_comparison。零改动 rag_eval/advisor/ 内部规则与三条写入入口；零新增持久化文件。

Tech Stack: Python 3.12、Pydantic v2、pandas（复用既有 run_reader helper）、pytest（-v 详细输出）、原生 JS（report.js IIFE 模块，复用既有 MetricPresenter.deltaInfo()）。

Global Constraints

不新增任何持久化文件（不写 diagnoses.json 或类似文件）；对比结果每次报告详情页请求时现场用 diagnose() 重新计算，始终反映当前代码里的最新阈值规则。
零改动 rag_eval/advisor/rules.py、llm_analyzer.py、writer.py、__init__.py。
零改动三条写入入口：rag_eval/execution/runner.py、webapp/services/score_job_manager.py、webapp/services/session_score_manager.py。
只对比"同 scenario_name、时间上最近的前一次运行"（不支持手动选任意两个 run，不做多跳历史链）。
对比区域只展示"有变化或仍有问题"的指标；两次都健康的指标不生成条目。
找不到上一次运行、数据缺失/损坏、diagnose() 内部异常等情况一律静默降级为 None，绝不影响报告详情页其余部分渲染或抛出异常。
设计依据：docs/superpowers/specs/2026-07-02-advisor-comparison-design.md（本计划每个任务对应该文档的具体章节，不要偏离）。
测试运行命令：C:\software\Python312\python.exe -m pytest tests/<file> -v（pytest 9.0.3，本机无 venv，需用此绝对路径）。

Task 1: `AdvisorComparisonEntry` / `AdvisorComparison` 模型 + `ReportData` 字段

Files:

Modify: webapp/models.py
Test: tests/test_advisor_comparison_models.py

Interfaces:

Produces:
- AdvisorComparisonEntry(metric: str, status: Literal["resolved","regressed","still_triggered","new_metric"], previous_score: float|None, previous_severity: str|None, current_score: float|None, current_severity: str|None)
- AdvisorComparison(previous_run_id: str, previous_finished_at: str, previous_judge_model: str, current_judge_model: str, judge_model_changed: bool, entries: list[AdvisorComparisonEntry])
- ReportData.advisor_comparison: AdvisorComparison | None (default None)
Step 1: Write the failing test

Create tests/test_advisor_comparison_models.py:

"""Tests for the AdvisorComparison Pydantic models and their default wiring."""
from __future__ import annotations

from webapp.models import AdvisorComparison, AdvisorComparisonEntry, ReportData


def test_advisor_comparison_entry_accepts_all_statuses():
    for status in ("resolved", "regressed", "still_triggered", "new_metric"):
        entry = AdvisorComparisonEntry(
            metric="faithfulness",
            status=status,
            previous_score=0.5,
            previous_severity="warning",
            current_score=0.9,
            current_severity=None,
        )
        assert entry.status == status


def test_advisor_comparison_entry_rejects_unknown_status():
    import pydantic

    try:
        AdvisorComparisonEntry(metric="faithfulness", status="unknown_status")
        raised = False
    except pydantic.ValidationError:
        raised = True
    assert raised


def test_advisor_comparison_defaults():
    comparison = AdvisorComparison(
        previous_run_id="run-1",
        previous_finished_at="2026-01-01T00:00:00+00:00",
    )
    assert comparison.previous_judge_model == ""
    assert comparison.current_judge_model == ""
    assert comparison.judge_model_changed is False
    assert comparison.entries == []


def test_report_data_defaults_advisor_comparison_to_none():
    report = ReportData()
    assert report.advisor_comparison is None


def test_report_data_accepts_advisor_comparison():
    comparison = AdvisorComparison(previous_run_id="run-1", previous_finished_at="t")
    report = ReportData(advisor_comparison=comparison)
    assert report.advisor_comparison == comparison

Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison_models.py -v Expected: FAIL with ImportError: cannot import name 'AdvisorComparison' from 'webapp.models'

Step 3: Write minimal implementation

In webapp/models.py, find the import line:

from typing import Any

Replace with:

from typing import Any, Literal

Add these two new model classes right before class ReportData(BaseModel)::

class AdvisorComparisonEntry(BaseModel):
    """One metric's diagnosis delta between the current run and its predecessor."""

    metric: str
    status: Literal["resolved", "regressed", "still_triggered", "new_metric"]
    previous_score: float | None = None
    previous_severity: str | None = None
    current_score: float | None = None
    current_severity: str | None = None


class AdvisorComparison(BaseModel):
    """Advisor-diagnosis comparison against the same scenario's previous run."""

    previous_run_id: str
    previous_finished_at: str
    previous_judge_model: str = ""
    current_judge_model: str = ""
    judge_model_changed: bool = False
    entries: list[AdvisorComparisonEntry] = Field(default_factory=list)

Then, inside the existing ReportData class, find its last field:

    token_usage: dict[str, dict[str, int]] = Field(
        default_factory=dict,
        description="按模型累计的 token 用量：{model: {input_tokens, output_tokens, calls}}。",
    )

Replace with (adds the new field immediately after, still inside the class body):

    token_usage: dict[str, dict[str, int]] = Field(
        default_factory=dict,
        description="按模型累计的 token 用量：{model: {input_tokens, output_tokens, calls}}。",
    )
    advisor_comparison: AdvisorComparison | None = Field(
        default=None,
        description="相比同场景上一次运行的顾问诊断差异；无可比对象时为 None。",
    )

Step 4: Run test to verify it passes

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison_models.py -v Expected: 5 passed

Step 5: Commit

git add webapp/models.py tests/test_advisor_comparison_models.py
git commit -m "feat(advisor-comparison): add AdvisorComparison models and ReportData field"

Task 2: `find_previous_run()` — locate the same-scenario predecessor run

Files:

Create: webapp/services/advisor_comparison.py
Test: tests/test_advisor_comparison.py

Interfaces:

Consumes: webapp.services.run_reader.list_run_summaries(extra_roots=None) -> list[RunSummary] (existing); webapp.models.RunSummary (existing, has run_id, scenario_name, finished_at, judge_model, metrics, output_path)
Produces: find_previous_run(scenario_name: str, current_run_id: str, current_finished_at: str, extra_roots: list[Path] | None = None) -> RunSummary | None, used by Task 3.
Step 1: Write the failing test

Create tests/test_advisor_comparison.py:

"""Tests for webapp.services.advisor_comparison: same-scenario run comparison."""
from __future__ import annotations

import json
from pathlib import Path

import pandas as pd

from webapp.services.advisor_comparison import find_previous_run


def _write_fake_run(
    root: Path,
    run_id: str,
    scenario_name: str,
    finished_at: str,
    judge_model: str,
    rows: list[dict],
) -> Path:
    """Write a minimal run directory discoverable by run_reader.list_run_summaries()."""
    run_dir = root / run_id
    run_dir.mkdir(parents=True, exist_ok=True)
    pd.DataFrame(rows).to_csv(run_dir / "scores.csv", index=False)
    metadata = {
        "run_id": run_id,
        "scenario_name": scenario_name,
        "judge_model": judge_model,
        "embedding_model": "embed-model",
        "finished_at": finished_at,
        "started_at": finished_at,
        "valid_samples": len(rows),
        "invalid_samples": 0,
    }
    (run_dir / "metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
    return run_dir


class TestFindPreviousRun:
    def test_finds_most_recent_prior_run_with_same_scenario(self, tmp_path: Path) -> None:
        _write_fake_run(tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.5}])
        _write_fake_run(tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.6}])
        _write_fake_run(tmp_path, "r3", "scn-a", "2026-01-03T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.9}])

        previous = find_previous_run(
            "scn-a", "r3", "2026-01-03T00:00:00+00:00", extra_roots=[tmp_path]
        )

        assert previous is not None
        assert previous.run_id == "r2"

    def test_excludes_runs_with_different_scenario_name(self, tmp_path: Path) -> None:
        _write_fake_run(tmp_path, "r1", "scn-other", "2026-01-01T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.5}])
        _write_fake_run(tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.6}])

        previous = find_previous_run(
            "scn-a", "r2", "2026-01-02T00:00:00+00:00", extra_roots=[tmp_path]
        )

        assert previous is None

    def test_returns_none_when_no_history(self, tmp_path: Path) -> None:
        _write_fake_run(tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.5}])

        previous = find_previous_run(
            "scn-a", "r1", "2026-01-01T00:00:00+00:00", extra_roots=[tmp_path]
        )

        assert previous is None

    def test_ignores_runs_at_or_after_current_time(self, tmp_path: Path) -> None:
        _write_fake_run(tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.5}])
        _write_fake_run(tmp_path, "r2", "scn-a", "2026-01-05T00:00:00+00:00", "gpt-5",
                         [{"sample_id": "s1", "faithfulness": 0.6}])

        # Current run finished at 2026-01-02, i.e. AFTER r1 but BEFORE r2.
        previous = find_previous_run(
            "scn-a", "r-current", "2026-01-02T00:00:00+00:00", extra_roots=[tmp_path]
        )

        assert previous is not None
        assert previous.run_id == "r1"

Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison.py -v Expected: FAIL with ModuleNotFoundError: No module named 'webapp.services.advisor_comparison'

Step 3: Write minimal implementation

Create webapp/services/advisor_comparison.py:

"""Advisor-diagnosis comparison against the same scenario's previous run.

The report detail page shows, for the run currently being viewed, whether
metrics flagged by the optimization advisor in the immediately preceding run
of the same scenario have since improved, regressed, or remain unresolved.

No new artifact files are written: diagnoses are recomputed on demand from
each run's scores.csv via the existing rag_eval.advisor.diagnose(), so the
comparison always reflects the current threshold rules in rules.py.
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

import pandas as pd

from webapp.models import AdvisorComparison, AdvisorComparisonEntry, RunSummary
from webapp.services import run_reader

logger = logging.getLogger("webapp.services.advisor_comparison")


def find_previous_run(
    scenario_name: str,
    current_run_id: str,
    current_finished_at: str,
    extra_roots: list[Path] | None = None,
) -> RunSummary | None:
    """Return the most recent prior run with the same scenario_name, or None.

    "Prior" means finished_at strictly earlier than current_finished_at, so a
    run that finished later (e.g. a concurrent run) is never mistaken for a
    historical baseline.
    """
    candidates = [
        summary
        for summary in run_reader.list_run_summaries(extra_roots)
        if summary.scenario_name == scenario_name
        and summary.run_id != current_run_id
        and (summary.finished_at or "") < (current_finished_at or "")
    ]
    if not candidates:
        return None
    candidates.sort(key=lambda summary: summary.finished_at or "", reverse=True)
    return candidates[0]

Step 4: Run test to verify it passes

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison.py -v Expected: 4 passed

Step 5: Commit

git add webapp/services/advisor_comparison.py tests/test_advisor_comparison.py
git commit -m "feat(advisor-comparison): add find_previous_run same-scenario lookup"

Task 3: `build_advisor_comparison()` — diagnosis diffing and status classification

Files:

Modify: webapp/services/advisor_comparison.py
Test: tests/test_advisor_comparison.py (extend)

Interfaces:

Consumes: find_previous_run(...) from Task 2; AdvisorComparison/AdvisorComparisonEntry from Task 1; rag_eval.advisor.diagnose(score_rows, metrics, top_low_samples=3) -> list[Diagnosis] (existing, Diagnosis has .metric, .mean_score, .severity); run_reader.read_scores_frame(run_dir) -> pd.DataFrame (existing).
Produces: build_advisor_comparison(run_dir: Path, scenario_name: str, metrics: list[str], extra_roots: list[Path] | None = None) -> AdvisorComparison | None, used by Task 4.
Step 1: Write the failing test

In tests/test_advisor_comparison.py, find the existing import line:

from webapp.services.advisor_comparison import find_previous_run

Replace with:

from webapp.services.advisor_comparison import build_advisor_comparison, find_previous_run

Append these test classes to the end of the file:

class TestBuildAdvisorComparison:
    def test_resolved_status_when_previously_triggered_now_healthy(self, tmp_path: Path) -> None:
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}, {"sample_id": "s2", "faithfulness": 0.5}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.95}, {"sample_id": "s2", "faithfulness": 0.95}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        assert len(comparison.entries) == 1
        entry = comparison.entries[0]
        assert entry.metric == "faithfulness"
        assert entry.status == "resolved"
        assert entry.previous_score == 0.5
        assert entry.current_score is None  # not triggered now → no Diagnosis on current side

    def test_regressed_status_when_previously_healthy_now_triggered(self, tmp_path: Path) -> None:
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.95}, {"sample_id": "s2", "faithfulness": 0.95}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}, {"sample_id": "s2", "faithfulness": 0.5}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        assert len(comparison.entries) == 1
        entry = comparison.entries[0]
        assert entry.status == "regressed"
        assert entry.previous_score is None
        assert entry.current_score == 0.5

    def test_still_triggered_status_shows_score_and_severity_change(self, tmp_path: Path) -> None:
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.55}, {"sample_id": "s2", "faithfulness": 0.55}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.45}, {"sample_id": "s2", "faithfulness": 0.45}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        entry = comparison.entries[0]
        assert entry.status == "still_triggered"
        assert entry.previous_severity == "warning"
        assert entry.current_severity == "critical"
        assert entry.previous_score == 0.55
        assert entry.current_score == 0.45

    def test_new_metric_status_when_metric_not_measured_before(self, tmp_path: Path) -> None:
        # Previous run only measured context_recall (healthy); faithfulness wasn't tracked at all.
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "context_recall": 0.95}, {"sample_id": "s2", "context_recall": 0.95}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}, {"sample_id": "s2", "faithfulness": 0.5}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        assert len(comparison.entries) == 1
        entry = comparison.entries[0]
        assert entry.metric == "faithfulness"
        assert entry.status == "new_metric"
        assert entry.previous_score is None
        assert entry.current_score == 0.5

    def test_metrics_healthy_in_both_are_omitted(self, tmp_path: Path) -> None:
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.95}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.96}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is None  # nothing to show → overall None

    def test_metric_dropped_from_current_scope_is_not_marked_resolved(self, tmp_path: Path) -> None:
        # Previous run triggered on context_precision, but current run doesn't
        # evaluate that metric at all — must NOT claim "resolved" without a
        # fair current-side measurement.
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "context_precision": 0.3}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.95}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is None

    def test_returns_none_when_no_previous_run(self, tmp_path: Path) -> None:
        current_dir = _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is None

    def test_judge_model_changed_flag_set_when_models_differ(self, tmp_path: Path) -> None:
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-4o",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        assert comparison.judge_model_changed is True
        assert comparison.previous_judge_model == "gpt-4o"
        assert comparison.current_judge_model == "gpt-5"

    def test_judge_model_changed_false_when_same(self, tmp_path: Path) -> None:
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        assert comparison.judge_model_changed is False

    def test_gracefully_returns_none_on_corrupt_previous_scores_csv(self, tmp_path: Path) -> None:
        previous_dir = _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )
        # Corrupt the previous run's scores.csv after it was written.
        (previous_dir / "scores.csv").write_text("not,a,valid\ncsv,,,", encoding="utf-8")

        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [{"sample_id": "s1", "faithfulness": 0.5}],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness"], extra_roots=[tmp_path]
        )

        assert comparison is None

    def test_worse_statuses_sorted_before_resolved(self, tmp_path: Path) -> None:
        """regressed/still_triggered/new_metric surface above resolved for visibility."""
        _write_fake_run(
            tmp_path, "r1", "scn-a", "2026-01-01T00:00:00+00:00", "gpt-5",
            [
                {"sample_id": "s1", "faithfulness": 0.5, "context_recall": 0.95},
            ],
        )
        current_dir = _write_fake_run(
            tmp_path, "r2", "scn-a", "2026-01-02T00:00:00+00:00", "gpt-5",
            [
                {"sample_id": "s1", "faithfulness": 0.95, "context_recall": 0.5},
            ],
        )

        comparison = build_advisor_comparison(
            current_dir, "scn-a", ["faithfulness", "context_recall"], extra_roots=[tmp_path]
        )

        assert comparison is not None
        statuses = [entry.status for entry in comparison.entries]
        # context_recall regressed → must appear before faithfulness resolved.
        assert statuses.index("regressed") < statuses.index("resolved")

Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison.py -v Expected: FAIL with ImportError: cannot import name 'build_advisor_comparison' from 'webapp.services.advisor_comparison'

Step 3: Write minimal implementation

In webapp/services/advisor_comparison.py, add these imports at the top (replacing the existing import block):

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

import pandas as pd

from rag_eval.advisor import diagnose
from webapp.models import AdvisorComparison, AdvisorComparisonEntry, RunSummary
from webapp.services import run_reader

logger = logging.getLogger("webapp.services.advisor_comparison")

# Worse-first ordering so regressions and unresolved issues surface above
# resolved ones in the rendered comparison list.
_STATUS_ORDER = {"regressed": 0, "still_triggered": 1, "new_metric": 2, "resolved": 3}

Keep the existing find_previous_run function unchanged, then append:

def _rows_as_records(frame: pd.DataFrame) -> list[dict[str, Any]]:
    """Convert a scores dataframe into plain dict records, NaN -> None."""
    if frame.empty:
        return []
    return frame.where(pd.notnull(frame), None).to_dict("records")


def build_advisor_comparison(
    run_dir: Path,
    scenario_name: str,
    metrics: list[str],
    extra_roots: list[Path] | None = None,
) -> AdvisorComparison | None:
    """Build the current run's advisor-diagnosis delta vs. its same-scenario predecessor.

    Returns None when there is no predecessor, the predecessor's data cannot be
    read, or there is nothing worth surfacing (no metric changed status). Never
    raises — any failure degrades to None so the report detail page is
    unaffected (mirrors run_advisor()'s defensive error handling).
    """
    try:
        metadata = run_reader._read_json(run_dir / "metadata.json")
        current_run_id = str(metadata.get("run_id") or run_dir.name)
        current_finished_at = str(metadata.get("finished_at") or "")
        current_judge_model = str(metadata.get("judge_model", ""))

        previous = find_previous_run(
            scenario_name, current_run_id, current_finished_at, extra_roots
        )
        if previous is None:
            return None

        previous_dir = Path(previous.output_path)
        previous_frame = run_reader.read_scores_frame(previous_dir)
        if previous_frame.empty:
            return None
        previous_rows = _rows_as_records(previous_frame)
        previous_metrics = previous.metrics

        current_frame = run_reader.read_scores_frame(run_dir)
        current_rows = _rows_as_records(current_frame)

        previous_diagnoses = {d.metric: d for d in diagnose(previous_rows, previous_metrics)}
        current_diagnoses = {d.metric: d for d in diagnose(current_rows, metrics)}

        entries: list[AdvisorComparisonEntry] = []
        for metric in sorted(set(previous_diagnoses) | set(current_diagnoses)):
            prev_d = previous_diagnoses.get(metric)
            curr_d = current_diagnoses.get(metric)

            if prev_d is not None and curr_d is None:
                if metric not in metrics:
                    # Dropped from the current scope entirely — cannot fairly
                    # claim "resolved" without a current-side measurement.
                    continue
                status = "resolved"
            elif prev_d is None and curr_d is not None:
                status = "regressed" if metric in previous_metrics else "new_metric"
            else:
                status = "still_triggered"

            entries.append(AdvisorComparisonEntry(
                metric=metric,
                status=status,
                previous_score=prev_d.mean_score if prev_d else None,
                previous_severity=prev_d.severity if prev_d else None,
                current_score=curr_d.mean_score if curr_d else None,
                current_severity=curr_d.severity if curr_d else None,
            ))

        if not entries:
            return None

        entries.sort(key=lambda e: (_STATUS_ORDER.get(e.status, 99), e.metric))

        previous_judge_model = previous.judge_model
        return AdvisorComparison(
            previous_run_id=previous.run_id,
            previous_finished_at=previous.finished_at,
            previous_judge_model=previous_judge_model,
            current_judge_model=current_judge_model,
            judge_model_changed=(
                bool(previous_judge_model)
                and bool(current_judge_model)
                and previous_judge_model != current_judge_model
            ),
            entries=entries,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "[advisor_comparison] failed to build comparison for run_dir=%s: %s",
            run_dir, exc,
        )
        return None

Step 4: Run test to verify it passes

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison.py -v Expected: 15 passed (4 from Task 2 + 11 new)

Step 5: Commit

git add webapp/services/advisor_comparison.py tests/test_advisor_comparison.py
git commit -m "feat(advisor-comparison): add build_advisor_comparison diagnosis diffing"

Task 4: Wire into `report_builder.build_report()`

Files:

Modify: webapp/services/report_builder.py
Test: tests/test_webapp_report_builder.py (extend)

Interfaces:

Consumes: advisor_comparison.build_advisor_comparison(run_dir, scenario_name, metrics) -> AdvisorComparison | None from Task 3.
Produces: ReportData.advisor_comparison correctly populated by build_report(), consumed by Task 5's frontend rendering.
Step 1: Write the failing test

Add to tests/test_webapp_report_builder.py. First add these imports at the top (extend the existing import block):

import json

from webapp.models import AdvisorComparison, AdvisorComparisonEntry
from webapp.services import report_builder

(report_builder may already be imported — if the line from webapp.services import report_builder already exists, do not duplicate it; only add the json import and the webapp.models import.)

Append these two tests to the end of the file:

def test_build_report_attaches_advisor_comparison(tmp_path: Path, monkeypatch) -> None:
    """build_report wires advisor_comparison.build_advisor_comparison() into ReportData."""
    run_dir = tmp_path / "run"
    _write_run_artifacts(run_dir)
    (run_dir / "metadata.json").write_text(
        json.dumps({"run_id": "run-1", "scenario_name": "my-scenario"}), encoding="utf-8"
    )

    fake_comparison = AdvisorComparison(
        previous_run_id="prev-1",
        previous_finished_at="2026-01-01T00:00:00+00:00",
        entries=[
            AdvisorComparisonEntry(
                metric="faithfulness",
                status="resolved",
                previous_score=0.5,
                previous_severity="warning",
                current_score=None,
                current_severity=None,
            )
        ],
    )

    captured_args = {}

    def _fake_build(run_dir_arg, scenario_name_arg, metrics_arg):
        captured_args["scenario_name"] = scenario_name_arg
        captured_args["metrics"] = metrics_arg
        return fake_comparison

    monkeypatch.setattr(
        report_builder.advisor_comparison, "build_advisor_comparison", _fake_build
    )

    report = build_report(run_dir, ["faithfulness", "context_recall"])

    assert report.advisor_comparison == fake_comparison
    assert captured_args["scenario_name"] == "my-scenario"
    assert captured_args["metrics"] == ["faithfulness", "context_recall"]


def test_build_report_advisor_comparison_none_when_no_previous_run(
    tmp_path: Path, monkeypatch
) -> None:
    """build_report leaves advisor_comparison as None when no predecessor exists."""
    run_dir = tmp_path / "run"
    _write_run_artifacts(run_dir)
    (run_dir / "metadata.json").write_text(
        json.dumps({"run_id": "run-1", "scenario_name": "my-scenario"}), encoding="utf-8"
    )

    monkeypatch.setattr(
        report_builder.advisor_comparison, "build_advisor_comparison", lambda *a, **k: None
    )

    report = build_report(run_dir, ["faithfulness", "context_recall"])

    assert report.advisor_comparison is None

Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_webapp_report_builder.py -v -k advisor_comparison Expected: FAIL — AttributeError: module 'webapp.services.report_builder' has no attribute 'advisor_comparison' (not imported yet in report_builder.py).

Step 3: Write minimal implementation

In webapp/services/report_builder.py, find the imports:

from webapp.services import question_history, run_reader

Replace with:

from webapp.services import advisor_comparison, question_history, run_reader

Find the build_report function's early-return branch:

def build_report(run_dir: Path, metrics: list[str]) -> ReportData:
    """Build the full aggregated report payload for one run directory."""
    frame = run_reader.read_scores_frame(run_dir)
    summary_markdown = run_reader.read_summary_markdown(run_dir)
    advice_markdown = run_reader.read_advice_markdown(run_dir)
    metric_weights, doc_weights = _read_weights_from_snapshot(run_dir)
    # Read once up front so both the empty-frame and full branches can surface it.
    metadata = run_reader._read_json(run_dir / "metadata.json")
    token_usage = metadata.get("token_usage") or {}

    if frame.empty or not metrics:
        return ReportData(
            metrics=metrics,
            metric_means={metric: None for metric in metrics},
            summary_markdown=summary_markdown,
            advice_markdown=advice_markdown,
            metric_weights=metric_weights,
            doc_weights=doc_weights,
            token_usage=token_usage,
        )

Replace with (adds advisor_comparison=None explicitly in the early-return branch — a run with no valid samples cannot fairly support a diagnosis comparison):

def build_report(run_dir: Path, metrics: list[str]) -> ReportData:
    """Build the full aggregated report payload for one run directory."""
    frame = run_reader.read_scores_frame(run_dir)
    summary_markdown = run_reader.read_summary_markdown(run_dir)
    advice_markdown = run_reader.read_advice_markdown(run_dir)
    metric_weights, doc_weights = _read_weights_from_snapshot(run_dir)
    # Read once up front so both the empty-frame and full branches can surface it.
    metadata = run_reader._read_json(run_dir / "metadata.json")
    token_usage = metadata.get("token_usage") or {}

    if frame.empty or not metrics:
        return ReportData(
            metrics=metrics,
            metric_means={metric: None for metric in metrics},
            summary_markdown=summary_markdown,
            advice_markdown=advice_markdown,
            metric_weights=metric_weights,
            doc_weights=doc_weights,
            token_usage=token_usage,
            advisor_comparison=None,
        )

Find the final return ReportData(...) at the end of the function:

    return ReportData(
        metrics=metrics,
        metric_means=rounded_means,
        distributions=distributions,
        groupings=_groupings(frame, metrics),
        lowest_samples=_lowest_samples(frame, metrics, history_index),
        summary_markdown=summary_markdown,
        advice_markdown=advice_markdown,
        weighted_score_mean=_round_or_none(overall_ws),
        metric_weights=metric_weights,
        doc_weights=doc_weights,
        token_usage=token_usage,
    )

Replace with:

    scenario_name = str(metadata.get("scenario_name") or "")
    comparison = (
        advisor_comparison.build_advisor_comparison(run_dir, scenario_name, metrics)
        if scenario_name
        else None
    )

    return ReportData(
        metrics=metrics,
        metric_means=rounded_means,
        distributions=distributions,
        groupings=_groupings(frame, metrics),
        lowest_samples=_lowest_samples(frame, metrics, history_index),
        summary_markdown=summary_markdown,
        advice_markdown=advice_markdown,
        weighted_score_mean=_round_or_none(overall_ws),
        metric_weights=metric_weights,
        doc_weights=doc_weights,
        token_usage=token_usage,
        advisor_comparison=comparison,
    )

Step 4: Run test to verify it passes

Run: C:\software\Python312\python.exe -m pytest tests/test_webapp_report_builder.py -v Expected: all pass (existing tests + 2 new)

Also re-run the full advisor-comparison suite to confirm no regression:

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison.py tests/test_advisor_comparison_models.py -v Expected: 20 passed (15 + 5)

Step 5: Commit

git add webapp/services/report_builder.py tests/test_webapp_report_builder.py
git commit -m "feat(advisor-comparison): wire build_advisor_comparison into report_builder"

Task 5: Report detail page UI — "相比上次运行" panel

Files:

Modify: webapp/static/index.html
Modify: webapp/static/js/report.js
Modify: webapp/static/css/app.css

Interfaces:

Consumes: ReportData.advisor_comparison (JSON field advisor_comparison from Task 4's API response); MetricPresenter.deltaInfo(metric, current, previous) (existing, in metric_presenter.js); App.escape, App.shortMetric, App.shortTime (existing, in app.js).

This task has no automated test — the repository has no JavaScript test harness (vanilla JS, no package.json/Jest). Verify via node --check syntax validation per Step 4 below.

Step 1: Add the panel container to index.html

In webapp/static/index.html, find:

          <!-- ⑤ 优化建议（optimization_advisor: true 时显示） -->
          <div id="advice-section" hidden>

Replace with:

          <!-- 相比上次运行（同 scenario_name 的顾问诊断对比，自动匹配，找不到则不显示） -->
          <div id="advisor-comparison-section" hidden>
            <div class="panel advisor-comparison-panel">
              <div class="advisor-comparison-header">
                <span class="section-label tight">相比上次运行</span>
                <span class="muted tiny" id="advisor-comparison-meta"></span>
              </div>
              <div id="advisor-comparison-body"></div>
            </div>
          </div>

          <!-- ⑤ 优化建议（optimization_advisor: true 时显示） -->
          <div id="advice-section" hidden>

Step 2: Add renderAdvisorComparison to report.js and call it from render()

In webapp/static/js/report.js, find:

      Report.renderAdvice(detail.summary, detail.report);
      Report.renderTokenUsage(detail.report);
      content.style.opacity = "1";

Replace with:

      Report.renderAdvisorComparison(detail.report);
      Report.renderAdvice(detail.summary, detail.report);
      Report.renderTokenUsage(detail.report);
      content.style.opacity = "1";

Add the new method right after renderTokenUsage (which itself sits right after _drawGroupTable — insert this new method immediately after renderTokenUsage's closing },):

  // 相比上次运行的顾问诊断对比（同 scenario_name 的最近一次前序运行，自动匹配）。
  renderAdvisorComparison(report) {
    const section = document.getElementById("advisor-comparison-section");
    const meta = document.getElementById("advisor-comparison-meta");
    const body = document.getElementById("advisor-comparison-body");
    if (!section || !meta || !body) return;

    const comparison = report.advisor_comparison;
    if (!comparison || !comparison.entries || comparison.entries.length === 0) {
      section.hidden = true;
      return;
    }

    section.hidden = false;
    meta.textContent = `对比：${comparison.previous_run_id}（${App.shortTime(comparison.previous_finished_at)}）`;

    const STATUS_LABEL = {
      resolved: "✅ 已改善",
      regressed: "⚠️ 新触发",
      still_triggered: "❌ 仍未解决",
      new_metric: "🆕 新增指标",
    };
    const STATUS_CLASS = {
      resolved: "delta-good",
      regressed: "delta-bad",
      still_triggered: "delta-bad",
      new_metric: "delta-flat",
    };

    const fmt = (v) => (v === null || v === undefined ? "—" : Number(v).toFixed(2));

    let rows = "";
    comparison.entries.forEach((entry) => {
      const label = STATUS_LABEL[entry.status] || entry.status;
      const cls = STATUS_CLASS[entry.status] || "delta-flat";
      const d = MetricPresenter.deltaInfo(entry.metric, entry.current_score, entry.previous_score);
      const deltaHtml = d.hasData && d.delta !== 0
        ? ` <span class="hist-delta ${d.cls}">${d.arrow}${d.magnitude}</span>`
        : "";
      rows += `
        <div class="advisor-comparison-row">
          <span class="advisor-comparison-status ${cls}">${label}</span>
          <span class="advisor-comparison-metric">${App.escape(App.shortMetric(entry.metric))}</span>
          <span class="advisor-comparison-scores">${fmt(entry.previous_score)} → ${fmt(entry.current_score)}${deltaHtml}</span>
        </div>`;
    });

    let caveat = "";
    if (comparison.judge_model_changed) {
      caveat = `<p class="muted tiny">⚠️ judge_model 不同（${App.escape(comparison.previous_judge_model)} → ${App.escape(comparison.current_judge_model)}），对比仅供参考。</p>`;
    }

    body.innerHTML = rows + caveat;
  },

Step 3: Add minimal CSS

In webapp/static/css/app.css, add this block right after the existing .hist-delta.delta-flat { color: var(--slate-light); } line (found via the existing .hist-delta/.delta-good/.delta-bad/.delta-flat rules already in the file):

.advisor-comparison-panel { border-left: 3px solid #0ea5e9; }
.advisor-comparison-header {
  display: flex; align-items: baseline; justify-content: space-between; gap: 10px;
  margin-bottom: 10px;
}
.advisor-comparison-row {
  display: flex; align-items: center; gap: 10px;
  padding: 6px 0; border-bottom: 1px solid var(--line);
  font-size: 13px;
}
.advisor-comparison-row:last-of-type { border-bottom: none; }
.advisor-comparison-status { font-weight: 600; white-space: nowrap; }
.advisor-comparison-metric { flex: 1; color: var(--slate); }
.advisor-comparison-scores { font-variant-numeric: tabular-nums; white-space: nowrap; }

Step 4: Verify JS syntax and run full backend regression

Run: node --check webapp/static/js/report.js Expected: no output (syntax OK)

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_comparison.py tests/test_advisor_comparison_models.py tests/test_webapp_report_builder.py -v Expected: all pass

Step 5: Commit

git add webapp/static/index.html webapp/static/js/report.js webapp/static/css/app.css
git commit -m "feat(advisor-comparison): add 相比上次运行 panel to report detail page"

Final Verification

Run the full test suite once all 5 tasks are complete

Run: C:\software\Python312\python.exe -m pytest tests/ -v Expected: All tests pass, including the new test_advisor_comparison_models.py (5), test_advisor_comparison.py (15), plus the 2 new tests in test_webapp_report_builder.py, and all pre-existing tests unchanged (the 6 pre-existing unrelated failures documented in prior sessions — test_settings_defaults, test_normalize_sample_pdf_offline_smoke_row, test_evaluator_and_reporting_write_run_assets, test_question_generator_rejects_invalid_json, test_question_generator_rejects_non_list_samples, test_execute_dataset_build_job_directly — remain as-is and are out of scope for this plan).

Spec coverage check

Confirm every section of docs/superpowers/specs/2026-07-02-advisor-comparison-design.md maps to a task:

§2 架构 (webapp-layer read-only comparison, reusing diagnose()) → Tasks 2, 3
§3.1 advisor_comparison.py functions and 4-state classification → Task 3
§3.2 webapp/models.py new models + ReportData field → Task 1
§4 数据流 (find previous run → diagnose both → classify → attach to ReportData → render) → Tasks 2, 3, 4, 5
§5 错误处理 (no previous run / corrupt data / diagnose() exception / empty entries) → covered inline in Task 3's tests (test_returns_none_when_no_previous_run, test_gracefully_returns_none_on_corrupt_previous_scores_csv, test_metrics_healthy_in_both_are_omitted)
§6 测试策略 (new test file + report_builder wiring tests, zero changes to existing advisor tests) → Tasks 2, 3, 4 (no test_advisor_rules.py/test_advisor_llm_analyzer.py/test_advisor_writer.py modified anywhere in this plan)
§7 非目标 (no manual run picker, no persisted diagnoses.json, no rules.py changes, no cross-scenario comparison, no multi-hop chain) → respected throughout; confirmed no task touches rag_eval/advisor/rules.py, llm_analyzer.py, or writer.py

44 KiB Raw Blame History Unescape Escape

优化建议历史对比 Implementation Plan

Global Constraints

Task 1: AdvisorComparisonEntry / AdvisorComparison 模型 + ReportData 字段

Task 2: find_previous_run() — locate the same-scenario predecessor run

Task 3: build_advisor_comparison() — diagnosis diffing and status classification

Task 4: Wire into report_builder.build_report()

Task 5: Report detail page UI — "相比上次运行" panel

Final Verification

44 KiB

Raw Blame History

Task 1: `AdvisorComparisonEntry` / `AdvisorComparison` 模型 + `ReportData` 字段

Task 2: `find_previous_run()` — locate the same-scenario predecessor run

Task 3: `build_advisor_comparison()` — diagnosis diffing and status classification

Task 4: Wire into `report_builder.build_report()`