AI 数字员工上岗验收：转人工率与异常分类统计脚本

{"time": "2026-10-20T09:31:00", "session_id": "s001", "resolved": true, "transferred": false, "rework": false, "error_type": "无"}
{"time": "2026-10-20T09:40:00", "session_id": "s002", "resolved": false, "transferred": true, "rework": false, "error_type": "超出知识范围"}

"""数字员工运行日志分析：一次通过率、转人工率、异常分布与质量闸门检查。"""
import json
from collections import Counter

# 质量闸门阈值，按岗位风险自行调整
GATE = {"first_pass_rate_min": 0.85, "transfer_rate_max": 0.15}


def analyze(log_path="digital_worker_log.jsonl"):
    with open(log_path, encoding="utf-8") as f:
        records = [json.loads(line) for line in f if line.strip()]
    total = len(records)
    if not total:
        print("暂无日志")
        return
    first_pass = sum(1 for r in records if r["resolved"] and not r["transferred"] and not r["rework"])
    transferred = sum(1 for r in records if r["transferred"])
    errors = Counter(r["error_type"] for r in records if r["error_type"] != "无")
    fp_rate, tr_rate = first_pass / total, transferred / total
    print(f"会话总数：{total}")
    print(f"一次通过率：{fp_rate:.1%}（闸门下限 {GATE['first_pass_rate_min']:.0%}）"
          f" -> {"通过" if fp_rate >= GATE["first_pass_rate_min"] else "未达标"}")
    print(f"转人工率：{tr_rate:.1%}（闸门上限 {GATE['transfer_rate_max']:.0%}）"
          f" -> {"通过" if tr_rate <= GATE["transfer_rate_max"] else "未达标"}")
    print("异常分布：" + ("、".join(f"{k} {v} 次" for k, v in errors.most_common()) or "无"))


if __name__ == "__main__":
    analyze()