siemens_ragas/docs/superpowers/plans/2026-07-02-token-usage-tracking.md

Judge LLM Token 用量追踪 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: 按 LLM 模型名累计记录 RAGAS 评分调用与优化顾问 LLM 分析调用的原始 token 用量（input/output/调用次数，不换算金额），持久化到 metadata.json/summary.md，并在 Web 报告详情页展示。

Architecture: 在 HTTP 层（httpx 响应钩子）拦截所有 AsyncOpenAI 请求的响应体，读取其中的 usage + model 字段。用 contextvars.ContextVar 保存"当前活跃的统计器"，因为 InlineScorer 会跨请求缓存 AsyncOpenAI 客户端，不能把统计器固定绑死在客户端构造时刻。三条落盘路径（CLI scenario 运行、/api/score/async、/api/score/session_async）各自在评分调用外包一层 track_token_usage()，把汇总结果写进 EvaluationResult.token_usage → metadata.json。

Tech Stack: Python 3.12、httpx（openai SDK 的传递依赖，已验证 0.28.1 可用）、pytest + unittest（现有测试框架混用）、pandas（既有报告聚合逻辑）。

Global Constraints

• 只记录原始 token 数（input_tokens / output_tokens / calls），不做金额换算，不改 LLM Profile 结构。
• 只统计落盘成 run 目录的路径：CLI main.py --scenario、/api/score/async、/api/score/session_async。不统计不落盘的同步 /api/score。
• 统计范围覆盖：RAGAS 评分调用（judge model + embedding model）+ 优化顾问 LLM 分析调用（advisor/llm_analyzer.py）。不统计 dataset_builder 题库生成路径。
• session_async 场景下 token 用量必须跨调用累加，不能每次覆盖为最后一次的值。
• HTTP 钩子解析失败、网关不返回 usage/model 字段时必须静默跳过，绝不能让评分调用报错或变慢。
• 已核实技术事实：openai==1.102.0 的 AsyncOpenAI()._client 是 httpx.AsyncClient 子类（AsyncHttpxClientWrapper），自带 event_hooks = {"request": [], "response": []} 可直接追加钩子。
• 设计依据：docs/superpowers/specs/2026-07-02-token-usage-tracking-design.md（本计划的每个任务都对应该文档的某一节，不要偏离）。
• 测试约定：本仓库用 asyncio.run(...) 包裹被测异步调用，而不是 @pytest.mark.asyncio（参考 tests/test_advisor_llm_analyzer.py）。运行测试用 C:\software\Python312\python.exe -m pytest tests/<file> -v。

---

Task 1: TokenUsageTracker 核心与 contextvar

Files:

• Create: rag_eval/metrics/token_tracker.py
• Test: tests/test_token_tracker.py

Interfaces:

• Produces:

  • class TokenUsageTracker — .record(model: str, input_tokens: int, output_tokens: int) -> None；.as_dict() -> dict[str, dict[str, int]]；.merge_into(existing: dict[str, dict[str, int]]) -> dict[str, dict[str, int]]
  • track_token_usage() -> AbstractContextManager[TokenUsageTracker]（@contextmanager）
  • get_current_tracker() -> TokenUsageTracker | None

• Step 1: Write the failing test

Create tests/test_token_tracker.py:

"""Tests for the per-run token usage accumulator and its context-scoped activation."""
from __future__ import annotations

from rag_eval.metrics.token_tracker import (
    TokenUsageTracker,
    get_current_tracker,
    track_token_usage,
)


def test_record_accumulates_input_output_and_calls():
    tracker = TokenUsageTracker()
    tracker.record("gpt-5", 100, 50)
    tracker.record("gpt-5", 20, 10)
    assert tracker.as_dict() == {
        "gpt-5": {"input_tokens": 120, "output_tokens": 60, "calls": 2}
    }


def test_record_groups_by_model_name():
    tracker = TokenUsageTracker()
    tracker.record("gpt-5", 100, 50)
    tracker.record("Qwen/Qwen3-Embedding-4B", 30, 0)
    result = tracker.as_dict()
    assert set(result.keys()) == {"gpt-5", "Qwen/Qwen3-Embedding-4B"}
    assert result["Qwen/Qwen3-Embedding-4B"] == {
        "input_tokens": 30, "output_tokens": 0, "calls": 1
    }


def test_record_defaults_blank_model_to_unknown():
    tracker = TokenUsageTracker()
    tracker.record("", 10, 5)
    assert "unknown" in tracker.as_dict()


def test_merge_into_sums_with_existing_totals():
    tracker = TokenUsageTracker()
    tracker.record("gpt-5", 100, 50)
    existing = {"gpt-5": {"input_tokens": 200, "output_tokens": 100, "calls": 3}}
    merged = tracker.merge_into(existing)
    assert merged == {"gpt-5": {"input_tokens": 300, "output_tokens": 150, "calls": 4}}


def test_merge_into_keeps_models_only_in_existing():
    tracker = TokenUsageTracker()
    tracker.record("gpt-5", 10, 5)
    existing = {"other-model": {"input_tokens": 1, "output_tokens": 1, "calls": 1}}
    merged = tracker.merge_into(existing)
    assert merged["other-model"] == {"input_tokens": 1, "output_tokens": 1, "calls": 1}
    assert merged["gpt-5"] == {"input_tokens": 10, "output_tokens": 5, "calls": 1}


def test_merge_into_does_not_mutate_existing_dict():
    tracker = TokenUsageTracker()
    tracker.record("gpt-5", 10, 5)
    existing = {"gpt-5": {"input_tokens": 1, "output_tokens": 1, "calls": 1}}
    tracker.merge_into(existing)
    assert existing == {"gpt-5": {"input_tokens": 1, "output_tokens": 1, "calls": 1}}


def test_get_current_tracker_returns_none_outside_context():
    assert get_current_tracker() is None


def test_track_token_usage_activates_and_resets_context():
    assert get_current_tracker() is None
    with track_token_usage() as tracker:
        assert get_current_tracker() is tracker
        tracker.record("gpt-5", 1, 1)
    assert get_current_tracker() is None


def test_track_token_usage_nested_contexts_are_isolated():
    with track_token_usage() as outer:
        outer.record("outer-model", 5, 5)
        with track_token_usage() as inner:
            inner.record("inner-model", 1, 1)
            assert get_current_tracker() is inner
        assert get_current_tracker() is outer
    assert outer.as_dict() == {
        "outer-model": {"input_tokens": 5, "output_tokens": 5, "calls": 1}
    }

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_token_tracker.py -v Expected: FAIL with ModuleNotFoundError: No module named 'rag_eval.metrics.token_tracker'

• Step 3: Write minimal implementation

Create rag_eval/metrics/token_tracker.py:

"""Per-run token usage accumulation, keyed by model name.

RAGAS 0.4.3's `ragas.metrics.collections` + instructor code path does not
expose real token counts (`ragas/cost.py` only serves the legacy langchain
`evaluate()` path). This module provides a context-scoped accumulator that
the HTTP response hook in `rag_eval.metrics.factory` feeds into, so token
counts survive across the AsyncOpenAI client caching used by InlineScorer
(see docs/superpowers/specs/2026-07-02-token-usage-tracking-design.md).
"""

from __future__ import annotations

from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Iterator


