Spaces:
Sleeping
Sleeping
File size: 2,675 Bytes
9bf4a3d | 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 | """
Same-input consistency test (spec #7).
Runs each configured provider on the same JD 3 times and reports score variance.
High-variance providers should get a lower production priority in
config.LLM_GENERATION.provider_order.
Usage:
PYTHONPATH=. python scripts/consistency_test.py
PYTHONPATH=. python scripts/consistency_test.py --jd airtel_pm --runs 3
"""
import os
import sys
import io
import json
import argparse
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
if hasattr(sys.stdout, "buffer"):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
from src.provider_eval import load_base_resume, load_jd_fixtures, consistency_runs
from src.providers import build_provider_chain
from src.llm_client import LLMClient
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--jd", type=str, default=None, help="JD fixture slug (default: first)")
ap.add_argument("--runs", type=int, default=3)
args = ap.parse_args()
base = load_base_resume()
if base is None:
print("ERROR: no base resume.")
sys.exit(1)
jobs = load_jd_fixtures()
if not jobs:
print("ERROR: no JD fixtures.")
sys.exit(1)
job = next((j for j in jobs if j["_jd_slug"] == args.jd), jobs[0]) if args.jd else jobs[0]
llm = LLMClient.__new__(LLMClient)
providers = build_provider_chain(llm)
real = [p for p in providers if getattr(p, "cfg", {}).get("api_key")]
if not real:
print("*** NO PROVIDER KEYS — StubProvider is deterministic, so variance")
print("*** will be 0 by construction. Run with Kimi/NVIDIA keys for a real test. ***\n")
print(f"JD under test: {job['_jd_slug']} | runs each: {args.runs}\n")
out_root = os.path.join("data", "output", "eval", "consistency")
results = []
print(f"{'Provider':<16} {'Run1':>5} {'Run2':>5} {'Run3':>5} {'Var':>6} {'Spread':>7} Stable?")
for p in providers:
res = consistency_runs(p, base, job, os.path.join(out_root, p.name),
n=args.runs, llm=llm)
results.append(res)
sc = res["scores"] + [""] * (3 - len(res["scores"]))
print(f"{p.name[:16]:<16} {str(sc[0]):>5} {str(sc[1]):>5} {str(sc[2]):>5} "
f"{res['variance']:>6} {res['spread']:>7} {'YES' if res['stable'] else 'NO'}")
os.makedirs(out_root, exist_ok=True)
with open(os.path.join(out_root, "consistency.json"), "w", encoding="utf-8") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"\nSaved -> {os.path.join(out_root, 'consistency.json')}")
if __name__ == "__main__":
main()
|