113 lines
7.9 KiB
Python
113 lines
7.9 KiB
Python
"""Isolated component regression gates. Evaluation samples never enter model context."""
|
|
from copy import deepcopy
|
|
from services.assistant_store import get, put, listing, uid, digest, now, event
|
|
from services import component_registry as registry
|
|
from services.skill_executor import execute_skill, execute_rules
|
|
|
|
# Independent, synthetic holdouts: not DEMO-A/B/C and not loaded by assistant_worker.
|
|
HOLDOUTS = [
|
|
{"name": "holdout-complete", "values": {"facility.amount": "200", "lease.total_value": "300", "financial.total_assets": "1000", "financial.total_liabilities": "400", "due_diligence.liabilities": "400", "financial.operating_cashflow": "50", "lease.duplicate_count": "0", "credit.overdue_count": "0", "credit.report_date": "2026-08-01"}, "coverage": "1.5000", "consistency": "pass", "cashflow": "pass", "duplicate": "pass"},
|
|
{"name": "holdout-incomplete", "values": {"facility.amount": "400", "lease.total_value": "360", "financial.total_assets": "1000", "financial.total_liabilities": "800", "due_diligence.liabilities": "600", "financial.operating_cashflow": None, "lease.duplicate_count": "2", "credit.overdue_count": "1", "credit.report_date": "2026-08-01"}, "coverage": "0.9000", "consistency": "issue", "cashflow": "unknown", "duplicate": "issue"},
|
|
]
|
|
|
|
|
|
def refs(values):
|
|
return {f: [{"id": "holdout." + f, "field": f, "value": v, "currency": "CNY", "unit": "万元", "period": "2025", "scope": "合并"}] for f, v in values.items()}
|
|
|
|
|
|
def check_candidate(body, db):
|
|
logical = dict(body, id=body.get("component_id", body["id"]))
|
|
registry.validate(logical)
|
|
checks = []
|
|
if body["kind"] == "skill":
|
|
if body["engine"] == "model":
|
|
return [{"case": "model-semantic-evaluation", "passed": False, "detail": "模型方法变更需要已标注语义评测及真实内网模型联调，不能用确定性单元测试替代"}]
|
|
aliases = {"corporate.consistency": "consistency", "corporate.cashflow": "cashflow", "corporate.assets": "duplicate"}
|
|
for sample in HOLDOUTS:
|
|
output, findings = execute_skill(logical, sample["values"], refs(sample["values"]))
|
|
if logical["id"] == "corporate.coverage":
|
|
passed = output.get("metrics.coverage") == sample["coverage"]
|
|
elif logical["id"] in aliases:
|
|
passed = len(findings) == 1 and findings[0]["result"] == sample[aliases[logical["id"]]]
|
|
else:
|
|
continue
|
|
checks.append({"case": sample["name"], "passed": passed, "outputs": output, "results": [f["result"] for f in findings]})
|
|
for sample in body.get("test_cases", []):
|
|
output, findings = execute_skill(logical, sample["values"], refs(sample["values"]))
|
|
passed = output == sample.get("expected_outputs", {}) and {f["code"]: f["result"] for f in findings} == sample.get("expected_checks", {})
|
|
checks.append({"case": sample["name"], "passed": passed, "outputs": output, "results": findings})
|
|
if not checks:
|
|
checks.append({"case": "golden-samples-required", "passed": False, "detail": "自定义 Skill 需提交独立测试样本 test_cases 和期望输出"})
|
|
elif body["kind"] == "rule_set":
|
|
for index, condition in enumerate(body.get("conditions", [])):
|
|
values = {condition["field"]: None, **condition.get("when", {})}
|
|
result = execute_rules(dict(logical, conditions=[condition]), values, refs(values))[0]
|
|
checks.append({"case": "missing-input-" + str(index), "passed": result["result"] == "unknown", "result": result["result"]})
|
|
for sample in body.get("test_cases", []):
|
|
result = execute_rules(logical, sample["values"], refs(sample["values"]))
|
|
checks.append({"case": sample["name"], "passed": {f["code"]: f["result"] for f in result} == sample["expected_checks"], "results": result})
|
|
if logical["id"] == "demo.coverage":
|
|
for sample in HOLDOUTS:
|
|
values = {"metrics.coverage": sample["coverage"]}
|
|
result = execute_rules(logical, values, refs(values))
|
|
checks.append({"case": sample["name"], "passed": len(result) == 1 and result[0]["result"] == ("pass" if sample["coverage"] == "1.5000" else "issue"), "results": result})
|
|
elif not body.get("test_cases"):
|
|
checks.append({"case": "rule-regression-expectations", "passed": False, "detail": "新规则需提供独立期望样本"})
|
|
elif body["kind"] == "plan":
|
|
skills = [registry.resolve(db, s) for s in logical["skills"]]
|
|
registry.ordered(skills)
|
|
for key in logical.get("rules", []) + [logical["template"]]:
|
|
registry.resolve(db, key)
|
|
checks.append({"case": "dependency-graph", "passed": True, "detail": "发布依赖存在且无循环/输出冲突"})
|
|
for sample in HOLDOUTS:
|
|
values, all_checks = deepcopy(sample["values"]), []
|
|
sources = refs(values)
|
|
for skill in registry.ordered(skills):
|
|
if skill["engine"] == "model":
|
|
continue
|
|
output, findings = execute_skill(skill, values, sources)
|
|
values.update(output)
|
|
all_checks.extend(findings)
|
|
checks.append({"case": sample["name"], "passed": bool(all_checks) and values.get("metrics.coverage", sample["coverage"]) == sample["coverage"], "outputs": values})
|
|
elif body["kind"] == "template":
|
|
from services.assistant_documents import section_content
|
|
snapshot = {"customer": "留出样本企业", "values": HOLDOUTS[0]["values"], "period": "2025", "scope": "合并"}
|
|
run = {"steps": [], "findings": [], "summary": {"coverage": "1.5000"}}
|
|
for section in body["sections"]:
|
|
content = section_content(section["id"], snapshot, run)
|
|
checks.append({"case": "section-" + section["id"], "passed": bool(content) and len(content) < 20000})
|
|
elif body["kind"] == "data_card":
|
|
from api.risk_agent_connectors import validate_query
|
|
source = registry.resolve(db, body["source_id"])
|
|
dialect = {"PostgreSQL": "postgres", "MySQL": "mysql"}.get(source["type"], "sqlite")
|
|
validate_query(body["sql"], source.get("allowed_tables", body.get("tables", [])), dialect)
|
|
checks.append({"case": "read-only-sql", "passed": True})
|
|
elif body["kind"] == "data_source":
|
|
checks.append({"case": "source-contract", "passed": body["type"] == "demo", "detail": "非演示连接需先通过旧数据接入页的只读连接测试，尚未自动批准"})
|
|
else:
|
|
checks.append({"case": "unsupported-publish-kind", "passed": False, "detail": "经验请通过项目反馈审核流程发布"})
|
|
return checks
|
|
|
|
|
|
def evaluate(db, draft_id, actor):
|
|
body = get(db, "component_draft", draft_id)
|
|
candidate_hash = digest({k: v for k, v in body.items() if k not in ("revision", "evaluation", "evaluation_id")})
|
|
try:
|
|
probes = check_candidate(body, db)
|
|
except (ValueError, KeyError, TypeError) as exc:
|
|
probes = [{"case": "validation", "passed": False, "detail": str(exc)}]
|
|
baseline = None
|
|
try:
|
|
old = registry.resolve(db, body["component_id"])
|
|
baseline = {"version": old["version"], "checks": check_candidate(old, db)}
|
|
except (KeyError, ValueError):
|
|
pass
|
|
evaluation = {"id": uid("component-eval"), "draft_id": draft_id, "candidate_hash": candidate_hash,
|
|
"suite_version": "component-holdout@1.0", "created_at": now(), "checks": probes,
|
|
"baseline": baseline, "passed": bool(probes) and all(p["passed"] for p in probes), "scope": "offline-regression-not-live-model-quality"}
|
|
put(db, "component_evaluation", evaluation, immutable=True)
|
|
body.update(evaluation_id=evaluation["id"], evaluation={k: v for k, v in evaluation.items() if k not in ("candidate_hash",)})
|
|
put(db, "component_draft", body)
|
|
event(db, "", "component.evaluated", actor, {"draft_id": draft_id, "passed": evaluation["passed"]})
|
|
return evaluation
|