Function _evaluate_single

src/benchmarks/writing_bench/evaluate.py:105–166

Evaluate a single result file. Returns the score dict or None if skipped.

(
    json_path: Path,
    eval_path: Path,
    benchmark: Dict[int, dict],
    client: LLMClient,
    judge_model: str,
    force: bool,
)

def _evaluate_single(
    json_path: Path,
    eval_path: Path,
    benchmark: Dict[int, dict],
    client: LLMClient,
    judge_model: str,
    force: bool,
) -> Optional[dict]:
    """Evaluate a single result file. Returns the score dict or None if skipped."""
    # Use cached eval if available
    if eval_path.exists() and not force:
        try:
            with eval_path.open("r") as f:
                return json.load(f)
        except Exception:
            pass

    with json_path.open("r") as f:
        run_data = json.load(f)

    index = run_data.get("index")
    if index not in benchmark:
        return None

    bm_entry = benchmark[index]
    query = bm_entry["query"]
    checklist = bm_entry["checklist"]
    response_text = run_data.get("response", "")
    is_completed = run_data.get("status") == "completed"

    if not response_text or not is_completed:
        result = {
            "index": index,
            "status": "skipped",
            "scores": {},
        }
    else:
        response_clean = _strip_think_prefix(response_text)
        scores: Dict[str, list] = {}

        for criteria in checklist:
            name = criteria["name"]
            scores[name] = []
            for _ in range(EVAL_TIMES):
                score_result = _call_judge(
                    client, query, response_clean, criteria, judge_model
                )
                if score_result:
                    scores[name].append(score_result)
                else:
                    scores[name].append({"score": 0, "reason": "Judge failed"})

        result = {
            "index": index,
            "status": "evaluated",
            "scores": scores,
        }