feat: 新增label快慢分流skill全套配置与校验脚本
This commit is contained in:
zhukai8
@@ -0,0 +1,352 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate fast/slow/ambiguous label records without external dependencies."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
POLICY_VERSION = "XA4.1-2026.6"
|
||||
|
||||
REASON_CODES_BY_LABEL: dict[str, set[str]] = {
|
||||
"快": {
|
||||
"DIRECT_ACTION",
|
||||
"DIRECT_DETERMINISTIC_QUERY",
|
||||
"EXPLICIT_DEVICE_MULTI_ACTION",
|
||||
"DIRECT_OPERATION_FLOW",
|
||||
},
|
||||
"慢": {
|
||||
"INTENT_INFERENCE",
|
||||
"SCENE_GOAL_MAPPING",
|
||||
"MULTI_STEP_PLANNING",
|
||||
"COMPLEX_REASONING",
|
||||
"TRUE_DISAMBIGUATION",
|
||||
"CONTEXT_REQUIRED",
|
||||
"MISSING_REQUIRED_SLOT",
|
||||
"AUTOMATION_SLOW_POLICY",
|
||||
},
|
||||
"模糊": {
|
||||
"OPEN_CHAT_OR_QA",
|
||||
"PRODUCT_KNOWLEDGE",
|
||||
"FAULT_DIAGNOSIS",
|
||||
"DEVICE_OR_ENV_STATE_QUERY",
|
||||
"POLICY_UNCERTAIN",
|
||||
},
|
||||
}
|
||||
|
||||
DEFAULT_ROUTE_BY_LABEL = {
|
||||
"快": "快系统",
|
||||
"慢": "慢系统",
|
||||
"模糊": "慢系统",
|
||||
}
|
||||
|
||||
LEGACY_LABELS = {
|
||||
"FAST_DIRECT",
|
||||
"SLOW_FILTER_RANK",
|
||||
"SLOW_DISAMBIGUATE",
|
||||
"SLOW_GOAL_STATE",
|
||||
"SLOW_WORKFLOW",
|
||||
"SLOW_CONTEXT_DEPENDENT",
|
||||
"SLOW_UNCLEAR",
|
||||
"SLOW_ADVANCED_CAPABILITY",
|
||||
}
|
||||
|
||||
CONFIDENCE_VALUES = {"high", "medium", "low"}
|
||||
|
||||
REQUIRED_FIELDS = {
|
||||
"query",
|
||||
"label",
|
||||
"default_route",
|
||||
"primary_reason_code",
|
||||
"reason_codes",
|
||||
"reason",
|
||||
"evidence",
|
||||
"context_used",
|
||||
"fast_system_pending",
|
||||
"confidence",
|
||||
"review_required",
|
||||
"uncertainties",
|
||||
"policy_version",
|
||||
}
|
||||
|
||||
OPTIONAL_FIELDS = {
|
||||
"id",
|
||||
"counterevidence",
|
||||
"domain_label",
|
||||
"candidate_domain_labels",
|
||||
"structural_signals",
|
||||
"legacy_eval_label",
|
||||
}
|
||||
|
||||
|
||||
def _is_nonempty_string(value: Any) -> bool:
|
||||
return isinstance(value, str) and bool(value.strip())
|
||||
|
||||
|
||||
def _check_string_list(
|
||||
value: Any,
|
||||
*,
|
||||
field: str,
|
||||
errors: list[str],
|
||||
require_nonempty: bool = False,
|
||||
unique: bool = False,
|
||||
) -> None:
|
||||
if not isinstance(value, list):
|
||||
errors.append(f"{field} 必须是数组")
|
||||
return
|
||||
if require_nonempty and not value:
|
||||
errors.append(f"{field} 至少包含一项")
|
||||
if any(not _is_nonempty_string(item) for item in value):
|
||||
errors.append(f"{field} 的每一项都必须是非空字符串")
|
||||
if unique and all(isinstance(item, str) for item in value):
|
||||
if len(value) != len(set(value)):
|
||||
errors.append(f"{field} 不允许重复项")
|
||||
|
||||
|
||||
def validate_record(record: Any, index: int) -> dict[str, Any]:
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
|
||||
if not isinstance(record, dict):
|
||||
return {
|
||||
"index": index,
|
||||
"valid": False,
|
||||
"errors": ["记录必须是 JSON 对象"],
|
||||
"warnings": [],
|
||||
}
|
||||
|
||||
if "__parse_error__" in record:
|
||||
return {
|
||||
"index": index,
|
||||
"valid": False,
|
||||
"errors": [str(record["__parse_error__"])],
|
||||
"warnings": [],
|
||||
}
|
||||
|
||||
missing = sorted(REQUIRED_FIELDS - set(record))
|
||||
if missing:
|
||||
errors.append(f"缺少必填字段: {', '.join(missing)}")
|
||||
|
||||
unknown = sorted(set(record) - REQUIRED_FIELDS - OPTIONAL_FIELDS)
|
||||
if unknown:
|
||||
errors.append(f"存在未知字段: {', '.join(unknown)}")
|
||||
|
||||
query = record.get("query")
|
||||
if not _is_nonempty_string(query):
|
||||
errors.append("query 必须是非空字符串")
|
||||
|
||||
label = record.get("label")
|
||||
if label not in REASON_CODES_BY_LABEL:
|
||||
errors.append("label 必须是 快、慢、模糊 之一")
|
||||
|
||||
default_route = record.get("default_route")
|
||||
expected_route = DEFAULT_ROUTE_BY_LABEL.get(label)
|
||||
if expected_route is not None and default_route != expected_route:
|
||||
errors.append(f"label={label} 时 default_route 必须是 {expected_route}")
|
||||
|
||||
primary_reason_code = record.get("primary_reason_code")
|
||||
reason_codes = record.get("reason_codes")
|
||||
allowed_reasons = REASON_CODES_BY_LABEL.get(label, set())
|
||||
if primary_reason_code not in allowed_reasons:
|
||||
errors.append(
|
||||
f"primary_reason_code={primary_reason_code!r} 与 label={label!r} 不兼容"
|
||||
)
|
||||
|
||||
if not isinstance(reason_codes, list):
|
||||
errors.append("reason_codes 必须是数组")
|
||||
else:
|
||||
if not reason_codes:
|
||||
errors.append("reason_codes 至少包含一项")
|
||||
if all(isinstance(code, str) for code in reason_codes):
|
||||
if len(reason_codes) != len(set(reason_codes)):
|
||||
errors.append("reason_codes 不允许重复")
|
||||
invalid_reason_codes = [
|
||||
code
|
||||
for code in reason_codes
|
||||
if not isinstance(code, str) or code not in allowed_reasons
|
||||
]
|
||||
if invalid_reason_codes:
|
||||
rendered_codes = ", ".join(repr(code) for code in invalid_reason_codes)
|
||||
errors.append(f"reason_codes 包含与 label 不兼容的值: {rendered_codes}")
|
||||
if primary_reason_code not in reason_codes:
|
||||
errors.append("primary_reason_code 必须同时出现在 reason_codes 中")
|
||||
|
||||
if not _is_nonempty_string(record.get("reason")):
|
||||
errors.append("reason 必须是非空字符串")
|
||||
|
||||
_check_string_list(
|
||||
record.get("evidence"),
|
||||
field="evidence",
|
||||
errors=errors,
|
||||
require_nonempty=True,
|
||||
)
|
||||
|
||||
if "counterevidence" in record:
|
||||
_check_string_list(
|
||||
record.get("counterevidence"),
|
||||
field="counterevidence",
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
if "domain_label" in record:
|
||||
domain_label = record.get("domain_label")
|
||||
if domain_label is not None and not _is_nonempty_string(domain_label):
|
||||
errors.append("domain_label 必须是非空字符串或 null")
|
||||
|
||||
if "candidate_domain_labels" in record:
|
||||
_check_string_list(
|
||||
record.get("candidate_domain_labels"),
|
||||
field="candidate_domain_labels",
|
||||
errors=errors,
|
||||
unique=True,
|
||||
)
|
||||
|
||||
structural_signals = record.get("structural_signals")
|
||||
if structural_signals is not None:
|
||||
if not isinstance(structural_signals, dict):
|
||||
errors.append("structural_signals 必须是对象")
|
||||
else:
|
||||
expected_keys = {"complex", "multi_instruction", "auto_task"}
|
||||
actual_keys = set(structural_signals)
|
||||
if actual_keys != expected_keys:
|
||||
errors.append(
|
||||
"structural_signals 必须且只能包含 "
|
||||
"complex、multi_instruction、auto_task"
|
||||
)
|
||||
for key in expected_keys:
|
||||
value = structural_signals.get(key)
|
||||
if value is not None and not isinstance(value, bool):
|
||||
errors.append(f"structural_signals.{key} 必须是布尔值或 null")
|
||||
|
||||
if not isinstance(record.get("context_used"), bool):
|
||||
errors.append("context_used 必须是布尔值")
|
||||
|
||||
fast_system_pending = record.get("fast_system_pending")
|
||||
if fast_system_pending is not None and not isinstance(fast_system_pending, bool):
|
||||
errors.append("fast_system_pending 必须是布尔值或 null")
|
||||
if fast_system_pending is True and label != "快":
|
||||
errors.append("只有 label=快 时 fast_system_pending 才能为 true")
|
||||
|
||||
legacy_eval_label = record.get("legacy_eval_label")
|
||||
if (
|
||||
"legacy_eval_label" in record
|
||||
and legacy_eval_label is not None
|
||||
and legacy_eval_label not in LEGACY_LABELS
|
||||
):
|
||||
errors.append("legacy_eval_label 不是受支持的旧八类标签")
|
||||
if label == "模糊" and legacy_eval_label is not None:
|
||||
warnings.append("模糊档通常无法无损映射到旧二值标签，请确认迁移依据")
|
||||
|
||||
confidence = record.get("confidence")
|
||||
if confidence not in CONFIDENCE_VALUES:
|
||||
errors.append("confidence 必须是 high、medium、low 之一")
|
||||
|
||||
review_required = record.get("review_required")
|
||||
if not isinstance(review_required, bool):
|
||||
errors.append("review_required 必须是布尔值")
|
||||
if confidence == "low" and review_required is not True:
|
||||
errors.append("confidence=low 时 review_required 必须为 true")
|
||||
|
||||
_check_string_list(
|
||||
record.get("uncertainties"),
|
||||
field="uncertainties",
|
||||
errors=errors,
|
||||
)
|
||||
uncertainties = record.get("uncertainties")
|
||||
if isinstance(uncertainties, list) and uncertainties and review_required is False:
|
||||
warnings.append("uncertainties 非空但 review_required=false，请确认不确定点不影响标签")
|
||||
|
||||
if record.get("policy_version") != POLICY_VERSION:
|
||||
errors.append(f"policy_version 必须是 {POLICY_VERSION}")
|
||||
|
||||
return {
|
||||
"index": index,
|
||||
"id": record.get("id"),
|
||||
"valid": not errors,
|
||||
"errors": errors,
|
||||
"warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
def _load_records_from_file(path: Path) -> list[Any]:
|
||||
text = path.read_text(encoding="utf-8-sig")
|
||||
if path.suffix.lower() == ".jsonl":
|
||||
records: list[Any] = []
|
||||
for line_number, line in enumerate(text.splitlines(), start=1):
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
records.append(json.loads(line))
|
||||
except json.JSONDecodeError as exc:
|
||||
records.append(
|
||||
{
|
||||
"__parse_error__": (
|
||||
f"JSONL 第 {line_number} 行解析失败: {exc.msg}"
|
||||
)
|
||||
}
|
||||
)
|
||||
return records
|
||||
|
||||
payload = json.loads(text)
|
||||
if isinstance(payload, list):
|
||||
return payload
|
||||
if isinstance(payload, dict) and isinstance(payload.get("records"), list):
|
||||
return payload["records"]
|
||||
return [payload]
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="校验 XA4.1-2026.6 快慢分流打标 JSON/JSONL"
|
||||
)
|
||||
source_group = parser.add_mutually_exclusive_group(required=True)
|
||||
source_group.add_argument("--record", help="单条 JSON 对象字符串")
|
||||
source_group.add_argument("--file", type=Path, help="JSON 或 JSONL 文件路径")
|
||||
parser.add_argument(
|
||||
"--strict",
|
||||
action="store_true",
|
||||
help="把 warning 也视为校验失败",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = _parse_args()
|
||||
try:
|
||||
if args.record is not None:
|
||||
records = [json.loads(args.record)]
|
||||
else:
|
||||
records = _load_records_from_file(args.file)
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
result = {
|
||||
"valid": False,
|
||||
"total": 0,
|
||||
"invalid": 1,
|
||||
"warning_count": 0,
|
||||
"errors": [str(exc)],
|
||||
"results": [],
|
||||
}
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 1
|
||||
|
||||
results = [validate_record(record, index) for index, record in enumerate(records)]
|
||||
invalid = sum(not result["valid"] for result in results)
|
||||
warning_count = sum(len(result["warnings"]) for result in results)
|
||||
valid = invalid == 0 and (not args.strict or warning_count == 0)
|
||||
summary = {
|
||||
"valid": valid,
|
||||
"total": len(records),
|
||||
"invalid": invalid,
|
||||
"warning_count": warning_count,
|
||||
"results": results,
|
||||
}
|
||||
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
||||
return 0 if valid else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user