# -*- coding: utf-8 -*- """ B-35 员工服务与制度问答（SSC 机器人）· 闸3 质检脚本 三类断言： A 数据完整性 —— 表在、列在、主键唯一、枚举值合法、关联可回查 B 规则正确性 —— 五道闸门判定自洽、独立重算零差异、红线可验证 C 指标达标 —— K1~K7 达标 + 每个指标补一条 0~1 区间断言（抓分母倒置） 用法：cd 02-skill && python verify.py """ import os import sys import datetime as dt try: import openpyxl except ImportError: print("需要 openpyxl：pip install openpyxl") sys.exit(1) HERE = os.path.dirname(os.path.abspath(__file__)) XLSX = os.path.join(HERE, "data", "b35_ssc_qa.xlsx") # ---- 与 PRD / rules.md / gen_data.py 三处一致的口径常量 ---- SIM_GATE = 0.72 Q_DIRECT = 0.65 Q_COND = 0.50 W_SIM, W_AUTH, W_FRESH, W_CROWD = 0.40, 0.25, 0.20, 0.15 AUTH_SCORE = {"国家/地方法规": 1.00, "集团制度": 0.85, "公司制度": 0.70, "部门细则": 0.55} SENSITIVE_CATS = ["薪酬明细", "绩效结果", "劳动合同", "离职补偿", "纪律处分", "考勤异常申诉"] HR_ROLES = ["HRBP", "HR管理员"] SLA_HOURS = 24 RECHECK_OVERDUE = 365 PASS = 0 FAIL_LIST = [] def ck(cond, msg): global PASS if cond: PASS += 1 else: FAIL_LIST.append(msg) def g(d, k): """空单元格读回是 None，统一转空串""" v = d.get(k, "") return "" if v is None else v def load(sheet): ws = WB[sheet] hdr = [c.value for c in ws[1]] out = [] for r in ws.iter_rows(min_row=2, values_only=True): if r is None or all(x is None for x in r): continue out.append(dict(zip(hdr, r))) return out def pdate(s): s = g({"x": s}, "x") if not s: return None for f in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): try: return dt.datetime.strptime(str(s), f) except ValueError: continue return None print("=" * 70) print("B-35 员工服务与制度问答 · 闸3 质检") print("=" * 70) if not os.path.exists(XLSX): print("❌ 数据源不存在：%s" % XLSX) sys.exit(1) WB = openpyxl.load_workbook(XLSX, data_only=True) KB = load("制度知识库") Q = load("员工提问记录") AG = load("代办申请单") RV = load("人工复核记录") TK = load("转人工工单") CF = load("制度冲突登记") RS = load("可预约资源池") MT = load("指标汇总") # ==================== A. 数据完整性 ==================== print("\n[A] 数据完整性") ck(set(WB.sheetnames) >= {"制度知识库", "员工提问记录", "代办申请单", "人工复核记录", "转人工工单", "制度冲突登记", "可预约资源池", "指标汇总"}, "A01 缺少标签页") ck(len(KB) == 60, "A02 制度条款应为 60 条，实际 %d" % len(KB)) ck(len(Q) == 240, "A03 提问应为 240 条，实际 %d" % len(Q)) ck(len(RV) == 65, "A04 人工复核应为 65 条，实际 %d" % len(RV)) ck(len(RS) == 18, "A05 资源池应为 18 条，实际 %d" % len(RS)) # 主键唯一 for name, rows, key in [("制度条款", KB, "条款号"), ("提问", Q, "提问编号"), ("代办单", AG, "申请单号"), ("工单", TK, "工单号"), ("冲突", CF, "冲突编号"), ("资源", RS, "资源编号")]: ids = [g(r, key) for r in rows] ck(len(ids) == len(set(ids)), "A06 %s 主键 %s 有重复" % (name, key)) # 关联可回查 qids = set(g(r, "提问编号") for r in Q) cid = set(g(r, "条款号") for r in KB) ck(all(g(a, "关联提问编号") in qids for a in AG), "A07 代办单关联提问编号有孤儿") ck(all(g(t, "关联提问编号") in qids for t in TK), "A08 工单关联提问编号有孤儿") ck(all(g(r, "关联提问编号") in qids for r in RV), "A09 复核关联提问编号有孤儿") ck(all(g(c, "关联提问编号") in qids for c in CF), "A10 冲突关联提问编号有孤儿") ck(all(g(q, "命中条款号") in cid for q in Q if g(q, "命中条数")), "A11 命中条款号有不在知识库里的") # 枚举合法 ck(all(g(k, "状态") in ("现行有效", "待复核", "已废止") for k in KB), "A12 条款状态枚举越界") ck(all(g(k, "效力层级") in AUTH_SCORE for k in KB), "A13 效力层级枚举越界") ck(all(g(q, "作答状态") in ("已作答", "已转人工") for q in Q), "A14 作答状态枚举越界") ck(all(g(q, "作答档位") in ("直接作答", "附条件作答", "降级作答", "不作答", "不出稿") for q in Q), "A15 作答档位枚举越界") ck(all(g(q, "命中闸门") in ("R1", "R2", "R3", "R4", "R5", "—") for q in Q), "A16 命中闸门枚举越界") ck(all(g(q, "意图类型") in ("制度咨询", "申请代办", "资源预约", "个人信息查询", "人事决策建议") for q in Q), "A17 意图类型枚举越界") # 待复核必须真的超期；已废止同理 for k in KB: if g(k, "状态") == "待复核": ck(int(g(k, "复核距今天数")) > RECHECK_OVERDUE, "A18 待复核条款 %s 复核距今天数未超期" % g(k, "条款号")) ck(all(int(g(k, "复核距今天数")) <= RECHECK_OVERDUE for k in KB if g(k, "状态") == "现行有效"), "A19 现行有效条款不应超期未复核") # ==================== B. 规则正确性 ==================== print("[B] 规则正确性") KBI = {g(k, "条款号"): k for k in KB} # B1 独立重算：新鲜度 + 加权质量分（逐条零差异） diff_fresh = diff_q = 0 for q in Q: if not g(q, "命中条款号"): continue k = KBI.get(g(q, "命中条款号")) if k is None: continue d = int(g(k, "复核距今天数")) f = round(max(0.0, 1.0 - d / 365.0), 4) if abs(f - float(g(k, "新鲜度得分"))) > 1e-6: diff_fresh += 1 if abs(f - float(g(q, "新鲜度得分"))) > 1e-6: diff_fresh += 1 rec = round(W_SIM * float(g(q, "最高相似度")) + W_AUTH * float(g(q, "效力层级得分")) + W_FRESH * float(g(q, "新鲜度得分")) + W_CROWD * float(g(q, "适用人群匹配得分")), 4) if abs(rec - float(g(q, "加权质量分"))) > 1e-6: diff_q += 1 ck(diff_fresh == 0, "B01 新鲜度得分独立重算有 %d 条差异" % diff_fresh) ck(diff_q == 0, "B02 加权质量分独立重算有 %d 条差异" % diff_q) # B2 效力层级得分与层级一致 ck(all(abs(float(g(k, "效力层级得分")) - AUTH_SCORE[g(k, "效力层级")]) < 1e-9 for k in KB), "B03 效力层级与得分映射不一致") # B3 R1：命中为空 或 相似度 <0.72 ck(all((int(g(q, "命中条数")) == 0 or float(g(q, "最高相似度")) < SIM_GATE) for q in Q if g(q, "命中闸门") == "R1"), "B04 R1 样本不满足无依据条件") ck(all(not (int(g(q, "命中条数")) and float(g(q, "最高相似度")) < SIM_GATE) for q in Q if g(q, "命中闸门") != "R1"), "B05 相似度低于门槛却未被 R1 拦下") # B4 敏感类非授权必被 R2 拦下（🔴 红线） sens_den = [q for q in Q if g(q, "是否敏感类") == "是" and g(q, "查询对象") != "本人" and g(q, "员工角色") not in HR_ROLES] ck(len(sens_den) > 0, "B06 敏感类非授权样本集为空，K7 无从验证") ck(all(g(q, "命中闸门") == "R2" and g(q, "作答状态") == "已转人工" for q in sens_den), "B07 敏感类非授权提问未被 R2 拦截") ck(not any(g(q, "引用条款") for q in sens_den), "B08 被 R2 拦截的提问不应带引用") # B5 敏感类 + HR 角色查询他人 = 放行（不是漏网） sens_ok = [q for q in Q if g(q, "是否敏感类") == "是" and g(q, "查询对象") != "本人" and g(q, "员工角色") in HR_ROLES] ck(all(g(q, "作答状态") == "已作答" for q in sens_ok), "B09 授权 HR 查询他人不应被拦截") # B6 人事决策建议必被 R4 拦下 ck(all(g(q, "命中闸门") == "R4" for q in Q if g(q, "意图类型") == "人事决策建议"), "B10 人事决策建议未被 R4 拦下") # B7 已废止条款不作答 ck(all(g(q, "作答状态") == "已转人工" and g(q, "转人工原因") == "条款已废止" for q in Q if g(q, "命中条款号") and KBI.get(g(q, "命中条款号")) and g(KBI[g(q, "命中条款号")], "状态") == "已废止"), "B11 命中已废止条款却仍作答") # B8 待复核条款 → 降级作答 pend_q = [q for q in Q if g(q, "命中条款号") and KBI.get(g(q, "命中条款号")) and g(KBI[g(q, "命中条款号")], "状态") == "待复核" and g(q, "作答状态") == "已作答"] ck(len(pend_q) > 0, "B12 无待复核降级样本") ck(all(g(q, "作答档位") == "降级作答" for q in pend_q), "B13 待复核条款未降级作答") ck(all("待复核" in g(q, "警示标注") for q in pend_q), "B14 降级作答未附待复核警示") # B9 出稿档位与质量分一致 for q in Q: if g(q, "作答状态") != "已作答": ck(float(g(q, "加权质量分")) >= Q_COND or g(q, "命中闸门") != "—" or g(q, "转人工原因") in ("质量分不足", "制度冲突待裁决"), "B15 转人工原因与质量分不自洽 %s" % g(q, "提问编号")) continue sc = float(g(q, "加权质量分")) if g(q, "作答档位") == "直接作答": ck(sc >= Q_DIRECT, "B16 直接作答质量分 %.4f < 0.65（%s）" % (sc, g(q, "提问编号"))) elif g(q, "作答档位") == "附条件作答": ck(Q_COND <= sc < Q_DIRECT, "B17 附条件作答质量分 %.4f 越界（%s）" % (sc, g(q, "提问编号"))) ck("建议向 HR 确认" in g(q, "警示标注"), "B18 附条件作答未附确认警示 %s" % g(q, "提问编号")) # B10 K4 引用可回溯（🔴 硬红线） ans = [q for q in Q if g(q, "作答状态") == "已作答"] ck(len(ans) == len([q for q in ans if g(q, "引用条款")]), "B19 存在已作答但无引用的提问（K4 硬红线）") ck(all(("｜" in g(q, "引用条款") and "生效" in g(q, "引用条款") and "适用" in g(q, "引用条款")) for q in ans), "B20 引用缺四要素之一（条款号/生效/适用/版本）") ck(not any(g(q, "引用条款") for q in Q if g(q, "作答状态") == "已转人工"), "B21 转人工的提问不应带引用") # B11 问题原文与条款摘要必须同分类（防止「问什么 ≠ 答什么」） ck(all(g(q, "问题分类") == g(KBI[g(q, "命中条款号")], "问题分类") for q in Q if g(q, "命中条款号") and KBI.get(g(q, "命中条款号"))), "B22 命中条款的分类与提问分类不一致") # B12 冲突：14 条，8 已处置 / 6 待裁决 ck(len(CF) == 14, "B23 冲突登记应为 14 条，实际 %d" % len(CF)) ck(len([c for c in CF if g(c, "处置状态") == "已处置"]) == 8, "B24 冲突已处置应为 8 条") ck(len([c for c in CF if g(c, "处置状态") == "待裁决"]) == 6, "B25 冲突待裁决应为 6 条") # 已处置：两条层级不同；待裁决：两条层级相同 for c in CF: a, b = g(c, "效力层级A"), g(c, "效力层级B") if g(c, "处置状态") == "已处置": ck(a != b, "B26 已处置冲突 %s 两条层级相同却自动选边" % g(c, "冲突编号")) else: ck(a == b, "B27 待裁决冲突 %s 两条层级不同" % g(c, "冲突编号")) ck(g(c, "适用人群是否重叠") == "是", "B28 冲突 %s 适用人群未重叠" % g(c, "冲突编号")) ck(all(g(q, "转人工原因") == "制度冲突待裁决" for q in Q if g(q, "提问编号") in set(g(c, "关联提问编号") for c in CF if g(c, "处置状态") == "待裁决")), "B29 待裁决冲突未转人工") # B13 代办：AI 不审批 —— 审批人角色必为人类角色 ck(all(g(a, "审批人角色") in ("直属上级", "部门经理", "行政专员", "HRBP", "HR管理员", "部门经理+HR", "HRBP+部门经理", "培训专员") for a in AG), "B30 代办单审批人角色异常") ck(not any(g(a, "审批人角色") in ("AI", "机器人", "系统") for a in AG), "B31 AI 出现在审批人位置") # 一次通过 ⇔ 重提次数 = 0 ck(all((g(a, "是否一次通过") == "是") == (int(g(a, "重提次数")) == 0) for a in AG if g(a, "是否一次通过")), "B32 一次通过标记与重提次数不自洽") ck(all(int(g(a, "重提次数")) >= 1 for a in AG if g(a, "是否一次通过") == "否"), "B33 非一次通过却重提次数为 0") # 已出结果的必须有完成时间 ck(all(g(a, "完成时间") for a in AG if g(a, "审批状态") in ("已通过", "已驳回")), "B34 已出审批结果却无完成时间") # R5：受限的不得生成代办单 r5_q = set(g(q, "提问编号") for q in Q if g(q, "命中闸门") == "R5") ck(len(r5_q) == 4, "B35 R5 受限样本应为 4 条，实际 %d" % len(r5_q)) ck(not any(g(a, "关联提问编号") in r5_q for a in AG), "B36 R5 受限提问却生成了代办单") ck(all(g(q, "是否可代办") == "否" and g(q, "代办受限原因") for q in Q if g(q, "命中闸门") == "R5"), "B37 R5 受限未填代办受限原因") # B14 转人工工单时效 for t in TK: st, ac = pdate(g(t, "转人工时间")), pdate(g(t, "受理时间")) due = pdate(g(t, "应闭环时限")) if st and ac and due: ck(abs((due - st).total_seconds() - SLA_HOURS * 3600) < 60, "B38 应闭环时限 ≠ 转人工时间 + 24h（%s）" % g(t, "工单号")) if g(t, "闭环状态") == "已闭环": ck(g(t, "是否超期") in ("是", "否"), "B39 已闭环工单未判超期") ck(bool(g(t, "闭环时间")), "B40 已闭环工单无闭环时间") if st and ac: ck((ac > due) == (g(t, "是否超期") == "是"), "B41 超期判定与受理时间不自洽（%s）" % g(t, "工单号")) else: ck(not g(t, "闭环时间"), "B42 未闭环工单不应有闭环时间") # B15 🔴 不出人名 ck(not any("姓名" in k for k in Q[0].keys()), "B43 提问表出现姓名字段") ck(not any("姓名" in k for k in TK[0].keys()), "B44 工单表出现姓名字段") ck(all("受理人角色" in t for t in TK), "B45 工单未记角色维度") # ==================== C. 指标达标 ==================== print("[C] 指标达标") def ratio(n, d): return n / d if d else 0.0 n_ans = len(ans) n_fb = len([q for q in Q if g(q, "反馈状态") == "已回收"]) n_solved = len([q for q in Q if g(q, "反馈结果") == "已解决"]) n_r1 = len([q for q in Q if g(q, "命中闸门") == "R1"]) n_cited = len([q for q in ans if g(q, "引用条款")]) n_rev_ok = len([r for r in RV if g(r, "复核结果") == "准确"]) n_decided = len([a for a in AG if g(a, "审批状态") in ("已通过", "已驳回")]) n_pass1 = len([a for a in AG if g(a, "是否一次通过") == "是"]) n_closed = len([t for t in TK if g(t, "闭环状态") == "已闭环"]) n_ontime = len([t for t in TK if g(t, "闭环状态") == "已闭环" and g(t, "是否超期") == "否"]) n_sens_block = len([q for q in sens_den if g(q, "作答状态") == "已转人工"]) K1 = ratio(n_solved, n_fb) K2 = ratio(n_rev_ok, len(RV)) K3 = ratio(n_r1, len(Q)) K4 = ratio(n_cited, n_ans) K5 = ratio(n_pass1, n_decided) K6 = ratio(n_ontime, n_closed) K7 = ratio(n_sens_block, len(sens_den)) ck(K1 >= 0.60, "C01 K1 自助解决率 %.4f < 60%%" % K1) ck(K2 >= 0.90, "C02 K2 首答准确率 %.4f < 90%%" % K2) ck(0.10 <= K3 <= 0.40, "C03 K3 无依据拦截率 %.4f 超出 10%%~40%% 健康区间" % K3) ck(K4 == 1.0, "C04 K4 引用可回溯率 %.4f ≠ 100%%（硬红线）" % K4) ck(K5 >= 0.70, "C05 K5 代办一次通过率 %.4f < 70%%" % K5) ck(K6 >= 0.85, "C06 K6 转人工及时率 %.4f < 85%%" % K6) ck(K7 == 1.0, "C07 K7 敏感信息拦截率 %.4f ≠ 100%%（硬红线）" % K7) # 每个指标补一条 0~1 区间断言（抓分母倒置 / 分子落在分母外） for name, v in [("K1", K1), ("K2", K2), ("K3", K3), ("K4", K4), ("K5", K5), ("K6", K6), ("K7", K7)]: ck(0.0 <= v <= 1.0, "C08 %s = %.4f 越出 0~1（分母可能被倒置）" % (name, v)) ck(n_solved <= n_fb, "C09 K1 分子 %d 落在分母 %d 之外" % (n_solved, n_fb)) ck(n_rev_ok <= len(RV), "C10 K2 分子落在分母之外") ck(n_r1 <= len(Q), "C11 K3 分子落在分母之外") ck(n_cited <= n_ans, "C12 K4 分子落在分母之外") ck(n_pass1 <= n_decided, "C13 K5 分子落在分母之外") ck(n_ontime <= n_closed, "C14 K6 分子落在分母之外") ck(n_sens_block <= len(sens_den), "C15 K7 分子落在分母之外") # 七个分母互不相同 denoms = [n_fb, len(RV), len(Q), n_ans, n_decided, n_closed, len(sens_den)] ck(len(set(denoms)) == 7, "C16 七个分母出现重复：%s" % denoms) # 指标汇总表与实测一致 mt = {g(m, "指标编号"): m for m in MT} for no, (num, den) in {"K1": (n_solved, n_fb), "K2": (n_rev_ok, len(RV)), "K3": (n_r1, len(Q)), "K4": (n_cited, n_ans), "K5": (n_pass1, n_decided), "K6": (n_ontime, n_closed), "K7": (n_sens_block, len(sens_den))}.items(): ck(no in mt, "C17 指标汇总缺 %s" % no) if no in mt: ck(int(g(mt[no], "分子")) == num and int(g(mt[no], "分母")) == den, "C18 %s 汇总表分子分母与实测不符" % no) ck(g(mt[no], "是否达标") == "达标", "C19 %s 汇总表未标达标" % no) # ==================== 输出 ==================== print("\n" + "-" * 70) print("实测指标：") print(" K1 自助解决率 %3d / %3d = %6.2f%% (目标 ≥60%%)" % (n_solved, n_fb, K1 * 100)) print(" K2 首答准确率 %3d / %3d = %6.2f%% (目标 ≥90%%)" % (n_rev_ok, len(RV), K2 * 100)) print(" K3 无依据拦截率 %3d / %3d = %6.2f%% (目标 10%%~40%%)" % (n_r1, len(Q), K3 * 100)) print(" K4 引用可回溯率 %3d / %3d = %6.2f%% (目标 =100%% 🔴)" % (n_cited, n_ans, K4 * 100)) print(" K5 代办一次通过率 %3d / %3d = %6.2f%% (目标 ≥70%%)" % (n_pass1, n_decided, K5 * 100)) print(" K6 转人工及时率 %3d / %3d = %6.2f%% (目标 ≥85%%)" % (n_ontime, n_closed, K6 * 100)) print(" K7 敏感信息拦截率 %3d / %3d = %6.2f%% (目标 =100%% 🔴)" % (n_sens_block, len(sens_den), K7 * 100)) print("-" * 70) print("样本分布：提问 %d（已作答 %d / 转人工 %d）｜代办 %d｜复核 %d｜工单 %d｜冲突 %d｜条款 %d" % (len(Q), n_ans, len(Q) - n_ans, len(AG), len(RV), len(TK), len(CF), len(KB))) if FAIL_LIST: print("\n❌ FAILED %d 项：" % len(FAIL_LIST)) for m in FAIL_LIST[:40]: print(" -", m) print("\n合计：PASS %d / FAIL %d" % (PASS, len(FAIL_LIST))) sys.exit(1) else: print("\n✅ 全部断言 PASS：%d 项（闸3 放行）" % PASS)