架构原则:确定性操作下沉到代码,模糊推理留给 LLM。AI 不做计算和判断,只做路由决策和报告翻译。
工资数据审核系统,全量对齐《工资审核标准流程 SOP》6 步流程,对飞书/SAP/ADP 导出的工资表执行自动化合规校验。SOP 覆盖率 95%(41/43 项)。
analyze_continuous_employees() 方法,自动识别连续在职人员,排除实习生/当月入离职/近期调薪转正/请假>2天/多主体发薪人员,对比12项计薪科目环比变化,自动标记 >10% 异常波动并分类原因deliver_audit_files.py,11个必发文件清单生成 + 审核结论摘要自动生成,流水线结束后 Phase 8/8 自动交付准备06_master_report.html)judgment 字段,包含规则逻辑、检查范围、排除人数及原因、实际检查人数、通过率、判定结论,让看报告的人能看到"怎么判的、排了谁、阈值是什么"data_index.json 作为报告和看板的关联核心,保证数据一致性run_full_pipeline.py 一键跑完不中断(数据扫描→审核→报告→看板→抽样→问题清单→总审核报告)不覆盖:实际算薪操作、薪酬市场调研、个税计算、社保核算。
| SOP 步骤 | 本 Skill 对应 | 说明 |
|---|---|---|
| --------- | ------------- | ------ |
| 第一步:审核流程(强制) | data_scan.py + 用户确认 | 禁止跳步 |
| 第二步:数据逻辑验证 | rules_engine.py 公式+业务逻辑 | 公式校验+5项业务规则 |
| 第三步:异常数据扫描 | rules_engine.py 红/黄/蓝线 | 4红+6黄+4蓝 |
| 第四步:总额对比分析 | rules_engine.py --prev 总额环比 | ±10%阈值 |
| 第五步:按人深入分析 | rules_engine.py --prev 按人分析 | 排除+6类变化 |
| 第六步:汇总审核报告 | generate_report.py / generate_report_v6.py | 结构化报告(v5)/ 表格化报告(v6) |
| 第七步:审核清单看板 | generate_kanban.py / generate_kanban_v6.py | 静态看板(v5)/ 动态交互看板(v6) |
| 第八步:抽样校验 | sampling_verify.py | 二次确认审核结果 |
| 数据支撑索引 | generate_data_index.py | 三者关联核心(报告↔看板↔数据) |
| 分段审核编排 | run_audit.py | 7阶段独立执行+断点续传 |
| 端到端流水线 | run_full_pipeline.py | 一键跑完不中断,始终生成所有输出(含总报告) |
| 总审核报告 | generate_master_report.py | 聚合所有输出为一份完整 HTML 总报告 |
| 连续在职环比分析 | rules_engine.py::analyze_continuous_employees() | 12项计薪科目环比,排除规则,异常标记 |
| 文件交付(Step 9) | deliver_to_feishu.py | 11个必发文件上传云盘 + 飞书文档创建 + 交付消息生成(自动化交付,LLM只负责发送) |
先判断用户需求属于哪类场景,再执行对应流程:
| 用户说的 | 匹配场景 | 执行 |
|---|---|---|
| --------- | --------- | ------ |
| "帮我审核本月工资数据" | 完整审核 (Step 1→6) | 先 data_scan → 用户确认 → run_full_audit |
| "快速看看有没有问题" | 红线校验 | --step red_lines |
| "发薪前帮我理一下数据" | 字段检查 | --step fields |
| "帮我验一下公式对不对" | 公式校验 | --step formulas |
| "帮我出个审核报告" | 报告生成 | generate_report.py |
| "帮我出个表格版审核报告" | 报告生成 v6 | generate_report_v6.py |
| "帮我出个审核清单/看板" | 审核清单看板 | generate_kanban.py |
| "帮我出个动态交互看板" | 动态看板 v6 | generate_kanban_v6.py(需要 data_index.json) |
| "分段审核/怕上下文截断" | 分段审核 | run_audit.py |
| "帮我二次确认审核结果" | 抽样校验 | sampling_verify.py |
| "从头到尾跑一遍审核" | 端到端流水线 | run_full_pipeline.py(含 v6 报告+看板+索引+总报告+交付摘要) |
| "审核完了把文件发给我" | 文件交付 | deliver_to_feishu.py(自动上传11个文件+创建飞书文档+生成交付消息,LLM只负责发送消息) |
⚠️ 禁止跳过此步骤直接执行审核!
出报告前必须先数据扫描确认,等待用户回复"确认无误"后才执行后续审核。
第一步:数据扫描
python3 scripts/data_scan.py \
--data <工资数据.csv> \
--output /tmp/data_scan_result.json
扫描输出以下信息供用户确认:
数据扫描输出示例:
============================================================
📊 工资数据扫描报告
============================================================
📋 总记录数: 2000
📅 最新发薪月: 2026-05
📅 发薪月分布:
2026-05: 2000 人 ← 最新
2026-04: 1980 人
🏢 公司主体分布:
道旅科技: 1500 人
道旅国际: 300 人
道旅服务: 200 人
💰 主要计薪项汇总:
应发合计: ¥15,234,567.89
实发金额合计: ¥12,345,678.90
加班费小计: ¥234,567.89
个人社保: ¥1,234,567.89
个人公积金: ¥987,654.32
👥 特殊人员:
新入职: 15 人
离职: 8 人
转正: 12 人
调薪: 5 人
🔍 数据质量检测:
工号重复: ✅ 无异常
姓名为空: ✅ 无异常
============================================================
请确认以上数据是否正确,回复'确认无误'后开始正式审核。
============================================================
第二步:用户确认
等待用户回复"确认无误"或指出问题。如果用户指出问题,先修正数据再继续。
第三步:正式审核
用户确认后才执行场景一的完整审核流程。
第一步:获取数据 — 要求用户提供本月工资数据文件(CSV/Excel),必须同时提供上月数据用于跨月对比。
第二步:数据扫描确认 — 见场景零。
第三步:执行审核
python3 -m scripts.rules_engine \
--data <本月工资数据.csv> \
--prev <上月工资数据.csv> \
--output /tmp/audit_result.json
第四步:确认门 — 如果 summary.blocked == true(触发红线),不要生成报告,先输出红线清单并要求用户确认数据修正后再继续。
第五步:生成报告 — 通过红线后,用脚本输出 JSON 生成人类可读报告:
python3 scripts/generate_report.py \
--input /tmp/audit_result.json \
--format both
输出文件:report.html + report.md 在工作目录。
python3 -m scripts.rules_engine \
--data <工资数据.csv> \
--step red_lines \
--output /tmp/red_lines_result.json
适用于发薪前快速检查,只需 10 秒。
python3 -m scripts.rules_engine \
--data <工资数据.csv> \
--step fields \
--output /tmp/fields_result.json
检查必填字段是否缺失、格式是否正确。
python3 -m scripts.rules_engine \
--data <工资数据.csv> \
--step formulas \
--output /tmp/formulas_result.json
验证应发/实发计算是否与公式一致。
# HTML 可视化报告
python3 scripts/generate_report.py \
--input /tmp/audit_result.json --format html --output report.html
# Markdown 报告
python3 scripts/generate_report.py \
--input /tmp/audit_result.json --format markdown --output report.md
# 两者都生成
python3 scripts/generate_report.py \
--input /tmp/audit_result.json --format both
审核清单看板 = 完整审核条目展开清单,每条包含审核结果 + 数据依据 + 处理建议。
python3 scripts/generate_kanban.py --input /tmp/audit_result.json --format both
超链接复核:--review-link "https://hr.example/emp/{emp_id}?row={row_index}"
输出 kanban.html + kanban.md,展开 rules.json 所有规则(28+条)。
审核完成后进行二次确认:python3 scripts/sampling_verify.py --data <数据.csv> --audit <审核结果.json> --sample-size 30 --threshold 0.05 --output <输出.json>
流程:随机抽样→独立重算(公式/红线/业务规则)→交叉对比→偏差率≤5%通过;>5%触发根因分析后重新审核。偏差超标时输出根因报告并使用 run_audit.py --resume 断点重审。
python3 scripts/run_audit.py --data <本月.csv> --prev <上月.csv> --output-dir /tmp/audit_phases
7个独立阶段(字段→公式→业务规则→红线阻断门→黄线→蓝线→汇总),每阶段输出独立 JSON,避免上下文截断。红线触发时自动阻断后续阶段。支持断点续传:--resume。
输出文件:00_data_scan.json 01_audit_result.json 02_report.html/md 02_report_v6.html/md(表格化,推荐)02a_data_index.json(三者关联核心)03_kanban.html/md 03_kanban_v6.html(动态交互,推荐)04_sampling_verify.json 05_issue_report.md(问题清单) 06_master_report.html(总审核报告) audit_summary.md(审核结论摘要)。
python3 scripts/run_full_pipeline.py --data <本月.csv> --prev <上月.csv> --output-dir /tmp/audit_output --review-link "https://hr.example/emp/{emp_id}?row={row_index}"
审核完成后必须执行此步骤,不可跳过。
python3 scripts/deliver_to_feishu.py --output-dir /tmp/audit_output --audit-result /tmp/audit_output/01_audit_result.json
自动化交付流程(LLM 不可干预、不可跳过):
02_report_v6.md import 为 docx 格式⚠️ LLM 层唯一职责:把 delivery_message.md 的内容通过消息工具发送到飞书。
⚠️ 禁止:自行挑选文件上传、自行创建文档格式、自行裁剪交付内容。
每个审核结论必须有数据依据,不凭空判断。 这是审核系统的核心原则——"审核通过"不是感觉出来的,是数据算出来的。
每个判定为"正常/通过"的审核项目,必须同时输出以下三项数据:
| 要素 | 说明 | 示例 |
|---|---|---|
| ------ | ------ | ------ |
| 检查数 | 共检查了多少条记录 | "2000条记录" |
| 通过数 | 多少条符合规则 | "1998条通过" |
| 阈值 | 判定标准是什么 | "容差0.01元" |
完整示例:
为什么必须这样:如果没有检查数和通过数,"通过"可能是只检查了1条、也可能是2000条——可信度完全不同。
必须列出具体人员、工号、异常值、预期值、违反的规则ID:
必须给出具体数值、变化幅度、驱动因素:
当某人被排除在异常分析外时,必须说明排除原因和依据:
红线触发时输出:触发人数/具体名单(工号+姓名+异常值)/违反规则(ID+名称)。
报告"结论与建议"结构:总体结论(检查数+各维度通过/异常数)→各维度详情表格(检查数/通过数/异常数/通过率/结论)→异常明细→建议行动(基于具体数据)。
脚本内置 COLUMN_ALIAS 字典,自动映射 30+ 种常见列名变体。如果列名无法识别,脚本返回 {"error": "Missing column: XXX"},LLM 必须原文转述。
金额计算使用 decimal.Decimal,避免浮点误差累积。公式校验容差 0.01(1 分钱)。
所有规则集中在 references/rules.json。修改规则只需编辑此文件,无需修改脚本。为什么这样设计:业务规则变化频繁(如最低工资调整),集中管理避免每次都要改代码。
rules.json 的 red_lines 数组中添加新对象,包含 id、name、condition、field、severity: "BLOCK"yellow_lines 数组中添加,action: "flag_and_explain"blue_lines 数组中添加,action: "note_only"policies 数组中添加,包含 condition(触发条件)和 effect(豁免效果)business_rules 数组中添加,支持 field1 <= field2 类双字段比较person_analysis_exclusions 对象中添加新 key,包含 field、condition、note规则 ID 命名规范:{类型}-{序号},如 RL-005(第5条红线)、YL-007(第7条黄线)、POL-006(第6条政策)、BR-005(第5条业务规则)
添加规则后不需要修改任何 Python 代码,引擎自动加载。
红线触发 → 立即阻断,不继续后续步骤,不生成报告。为什么:红线代表严重违规(如负实发、低于最低工资),必须先修正数据再审核,否则报告无意义。
本 Skill 被 Workflow Orchestration Skill 编排时,必须作为单节点调用。
根因:run_full_pipeline.py 已是端到端流水线,内部按序完成所有 8 步(扫描→规则→红黄蓝线→报告→看板→抽样→问题清单→总报告)。拆成多节点编排无收益——每个节点仍调用同一个 skill,内部全量执行一遍。
| 错误做法 | 正确做法 |
|---|---|
| ---------- | ---------- |
| N1 data_scan → N2 rules_engine → N3 generate_report → ... | N1 run_full_pipeline.py 一键跑完 |
唯一例外:当业务确实需要中间状态可见性或独立重试时,才使用 run_audit.py 分段模式(7阶段+断点续传),但仍然是同一 skill 的多次调用,不是拆成多个 skill。
| 场景 | 处理 |
|---|---|
| ------ | ------ |
| 数据源字段缺失 | 列出缺失字段清单,要求用户补充后重试 |
| 日期格式混乱 | 脚本自动尝试多种格式解析,失败则返回错误 |
| 无上月数据 | 跨月对比/按人分析跳过,在报告中标注"无基准数据" |
| Python 环境无 pandas | 提示用户 pip install pandas numpy openpyxl |
| 红线触发 | 立即阻断,不继续后续步骤 |
payroll-data-audit/
├── SKILL.md
├── _meta.json
├── scripts/
│ ├── data_scan.py # 数据扫描(SOP 第一步)
│ ├── rules_engine.py # 统一规则引擎(OOP)
│ ├── generate_report.py # HTML/Markdown 报告生成器(v5,兼容)
│ ├── generate_report_v6.py # 表格化报告生成器(v6,推荐)
│ ├── generate_kanban.py # 审核清单看板生成器(Step 7,静态)
│ ├── generate_kanban_v6.py # 动态交互看板生成器(v6,推荐)
│ ├── generate_data_index.py # 数据支撑索引生成器(v6,三者关联核心)
│ ├── run_audit.py # 分段审核编排器(7阶段+断点续传)
│ ├── run_full_pipeline.py # 端到端流水线(含 v6 输出 + 总报告)
│ ├── generate_master_report.py # 总审核报告生成器(v6.2,聚合所有输出)
│ ├── sampling_verify.py # 抽样校验器(二次确认)
│ ├── deliver_audit_files.py # 文件交付清单生成器(v6.3,仅生成清单,已废弃)
│ ├── deliver_to_feishu.py # 文件交付到飞书(v7.4,自动上传+文档+消息,推荐)
│ └── tests/ # pytest 测试套件
│ ├── test_rules_engine.py
│ └── test_parameterized.py
├── references/
│ └── rules.json # 所有规则声明(唯一真相来源)
└── .github/workflows/
└── test.yml # CI/CD
pip install pandas numpy openpyxl
cd payroll-data-audit
python -m pytest scripts/tests/ -v --cov=scripts.rules_engine --cov-report=term-missing
CI/CD 自动运行:push/PR 时触发 pytest + coverage(80% 门槛)+ lint。
共 8 个版本