@dataclass
class TokenUsageTracker:
    """Accumulates input/output token counts and call counts, grouped by model name."""

    _totals: dict[str, dict[str, int]] = field(default_factory=dict)

    def record(self, model: str, input_tokens: int, output_tokens: int) -> None:
        """Add one API call's usage to the running total for `model`."""
        key = model or "unknown"
        bucket = self._totals.setdefault(
            key, {"input_tokens": 0, "output_tokens": 0, "calls": 0}
        )
        bucket["input_tokens"] += int(input_tokens)
        bucket["output_tokens"] += int(output_tokens)
        bucket["calls"] += 1

    def as_dict(self) -> dict[str, dict[str, int]]:
        """Return a plain-dict snapshot: {model: {input_tokens, output_tokens, calls}}."""
        return {model: dict(usage) for model, usage in self._totals.items()}

    def merge_into(self, existing: dict[str, dict[str, int]]) -> dict[str, dict[str, int]]:
        """Return a new dict combining `existing` accumulated totals with this tracker's totals.

        Used by session-scoped scoring (one call at a time) to keep a running
        total across multiple calls instead of overwriting with just the latest call.
        Does not mutate `existing`.
        """
        merged: dict[str, dict[str, int]] = {
            model: dict(usage) for model, usage in existing.items()
        }
        for model, usage in self.as_dict().items():
            bucket = merged.setdefault(
                model, {"input_tokens": 0, "output_tokens": 0, "calls": 0}
            )
            bucket["input_tokens"] += usage["input_tokens"]
            bucket["output_tokens"] += usage["output_tokens"]
            bucket["calls"] += usage["calls"]
        return merged


_current_tracker: ContextVar[TokenUsageTracker | None] = ContextVar(
    "_current_tracker", default=None
)


@contextmanager
def track_token_usage() -> Iterator[TokenUsageTracker]:
    """Activate a fresh TokenUsageTracker for the duration of the `with` block.

    Any AsyncOpenAI client with `attach_usage_hook()` applied that makes a
    call while this context is active will have its usage recorded here.
    Safe to nest; the innermost tracker is active within its own block.
    """
    tracker = TokenUsageTracker()
    token = _current_tracker.set(tracker)
    try:
        yield tracker
    finally:
        _current_tracker.reset(token)


def get_current_tracker() -> TokenUsageTracker | None:
    """Return the currently active tracker, or None if no `track_token_usage()` block is active."""
    return _current_tracker.get()

• Step 4: Run test to verify it passes

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

• Step 5: Commit

git add rag_eval/metrics/token_tracker.py tests/test_token_tracker.py
git commit -m "feat(token-tracking): add TokenUsageTracker with context-scoped activation"

---

Task 2: HTTP 响应钩子 + attach_usage_hook()，接入 build_models()

Files:

• Modify: rag_eval/metrics/factory.py
• Test: tests/test_token_usage_hook.py

Interfaces:

• Consumes: get_current_tracker(), TokenUsageTracker, track_token_usage() from Task 1 (rag_eval.metrics.token_tracker)

• Produces:

  • attach_usage_hook(client: AsyncOpenAI) -> None — idempotent; other tasks (3, and indirectly 6/7/8 via build_models) call this.
  • _usage_response_hook(response: httpx.Response) -> None (module-private, but imported directly by tests for unit coverage)

• Step 1: Write the failing test

Create tests/test_token_usage_hook.py:

"""Tests for the token-usage HTTP response hook and attach_usage_hook wiring."""
from __future__ import annotations

import asyncio
import json

import httpx

from rag_eval.metrics.factory import _usage_response_hook, attach_usage_hook
from rag_eval.metrics.token_tracker import track_token_usage


def _fake_response(payload: dict | None) -> httpx.Response:
    """Build a real httpx.Response with a JSON (or broken) body for hook testing."""
    content = b"not json" if payload is None else json.dumps(payload).encode("utf-8")
    return httpx.Response(200, content=content, request=httpx.Request("POST", "http://test/x"))


class TestUsageResponseHook:
    def test_records_usage_when_tracker_active(self):
        with track_token_usage() as tracker:
            response = _fake_response({
                "model": "gpt-5",
                "usage": {"prompt_tokens": 120, "completion_tokens": 45},
            })
            asyncio.run(_usage_response_hook(response))
        assert tracker.as_dict() == {
            "gpt-5": {"input_tokens": 120, "output_tokens": 45, "calls": 1}
        }

    def test_noop_when_no_tracker_active(self):
        response = _fake_response({"model": "gpt-5", "usage": {"prompt_tokens": 1, "completion_tokens": 1}})
        # Must not raise even though no tracker is active.
        asyncio.run(_usage_response_hook(response))

    def test_noop_when_response_has_no_usage_field(self):
        with track_token_usage() as tracker:
            response = _fake_response({"model": "gpt-5"})
            asyncio.run(_usage_response_hook(response))
        assert tracker.as_dict() == {}

    def test_noop_on_non_json_response(self):
        with track_token_usage() as tracker:
            response = _fake_response(None)
            asyncio.run(_usage_response_hook(response))
        assert tracker.as_dict() == {}

    def test_embedding_response_without_completion_tokens_defaults_output_to_zero(self):
        """Embeddings responses omit completion_tokens; output should default to 0."""
        with track_token_usage() as tracker:
            response = _fake_response({
                "model": "Qwen/Qwen3-Embedding-4B",
                "usage": {"prompt_tokens": 30, "total_tokens": 30},
            })
            asyncio.run(_usage_response_hook(response))
        assert tracker.as_dict() == {
            "Qwen/Qwen3-Embedding-4B": {"input_tokens": 30, "output_tokens": 0, "calls": 1}
        }


class TestAttachUsageHook:
    def test_attaches_hook_to_client_event_hooks(self):
        from openai import AsyncOpenAI

        client = AsyncOpenAI(api_key="sk-test", base_url="http://localhost:1")
        attach_usage_hook(client)
        assert _usage_response_hook in client._client.event_hooks["response"]

    def test_idempotent_when_called_twice_on_same_client(self):
        from openai import AsyncOpenAI

        client = AsyncOpenAI(api_key="sk-test", base_url="http://localhost:1")
        attach_usage_hook(client)
        attach_usage_hook(client)
        assert client._client.event_hooks["response"].count(_usage_response_hook) == 1

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_token_usage_hook.py -v Expected: FAIL with ImportError: cannot import name '_usage_response_hook' from 'rag_eval.metrics.factory'

• Step 3: Write minimal implementation

Edit rag_eval/metrics/factory.py. Find the existing single import line:

from openai import AsyncOpenAI

Replace with (adds the new httpx import right before it):

import httpx
from openai import AsyncOpenAI

Find:

from rag_eval.shared.models import Scenario

Replace with (adds the new token_tracker import right after it):

from rag_eval.shared.models import Scenario

from .token_tracker import get_current_tracker

Add these two functions right after the module-level logger = logging.getLogger(...) line and before _resolve_openai_client_kwargs:

async def _usage_response_hook(response: httpx.Response) -> None:
    """Record token usage from an OpenAI-compatible HTTP response, if a tracker is active.

    Applies to both chat-completions and embeddings responses since both
    return top-level `model` and `usage` fields in OpenAI-compatible APIs.
    Never raises — a broken/incompatible gateway response must not affect scoring.
    """
    tracker = get_current_tracker()
    if tracker is None:
        return
    try:
        await response.aread()
        data = response.json()
        usage = data.get("usage")
        if not usage:
            # Gateway did not report usage at all — skip rather than record a
            # misleading 0/0 call.
            return
        model = data.get("model") or "unknown"
        tracker.record(
            model,
            int(usage.get("prompt_tokens", 0) or 0),
            int(usage.get("completion_tokens", 0) or 0),
        )
    except Exception:  # noqa: BLE001
        logger.debug("[factory] usage hook failed to parse response", exc_info=True)


def attach_usage_hook(client: AsyncOpenAI) -> None:
    """Attach the token-usage response hook to an AsyncOpenAI client (idempotent).

    Safe to call multiple times on the same client (e.g. when judge and
    embedding models share one client) — the hook is only appended once.
    """
    httpx_client = getattr(client, "_client", None)
    if httpx_client is None or not hasattr(httpx_client, "event_hooks"):
        return
    hooks = httpx_client.event_hooks.setdefault("response", [])
    if _usage_response_hook not in hooks:
        hooks.append(_usage_response_hook)

