File size: 4,304 Bytes
0d1866c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b207cb0
 
 
0d1866c
 
 
 
 
 
 
 
b207cb0
 
 
0d1866c
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
from pathlib import Path
from statistics import median

from config import CARDS_PATH, DEFAULT_OLLAMA_URL, DEFAULT_TIMEOUT_SECONDS, RESULTS_DIR, load_cards, utc_timestamp
from grader import grade_card
from providers import create_provider
from template import build_messages, build_prompt_text


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run local-intelligence evals against a provider.")
    parser.add_argument("--provider", choices=["replay", "apple", "mlx", "mlx_vlm", "ollama", "hf"], required=True)
    parser.add_argument("--cards", default=str(CARDS_PATH))
    parser.add_argument("--card")
    parser.add_argument("--tier")
    parser.add_argument("--limit", type=int)
    parser.add_argument("--model")
    parser.add_argument("--adapter")
    parser.add_argument("--apple-command")
    parser.add_argument("--ollama-url", default=DEFAULT_OLLAMA_URL)
    parser.add_argument("--hf-token")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SECONDS)
    parser.add_argument("--max-tokens", type=int, default=400)
    parser.add_argument("--replay-file")
    parser.add_argument("--output")
    parser.add_argument("--verbose", action="store_true")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    cards = load_cards(Path(args.cards))
    if args.card:
        cards = [card for card in cards if card.id == args.card]
    if args.tier:
        cards = [card for card in cards if card.tier == args.tier]
    if args.limit:
        cards = cards[: args.limit]
    if not cards:
        raise SystemExit("No cards selected.")

    provider = create_provider(args)
    rows = []

    for index, card in enumerate(cards, start=1):
        payload = card.payload
        messages = build_messages(payload, card.test_input)
        prompt_text = build_prompt_text(payload, card.test_input)
        response = provider.generate(payload, messages, prompt_text)
        grade = grade_card(payload, response.raw_text)

        row = {
            "card_id": card.id,
            "title": card.title,
            "tier": card.tier,
            "provider": args.provider,
            "model": args.model,
            "latency_ms": round(response.latency_ms, 2),
            "raw_text": response.raw_text,
            "metadata": response.metadata,
            "grade": grade,
        }
        rows.append(row)

        if args.verbose:
            status = "PASS" if grade["passed"] else "FAIL"
            print(f"[{status}] {index:02d}. {card.id} ({card.tier})")
            if not grade["passed"]:
                for assertion in grade["assertions"]:
                    if not assertion["passed"]:
                        print(f"  - {assertion['assertion']}: {assertion['details']}")

    passed = sum(1 for row in rows if row["grade"]["passed"])
    average_score = sum(row["grade"]["score"] for row in rows) / len(rows)
    latencies = [row["latency_ms"] for row in rows]
    exact_parse_count = sum(1 for row in rows if row["grade"].get("parse_mode") == "exact")
    normalized_parse_count = sum(1 for row in rows if row["grade"].get("parse_mode") == "normalized")
    parse_failure_count = sum(1 for row in rows if row["grade"].get("parse_error"))
    summary = {
        "provider": args.provider,
        "model": args.model,
        "cards": len(rows),
        "passed": passed,
        "pass_rate": round(passed / len(rows), 4),
        "average_score": round(average_score, 4),
        "median_latency_ms": round(median(latencies), 2),
        "exact_parse_rate": round(exact_parse_count / len(rows), 4),
        "normalized_parse_rate": round(normalized_parse_count / len(rows), 4),
        "parse_failure_rate": round(parse_failure_count / len(rows), 4),
    }

    print(json.dumps(summary, indent=2))

    output_path = Path(args.output) if args.output else RESULTS_DIR / f"{args.provider}-{utc_timestamp()}.json"
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps({"summary": summary, "rows": rows}, indent=2))
    print(f"Wrote results to {output_path}")
    return 0 if passed == len(rows) else 1


if __name__ == "__main__":
    raise SystemExit(main())