131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from run_selector_demo import load_selector_rules, model_hard_exclusion_hits, recommend
|
|
|
|
|
|
def read_json(path):
|
|
return json.loads(Path(path).read_text(encoding="utf-8"))
|
|
|
|
|
|
def selected_model_ids(result):
|
|
return [item["model_id"] for item in result.get("selected_models", [])]
|
|
|
|
|
|
def has_ia_gate(input_text, selector_rules):
|
|
global_rules = selector_rules.get("global_rules", {})
|
|
heavy_signals = global_rules.get("ia_heavy_signals", [])
|
|
return any(signal in input_text for signal in heavy_signals) or "QPI 已判断" in input_text
|
|
|
|
|
|
def evaluate_input(root, selector_rules, item):
|
|
result = recommend(root, item["input"])
|
|
selected_ids = selected_model_ids(result)
|
|
expected = item.get("expected_selector_behavior")
|
|
errors = []
|
|
warnings = []
|
|
|
|
model_rules = {
|
|
model_rule["model_id"]: model_rule
|
|
for model_rule in selector_rules.get("models", [])
|
|
}
|
|
|
|
if expected == "no_call":
|
|
if selected_ids:
|
|
errors.append(f"expected no selected models, got {selected_ids}")
|
|
if result.get("no_call") is not True:
|
|
errors.append("expected no_call=true")
|
|
|
|
if expected == "no_call_or_low_priority":
|
|
if "qpi" in selected_ids:
|
|
errors.append("expected no QPI selection")
|
|
if "intellectual_archaeology" in selected_ids:
|
|
errors.append("expected no IA selection")
|
|
|
|
if expected in {"select_qpi", "select_qpi_low_confidence", "select_qpi_reject_ia"}:
|
|
if "qpi" not in selected_ids:
|
|
errors.append("expected QPI selection")
|
|
|
|
if expected == "select_qpi_reject_ia" and "intellectual_archaeology" in selected_ids:
|
|
errors.append("expected IA rejection")
|
|
|
|
if expected == "select_intellectual_archaeology":
|
|
if "intellectual_archaeology" not in selected_ids:
|
|
errors.append("expected IA selection")
|
|
if not has_ia_gate(item["input"], selector_rules):
|
|
errors.append("IA selected expectation lacks heavy-depth or QPI-completed gate")
|
|
ia_hard_exclusion_hits = model_hard_exclusion_hits(
|
|
item["input"],
|
|
model_rules.get("intellectual_archaeology", {}),
|
|
)
|
|
if ia_hard_exclusion_hits:
|
|
errors.append("IA selection expected despite IA hard exclusion signal")
|
|
|
|
if expected == "select_qpi" and set(selected_ids) - {"qpi"}:
|
|
warnings.append("unexpected extra selected models: " + ", ".join(sorted(set(selected_ids) - {"qpi"})))
|
|
|
|
return {
|
|
"case_id": item["case_id"],
|
|
"expected_selector_behavior": expected,
|
|
"selected_models": selected_ids,
|
|
"no_call": result.get("no_call"),
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def write_report(root, results):
|
|
report_path = root / "reports" / "selector_calibration_smoke_report.md"
|
|
failures = [result for result in results if result["errors"]]
|
|
warnings = [result for result in results if result["warnings"]]
|
|
lines = [
|
|
"# Selector Calibration Smoke Report",
|
|
"",
|
|
f"Status: `{'PASS' if not failures else 'FAIL'}`",
|
|
"",
|
|
"Command: `python scripts/run_selector_calibration_smoke.py`",
|
|
"",
|
|
f"Calibration inputs checked: {len(results)}",
|
|
f"Failures: {len(failures)}",
|
|
f"Warnings: {len(warnings)}",
|
|
"",
|
|
"## Cases",
|
|
"",
|
|
]
|
|
for result in results:
|
|
status = "PASS" if not result["errors"] else "FAIL"
|
|
lines.append(
|
|
f"- `{result['case_id']}`: {status}; expected={result['expected_selector_behavior']}; "
|
|
f"selected={result['selected_models']}; no_call={result['no_call']}"
|
|
)
|
|
for error in result["errors"]:
|
|
lines.append(f" - {error}")
|
|
for warning in result["warnings"]:
|
|
lines.append(f" - WARNING: {warning}")
|
|
report_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return report_path
|
|
|
|
|
|
def main():
|
|
root = Path(__file__).resolve().parents[1]
|
|
selector_rules = load_selector_rules(root)
|
|
calibration = read_json(root / "selector" / "selector_calibration_inputs.json")
|
|
results = [
|
|
evaluate_input(root, selector_rules, item)
|
|
for item in calibration.get("inputs", [])
|
|
]
|
|
report_path = write_report(root, results)
|
|
print(f"selector calibration smoke report written to {report_path}")
|
|
failures = [result for result in results if result["errors"]]
|
|
if failures:
|
|
for result in failures:
|
|
print(f"ERROR: {result['case_id']}: {'; '.join(result['errors'])}")
|
|
return 1
|
|
print("selector calibration smoke passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|