Now wire it into build_models(). Find:

    llm_client = AsyncOpenAI(**llm_kwargs)
    # Only allocate a second client when the embedding model needs different settings.
    emb_client = AsyncOpenAI(**emb_kwargs) if emb_kwargs != llm_kwargs else llm_client

Replace with:

    llm_client = AsyncOpenAI(**llm_kwargs)
    attach_usage_hook(llm_client)
    # Only allocate a second client when the embedding model needs different settings.
    emb_client = AsyncOpenAI(**emb_kwargs) if emb_kwargs != llm_kwargs else llm_client
    attach_usage_hook(emb_client)

• Step 4: Run test to verify it passes

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

Also re-run the pre-existing build_models tests to confirm no regression (they patch AsyncOpenAI with a fake class that has no _client attribute, so attach_usage_hook must degrade gracefully via the getattr(client, "_client", None) guard):

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

• Step 5: Commit

git add rag_eval/metrics/factory.py tests/test_token_usage_hook.py
git commit -m "feat(token-tracking): add HTTP response hook and attach_usage_hook, wire into build_models"

---

Task 3: 优化顾问 LLM 分析调用接入用量钩子

Files:

• Modify: rag_eval/advisor/llm_analyzer.py
• Test: tests/test_advisor_llm_analyzer.py (extend existing file)

Interfaces:

• Consumes: attach_usage_hook(client: AsyncOpenAI) -> None from Task 2 (rag_eval.metrics.factory)

• Step 1: Write the failing test

Add to tests/test_advisor_llm_analyzer.py (after test_analyze_does_not_close_injected_client, before test_is_reasoning_model_detection):

def test_analyze_attaches_usage_hook_to_self_created_client(monkeypatch) -> None:
    """A self-created client gets the token-usage hook attached (not the injected-client path)."""
    captured: dict = {}
    fake = _FakeClient(captured)
    hook_calls = []

    import openai
    import rag_eval.metrics.factory as factory_mod

    monkeypatch.setattr(openai, "AsyncOpenAI", lambda **kwargs: fake)
    monkeypatch.setattr(
        factory_mod, "resolve_openai_client_kwargs", lambda *a, **k: {"api_key": "x"}
    )
    monkeypatch.setattr(factory_mod, "attach_usage_hook", lambda c: hook_calls.append(c))

    asyncio.run(analyze([_diagnosis()], "scn", "gpt-4o", _Settings()))

    assert hook_calls == [fake]


def test_analyze_does_not_attach_hook_for_injected_client() -> None:
    """An injected chat_client is assumed to already have the hook attached by its owner."""
    hook_calls = []
    import rag_eval.metrics.factory as factory_mod
    import unittest.mock as mock

    with mock.patch.object(factory_mod, "attach_usage_hook", lambda c: hook_calls.append(c)):
        fake = _FakeClient({})
        asyncio.run(analyze([_diagnosis()], "scn", "gpt-4o", _Settings(), chat_client=fake))

    assert hook_calls == []

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_advisor_llm_analyzer.py -v -k usage_hook Expected: FAIL — test_analyze_attaches_usage_hook_to_self_created_client fails with assert [] == [fake] (hook never called).

• Step 3: Write minimal implementation

In rag_eval/advisor/llm_analyzer.py, find the self-created client branch inside analyze():

        client = chat_client
        owns_client = False
        if client is None:
            from openai import AsyncOpenAI

            from rag_eval.metrics.factory import resolve_openai_client_kwargs

            client = AsyncOpenAI(**resolve_openai_client_kwargs(judge_model, settings))
            owns_client = True

Replace with:

        client = chat_client
        owns_client = False
        if client is None:
            from openai import AsyncOpenAI

            from rag_eval.metrics.factory import attach_usage_hook, resolve_openai_client_kwargs

            client = AsyncOpenAI(**resolve_openai_client_kwargs(judge_model, settings))
            attach_usage_hook(client)
            owns_client = True

• Step 4: Run test to verify it passes

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

• Step 5: Commit

git add rag_eval/advisor/llm_analyzer.py tests/test_advisor_llm_analyzer.py
git commit -m "feat(token-tracking): attach usage hook to advisor's self-created LLM client"

---

Task 4: EvaluationResult.token_usage 字段 + metadata.json 持久化

Files:

• Modify: rag_eval/shared/models.py
• Modify: rag_eval/reporting/writers.py
• Test: tests/test_token_usage_persistence.py

Interfaces:

• Produces: EvaluationResult.token_usage: dict[str, dict[str, int]] (default {}), consumed by Tasks 5, 6, 7, 8, 9.

• Step 1: Write the failing test

Create tests/test_token_usage_persistence.py:

"""Tests that EvaluationResult.token_usage is persisted into metadata.json."""
from __future__ import annotations

import json
from pathlib import Path

from rag_eval.reporting.writers import write_run_artifacts
from rag_eval.shared.models import DatasetConfig, EvaluationResult, RuntimeConfig, Scenario


def _scenario(tmp_path: Path) -> Scenario:
    return Scenario(
        scenario_name="token-persist-test",
        mode="offline",
        dataset=DatasetConfig(path=tmp_path / "dataset.csv"),
        judge_model="gpt-5",
        embedding_model="embedding-model",
        metrics=["faithfulness"],
        output_dir=tmp_path / "outputs",
        runtime=RuntimeConfig(batch_size=1),
    )


def test_evaluation_result_defaults_token_usage_to_empty_dict(tmp_path: Path) -> None:
    result = EvaluationResult(
        scenario=_scenario(tmp_path),
        run_id="run-1",
        started_at="t0",
        finished_at="t1",
        valid_samples=[],
        invalid_samples=[],
        score_rows=[],
    )
    assert result.token_usage == {}


def test_write_run_artifacts_persists_token_usage(tmp_path: Path) -> None:
    scenario = _scenario(tmp_path)
    result = EvaluationResult(
        scenario=scenario,
        run_id="run-2",
        started_at="t0",
        finished_at="t1",
        valid_samples=[],
        invalid_samples=[],
        score_rows=[],
        token_usage={"gpt-5": {"input_tokens": 100, "output_tokens": 40, "calls": 2}},
    )

    write_run_artifacts(result)

    metadata_path = scenario.output_dir / "run-2" / "metadata.json"
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["token_usage"] == {
        "gpt-5": {"input_tokens": 100, "output_tokens": 40, "calls": 2}
    }


def test_write_run_artifacts_writes_empty_token_usage_when_unset(tmp_path: Path) -> None:
    scenario = _scenario(tmp_path)
    result = EvaluationResult(
        scenario=scenario,
        run_id="run-3",
        started_at="t0",
        finished_at="t1",
        valid_samples=[],
        invalid_samples=[],
        score_rows=[],
    )

    write_run_artifacts(result)

    metadata_path = scenario.output_dir / "run-3" / "metadata.json"
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["token_usage"] == {}

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_token_usage_persistence.py -v Expected: FAIL — test_evaluation_result_defaults_token_usage_to_empty_dict fails with TypeError: EvaluationResult.__init__() got an unexpected keyword argument 'token_usage' (for the other two tests, KeyError: 'token_usage' on the metadata assertion).

• Step 3: Write minimal implementation

In rag_eval/shared/models.py, find the EvaluationResult dataclass:

@dataclass(slots=True)
class EvaluationResult:
    """Aggregate result object returned after a scenario completes."""

    scenario: Scenario
    run_id: str
    started_at: str
    finished_at: str
    valid_samples: list[NormalizedSample]
    invalid_samples: list[InvalidSample]
    score_rows: list[dict[str, Any]]

Replace with:

@dataclass(slots=True)
class EvaluationResult:
    """Aggregate result object returned after a scenario completes."""

    scenario: Scenario
    run_id: str
    started_at: str
    finished_at: str
    valid_samples: list[NormalizedSample]
    invalid_samples: list[InvalidSample]
    score_rows: list[dict[str, Any]]
    # Token usage grouped by model name: {model: {input_tokens, output_tokens, calls}}.
    # Populated by callers via rag_eval.metrics.token_tracker.track_token_usage().
    token_usage: dict[str, dict[str, int]] = field(default_factory=dict)

