Claude
7a510a926d
Group review, feedback, evaluation, observability, and entrypoint code into packages. Keep thin top-level compatibility shims for existing scripts and imports, and mirror the structure in the tests.
351 lines
14 KiB
Python
351 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""pragent pilot — one-time Langfuse project setup for evaluation.
|
|
|
|
Three jobs, each idempotent so it can be re-run after any change:
|
|
|
|
1. **Score configs.** Registers the schema for every score pragent emits
|
|
(`eval_scores.SCORE_CONFIGS` + `feedback_scores.SCORE_CONFIGS`). Without
|
|
these the scores still ingest, but nothing stops a later scorer writing
|
|
`severity_max="HIGH"` beside today's `"high"` and quietly splitting one
|
|
series into two. Configs are immutable in Langfuse — a name that already
|
|
exists is left alone rather than updated.
|
|
|
|
2. **Dataset.** Seeds `pragent-reviews` from `feedback.db`: one item per PR
|
|
the reviewer has actually run on, carrying the repo/PR/sha as input and
|
|
the findings it posted as `expectedOutput`.
|
|
|
|
Read `expectedOutput` here as "what the reviewer said last time", not "what
|
|
is correct" — no human has labelled any of it. It is a regression baseline:
|
|
re-run a candidate model over these PRs and the diff against this column is
|
|
the behaviour change. Promoting an item to real ground truth means a human
|
|
editing it after reviewing the PR, which is what the dataset view is for.
|
|
|
|
3. **Trace backfill** (`--backfill-traces`). Scores only ride along with new
|
|
reviews, so without this the charts stay empty until the next PR lands.
|
|
Every trace `langfuse_trace` has ever written already carries the finding
|
|
count, the severity histogram and the cost in its metadata, which is
|
|
everything four of the five scorers need. `dropped_findings` is absent from
|
|
historical traces and is left unscored rather than backfilled as zero.
|
|
|
|
4. **Reports** what it found, so the gap between "reviews recorded" and
|
|
"reviews with human feedback" is visible rather than assumed.
|
|
|
|
Usage:
|
|
LANGFUSE_HOST=... LANGFUSE_PUBLIC_KEY=... LANGFUSE_SECRET_KEY=... \\
|
|
python3 eval_bootstrap.py --db /data/feedback.db
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import eval_scores # noqa: E402
|
|
import feedback_scores # noqa: E402
|
|
|
|
DATASET_NAME = "pragent-reviews"
|
|
|
|
|
|
def _conf() -> tuple[str, str, str]:
|
|
host = (os.environ.get("LANGFUSE_HOST") or "").strip().rstrip("/")
|
|
pk = (os.environ.get("LANGFUSE_PUBLIC_KEY") or "").strip()
|
|
sk = (os.environ.get("LANGFUSE_SECRET_KEY") or "").strip()
|
|
if not host or not pk or not sk:
|
|
raise SystemExit("LANGFUSE_HOST / LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY must be set")
|
|
return host, pk, sk
|
|
|
|
|
|
def _call(method: str, path: str, body: dict | None = None, timeout: float = 20.0):
|
|
host, pk, sk = _conf()
|
|
auth = base64.b64encode(f"{pk}:{sk}".encode()).decode("ascii")
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(
|
|
host + path,
|
|
data=data,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Basic {auth}",
|
|
"User-Agent": "pragent-pilot/1.0",
|
|
},
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
raw = resp.read()
|
|
return resp.status, (json.loads(raw) if raw else None)
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()[:400].decode("utf-8", "replace")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Score configs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def ensure_score_configs() -> dict:
|
|
status, existing = _call("GET", "/api/public/score-configs?limit=100")
|
|
have = set()
|
|
if status == 200 and isinstance(existing, dict):
|
|
have = {c.get("name") for c in existing.get("data", [])}
|
|
|
|
created, skipped, failed = [], [], []
|
|
for cfg in list(eval_scores.SCORE_CONFIGS) + list(feedback_scores.SCORE_CONFIGS):
|
|
if cfg["name"] in have:
|
|
skipped.append(cfg["name"])
|
|
continue
|
|
st, resp = _call("POST", "/api/public/score-configs", cfg)
|
|
if st in (200, 201):
|
|
created.append(cfg["name"])
|
|
else:
|
|
failed.append({"name": cfg["name"], "status": st, "error": resp})
|
|
return {"created": created, "already_present": skipped, "failed": failed}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Dataset from recorded reviews
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def item_id(repo: str, pr) -> str:
|
|
"""A dataset-item id that survives being put in a URL path.
|
|
|
|
The obvious `{repo}#{pr}` is unusable: the UI routes items as
|
|
`/datasets/{id}/items/{item_id}`, so the `/` in `owner/repo` splits into
|
|
extra path segments and everything after the `#` is a fragment the browser
|
|
never sends. The item is created fine and then 404s when opened.
|
|
|
|
Session ids elsewhere keep the `{repo}#{pr}` form — those are never path
|
|
segments, and `feedback_scores` depends on that shape.
|
|
"""
|
|
return f"{repo.replace('/', '__')}__pr{pr}"
|
|
|
|
|
|
def _item_metadata(*, repo, pr, head_sha, reviews_run, last_seen, findings) -> dict:
|
|
"""Filterable facets for one dataset item.
|
|
|
|
Kept flat and primitive: the filter bar matches a metadata key against a
|
|
literal, so a nested object or a list is not reachable from the UI.
|
|
"""
|
|
owner, _, repo_name = str(repo).partition("/")
|
|
sevs = [str(f["severity"] or "").lower() for f in findings]
|
|
ranked = [s for s in sevs if s in eval_scores.SEVERITY_RANK]
|
|
return {
|
|
"repo": repo,
|
|
"owner": owner or repo,
|
|
"repo_name": repo_name or repo,
|
|
"pr": int(pr),
|
|
"head_sha": head_sha,
|
|
"reviews_run": reviews_run,
|
|
"last_reviewed_at": last_seen,
|
|
"last_reviewed_iso": datetime.fromtimestamp(last_seen, timezone.utc).isoformat(),
|
|
"finding_count": len(findings),
|
|
"has_findings": bool(findings),
|
|
# "none" rather than omitting the key: a filter for silent reviews needs
|
|
# something to match, and an absent key matches nothing.
|
|
"max_severity": (
|
|
max(ranked, key=lambda s: eval_scores.SEVERITY_RANK[s]) if ranked else "none"
|
|
),
|
|
# Flags that this row is the reviewer's own past output, not a human
|
|
# judgement. Filter on it before anyone treats the dataset as truth.
|
|
"labelled_by_human": False,
|
|
}
|
|
|
|
|
|
def read_review_items(db_path: str) -> list[dict]:
|
|
"""One dataset item per (repo, pr) the reviewer has run on.
|
|
|
|
Keyed on the PR rather than on each individual review row: the same PR is
|
|
re-reviewed on every push, and 113 rows over 26 PRs would make a benchmark
|
|
that is 4x redundant and weighted towards whichever PR churned most.
|
|
"""
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
prs = conn.execute(
|
|
"""
|
|
SELECT repo, pr, MAX(posted_at) AS last_seen, COUNT(*) AS reviews,
|
|
MAX(head_sha) AS head_sha
|
|
FROM review GROUP BY repo, pr ORDER BY repo, pr
|
|
"""
|
|
).fetchall()
|
|
items = []
|
|
for row in prs:
|
|
findings = conn.execute(
|
|
"""
|
|
SELECT path, line, severity, problem, fix
|
|
FROM inline_finding WHERE repo = ? AND pr = ?
|
|
ORDER BY path, line
|
|
""",
|
|
(row["repo"], row["pr"]),
|
|
).fetchall()
|
|
items.append(
|
|
{
|
|
"id": item_id(row["repo"], row["pr"]),
|
|
"input": {
|
|
"repo": row["repo"],
|
|
"pr": int(row["pr"]),
|
|
"head_sha": row["head_sha"],
|
|
},
|
|
"expectedOutput": {
|
|
"findings": [dict(f) for f in findings],
|
|
"finding_count": len(findings),
|
|
},
|
|
# The UI's filter bar reads metadata and nothing else, so
|
|
# anything worth slicing on is a top-level key here even
|
|
# where it duplicates `input`. `owner` and `repo_name` are
|
|
# split out because a filter on the joined `repo` can only
|
|
# match one repo at a time, never a whole org.
|
|
"metadata": _item_metadata(
|
|
repo=row["repo"],
|
|
pr=row["pr"],
|
|
head_sha=row["head_sha"],
|
|
reviews_run=int(row["reviews"]),
|
|
last_seen=int(row["last_seen"]),
|
|
findings=findings,
|
|
),
|
|
}
|
|
)
|
|
return items
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def ensure_dataset(items: list[dict], name: str = DATASET_NAME) -> dict:
|
|
st, _ = _call(
|
|
"POST",
|
|
"/api/public/datasets",
|
|
{
|
|
"name": name,
|
|
"description": (
|
|
"PRs the pragent pilot has reviewed, seeded from feedback.db. "
|
|
"expectedOutput is the reviewer's own prior output — a regression "
|
|
"baseline, not human-verified ground truth."
|
|
),
|
|
"metadata": {"source": "feedback.db", "seeded_by": "eval_bootstrap.py"},
|
|
},
|
|
)
|
|
# A duplicate name is fine: the dataset already exists from an earlier run.
|
|
dataset_ok = st in (200, 201, 409)
|
|
|
|
created, failed = 0, []
|
|
for item in items:
|
|
body = {
|
|
"datasetName": name,
|
|
"id": item["id"], # idempotent: same PR updates rather than duplicates
|
|
"input": item["input"],
|
|
"expectedOutput": item["expectedOutput"],
|
|
"metadata": item["metadata"],
|
|
}
|
|
ist, resp = _call("POST", "/api/public/dataset-items", body)
|
|
if ist in (200, 201):
|
|
created += 1
|
|
else:
|
|
failed.append({"item": item["id"], "status": ist, "error": resp})
|
|
return {"dataset": name, "dataset_created": dataset_ok, "items_upserted": created, "failed": failed}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Backfill scores onto traces that predate the scorers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _synth_findings(severities: dict) -> list[dict]:
|
|
"""Rebuild a findings list from a trace's severity histogram.
|
|
|
|
Only severity matters to the scorers, and that is all the histogram kept.
|
|
Reconstructing placeholders is honest here because every scorer being
|
|
backfilled reads nothing else off a finding.
|
|
"""
|
|
out = []
|
|
for sev, count in (severities or {}).items():
|
|
out.extend({"severity": sev} for _ in range(int(count)))
|
|
return out
|
|
|
|
|
|
def backfill_traces(limit_pages: int = 20) -> dict:
|
|
import eval_scores as es
|
|
|
|
scored, skipped, events = 0, 0, []
|
|
page = 1
|
|
while page <= limit_pages:
|
|
st, resp = _call("GET", f"/api/public/traces?limit=50&page={page}&name=pr-review")
|
|
if st != 200 or not isinstance(resp, dict):
|
|
break
|
|
rows = resp.get("data") or []
|
|
if not rows:
|
|
break
|
|
for tr in rows:
|
|
meta = tr.get("metadata") or {}
|
|
severities = meta.get("severities") or {}
|
|
count = meta.get("findings")
|
|
if count is None:
|
|
skipped += 1
|
|
continue
|
|
findings = _synth_findings(severities)
|
|
# The histogram is authoritative when present; a trace that recorded
|
|
# a count but no histogram still scores its rate.
|
|
if not findings and count:
|
|
findings = [{"severity": "medium"} for _ in range(int(count))]
|
|
batch = es.build_scores(
|
|
trace_id=tr["id"],
|
|
findings=findings,
|
|
environment=tr.get("environment") or "default",
|
|
cost_usd=(tr.get("totalCost") or meta.get("provider_cost_usd")),
|
|
timestamp=tr.get("timestamp"),
|
|
comment="backfilled from trace metadata",
|
|
)
|
|
events.extend(batch)
|
|
scored += 1
|
|
page += 1
|
|
|
|
posted = False
|
|
status = None
|
|
if events:
|
|
import langfuse_trace
|
|
|
|
host, pk, sk = _conf()
|
|
# Chunked: one 2000-event POST is refused, and a partial backfill that
|
|
# reports success is worse than a slow one.
|
|
for i in range(0, len(events), 200):
|
|
status = langfuse_trace._post(host, pk, sk, events[i:i + 200], 30.0)
|
|
posted = status in (200, 201, 207)
|
|
if not posted:
|
|
break
|
|
return {"traces_scored": scored, "traces_skipped": skipped, "scores": len(events),
|
|
"posted": posted, "http_status": status}
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="Bootstrap Langfuse evaluation for the pragent pilot")
|
|
ap.add_argument("--db", default=os.environ.get("PRAGENT_FEEDBACK_DB", "/data/feedback.db"))
|
|
ap.add_argument("--skip-dataset", action="store_true")
|
|
ap.add_argument("--skip-configs", action="store_true")
|
|
ap.add_argument("--backfill-traces", action="store_true",
|
|
help="score traces written before the scorers existed")
|
|
args = ap.parse_args()
|
|
|
|
out: dict = {}
|
|
if not args.skip_configs:
|
|
out["score_configs"] = ensure_score_configs()
|
|
if not args.skip_dataset:
|
|
items = read_review_items(args.db)
|
|
out["dataset"] = ensure_dataset(items)
|
|
out["dataset"]["items_read"] = len(items)
|
|
if args.backfill_traces:
|
|
out["trace_backfill"] = backfill_traces()
|
|
print(json.dumps(out, indent=2))
|
|
|
|
failed = (out.get("score_configs", {}).get("failed") or []) + (
|
|
out.get("dataset", {}).get("failed") or []
|
|
)
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|