ml-services/evaluation/consistency.py

"""
Consistency harness for the graph evaluation pipeline.

A scoring system that produces different results on the same input across
runs is not dependable. This harness runs the SAME call through the graph N
times and measures agreement:

  - compliance : per-check agreement rate (fraction of runs matching the
                 modal pass/fail/null verdict)
  - quality    : per-dimension score min / max / range / mean / stdev
  - escalation : risk_level + emotion agreement rate
  - anchoring  : evidence anchor rate per run

Usage:
  python consistency.py --call_id en_CA_Banking_1586889 --runs 5
"""
import os
import json
import time
import argparse
import statistics
from collections import Counter

from graph import build_graph

COMPLIANCE_KEYS = [
    "name_announced", "company_announced", "recording_disclosure",
    "identity_verified", "resolution_provided", "transfer_next_steps",
]
QUALITY_KEYS = [
    "efficiency", "problem_resolution", "clarity",
    "professionalism", "empathy", "customer_satisfaction",
]


def agreement(values):
    """Fraction of values matching the mode. 1.0 = perfect agreement."""
    if not values:
        return None
    most_common = Counter(values).most_common(1)[0][1]
    return most_common / len(values)


def run_sweep(call_id, runs, results_dir):
    graph = build_graph()
    evaluations = []

    for i in range(1, runs + 1):
        print(f"\n--- run {i}/{runs} " + "-" * 50)
        t0 = time.time()
        final = graph.invoke({"call_id": call_id, "results_dir": results_dir})
        dt = time.time() - t0
        ev = final["evaluation"]
        evaluations.append(ev)
        stats = ev.get("_anchor_stats", {})
        print(f"--- run {i} done in {dt:.1f}s "
              f"(anchored {stats.get('anchored')}/{stats.get('total')})")

    return evaluations


def analyze(evaluations):
    report = {"n_runs": len(evaluations), "compliance": {}, "quality": {},
              "escalation": {}, "anchoring": {}}

    # compliance agreement per check
    for key in COMPLIANCE_KEYS:
        verdicts = [str(ev["compliance"][key]["passed"]) for ev in evaluations]
        report["compliance"][key] = {
            "verdicts": verdicts,
            "agreement": agreement(verdicts),
        }

    # quality score variance per dimension
    for key in QUALITY_KEYS:
        scores = [ev["quality"][key]["score"] for ev in evaluations
                  if ev["quality"].get(key)]
        if not scores:
            continue
        report["quality"][key] = {
            "scores": scores,
            "min": min(scores),
            "max": max(scores),
            "range": max(scores) - min(scores),
            "mean": round(statistics.mean(scores), 2),
            "stdev": round(statistics.stdev(scores), 3) if len(scores) > 1 else 0.0,
        }

    # escalation agreement
    risks = [ev["escalation"]["risk_level"] for ev in evaluations]
    emotions = [ev["escalation"]["customer_emotion_text"] for ev in evaluations]
    report["escalation"] = {
        "risk_levels": risks,
        "risk_agreement": agreement(risks),
        "emotions": emotions,
        "emotion_agreement": agreement(emotions),
    }

    # anchor rates
    rates = []
    for ev in evaluations:
        s = ev.get("_anchor_stats", {})
        if s.get("total"):
            rates.append(round(s["anchored"] / s["total"], 3))
    report["anchoring"] = {"rates": rates}

    return report


def print_report(report, call_id):
    n = report["n_runs"]
    print("\n" + "=" * 68)
    print(f"  CONSISTENCY REPORT  {call_id}  ({n} runs)")
    print("=" * 68)

    print("\n  COMPLIANCE (agreement with modal verdict)")
    worst_c = 1.0
    for key, r in report["compliance"].items():
        a = r["agreement"]
        worst_c = min(worst_c, a)
        flag = "" if a == 1.0 else f"  <-- UNSTABLE {r['verdicts']}"
        print(f"    {a * 100:5.1f}%  {key}{flag}")

    print("\n  QUALITY (score spread across runs)")
    worst_range = 0
    for key, r in report["quality"].items():
        worst_range = max(worst_range, r["range"])
        flag = "" if r["range"] <= 1 else "  <-- UNSTABLE"
        print(f"    {key:<24} scores={r['scores']}  "
              f"range={r['range']}  stdev={r['stdev']}{flag}")

    e = report["escalation"]
    print(f"\n  ESCALATION  risk agreement: {e['risk_agreement'] * 100:.1f}% "
          f"{e['risk_levels']}")
    print(f"              emotion agreement: {e['emotion_agreement'] * 100:.1f}% "
          f"{e['emotions']}")

    rates = report["anchoring"]["rates"]
    if rates:
        print(f"\n  ANCHOR RATES per run: {rates}")

    print("\n  VERDICT: ", end="")
    if worst_c == 1.0 and worst_range <= 1 and e["risk_agreement"] == 1.0:
        print("stable (all compliance unanimous, quality range <= 1, "
              "risk unanimous)")
    else:
        print("instability detected -- see UNSTABLE flags above")
    print("=" * 68)


def main():
    ap = argparse.ArgumentParser(
        description="Run the graph N times on one call, measure agreement")
    ap.add_argument("--call_id", required=True)
    ap.add_argument("--runs", type=int, default=5)
    ap.add_argument("--results_dir", default="results_channels")
    args = ap.parse_args()

    evaluations = run_sweep(args.call_id, args.runs, args.results_dir)
    report = analyze(evaluations)
    print_report(report, args.call_id)

    here = os.path.dirname(os.path.abspath(__file__))
    out = os.path.join(here, "results", f"{args.call_id}_consistency.json")
    with open(out, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2)
    print(f"\nSaved -> {out}")


if __name__ == "__main__":
    main()