In rag_eval/reporting/writers.py, find:

    metadata = {
        "run_id": result.run_id,
        "scenario_name": result.scenario.scenario_name,
        "mode": result.scenario.mode,
        "judge_model": result.scenario.judge_model,
        "embedding_model": result.scenario.embedding_model,
        "started_at": result.started_at,
        "finished_at": result.finished_at,
        "dataset": result.scenario.dataset.path.as_posix(),
        "valid_samples": len(result.valid_samples),
        "invalid_samples": len(result.invalid_samples),
    }

Replace with:

    metadata = {
        "run_id": result.run_id,
        "scenario_name": result.scenario.scenario_name,
        "mode": result.scenario.mode,
        "judge_model": result.scenario.judge_model,
        "embedding_model": result.scenario.embedding_model,
        "started_at": result.started_at,
        "finished_at": result.finished_at,
        "dataset": result.scenario.dataset.path.as_posix(),
        "valid_samples": len(result.valid_samples),
        "invalid_samples": len(result.invalid_samples),
        "token_usage": result.token_usage,
    }

• Step 4: Run test to verify it passes

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

Also re-run the broader offline/online suites to confirm the new dataclass field (with a default) doesn't break any existing EvaluationResult(...) construction call sites:

Run: C:\software\Python312\python.exe -m pytest tests/test_offline_eval.py tests/test_online_eval.py -v Expected: all pass (unchanged)

• Step 5: Commit

git add rag_eval/shared/models.py rag_eval/reporting/writers.py tests/test_token_usage_persistence.py
git commit -m "feat(token-tracking): add EvaluationResult.token_usage and persist to metadata.json"

---

Task 5: summary.md 新增 "Token 用量" 小节

Files:

• Modify: rag_eval/reporting/summary.py
• Test: tests/test_reporting_summary_token_usage.py

Interfaces:

• Consumes: EvaluationResult.token_usage from Task 4

• Produces: _token_usage_section(token_usage: dict[str, dict[str, int]]) -> list[str] (module-private helper, used by build_summary_markdown)

• Step 1: Write the failing test

Create tests/test_reporting_summary_token_usage.py:

"""Tests for the '## Token 用量' section rendered by build_summary_markdown."""
from __future__ import annotations

from pathlib import Path

from rag_eval.reporting.summary import build_summary_markdown
from rag_eval.shared.models import DatasetConfig, EvaluationResult, RuntimeConfig, Scenario


def _scenario(tmp_path: Path) -> Scenario:
    return Scenario(
        scenario_name="summary-token-test",
        mode="offline",
        dataset=DatasetConfig(path=tmp_path / "dataset.csv"),
        judge_model="gpt-5",
        embedding_model="embedding-model",
        metrics=["faithfulness"],
        output_dir=tmp_path / "outputs",
        runtime=RuntimeConfig(batch_size=1),
    )


def test_summary_includes_token_usage_table(tmp_path: Path) -> None:
    scenario = _scenario(tmp_path)
    result = EvaluationResult(
        scenario=scenario,
        run_id="run-1",
        started_at="t0",
        finished_at="t1",
        valid_samples=[],
        invalid_samples=[],
        score_rows=[{"sample_id": "s1", "faithfulness": 0.9, "error": ""}],
        token_usage={
            "gpt-5": {"input_tokens": 12450, "output_tokens": 3200, "calls": 60},
            "Qwen/Qwen3-Embedding-4B": {"input_tokens": 45000, "output_tokens": 0, "calls": 30},
        },
    )

    markdown = build_summary_markdown(result)

    assert "## Token 用量" in markdown
    assert "gpt-5" in markdown
    assert "12450" in markdown
    assert "Qwen/Qwen3-Embedding-4B" in markdown
    assert "45000" in markdown


def test_summary_shows_fallback_text_when_token_usage_empty(tmp_path: Path) -> None:
    scenario = _scenario(tmp_path)
    result = EvaluationResult(
        scenario=scenario,
        run_id="run-2",
        started_at="t0",
        finished_at="t1",
        valid_samples=[],
        invalid_samples=[],
        score_rows=[{"sample_id": "s1", "faithfulness": 0.9, "error": ""}],
    )

    markdown = build_summary_markdown(result)

    assert "## Token 用量" in markdown
    assert "未记录 token 用量" in markdown

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_reporting_summary_token_usage.py -v Expected: FAIL — assert "## Token 用量" in markdown fails (section not present yet).

• Step 3: Write minimal implementation

In rag_eval/reporting/summary.py, add this helper function after _table_from_frame and before build_summary_markdown:

def _token_usage_section(token_usage: dict[str, dict[str, int]]) -> list[str]:
    """Render the '## Token 用量' section as a list of markdown lines."""
    lines = ["", "## Token 用量", ""]
    if not token_usage:
        lines.append("未记录 token 用量。")
        return lines
    lines.append("| 模型 | input_tokens | output_tokens | 调用次数 |")
    lines.append("|---|---|---|---|")
    for model in sorted(token_usage):
        usage = token_usage[model]
        lines.append(
            f"| {model} | {usage.get('input_tokens', 0)} "
            f"| {usage.get('output_tokens', 0)} | {usage.get('calls', 0)} |"
        )
    return lines

Then find the two return statements in build_summary_markdown. First, the empty-scores early return:

    if scores.empty:
        lines.append("No valid samples were scored.")
        return "\n".join(lines) + "\n"

Replace with:

    if scores.empty:
        lines.append("No valid samples were scored.")
        lines.extend(_token_usage_section(result.token_usage))
        return "\n".join(lines) + "\n"

Second, the final return at the end of the function:

    detail_columns = ["sample_id", *result.scenario.metrics, "weighted_score", "error"]
    existing_columns = [c for c in detail_columns if c in scores.columns]
    detail = scores[existing_columns]
    lines.extend([
        "",
        "## Per-sample Scores",
        "",
        "```text",
        _table_from_frame(detail),
        "```",
    ])
    return "\n".join(lines) + "\n"

Replace with:

    detail_columns = ["sample_id", *result.scenario.metrics, "weighted_score", "error"]
    existing_columns = [c for c in detail_columns if c in scores.columns]
    detail = scores[existing_columns]
    lines.extend([
        "",
        "## Per-sample Scores",
        "",
        "```text",
        _table_from_frame(detail),
        "```",
    ])
    lines.extend(_token_usage_section(result.token_usage))
    return "\n".join(lines) + "\n"

• Step 4: Run test to verify it passes

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

• Step 5: Commit

git add rag_eval/reporting/summary.py tests/test_reporting_summary_token_usage.py
git commit -m "feat(token-tracking): add Token usage section to summary.md"

---

Task 6: CLI main.py --scenario 评测流程接入统计

Files:

• Modify: rag_eval/execution/evaluator.py
• Test: tests/test_evaluator_token_usage.py

Interfaces:

• Consumes: track_token_usage() from Task 1; EvaluationResult.token_usage field from Task 4.

• Step 1: Write the failing test

Create tests/test_evaluator_token_usage.py:

"""Tests verifying the CLI evaluation flow captures token usage from metric scoring."""
from __future__ import annotations

import shutil
import unittest
from pathlib import Path

import pandas as pd

from rag_eval.execution.evaluator import Evaluator
from rag_eval.metrics.pipeline import MetricPipeline
from rag_eval.metrics.token_tracker import get_current_tracker
from rag_eval.shared.models import DatasetConfig, RuntimeConfig, Scenario


