gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

pilot(eval): switch rule target from observation to trace

Browse Source 
The standard /api/public/ingestion path feeds only the trace-upsert
queue; evalService.createEvalJobs only dispatches targetObject in
{TRACE, DATASET}. Observation rules fire exclusively from the OTel
pipeline, which this pilot does not use. The trace body already carries
review input/output via langfuse_trace, so a trace rule sees the same
material an observation rule would.
This commit is contained in:
Claude
2026-08-31 19:18:59 +00:00
parent d746b1fdc2
commit 3109bd7c2d
2 changed files with 446 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/eval_judges.py
+314
View File
@@ -0,0 +1,314 @@
#!/usr/bin/env python3
"""pragent pilot — LLM-as-a-judge evaluators for the reviewer.
The deterministic scorers in `eval_scores.py` measure *behaviour*: how many
findings, how severe, how much they cost. None of them can say whether a
finding was any good. With no human labels in `feedback.db`, a judge is the
only thing that can — so these two ask the questions that need no ground truth,
only the review itself:
`finding_actionability` — is each finding concrete enough to act on? A
reviewer that says "consider improving error handling" at file level is
indistinguishable from a useful one by finding count alone. This is the
failure mode a cheap model degrades into first.
`review_self_consistency` — does the summary agree with the findings it
posted? Claiming "no issues found" above a list of two criticals, or
describing a problem in prose that never became a finding, is a defect the
reviewer can commit entirely on its own.
Neither judge is asked whether a finding is *correct*. That needs the diff,
which these traces do not carry, and a judge asked to rule on correctness from
a summary alone will confabulate. Accuracy stays an open question until humans
start labelling — which is what `feedback_scores.py` is there to capture.
**The judge is a different model from the reviewer.** The reviewer runs
MiniMax-M2.7; the judge runs kimi-k2.7-code through the same headroom hub. A
model grading its own output agrees with itself for reasons that have nothing
to do with quality.
Evaluators score *observations*, and their variable mapping reads the
observation's own input/output — which is why `langfuse_trace` now writes the
review onto the generation and not just onto the trace.
Usage:
LANGFUSE_HOST=... LANGFUSE_PUBLIC_KEY=... LANGFUSE_SECRET_KEY=... \\
python3 eval_judges.py --dry-run
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import eval_bootstrap as eb # noqa: E402
# The headroom hub in front of the local Ollama, plus a small pass-through
# proxy (`judge-proxy` on 8802) that patches every `thinking` content block
# to carry the `signature` field Langfuse's Anthropic adapter requires. The
# underlying model is kimi-k2.7-code through the hub on 8790; the proxy fixes
# the shape so Mastra's Zod parse stops failing.
JUDGE_PROVIDER = "headroom-ollama"
JUDGE_BASE_URL = os.environ.get("PRAGENT_JUDGE_BASE_URL", "http://100.74.17.70:8802")
JUDGE_API_KEY = os.environ.get("PRAGENT_JUDGE_API_KEY", "ollama")
JUDGE_MODEL = os.environ.get("PRAGENT_JUDGE_MODEL", "kimi-k2.7-code:cloud")
# The trace names this project emits (`pr-review` on the trace, `opencode-review`
# on the generation). Filter on `traceName` rather than observation `name` — the
# observation-rule schema only exposes `traceName` as a stringOptions column, and
# every observation inside these traces is the review itself, so the narrowness
# is the same.
REVIEW_TRACE_NAMES = ["pr-review", "opencode-review"]
def _model_config() -> dict:
return {"provider": JUDGE_PROVIDER, "model": JUDGE_MODEL}
JUDGES = [
{
"name": "finding_actionability",
"prompt": (
"You are auditing the output of an automated code reviewer.\n\n"
"PR under review:\n{{input}}\n\n"
"What the reviewer produced:\n{{output}}\n\n"
"Rate how ACTIONABLE the findings are, from 0 to 1. A finding is "
"actionable when a developer could act on it without asking a "
"follow-up question: it points at a specific location, names a "
"concrete problem, and proposes a fix that could be applied.\n\n"
"Score 1.0 when every finding is specific and fixable. Score around "
"0.5 when findings identify a real area but leave the developer to "
"work out what to change. Score near 0.0 when findings are generic "
"advice that would apply to almost any pull request.\n\n"
"Judge only specificity and actionability. You cannot see the diff, "
"so do NOT attempt to judge whether a finding is factually correct, "
"and do not penalise a finding for being one you cannot verify.\n\n"
"If the reviewer reported no findings at all, return 1.0 and say in "
"your reasoning that there was nothing to judge — a silent review is "
"measured by finding_rate, not here."
),
"outputDefinition": {
"dataType": "NUMERIC",
"minValue": 0,
"maxValue": 1,
"reasoning": {
"description": (
"Name the least actionable finding and say what it would "
"need in order to be acted on."
)
},
"score": {"description": "0 = generic advice, 1 = every finding is specific and fixable."},
},
},
{
"name": "review_self_consistency",
"prompt": (
"You are auditing the output of an automated code reviewer.\n\n"
"PR under review:\n{{input}}\n\n"
"What the reviewer produced:\n{{output}}\n\n"
"The output contains a prose `summary` and a list of `findings`. "
"Decide whether the summary is CONSISTENT with the findings.\n\n"
"Inconsistent means, for example: the summary says no issues were "
"found while findings are listed; the summary describes a problem "
"that never became a finding; the summary characterises the severity "
"of the findings in a way the findings themselves contradict; or the "
"summary refers to files that appear in no finding and in no part of "
"the PR description.\n\n"
"A summary that adds context beyond the findings is NOT inconsistent "
"as long as nothing in it contradicts them. A review that found "
"nothing and says so is consistent.\n\n"
"You cannot see the diff. Judge the summary against the findings and "
"the PR title only — never against what you imagine the code does."
),
"outputDefinition": {
"dataType": "BOOLEAN",
"reasoning": {
"description": "Quote the part of the summary that conflicts with the findings, if any."
},
"score": {"description": "true = summary agrees with the findings, false = it contradicts them."},
},
},
]
# Both judges read the observation's own input/output.
MAPPING = [
{"variable": "input", "source": "input"},
{"variable": "output", "source": "output"},
]
# ---------------------------------------------------------------------------
# LLM connection
# ---------------------------------------------------------------------------
def ensure_llm_connection() -> dict:
"""Point the project at the judge model. Upserted on `provider`."""
body = {
"provider": JUDGE_PROVIDER,
"adapter": "anthropic",
"baseURL": JUDGE_BASE_URL,
"secretKey": JUDGE_API_KEY,
"customModels": [JUDGE_MODEL],
# The hub serves two local models and none of Anthropic's, so the
# default catalogue would be a list of models that all fail on use.
"withDefaultModels": False,
}
st, resp = eb._call("PUT", "/api/public/llm-connections", body)
return {"status": st, "ok": st in (200, 201), "provider": JUDGE_PROVIDER,
"error": None if st in (200, 201) else resp}
# ---------------------------------------------------------------------------
# Evaluators
# ---------------------------------------------------------------------------
def existing_evaluators() -> dict[str, str]:
"""name -> id for evaluators already in the project."""
out: dict[str, str] = {}
st, body = eb._call("GET", "/api/public/unstable/evaluators?limit=100")
if st == 200 and isinstance(body, dict):
for ev in body.get("data") or []:
out[ev.get("name")] = ev.get("id")
return out
def ensure_evaluators() -> dict:
"""Create each judge if no version exists for the name yet.
POST /evaluators with a name that already exists creates a new version, not
a no-op — re-running this script would pile up versions until the page
listing them is unreadable. Skip when an evaluator of that name is present.
"""
created, skipped, failed = {}, [], []
existing = set(existing_evaluators())
for judge in JUDGES:
if judge["name"] in existing:
skipped.append(judge["name"])
continue
body = {
"type": "llm_as_judge",
"name": judge["name"],
"prompt": judge["prompt"],
"outputDefinition": judge["outputDefinition"],
"modelConfig": _model_config(),
}
st, resp = eb._call("POST", "/api/public/unstable/evaluators", body, timeout=60.0)
if st in (200, 201) and isinstance(resp, dict):
created[judge["name"]] = resp.get("id")
else:
failed.append({"name": judge["name"], "status": st, "error": resp})
return {"created": created, "skipped": skipped, "failed": failed}
# ---------------------------------------------------------------------------
# Rules — what gets judged, and how often
# ---------------------------------------------------------------------------
def rule_body(name: str, judge_name: str, sampling: float) -> dict:
"""POST /evaluation-rules shape for an LLM-as-judge trace rule.
Target is `trace` rather than `observation` on purpose: the standard
`/api/public/ingestion` path that ships review traces here feeds only
the trace-upsert queue, and `evalService.createEvalJobs` only creates
jobs for `targetObject ∈ {TRACE, DATASET}`. Observation rules are
triggered exclusively from the OTel ingestion pipeline, which this
pilot does not use. A trace rule reads the trace's own input/output —
`langfuse_trace` already writes `_review_input`/`_review_output` onto
the trace body for exactly this reason.
Mapping is required at both the rule root (server validates it there)
and inside `evaluator` (the API echoes it back).
"""
return {
"name": name,
"enabled": True,
"target": "trace",
"sampling": sampling,
"filter": [
{"column": "traceName", "operator": "any of",
"value": REVIEW_TRACE_NAMES, "type": "stringOptions"},
],
"evaluator": {
"name": judge_name,
"scope": "project",
"variableMapping": MAPPING,
},
"mapping": MAPPING,
}
def ensure_rules(evaluator_ids: dict[str, str], sampling: float) -> dict:
"""Idempotent: existing rules with the same name are skipped, not duplicated.
The API has no `name`-keyed upsert; the convention is to POST once and
re-run the script to verify the response. A duplicate POST raises 409.
"""
created, failed, skipped = [], [], []
existing = existing_rule_names()
for name, eid in evaluator_ids.items():
if not eid:
continue
rule_name = f"{name}-on-reviews"
if rule_name in existing:
skipped.append(name)
continue
st, resp = eb._call(
"POST", "/api/public/unstable/evaluation-rules",
rule_body(rule_name, name, sampling), timeout=60.0,
)
if st in (200, 201):
created.append(name)
else:
failed.append({"rule": name, "status": st, "error": resp})
return {"created": created, "failed": failed, "skipped": skipped}
def existing_rule_names() -> set[str]:
"""Names of observation-target rules already in the project."""
out: set[str] = set()
st, body = eb._call("GET", "/api/public/unstable/evaluation-rules?limit=100")
if st == 200 and isinstance(body, dict):
for r in body.get("data") or []:
if r.get("target") == "observation":
out.add(r.get("name"))
return out
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--sampling", type=float, default=1.0,
help="fraction of matching observations to judge (default: all)")
ap.add_argument("--skip-connection", action="store_true")
ap.add_argument("--dry-run", action="store_true")
args = ap.parse_args(argv)
if args.dry_run:
print(json.dumps({
"would_connect": {"provider": JUDGE_PROVIDER, "baseURL": JUDGE_BASE_URL,
"model": JUDGE_MODEL},
"would_create": [j["name"] for j in JUDGES],
"existing_evaluators": sorted(existing_evaluators()),
"sampling": args.sampling,
}, indent=2))
return 0
report = {}
if not args.skip_connection:
report["llm_connection"] = ensure_llm_connection()
report["evaluators"] = ensure_evaluators()
ids = dict(report["evaluators"]["created"])
# Fall back to whatever is already registered, so a re-run still wires rules.
for name, eid in existing_evaluators().items():
ids.setdefault(name, eid)
report["rules"] = ensure_rules(
{j["name"]: ids.get(j["name"]) for j in JUDGES}, args.sampling
)
print(json.dumps(report, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/pilot/test_eval_judges.py
+132
View File
@@ -0,0 +1,132 @@
"""Tests for the LLM-as-judge evaluator bootstrap."""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))
import eval_judges as ej # noqa: E402
# --- rule_body ------------------------------------------------------------
def test_rule_body_targets_traces():
"""Trace target matches the path `/api/public/ingestion` triggers.
Observation rules only fire from the OTel ingestion pipeline; this
pilot uses standard ingestion, so its jobs only come from
`evalService.createEvalJobs` and that dispatcher handles
`targetObject ∈ {TRACE, DATASET}`.
"""
body = ej.rule_body("rule-x", "finding_actionability", 1.0)
assert body["target"] == "trace"
assert body["enabled"] is True
def test_rule_body_filters_on_trace_name():
"""`name` isn't a stringOptions column; only `traceName` is."""
body = ej.rule_body("rule-x", "finding_actionability", 1.0)
f = body["filter"][0]
assert f["column"] == "traceName"
assert f["operator"] == "any of"
assert f["type"] == "stringOptions"
assert "pr-review" in f["value"]
def test_rule_body_references_evaluator_by_name():
"""Ids are version-specific; rules must name the evaluator across versions."""
body = ej.rule_body("rule-x", "finding_actionability", 1.0)
assert body["evaluator"]["name"] == "finding_actionability"
assert body["evaluator"]["scope"] == "project"
def test_rule_body_maps_input_and_output():
"""Both judges read the observation's own input/output."""
body = ej.rule_body("rule-x", "any", 1.0)
sources = {m["source"] for m in body["mapping"]}
assert sources == {"input", "output"}
def test_rule_body_carries_mapping_at_both_levels():
"""The server validates `mapping` at the rule root and echoes it on the evaluator."""
body = ej.rule_body("rule-x", "any", 1.0)
assert body["mapping"]
assert body["evaluator"]["variableMapping"] == body["mapping"]
def test_rule_body_passes_sampling_through():
assert ej.rule_body("r", "any", 0.25)["sampling"] == 0.25
# --- ensure_evaluators idempotency ---------------------------------------
def test_ensure_evaluators_skips_existing(monkeypatch):
seen = []
def fake_call(method, path, body=None, timeout=20.0):
seen.append(path)
return 200, {}
monkeypatch.setattr(ej.eb, "_call", fake_call)
monkeypatch.setattr(ej, "existing_evaluators",
lambda: {"finding_actionability": "id-1", "review_self_consistency": "id-2"})
res = ej.ensure_evaluators()
assert res["created"] == {}
assert sorted(res["skipped"]) == ["finding_actionability", "review_self_consistency"]
assert res["failed"] == []
assert seen == []
def test_ensure_evaluators_records_failures(monkeypatch):
def fake_call(method, path, body=None, timeout=20.0):
return 422, "boom"
monkeypatch.setattr(ej.eb, "_call", fake_call)
monkeypatch.setattr(ej, "existing_evaluators", lambda: {})
res = ej.ensure_evaluators()
assert res["created"] == {}
assert res["failed"][0]["status"] == 422
# --- ensure_rules idempotency --------------------------------------------
def test_ensure_rules_skips_existing(monkeypatch):
calls = []
monkeypatch.setattr(ej.eb, "_call",
lambda *a, **k: calls.append(a) or (200, {}))
monkeypatch.setattr(ej, "existing_evaluators",
lambda: {"finding_actionability": "id-1",
"review_self_consistency": "id-2"})
monkeypatch.setattr(ej, "existing_rule_names",
lambda: {"finding_actionability-on-reviews",
"review_self_consistency-on-reviews"})
res = ej.ensure_rules({"finding_actionability": "id-1",
"review_self_consistency": "id-2"}, 1.0)
assert res["created"] == []
assert sorted(res["skipped"]) == ["finding_actionability", "review_self_consistency"]
assert calls == []
def test_ensure_rules_creates_when_missing(monkeypatch):
calls = []
monkeypatch.setattr(ej.eb, "_call",
lambda *a, **k: calls.append(a) or (201, {}))
monkeypatch.setattr(ej, "existing_rule_names", lambda: set())
res = ej.ensure_rules({"finding_actionability": "id-1"}, 1.0)
assert res["created"] == ["finding_actionability"]
assert calls[0][0] == "POST"
assert calls[0][1] == "/api/public/unstable/evaluation-rules"
# --- judge shape ----------------------------------------------------------
def test_judges_have_required_keys():
for j in ej.JUDGES:
assert j["prompt"]
assert j["outputDefinition"]["dataType"] in ("NUMERIC", "BOOLEAN", "CATEGORICAL")
def test_default_base_url_points_at_the_thinking_patch_proxy():
"""`8802` is the judge-proxy that adds a `signature` to thinking blocks."""
assert "8802" in ej.JUDGE_BASE_URL