#!/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 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 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": f'{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), }, "metadata": { "reviews_run": int(row["reviews"]), "last_reviewed_at": int(row["last_seen"]), # Flags that this row is the reviewer's own past output, # not a human judgement. Filter on it before anyone # treats the dataset as ground truth. "labelled_by_human": False, }, } ) 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())