class FakeMetricWithUsage:
    """Fake RAGAS metric that records token usage like a real HTTP-hooked call would."""

    def __init__(self, value: float, model: str, input_tokens: int, output_tokens: int):
        self.value = value
        self.model = model
        self.input_tokens = input_tokens
        self.output_tokens = output_tokens

    async def ascore(self, **kwargs):
        tracker = get_current_tracker()
        if tracker is not None:
            tracker.record(self.model, self.input_tokens, self.output_tokens)

        class Result:
            def __init__(self, value: float):
                self.value = value

        return Result(self.value)


class PlainFakeMetric:
    """Fake metric that never records usage (simulates a hook that captured nothing)."""

    async def ascore(self, **kwargs):
        class Result:
            value = 0.9

        return Result()


class EvaluatorTokenUsageTests(unittest.TestCase):
    def setUp(self) -> None:
        root = Path("tests/.tmp").resolve()
        root.mkdir(parents=True, exist_ok=True)
        self.temp_dir = root / self._testMethodName
        shutil.rmtree(self.temp_dir, ignore_errors=True)
        self.temp_dir.mkdir(parents=True, exist_ok=True)

    def tearDown(self) -> None:
        shutil.rmtree(self.temp_dir, ignore_errors=True)

    def _write_offline_dataset(self, path: Path, rows: list[dict]) -> None:
        pd.DataFrame(rows).to_csv(path, index=False)

    def test_evaluate_populates_token_usage_from_metric_calls(self) -> None:
        dataset_path = self.temp_dir / "offline.csv"
        self._write_offline_dataset(dataset_path, [
            {
                "sample_id": "sample-1",
                "question": "What is the policy scope?",
                "answer": "It covers all employees.",
                "contexts": '["Context A"]',
                "ground_truth": "It covers all employees.",
            },
            {
                "sample_id": "sample-2",
                "question": "What about contractors?",
                "answer": "Contractors are excluded.",
                "contexts": '["Context B"]',
                "ground_truth": "Contractors are excluded.",
            },
        ])

        scenario = Scenario(
            scenario_name="token-usage-test",
            mode="offline",
            dataset=DatasetConfig(path=dataset_path),
            judge_model="gpt-5",
            embedding_model="embedding-model",
            metrics=["faithfulness"],
            output_dir=self.temp_dir / "outputs",
            runtime=RuntimeConfig(batch_size=1),
        )
        pipeline = MetricPipeline(
            metrics={"faithfulness": FakeMetricWithUsage(0.8, "gpt-5", 100, 40)}
        )
        evaluator = Evaluator(scenario=scenario, metric_pipeline=pipeline)

        result = evaluator.evaluate()

        # Two samples each recorded one call → totals sum across both.
        self.assertEqual(
            result.token_usage,
            {"gpt-5": {"input_tokens": 200, "output_tokens": 80, "calls": 2}},
        )

    def test_evaluate_defaults_to_empty_token_usage_when_nothing_recorded(self) -> None:
        dataset_path = self.temp_dir / "offline.csv"
        self._write_offline_dataset(dataset_path, [
            {
                "sample_id": "sample-1",
                "question": "What is the policy scope?",
                "answer": "It covers all employees.",
                "contexts": '["Context A"]',
                "ground_truth": "It covers all employees.",
            },
        ])

        scenario = Scenario(
            scenario_name="token-usage-empty-test",
            mode="offline",
            dataset=DatasetConfig(path=dataset_path),
            judge_model="gpt-5",
            embedding_model="embedding-model",
            metrics=["faithfulness"],
            output_dir=self.temp_dir / "outputs",
            runtime=RuntimeConfig(batch_size=1),
        )
        pipeline = MetricPipeline(metrics={"faithfulness": PlainFakeMetric()})
        evaluator = Evaluator(scenario=scenario, metric_pipeline=pipeline)

        result = evaluator.evaluate()
        self.assertEqual(result.token_usage, {})

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_evaluator_token_usage.py -v Expected: FAIL — AssertionError: {} != {'gpt-5': {...}} (evaluator does not wrap scoring in track_token_usage() yet, so nothing gets captured and result.token_usage stays {} from the Task 4 default).

• Step 3: Write minimal implementation

In rag_eval/execution/evaluator.py, add the import near the other rag_eval imports:

from rag_eval.metrics.pipeline import MetricPipeline
from rag_eval.metrics.token_tracker import track_token_usage
from rag_eval.metrics.weights import compute_weighted_score, resolve_weight

Find:

        logger.info("[eval] scoring %d samples with metric pipeline ...", len(samples))
        t0 = time.monotonic()
        metric_scores = asyncio.run(
            self.metric_pipeline.score_samples(
                samples,
                max_concurrency=self.scenario.runtime.metric_limit(),
            )
        )
        elapsed = time.monotonic() - t0
        logger.info("[eval] metric scoring done  elapsed=%.1fs", elapsed)

Replace with:

        logger.info("[eval] scoring %d samples with metric pipeline ...", len(samples))
        t0 = time.monotonic()
        with track_token_usage() as usage_tracker:
            metric_scores = asyncio.run(
                self.metric_pipeline.score_samples(
                    samples,
                    max_concurrency=self.scenario.runtime.metric_limit(),
                )
            )
        elapsed = time.monotonic() - t0
        logger.info("[eval] metric scoring done  elapsed=%.1fs", elapsed)
        logger.info("[eval] token_usage=%s", usage_tracker.as_dict())

Find the final return EvaluationResult(...):

        return EvaluationResult(
            scenario=self.scenario,
            run_id=run_id,
            started_at=started_at,
            finished_at=finished_at,
            valid_samples=samples,
            invalid_samples=invalid_samples,
            score_rows=score_rows,
        )

Replace with:

        return EvaluationResult(
            scenario=self.scenario,
            run_id=run_id,
            started_at=started_at,
            finished_at=finished_at,
            valid_samples=samples,
            invalid_samples=invalid_samples,
            score_rows=score_rows,
            token_usage=usage_tracker.as_dict(),
        )

• Step 4: Run test to verify it passes

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

Also re-run the broader offline/online suites to confirm no regression:

Run: C:\software\Python312\python.exe -m pytest tests/test_offline_eval.py tests/test_online_eval.py -v Expected: all pass (unchanged)

• Step 5: Commit

git add rag_eval/execution/evaluator.py tests/test_evaluator_token_usage.py
git commit -m "feat(token-tracking): wrap CLI evaluator metric scoring in track_token_usage"

---

Task 7: /api/score/async（score_job_manager.py）接入统计

Files:

• Modify: webapp/services/score_job_manager.py
• Test: tests/webapp/test_score_job_manager_token_usage.py

Interfaces:

• Consumes: track_token_usage() from Task 1; EvaluationResult.token_usage from Task 4.

• Step 1: Write the failing test

Create tests/webapp/test_score_job_manager_token_usage.py:

"""Tests that /api/score/async persists token usage captured during scoring."""
from __future__ import annotations

import json
import time

from webapp.models import ScoreRequest
from webapp.services.score_job_manager import ScoreJobManager


