Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Run offline golden evals for the Customer Agent domain layer.""" | |
| from __future__ import annotations | |
| import argparse | |
| import importlib.util | |
| import json | |
| import sys | |
| from pathlib import Path | |
| from types import ModuleType | |
| def load_customer_agent() -> ModuleType: | |
| path = Path(__file__).resolve().parents[2] / "scripts" / "customer_agent.py" | |
| spec = importlib.util.spec_from_file_location("customer_agent_eval", path) | |
| if spec is None or spec.loader is None: | |
| raise RuntimeError(f"cannot load {path}") | |
| module = importlib.util.module_from_spec(spec) | |
| sys.modules[spec.name] = module | |
| spec.loader.exec_module(module) | |
| return module | |
| def fake_agent(ca: ModuleType, data: dict) -> object: | |
| return ca.CustomerAgentRecord( | |
| id="eval", | |
| customer_id="cust", | |
| name=str(data.get("name") or "Shop"), | |
| greeting="", | |
| voice_instructions="", | |
| escalate_email=str(data.get("escalate_email") or ""), | |
| escalate_url="", | |
| escalate_webhook_url=str(data.get("escalate_webhook_url") or ""), | |
| allowed_origins=list(data.get("allowed_origins") or []), | |
| site_key_prefix="pk_live_eval", | |
| status="active", | |
| created_at="", | |
| updated_at="", | |
| ) | |
| def run_case(ca: ModuleType, knowledge_chunks: list[dict], case: dict) -> tuple[bool, str]: | |
| case_type = case.get("type") | |
| case_id = case.get("id", "?") | |
| if case_type == "retrieval": | |
| hits = ca.retrieve_chunks(knowledge_chunks, str(case["question"])) | |
| hit = ca.knowledge_is_hit(hits) | |
| if bool(case.get("expect_knowledge_hit")) != hit: | |
| return False, f"{case_id}: knowledge_hit={hit} expected {case.get('expect_knowledge_hit')}" | |
| needle = str(case.get("expect_title_contains") or "").casefold() | |
| if needle: | |
| title = str(hits[0].get("title") or "").casefold() if hits else "" | |
| if needle not in title and needle not in str(hits[0].get("text") or "").casefold(): | |
| return False, f"{case_id}: top hit missing '{needle}' (got title={title!r})" | |
| return True, case_id | |
| if case_type == "guardrail": | |
| q = str(case["question"]) | |
| hard = ca.looks_hard_refuse(q) | |
| sensitive = ca.looks_sensitive(q) | |
| if "expect_hard_refuse" in case and bool(case["expect_hard_refuse"]) != hard: | |
| return False, f"{case_id}: hard_refuse={hard}" | |
| if "expect_sensitive" in case and bool(case["expect_sensitive"]) != sensitive: | |
| return False, f"{case_id}: sensitive={sensitive}" | |
| answer = ca.hard_refuse_answer("help@example.com") if hard else "I don't know that from our FAQ." | |
| escalate = ca.should_recommend_escalation(q, answer, knowledge_hit=False) | |
| if "expect_escalate" in case and bool(case["expect_escalate"]) != escalate: | |
| return False, f"{case_id}: escalate={escalate}" | |
| return True, case_id | |
| if case_type == "origin": | |
| allowed = list(case.get("allowed") or []) | |
| origin = case.get("request_origin") | |
| ok = ca.origin_allowed(allowed, origin) | |
| if bool(case.get("expect_allowed")) != ok: | |
| return False, f"{case_id}: allowed={ok}" | |
| return True, case_id | |
| if case_type == "readiness": | |
| agent_data = dict(case.get("agent") or {}) | |
| knowledge_count = int(agent_data.pop("knowledge_count", 0)) | |
| agent = fake_agent(ca, agent_data) | |
| report = ca.readiness_checklist(agent, knowledge_count=knowledge_count) | |
| ready = bool(report.get("ready_for_production")) | |
| if bool(case.get("expect_ready_for_production")) != ready: | |
| return False, f"{case_id}: ready_for_production={ready}" | |
| return True, case_id | |
| return False, f"{case_id}: unknown type {case_type!r}" | |
| def main(argv: list[str] | None = None) -> int: | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument( | |
| "--cases", | |
| default=str(Path(__file__).resolve().parents[1] / "evals" / "golden_cases.json"), | |
| ) | |
| args = parser.parse_args(argv) | |
| cases_path = Path(args.cases) | |
| payload = json.loads(cases_path.read_text(encoding="utf-8")) | |
| ca = load_customer_agent() | |
| knowledge = payload.get("knowledge") or [] | |
| chunks = [{"title": item["title"], "text": item["text"]} for item in knowledge] | |
| passed = 0 | |
| failed: list[str] = [] | |
| for case in payload.get("cases") or []: | |
| ok, detail = run_case(ca, chunks, case) | |
| if ok: | |
| passed += 1 | |
| print(f"PASS {detail}") | |
| else: | |
| failed.append(detail) | |
| print(f"FAIL {detail}") | |
| total = passed + len(failed) | |
| rate = (passed / total) if total else 0.0 | |
| print(f"\n{passed}/{total} passed ({rate:.0%})") | |
| if failed: | |
| print("Failures:") | |
| for item in failed: | |
| print(f" - {item}") | |
| return 1 | |
| if rate < 1.0: | |
| return 1 | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |