|
| 1 | +""" |
| 2 | +LLM-as-judge rubric scoring for agent turns (subjective prompt-quality signal). |
| 3 | +
|
| 4 | +The deterministic scorer in ``agent_eval`` measures *behavior* (did the agent follow the |
| 5 | +edit contract, loop, error out). This module covers the *subjective* half — tone, scope |
| 6 | +discipline, directness, and whether the final summary is useful — by asking a model to |
| 7 | +grade the turn transcript against a fixed rubric and return structured JSON. |
| 8 | +
|
| 9 | +It is intentionally separate and opt-in: nothing here runs in the default gate, and the |
| 10 | +judge model is supplied by the caller (no judge dependency is pinned). Use a capable model |
| 11 | +as the judge — grading is easier than coding, but a 3b model makes a noisy judge. |
| 12 | +
|
| 13 | +Usage (see ``tests/core/test_agent_prompt_eval.py``):: |
| 14 | +
|
| 15 | + from cecli import models |
| 16 | + from bright_vision_core.agent_judge import judge_transcript, transcript_from_events |
| 17 | + judge_model = models.Model("ollama_chat/qwen3-coder:30b") |
| 18 | + verdict = await judge_transcript(judge_model, task, transcript_from_events(events)) |
| 19 | + assert verdict.overall >= 3 |
| 20 | +""" |
| 21 | + |
| 22 | +from __future__ import annotations |
| 23 | + |
| 24 | +import json |
| 25 | +import re |
| 26 | +from dataclasses import dataclass, asdict, field |
| 27 | +from typing import Any |
| 28 | + |
| 29 | +# Rubric dimensions scored 1 (poor) .. 5 (excellent). Keep names stable: tests and |
| 30 | +# eval logs key on them. |
| 31 | +RUBRIC_DIMENSIONS: dict[str, str] = { |
| 32 | + "scope_discipline": ( |
| 33 | + "Did the agent do only what the task asked, without unrequested refactors, " |
| 34 | + "reformatting, or edits to unrelated code? 5 = perfectly scoped; 1 = sprawling." |
| 35 | + ), |
| 36 | + "directness": ( |
| 37 | + "Was the agent's communication concise and confident, leading with substance, " |
| 38 | + "without filler, hedging, or repeated restatements? 5 = crisp; 1 = rambling." |
| 39 | + ), |
| 40 | + "investigation": ( |
| 41 | + "Did the agent read/inspect relevant code before acting or claiming, rather than " |
| 42 | + "guessing? 5 = grounded in what it actually read; 1 = unfounded assertions." |
| 43 | + ), |
| 44 | + "summary_quality": ( |
| 45 | + "Was the final summary accurate, useful, and free of tool-call syntax or internal " |
| 46 | + "jargon — something the user can act on? 5 = clear and honest; 1 = absent/misleading." |
| 47 | + ), |
| 48 | +} |
| 49 | + |
| 50 | +_SCALE_MIN = 1 |
| 51 | +_SCALE_MAX = 5 |
| 52 | + |
| 53 | +_SYSTEM_PROMPT = ( |
| 54 | + "You are a strict, fair evaluator of an AI software-engineering agent's turn. " |
| 55 | + "You are grading the agent's behavior and communication quality against a rubric — " |
| 56 | + "NOT redoing the task. Be objective and cite the transcript. " |
| 57 | + "Respond with ONLY a single JSON object, no prose, no code fences." |
| 58 | +) |
| 59 | + |
| 60 | + |
| 61 | +@dataclass |
| 62 | +class JudgeVerdict: |
| 63 | + """Rubric scores (1..5 per dimension) plus an overall and the judge's notes.""" |
| 64 | + |
| 65 | + scores: dict[str, int] |
| 66 | + overall: float |
| 67 | + notes: str = "" |
| 68 | + raw: str = "" |
| 69 | + parse_error: str | None = None |
| 70 | + dimensions: dict[str, str] = field(default_factory=lambda: dict(RUBRIC_DIMENSIONS)) |
| 71 | + |
| 72 | + def as_dict(self) -> dict[str, Any]: |
| 73 | + return asdict(self) |
| 74 | + |
| 75 | + @property |
| 76 | + def ok(self) -> bool: |
| 77 | + """True when the judge returned usable scores for every rubric dimension.""" |
| 78 | + return self.parse_error is None and set(self.scores) == set(RUBRIC_DIMENSIONS) |
| 79 | + |
| 80 | + |
| 81 | +def transcript_from_events(events: list[dict], *, max_chars: int = 12_000) -> str: |
| 82 | + """ |
| 83 | + Render an SSE event stream into a compact transcript for the judge. |
| 84 | +
|
| 85 | + Keeps the user message, assistant prose, tool calls, tool output/errors, and the final |
| 86 | + summary — the things a reviewer would read. Token-usage footers and progress pulses are |
| 87 | + dropped. Truncated from the middle if oversized so both the task and the ending survive. |
| 88 | + """ |
| 89 | + lines: list[str] = [] |
| 90 | + for event in events: |
| 91 | + kind = event.get("type") |
| 92 | + text = str(event.get("text") or "").strip() |
| 93 | + if kind == "user_message": |
| 94 | + lines.append(f"USER: {text}") |
| 95 | + elif kind == "tool_output": |
| 96 | + if text and not _is_noise(text): |
| 97 | + lines.append(f"TOOL: {text}") |
| 98 | + elif kind == "tool_error": |
| 99 | + if text: |
| 100 | + lines.append(f"TOOL_ERROR: {text}") |
| 101 | + elif kind == "error": |
| 102 | + if text: |
| 103 | + lines.append(f"ERROR: {text}") |
| 104 | + elif kind == "done": |
| 105 | + summary = str(event.get("assistant_text") or "").strip() |
| 106 | + if summary: |
| 107 | + lines.append(f"ASSISTANT_SUMMARY: {summary}") |
| 108 | + transcript = "\n".join(lines).strip() |
| 109 | + if len(transcript) <= max_chars: |
| 110 | + return transcript |
| 111 | + head = transcript[: max_chars // 2] |
| 112 | + tail = transcript[-max_chars // 2 :] |
| 113 | + return f"{head}\n…[transcript truncated]…\n{tail}" |
| 114 | + |
| 115 | + |
| 116 | +_NOISE_PREFIXES = ("Recovered prose shell", "Running ") |
| 117 | +_TOKEN_FOOTER = re.compile(r"^[\d.]+k?\s+[↑↓]") |
| 118 | + |
| 119 | + |
| 120 | +def _is_noise(text: str) -> bool: |
| 121 | + if _TOKEN_FOOTER.match(text): |
| 122 | + return True |
| 123 | + return any(text.startswith(p) for p in _NOISE_PREFIXES) |
| 124 | + |
| 125 | + |
| 126 | +def build_judge_messages(task: str, transcript: str) -> list[dict[str, str]]: |
| 127 | + """Build the chat messages for the judge model.""" |
| 128 | + rubric_lines = "\n".join(f"- {name}: {desc}" for name, desc in RUBRIC_DIMENSIONS.items()) |
| 129 | + keys = ", ".join(f'"{k}"' for k in RUBRIC_DIMENSIONS) |
| 130 | + user = ( |
| 131 | + f"# Task given to the agent\n{task}\n\n" |
| 132 | + f"# Agent turn transcript\n{transcript}\n\n" |
| 133 | + f"# Rubric (score each {_SCALE_MIN}-{_SCALE_MAX}, integers only)\n{rubric_lines}\n\n" |
| 134 | + "# Output\n" |
| 135 | + "Return ONLY this JSON object (no markdown, no fences):\n" |
| 136 | + '{"scores": {' + keys + ': <int>}, "notes": "<one or two sentences citing the transcript>"}' |
| 137 | + ) |
| 138 | + return [ |
| 139 | + {"role": "system", "content": _SYSTEM_PROMPT}, |
| 140 | + {"role": "user", "content": user}, |
| 141 | + ] |
| 142 | + |
| 143 | + |
| 144 | +def _clamp(value: Any) -> int | None: |
| 145 | + try: |
| 146 | + n = int(round(float(value))) |
| 147 | + except (TypeError, ValueError): |
| 148 | + return None |
| 149 | + return max(_SCALE_MIN, min(_SCALE_MAX, n)) |
| 150 | + |
| 151 | + |
| 152 | +def parse_judge_response(raw: str) -> JudgeVerdict: |
| 153 | + """Parse the judge's JSON reply into a verdict, tolerating fences and stray prose.""" |
| 154 | + text = (raw or "").strip() |
| 155 | + # Strip accidental code fences. |
| 156 | + fence = re.match(r"^```[a-zA-Z]*\s*(.*?)\s*```$", text, re.DOTALL) |
| 157 | + if fence: |
| 158 | + text = fence.group(1).strip() |
| 159 | + # Grab the first {...} block if there is surrounding prose. |
| 160 | + if not text.startswith("{"): |
| 161 | + brace = re.search(r"\{.*\}", text, re.DOTALL) |
| 162 | + if brace: |
| 163 | + text = brace.group(0) |
| 164 | + try: |
| 165 | + data = json.loads(text) |
| 166 | + except (json.JSONDecodeError, TypeError) as err: |
| 167 | + return JudgeVerdict(scores={}, overall=0.0, raw=raw, parse_error=f"json: {err}") |
| 168 | + |
| 169 | + raw_scores = data.get("scores") if isinstance(data, dict) else None |
| 170 | + if not isinstance(raw_scores, dict): |
| 171 | + return JudgeVerdict(scores={}, overall=0.0, raw=raw, parse_error="missing 'scores' object") |
| 172 | + |
| 173 | + scores: dict[str, int] = {} |
| 174 | + for dim in RUBRIC_DIMENSIONS: |
| 175 | + val = _clamp(raw_scores.get(dim)) |
| 176 | + if val is not None: |
| 177 | + scores[dim] = val |
| 178 | + missing = set(RUBRIC_DIMENSIONS) - set(scores) |
| 179 | + parse_error = f"missing dimensions: {sorted(missing)}" if missing else None |
| 180 | + overall = round(sum(scores.values()) / len(scores), 2) if scores else 0.0 |
| 181 | + notes = str(data.get("notes") or "").strip() if isinstance(data, dict) else "" |
| 182 | + return JudgeVerdict( |
| 183 | + scores=scores, overall=overall, notes=notes, raw=raw, parse_error=parse_error |
| 184 | + ) |
| 185 | + |
| 186 | + |
| 187 | +async def judge_transcript(judge_model, task: str, transcript: str) -> JudgeVerdict: |
| 188 | + """ |
| 189 | + Score a turn transcript with ``judge_model`` against the rubric. |
| 190 | +
|
| 191 | + ``judge_model`` is any cecli ``models.Model`` exposing the async |
| 192 | + ``simple_send_with_retries(messages)`` API. Returns a :class:`JudgeVerdict`; on a model |
| 193 | + or parse failure the verdict carries ``parse_error`` and ``ok == False`` rather than |
| 194 | + raising, so eval callers can degrade gracefully. |
| 195 | + """ |
| 196 | + messages = build_judge_messages(task, transcript) |
| 197 | + try: |
| 198 | + reply = await judge_model.simple_send_with_retries(messages) |
| 199 | + except Exception as err: # network/model errors should not crash an eval run |
| 200 | + return JudgeVerdict(scores={}, overall=0.0, parse_error=f"model: {err}") |
| 201 | + if not reply: |
| 202 | + return JudgeVerdict(scores={}, overall=0.0, parse_error="empty judge response") |
| 203 | + return parse_judge_response(reply) |
| 204 | + |
| 205 | + |
| 206 | +def summarize_verdict(label: str, verdict: JudgeVerdict) -> str: |
| 207 | + """One-line human summary for eval logs / CI output.""" |
| 208 | + if not verdict.ok: |
| 209 | + return f"[{label}] judge: UNAVAILABLE ({verdict.parse_error})" |
| 210 | + dims = " ".join(f"{k}={v}" for k, v in verdict.scores.items()) |
| 211 | + return f"[{label}] judge overall={verdict.overall} {dims}" |
0 commit comments