def _wait_for_status(mgr: ScoreJobManager, job_id: str, timeout: float = 2.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        status = mgr.get(job_id)
        if status is not None and status.status in ("completed", "failed"):
            return status
        time.sleep(0.02)
    raise TimeoutError(f"job {job_id} did not complete in time")


def test_run_writes_token_usage_to_metadata(tmp_path, monkeypatch):
    """_run() wraps inline_scorer.score in track_token_usage and persists totals."""
    from rag_eval.metrics.token_tracker import get_current_tracker

    mgr = ScoreJobManager(
        output_dir=tmp_path / "score-async",
        index_dir=tmp_path / "score-jobs",
        max_workers=1,
    )

    def _fake_score(**kwargs):
        tracker = get_current_tracker()
        if tracker is not None:
            tracker.record("gpt-5", 120, 45)
        return {m: 0.9 for m in kwargs["metrics"]}

    monkeypatch.setattr(
        "webapp.services.inline_scorer.inline_scorer.score", _fake_score
    )

    request = ScoreRequest(question="q?", answer="a.", metrics=["answer_relevancy"])
    status = mgr.submit(request)
    final_status = _wait_for_status(mgr, status.job_id)

    assert final_status.status == "completed"
    run_dir = tmp_path / "score-async" / final_status.run_id
    metadata = json.loads((run_dir / "metadata.json").read_text(encoding="utf-8"))
    assert metadata["token_usage"] == {
        "gpt-5": {"input_tokens": 120, "output_tokens": 45, "calls": 1}
    }


def test_run_writes_empty_token_usage_when_nothing_recorded(tmp_path, monkeypatch):
    mgr = ScoreJobManager(
        output_dir=tmp_path / "score-async",
        index_dir=tmp_path / "score-jobs",
        max_workers=1,
    )

    def _fake_score(**kwargs):
        return {m: 0.9 for m in kwargs["metrics"]}

    monkeypatch.setattr(
        "webapp.services.inline_scorer.inline_scorer.score", _fake_score
    )

    request = ScoreRequest(question="q?", answer="a.", metrics=["answer_relevancy"])
    status = mgr.submit(request)
    final_status = _wait_for_status(mgr, status.job_id)

    run_dir = tmp_path / "score-async" / final_status.run_id
    metadata = json.loads((run_dir / "metadata.json").read_text(encoding="utf-8"))
    assert metadata["token_usage"] == {}

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/webapp/test_score_job_manager_token_usage.py -v Expected: FAIL — KeyError: 'token_usage' is not raised (Task 4 already ensures the key exists), but the first test fails with assert {} == {'gpt-5': {...}} since _run() does not yet wrap scoring in track_token_usage().

• Step 3: Write minimal implementation

In webapp/services/score_job_manager.py, inside _run(), add the import to the existing lazy-import block:

        from rag_eval.advisor import run_advisor
        from rag_eval.metrics.weights import compute_weighted_score
        from rag_eval.reporting.writers import write_run_artifacts
        from rag_eval.settings import EvaluationSettings

becomes:

        from rag_eval.advisor import run_advisor
        from rag_eval.metrics.token_tracker import track_token_usage
        from rag_eval.metrics.weights import compute_weighted_score
        from rag_eval.reporting.writers import write_run_artifacts
        from rag_eval.settings import EvaluationSettings

Find:

        try:
            if effective:
                raw_scores = inline_scorer.score(
                    question=request.question,
                    answer=request.answer,
                    contexts=request.contexts_as_list(),
                    ground_truth=request.ground_truth,
                    metrics=effective,
                    judge_model=judge_model,
                    embedding_model=embedding_model,
                    settings=settings,
                    judge_language=judge_language,
                )
            else:
                raw_scores = {}

Replace with:

        try:
            with track_token_usage() as usage_tracker:
                if effective:
                    raw_scores = inline_scorer.score(
                        question=request.question,
                        answer=request.answer,
                        contexts=request.contexts_as_list(),
                        ground_truth=request.ground_truth,
                        metrics=effective,
                        judge_model=judge_model,
                        embedding_model=embedding_model,
                        settings=settings,
                        judge_language=judge_language,
                    )
                else:
                    raw_scores = {}

Find the EvaluationResult( construction:

            result = EvaluationResult(
                scenario=scenario,
                run_id=run_id,
                started_at=started_at,
                finished_at=finished_at,
                valid_samples=[sample],
                invalid_samples=[],
                score_rows=[score_row],
            )

Replace with:

            result = EvaluationResult(
                scenario=scenario,
                run_id=run_id,
                started_at=started_at,
                finished_at=finished_at,
                valid_samples=[sample],
                invalid_samples=[],
                score_rows=[score_row],
                token_usage=usage_tracker.as_dict(),
            )

• Step 4: Run test to verify it passes

Run: C:\software\Python312\python.exe -m pytest tests/webapp/test_score_job_manager_token_usage.py -v Expected: 2 passed

Also re-run the existing async score jobs API tests to confirm no regression:

Run: C:\software\Python312\python.exe -m pytest tests/webapp/test_score_jobs_api.py -v Expected: all pass (unchanged)

• Step 5: Commit

git add webapp/services/score_job_manager.py tests/webapp/test_score_job_manager_token_usage.py
git commit -m "feat(token-tracking): capture token usage in /api/score/async job runs"

---

Task 8: /api/score/session_async（session_score_manager.py）接入统计并跨调用累加

Files:

• Modify: webapp/services/session_score_manager.py
• Test: tests/webapp/test_session_score_manager_token_usage.py

Interfaces:

• Consumes: track_token_usage(), TokenUsageTracker.merge_into(...) from Task 1; EvaluationResult.token_usage from Task 4.

• Produces: SessionScoreJobManager._read_metadata(run_dir: Path) -> dict[str, Any] (private helper, mirrors the existing _read_score_rows).

• Step 1: Write the failing test

Create tests/webapp/test_session_score_manager_token_usage.py:

"""Tests that session-grouped async scoring accumulates token usage across calls."""
from __future__ import annotations

import json
import time

from webapp.models import ScoreRequest
from webapp.services.session_score_manager import SessionScoreJobManager


def _wait_for_call_count(mgr: SessionScoreJobManager, session_id: str, expected: int, timeout: float = 2.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        session = mgr.get_session(session_id)
        if session is not None and session.call_count >= expected:
            all_done = all(j.status in ("completed", "failed") for j in session.jobs)
            if all_done:
                return session
        time.sleep(0.02)
    raise TimeoutError(f"session {session_id} did not reach {expected} completed calls in time")


def test_session_accumulates_token_usage_across_calls(tmp_path, monkeypatch):
    from rag_eval.metrics.token_tracker import get_current_tracker

    mgr = SessionScoreJobManager(
        output_dir=tmp_path / "score-session",
        index_dir=tmp_path / "score-session-jobs",
        max_workers=1,
    )

    call_usages = iter([(100, 40), (30, 10)])

    def _fake_score(**kwargs):
        tracker = get_current_tracker()
        input_tok, output_tok = next(call_usages)
        if tracker is not None:
            tracker.record("gpt-5", input_tok, output_tok)
        return {m: 0.9 for m in kwargs["metrics"]}

    monkeypatch.setattr(
        "webapp.services.inline_scorer.inline_scorer.score", _fake_score
    )

    request = ScoreRequest(question="q?", answer="a.", metrics=["answer_relevancy"])

    _, run_id = mgr.submit("session-token-test", request)
    _wait_for_call_count(mgr, "session-token-test", 1)
    mgr.submit("session-token-test", request)
    _wait_for_call_count(mgr, "session-token-test", 2)

    run_dir = tmp_path / "score-session" / run_id
    metadata = json.loads((run_dir / "metadata.json").read_text(encoding="utf-8"))
    assert metadata["token_usage"] == {
        "gpt-5": {"input_tokens": 130, "output_tokens": 50, "calls": 2}
    }


def test_session_first_call_writes_token_usage_from_scratch(tmp_path, monkeypatch):
    from rag_eval.metrics.token_tracker import get_current_tracker

    mgr = SessionScoreJobManager(
        output_dir=tmp_path / "score-session",
        index_dir=tmp_path / "score-session-jobs",
        max_workers=1,
    )

    def _fake_score(**kwargs):
        tracker = get_current_tracker()
        if tracker is not None:
            tracker.record("gpt-5", 50, 20)
        return {m: 0.9 for m in kwargs["metrics"]}

    monkeypatch.setattr(
        "webapp.services.inline_scorer.inline_scorer.score", _fake_score
    )

    request = ScoreRequest(question="q?", answer="a.", metrics=["answer_relevancy"])
    _, run_id = mgr.submit("session-first-call-test", request)
    _wait_for_call_count(mgr, "session-first-call-test", 1)

    run_dir = tmp_path / "score-session" / run_id
    metadata = json.loads((run_dir / "metadata.json").read_text(encoding="utf-8"))
    assert metadata["token_usage"] == {"gpt-5": {"input_tokens": 50, "output_tokens": 20, "calls": 1}}

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/webapp/test_session_score_manager_token_usage.py -v Expected: FAIL — both tests fail with assert {} == {'gpt-5': {...}} (the manager does not wrap scoring in track_token_usage() yet).

• Step 3: Write minimal implementation

In webapp/services/session_score_manager.py, add _read_metadata right after the existing _read_score_rows method:

    def _read_score_rows(self, run_dir: Path) -> list[dict[str, Any]]:
        """Read existing scores.csv rows, returning empty list if file doesn't exist."""
        scores_path = run_dir / "scores.csv"
        if not scores_path.is_file():
            return []
        try:
            frame = pd.read_csv(scores_path)
            return frame.where(pd.notnull(frame), None).to_dict("records")
        except (OSError, ValueError):
            return []

    def _read_metadata(self, run_dir: Path) -> dict[str, Any]:
        """Read this session's existing metadata.json, returning {} if absent/unreadable."""
        metadata_path = run_dir / "metadata.json"
        if not metadata_path.is_file():
            return {}
        try:
            return json.loads(metadata_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return {}

Add the import to the existing lazy-import block inside _run():

        from rag_eval.advisor import run_advisor
        from rag_eval.metrics.weights import compute_weighted_score
        from rag_eval.reporting.writers import write_run_artifacts

becomes:

        from rag_eval.advisor import run_advisor
        from rag_eval.metrics.token_tracker import track_token_usage
        from rag_eval.metrics.weights import compute_weighted_score
        from rag_eval.reporting.writers import write_run_artifacts

Find:

        try:
            # --- Scoring (can run concurrently for the same session) ----------
            if effective:
                raw_scores = inline_scorer.score(
                    question=request.question,
                    answer=request.answer,
                    contexts=request.contexts_as_list(),
                    ground_truth=request.ground_truth,
                    metrics=effective,
                    judge_model=judge_model,
                    embedding_model=embedding_model,
                    settings=settings,
                    judge_language=judge_language,
                )
            else:
                raw_scores = {}

Replace with:

        try:
            # --- Scoring (can run concurrently for the same session) ----------
            with track_token_usage() as usage_tracker:
                if effective:
                    raw_scores = inline_scorer.score(
                        question=request.question,
                        answer=request.answer,
                        contexts=request.contexts_as_list(),
                        ground_truth=request.ground_truth,
                        metrics=effective,
                        judge_model=judge_model,
                        embedding_model=embedding_model,
                        settings=settings,
                        judge_language=judge_language,
                    )
                else:
                    raw_scores = {}

Find, inside the with session_lock: block:

            session_lock = self._get_session_lock(session_id)
            with session_lock:
                run_dir = self._output_dir / run_id
                run_dir.mkdir(parents=True, exist_ok=True)

                # Read all existing rows, then append the new one
                existing_rows = self._read_score_rows(run_dir)

Replace with:

            session_lock = self._get_session_lock(session_id)
            with session_lock:
                run_dir = self._output_dir / run_id
                run_dir.mkdir(parents=True, exist_ok=True)

                # Merge this call's token usage into the session's running total, so
                # repeated calls accumulate instead of overwriting (mirrors the
                # scores.csv append-only accumulation below).
                existing_metadata = self._read_metadata(run_dir)
                merged_token_usage = usage_tracker.merge_into(
                    existing_metadata.get("token_usage", {})
                )

                # Read all existing rows, then append the new one
                existing_rows = self._read_score_rows(run_dir)

Find the EvaluationResult( construction:

                result = EvaluationResult(
                    scenario=scenario,
                    run_id=run_id,
                    started_at=started_at_val if isinstance(started_at_val, str) else finished_at,
                    finished_at=finished_at,
                    valid_samples=valid_samples,
                    invalid_samples=[],
                    score_rows=all_rows,
                )

Replace with:

                result = EvaluationResult(
                    scenario=scenario,
                    run_id=run_id,
                    started_at=started_at_val if isinstance(started_at_val, str) else finished_at,
                    finished_at=finished_at,
                    valid_samples=valid_samples,
                    invalid_samples=[],
                    score_rows=all_rows,
                    token_usage=merged_token_usage,
                )

• Step 4: Run test to verify it passes

Run: C:\software\Python312\python.exe -m pytest tests/webapp/test_session_score_manager_token_usage.py -v Expected: 2 passed

Also re-run the existing session score jobs API tests to confirm no regression:

Run: C:\software\Python312\python.exe -m pytest tests/webapp/test_session_score_jobs_api.py -v Expected: all pass (unchanged)

• Step 5: Commit

git add webapp/services/session_score_manager.py tests/webapp/test_session_score_manager_token_usage.py
git commit -m "feat(token-tracking): accumulate token usage across session_async calls"

---

Task 9: Web 报告层透传 token_usage

Files:

• Modify: webapp/models.py
• Modify: webapp/services/report_builder.py
• Test: tests/test_report_builder_token_usage.py

Interfaces:

• Consumes: metadata.json's token_usage key (written by Tasks 4/7/8)

• Produces: ReportData.token_usage: dict[str, dict[str, int]], consumed by Task 10's report.js.

• Step 1: Write the failing test

Create tests/test_report_builder_token_usage.py:

"""Tests for token_usage passthrough in the webapp report builder."""
from __future__ import annotations

import json
from pathlib import Path

from webapp.services.report_builder import build_report


def _write_minimal_run(run_dir: Path, token_usage: dict | None) -> None:
    run_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / "scores.csv").write_text(
        "sample_id,faithfulness\ns1,0.9\n", encoding="utf-8"
    )
    (run_dir / "summary.md").write_text("summary", encoding="utf-8")
    metadata = {"run_id": run_dir.name}
    if token_usage is not None:
        metadata["token_usage"] = token_usage
    (run_dir / "metadata.json").write_text(json.dumps(metadata), encoding="utf-8")


def test_build_report_passes_through_token_usage(tmp_path: Path) -> None:
    run_dir = tmp_path / "run"
    _write_minimal_run(
        run_dir,
        token_usage={"gpt-5": {"input_tokens": 100, "output_tokens": 50, "calls": 2}},
    )

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

    assert report.token_usage == {
        "gpt-5": {"input_tokens": 100, "output_tokens": 50, "calls": 2}
    }


def test_build_report_defaults_token_usage_to_empty_dict(tmp_path: Path) -> None:
    run_dir = tmp_path / "run"
    _write_minimal_run(run_dir, token_usage=None)

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

    assert report.token_usage == {}


def test_build_report_early_return_branch_still_surfaces_token_usage(tmp_path: Path) -> None:
    """metrics=[] forces the early-return branch; token_usage must still surface."""
    run_dir = tmp_path / "run"
    _write_minimal_run(
        run_dir,
        token_usage={"gpt-5": {"input_tokens": 5, "output_tokens": 5, "calls": 1}},
    )

    report = build_report(run_dir, [])

    assert report.token_usage == {"gpt-5": {"input_tokens": 5, "output_tokens": 5, "calls": 1}}

• Step 2: Run test to verify it fails

Run: C:\software\Python312\python.exe -m pytest tests/test_report_builder_token_usage.py -v Expected: FAIL — AttributeError: 'ReportData' object has no attribute 'token_usage'

• Step 3: Write minimal implementation

In webapp/models.py, add the field to ReportData (after doc_weights):

class ReportData(BaseModel):
    """Aggregated report payload rendered by the report detail page."""

    metrics: list[str] = Field(default_factory=list)
    metric_means: dict[str, float | None] = Field(default_factory=dict)
    distributions: dict[str, list[DistributionBin]] = Field(default_factory=dict)
    groupings: dict[str, list[GroupStat]] = Field(default_factory=dict)
    lowest_samples: list[SampleScore] = Field(default_factory=list)
    summary_markdown: str = ""
    advice_markdown: str = ""  # optimization_advice.md content (empty if not generated)
    weighted_score_mean: float | None = Field(
        default=None,
        description="加权综合得分均值（metric_weights × doc_weights 共同作用）。",
    )
    metric_weights: dict[str, float] = Field(
        default_factory=dict,
        description="该次运行使用的指标权重配置（来自 scenario.snapshot.yaml）。",
    )
    doc_weights: dict[str, float] = Field(
        default_factory=dict,
        description="该次运行使用的文档权重配置（来自 scenario.snapshot.yaml）。",
    )
    token_usage: dict[str, dict[str, int]] = Field(
        default_factory=dict,
        description="按模型累计的 token 用量：{model: {input_tokens, output_tokens, calls}}。",
    )

In webapp/services/report_builder.py, find build_report:

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)

    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,
        )

Replace with:

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,
        )

Further down in the same function, find:

    # Cross-run history: scores of the same question in *other* runs (Approach A —
    # on-demand global scan, excluding the run currently being viewed).
    metadata = run_reader._read_json(run_dir / "metadata.json")
    current_run_id = str(metadata.get("run_id") or run_dir.name)
    history_index = question_history.build_question_history_index(
        exclude_run_id=current_run_id
    )

    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,
    )

