| |
| """Evaluate a System One compatible endpoint on LocalLLaMA/typed-decisions. |
| |
| The runner intentionally uses only Python's standard library. It sends all five |
| questions in a case together, writes one durable JSONL record per case, and can |
| resume an interrupted run. |
| """ |
|
|
| import argparse |
| import json |
| import math |
| import os |
| import sys |
| import time |
| import urllib.error |
| import urllib.request |
| from collections import defaultdict |
| from pathlib import Path |
|
|
|
|
| EPSILON = 1e-12 |
| WORKFLOWS = ( |
| "agent_trace_observability", |
| "customer_service", |
| "invoice_processing", |
| "security_incidents", |
| ) |
|
|
|
|
| def parse_args(): |
| here = Path(__file__).resolve().parent |
| parser = argparse.ArgumentParser( |
| description="Run typed-decisions against a /v1/systemone endpoint." |
| ) |
| parser.add_argument("--endpoint", default="http://127.0.0.1:8741/v1/systemone") |
| parser.add_argument("--data", type=Path, default=here / "data" / "test.jsonl") |
| parser.add_argument("--output-dir", type=Path, default=here / "runs" / "default") |
| parser.add_argument("--workflow", action="append", choices=WORKFLOWS, |
| help="Only run this workflow; repeat to select more than one.") |
| parser.add_argument("--limit", type=int, default=None, |
| help="Run at most N selected cases (useful for an explicit smoke run).") |
| parser.add_argument("--timeout", type=float, default=120.0) |
| parser.add_argument("--retries", type=int, default=2, |
| help="Retries after the first attempt for transient HTTP/network errors.") |
| parser.add_argument("--retry-delay", type=float, default=2.0) |
| parser.add_argument("--no-resume", action="store_true", |
| help="Refuse to reuse an existing predictions.jsonl.") |
| parser.add_argument("--api-key-env", default="", |
| help="Optional environment variable containing a Bearer token.") |
| return parser.parse_args() |
|
|
|
|
| def read_rows(path, workflows=None, limit=None): |
| selected = [] |
| with path.open(encoding="utf-8") as handle: |
| for line_number, line in enumerate(handle, 1): |
| try: |
| row = json.loads(line) |
| except json.JSONDecodeError as exc: |
| raise ValueError("invalid JSON on {}:{}: {}".format(path, line_number, exc)) |
| if workflows and row.get("workflow") not in workflows: |
| continue |
| selected.append(row) |
| if limit is not None and len(selected) >= limit: |
| break |
| return selected |
|
|
|
|
| def load_completed(path): |
| completed = {} |
| if not path.exists(): |
| return completed |
| with path.open(encoding="utf-8") as handle: |
| for line_number, line in enumerate(handle, 1): |
| try: |
| item = json.loads(line) |
| except json.JSONDecodeError as exc: |
| raise ValueError("invalid JSON on {}:{}: {}".format(path, line_number, exc)) |
| |
| if item.get("ok") and item.get("id"): |
| completed[item["id"]] = item |
| return completed |
|
|
|
|
| def decode_row(row): |
| return tuple(json.loads(row[name]) for name in ("state", "questions", "gold")) |
|
|
|
|
| def post_json(endpoint, body, timeout, token): |
| headers = {"Content-Type": "application/json", "Accept": "application/json"} |
| if token: |
| headers["Authorization"] = "Bearer " + token |
| request = urllib.request.Request( |
| endpoint, |
| data=json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), |
| headers=headers, |
| method="POST", |
| ) |
| started = time.perf_counter() |
| try: |
| with urllib.request.urlopen(request, timeout=timeout) as response: |
| raw = response.read() |
| status = response.status |
| except urllib.error.HTTPError as exc: |
| detail = exc.read().decode("utf-8", "replace") |
| raise RuntimeError("HTTP {}: {}".format(exc.code, detail[:1000])) from exc |
| elapsed_ms = (time.perf_counter() - started) * 1000.0 |
| try: |
| payload = json.loads(raw) |
| except json.JSONDecodeError as exc: |
| raise RuntimeError("endpoint returned non-JSON data: {!r}".format(raw[:200])) from exc |
| if status < 200 or status >= 300: |
| raise RuntimeError("HTTP {}: {}".format(status, payload)) |
| return payload, elapsed_ms |
|
|
|
|
| def probabilities_for(question, answer, labels): |
| typ = question.get("type", "choice") |
| if typ == "noul": |
| if "noul" in answer: |
| p_true = float(answer["noul"]) |
| raw = {"false": 1.0 - p_true, "true": p_true} |
| else: |
| raw = answer.get("probabilities", {}) |
| else: |
| raw = answer.get("probabilities", {}) |
| try: |
| values = [float(raw[label]) for label in labels] |
| except (KeyError, TypeError, ValueError) as exc: |
| raise ValueError("missing or invalid probabilities for labels {}".format(labels)) from exc |
| if any(not math.isfinite(value) or value < 0.0 for value in values): |
| raise ValueError("probabilities must be finite and non-negative") |
| total = sum(values) |
| if total <= 0.0: |
| raise ValueError("probabilities sum to zero") |
| return [value / total for value in values] |
|
|
|
|
| def decision_records(result): |
| response = result["response"] |
| answers = response.get("answers", response) |
| questions = result["questions"] |
| gold = result["gold"] |
| if not isinstance(answers, dict): |
| raise ValueError("response has no answers object") |
| for name, question in questions.items(): |
| if name not in gold or name not in answers: |
| raise ValueError("response is missing question {!r}".format(name)) |
| gold_answer = gold[name] |
| gold_probs_map = gold_answer["probabilities"] |
| labels = list(gold_probs_map) |
| gold_probs = [float(gold_probs_map[label]) for label in labels] |
| pred_probs = probabilities_for(question, answers[name], labels) |
| pred_index = max(range(len(labels)), key=pred_probs.__getitem__) |
| predicted = labels[pred_index] |
| expected = str(gold_answer.get("label", labels[max( |
| range(len(labels)), key=gold_probs.__getitem__ |
| )])) |
| correct = predicted == expected |
| kl = sum(g * math.log(g / max(p, EPSILON)) |
| for g, p in zip(gold_probs, pred_probs) if g > 0.0) |
| brier = sum((p - g) ** 2 for p, g in zip(pred_probs, gold_probs)) |
| yield { |
| "workflow": result["workflow"], |
| "question": name, |
| "type": question.get("type", "choice"), |
| "correct": correct, |
| "confidence": max(pred_probs), |
| "kl": kl, |
| "brier": brier, |
| } |
|
|
|
|
| def expected_calibration_error(items, bins=15): |
| if not items: |
| return None |
| buckets = [[] for _ in range(bins)] |
| for item in items: |
| index = min(bins - 1, int(item["confidence"] * bins)) |
| buckets[index].append(item) |
| total = len(items) |
| return sum( |
| (len(bucket) / total) * abs( |
| sum(item["confidence"] for item in bucket) / len(bucket) |
| - sum(float(item["correct"]) for item in bucket) / len(bucket) |
| ) |
| for bucket in buckets if bucket |
| ) |
|
|
|
|
| def aggregate(items): |
| if not items: |
| return {"decisions": 0, "accuracy": None, "kl": None, "brier": None, "ece_15": None} |
| return { |
| "decisions": len(items), |
| "correct": sum(item["correct"] for item in items), |
| "accuracy": sum(item["correct"] for item in items) / len(items), |
| "kl": sum(item["kl"] for item in items) / len(items), |
| "brier": sum(item["brier"] for item in items) / len(items), |
| "ece_15": expected_calibration_error(items, 15), |
| } |
|
|
|
|
| def percentile(values, fraction): |
| if not values: |
| return None |
| ordered = sorted(values) |
| index = max(0, min(len(ordered) - 1, math.ceil(fraction * len(ordered)) - 1)) |
| return ordered[index] |
|
|
|
|
| def build_report(rows, results): |
| decisions = [] |
| scoring_errors = [] |
| successful = [result for result in results if result.get("ok")] |
| for result in successful: |
| try: |
| decisions.extend(decision_records(result)) |
| except (KeyError, TypeError, ValueError) as exc: |
| scoring_errors.append({"id": result.get("id"), "error": str(exc)}) |
|
|
| groups = { |
| "by_workflow": lambda item: item["workflow"], |
| "by_type": lambda item: item["type"], |
| "by_question": lambda item: item["workflow"] + "/" + item["question"], |
| } |
| report = { |
| "dataset": "LocalLLaMA/typed-decisions", |
| "split": "test", |
| "selected_cases": len(rows), |
| "successful_cases": len(successful), |
| "failed_cases": len(rows) - len(successful), |
| "scored_cases": len(successful) - len(scoring_errors), |
| "overall": aggregate(decisions), |
| "metric_notes": { |
| "accuracy": "argmax prediction against the gold argmax label", |
| "kl": "mean KL(gold || prediction), natural logarithm", |
| "brier": "mean sum over classes of (prediction - soft gold)^2", |
| "ece_15": "15 equal-width bins using top probability and hard argmax correctness", |
| }, |
| "latency_ms": { |
| "p50": percentile([r["latency_ms"] for r in successful], 0.50), |
| "p95": percentile([r["latency_ms"] for r in successful], 0.95), |
| }, |
| "scoring_errors": scoring_errors, |
| } |
| for title, key_fn in groups.items(): |
| grouped = defaultdict(list) |
| for item in decisions: |
| grouped[key_fn(item)].append(item) |
| report[title] = {key: aggregate(value) for key, value in sorted(grouped.items())} |
| return report |
|
|
|
|
| def main(): |
| args = parse_args() |
| if args.limit is not None and args.limit < 1: |
| raise SystemExit("--limit must be at least 1") |
| rows = read_rows(args.data, set(args.workflow or []), args.limit) |
| if not rows: |
| raise SystemExit("no cases selected") |
| args.output_dir.mkdir(parents=True, exist_ok=True) |
| predictions_path = args.output_dir / "predictions.jsonl" |
| if args.no_resume and predictions_path.exists(): |
| raise SystemExit("{} already exists; choose a new --output-dir or omit --no-resume".format( |
| predictions_path |
| )) |
| completed = load_completed(predictions_path) |
| token = os.environ.get(args.api_key_env, "") if args.api_key_env else "" |
| results = dict(completed) |
| pending = [row for row in rows if row["id"] not in completed] |
| print("selected {} cases; {} complete; {} pending".format(len(rows), len(completed), len(pending))) |
|
|
| with predictions_path.open("a", encoding="utf-8", buffering=1) as output: |
| for position, row in enumerate(pending, 1): |
| state, questions, gold = decode_row(row) |
| error = None |
| for attempt in range(args.retries + 1): |
| try: |
| response, elapsed_ms = post_json( |
| args.endpoint, {"state": state, "questions": questions}, args.timeout, token |
| ) |
| item = { |
| "id": row["id"], "workflow": row["workflow"], "ok": True, |
| "latency_ms": elapsed_ms, "questions": questions, "gold": gold, |
| "response": response, |
| } |
| results[row["id"]] = item |
| output.write(json.dumps(item, ensure_ascii=False, separators=(",", ":")) + "\n") |
| print("[{}/{}] {} {:.1f} ms".format(position, len(pending), row["id"], elapsed_ms)) |
| break |
| except (OSError, RuntimeError, TimeoutError) as exc: |
| error = str(exc) |
| if attempt < args.retries: |
| time.sleep(args.retry_delay * (attempt + 1)) |
| else: |
| item = {"id": row["id"], "workflow": row["workflow"], "ok": False, |
| "error": error} |
| output.write(json.dumps(item, ensure_ascii=False, separators=(",", ":")) + "\n") |
| print("[{}/{}] {} ERROR {}".format(position, len(pending), row["id"], error), |
| file=sys.stderr) |
|
|
| ordered_results = [results[row["id"]] for row in rows if row["id"] in results] |
| report = build_report(rows, ordered_results) |
| report_path = args.output_dir / "report.json" |
| report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") |
| overall = report["overall"] |
| print("report: {}".format(report_path)) |
| print("accuracy={:.4f} KL={:.4f} Brier={:.4f} ECE-15={:.4f}".format( |
| overall["accuracy"], overall["kl"], overall["brier"], overall["ece_15"] |
| )) |
| return 0 if report["failed_cases"] == 0 and not report["scoring_errors"] else 2 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