Replace with (removes the now-duplicate metadata read, keeps current_run_id derivation):

    # Cross-run history: scores of the same question in *other* runs (Approach A —
    # on-demand global scan, excluding the run currently being viewed).
    current_run_id = str(metadata.get("run_id") or run_dir.name)
    history_index = question_history.build_question_history_index(
        exclude_run_id=current_run_id
    )

    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,
    )

• Step 4: Run test to verify it passes

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

Also re-run the existing report builder tests to confirm no regression:

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

• Step 5: Commit

git add webapp/models.py webapp/services/report_builder.py tests/test_report_builder_token_usage.py
git commit -m "feat(token-tracking): surface token_usage in ReportData"

---

Task 10: Web 报告详情页新增 "Token 用量" 面板

Files:

• Modify: webapp/static/index.html
• Modify: webapp/static/js/report.js

Interfaces:

• Consumes: ReportData.token_usage (JSON field token_usage from Task 9's API response)

This task has no automated test — the repository has no JavaScript test harness (vanilla JS, no package.json/Jest). Verify manually per Step 3 below.

• Step 1: Add the panel container to index.html

In webapp/static/index.html, find the optimization-advice section closing tag inside #report-content:

          <!-- ⑤ 优化建议（optimization_advisor: true 时显示） -->
          <div id="advice-section" hidden>
            <div class="section-label">⑤ 优化建议 OPTIMIZATION ADVICE</div>
            <div class="panel advice-panel">
              <div class="advice-header">
                <span class="advice-badge">AI 诊断报告</span>
                <span class="advice-model" id="advice-model-label"></span>
              </div>
              <div class="advice-body" id="advice-body"></div>
            </div>
          </div>
        </div>
      </section>

Replace with:

          <!-- ⑤ 优化建议（optimization_advisor: true 时显示） -->
          <div id="advice-section" hidden>
            <div class="section-label">⑤ 优化建议 OPTIMIZATION ADVICE</div>
            <div class="panel advice-panel">
              <div class="advice-header">
                <span class="advice-badge">AI 诊断报告</span>
                <span class="advice-model" id="advice-model-label"></span>
              </div>
              <div class="advice-body" id="advice-body"></div>
            </div>
          </div>

          <!-- ⑥ Token 用量（按模型分组，不换算金额） -->
          <div class="section-label">⑥ Token 用量</div>
          <div class="panel" id="token-usage-wrap"></div>
        </div>
      </section>

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

In webapp/static/js/report.js, find the render() method's body:

      const detail = await API.runDetail(runId);
      Report.currentDetail = detail;
      Report.renderMeta(detail.summary);
      Report.renderMetricCards(detail.summary, detail.report);
      Report.renderDistribution(detail.report);
      Report.renderGroupings(detail.report);
      Report.renderLowest(detail.report);
      Report.renderAdvice(detail.summary, detail.report);
      content.style.opacity = "1";

Replace with:

      const detail = await API.runDetail(runId);
      Report.currentDetail = detail;
      Report.renderMeta(detail.summary);
      Report.renderMetricCards(detail.summary, detail.report);
      Report.renderDistribution(detail.report);
      Report.renderGroupings(detail.report);
      Report.renderLowest(detail.report);
      Report.renderAdvice(detail.summary, detail.report);
      Report.renderTokenUsage(detail.report);
      content.style.opacity = "1";

Add the new method right after _drawGroupTable (reuses the existing group-table CSS class for visual consistency):

  // 渲染"Token 用量"面板：按模型分组的 input/output/调用次数表格。
  renderTokenUsage(report) {
    const wrap = document.getElementById("token-usage-wrap");
    if (!wrap) return;
    const usage = report.token_usage || {};
    const models = Object.keys(usage).sort();
    if (models.length === 0) {
      wrap.innerHTML = '<p class="muted tiny">暂无 token 用量数据。</p>';
      return;
    }
    let head = "<tr><th>模型</th><th>input_tokens</th><th>output_tokens</th><th>调用次数</th></tr>";
    let body = "";
    models.forEach((model) => {
      const u = usage[model] || {};
      body += `<tr><td>${App.escape(model)}</td><td>${u.input_tokens ?? 0}</td>` +
        `<td>${u.output_tokens ?? 0}</td><td>${u.calls ?? 0}</td></tr>`;
    });
    wrap.innerHTML = `<table class="group-table">${head}${body}</table>`;
  },

• Step 3: Manual verification

Run: C:\software\Python312\python.exe -m pytest tests/ -v -k token_usage (sanity check: all backend token-usage tests from Tasks 1–9 still pass together) Expected: all pass

Then start the web server and manually confirm the panel renders:

C:\software\Python312\python.exe -m uvicorn webapp.server:app --reload --port 8800

Open http://localhost:8800, submit any scoring request (e.g. via /api/score/async), open「运行列表」→ pick that run → 报告详情页 should show a new "⑥ Token 用量" panel: either the fallback text "暂无 token 用量数据。" (if the configured LLM gateway doesn't echo a usage field) or a table with model/input/output/调用次数 rows.

• Step 4: Commit

git add webapp/static/index.html webapp/static/js/report.js
git commit -m "feat(token-tracking): add Token usage panel to the report detail page"

---

Final Verification

• Run the full test suite once all 10 tasks are complete

Run: C:\software\Python312\python.exe -m pytest tests/ -v Expected: All tests pass, including the new test_token_tracker.py, test_token_usage_hook.py, test_token_usage_persistence.py, test_reporting_summary_token_usage.py, test_evaluator_token_usage.py, tests/webapp/test_score_job_manager_token_usage.py, tests/webapp/test_session_score_manager_token_usage.py, test_report_builder_token_usage.py, plus all pre-existing tests unchanged (any pre-existing failures unrelated to this feature are expected to remain as-is — do not attempt to fix them as part of this plan).

• Spec coverage check

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

• §3.1 (TokenUsageTracker + contextvar) → Task 1
• §3.2 (HTTP hook + attach_usage_hook) → Task 2
• §3.3 (挂载点: build_models + llm_analyzer) → Tasks 2, 3
• §4.1 (CLI evaluator) → Task 6
• §4.2 (/api/score/async) → Task 7
• §4.3 (/api/score/session_async 累加) → Task 8
• §4.4 (write_run_artifacts + summary.md) → Tasks 4, 5
• §5 (报告层与 Web UI) → Tasks 9, 10
• §6 (错误处理与边界情况) → covered inline in Tasks 2 (hook silently no-ops), 8 (session lock reuse), 9 (missing-key defaults)
• §7 (测试策略) → one test file per task, matching the spec's proposed file list
• §8 (非目标) → respected: no cost/$ conversion, no /api/score sync coverage, no dataset_builder coverage