制造业数字员工建设（三）：监测与考核方法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# Kong AI Proxy 插件配置（记录 LLM 操作）
plugins:
  - name: ai-proxy
    config:
      route_type: llm/v1/chat
      logging:
        log_statistics: true           # 开启统计日志
        log_payloads: true             # 记录请求/响应内容
        log_errors: true               # 记录错误信息
      model:
        provider: openai
        name: gpt-4

# Kong File Log 插件配置（记录所有 API 操作）
  - name: file-log
    config:
      path: /var/log/kong/agent-audit.log
      reopen: true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# skills/kpi_data_collector.py
from openclaw.skills import Skill, register
from openclaw.api import KongLogClient, ERPClient

@register("kpi-data-collector")
class KPIDataCollector(Skill):
    """从日志系统和业务系统采集 KPI 原始数据"""
    
    def collect_daily_data(self, agent_id: str, date: str):
        # 1. 从 Kong 日志拉取 Agent 的 API 调用统计
        kong_stats = self.kong_log.query(f"""
            SELECT 
                count(*) as total_requests,
                count(CASE WHEN status >= 500 THEN 1 END) as errors,
                avg(duration) as avg_duration,
                percentile_cont(0.95) WITHIN GROUP (ORDER BY duration) as p95_duration
            FROM agent_logs
            WHERE agent_id = '{agent_id}'
              AND date = '{date}'
        """)
        
        # 2. 从 ERP 拉取订单修正记录（用于准确率反查）
        erp_corrections = self.erp.query(f"""
            SELECT count(*) as corrected_orders
            FROM sales_order_audit
            WHERE corrected_by = 'manual'
              AND original_agent = '{agent_id}'
              AND correction_date = '{date}'
        """)
        
        # 3. 从告警系统拉取漏报/误报记录
        alert_quality = self.alert_system.query(f"""
            SELECT 
                count(CASE WHEN alert_is_correct THEN 1 END) as correct,
                count(CASE WHEN NOT alert_is_correct THEN 1 END) as false_positive,
                count(CASE WHEN missed THEN 1 END) as false_negative
            FROM alert_quality
            WHERE agent_id = '{agent_id}'
              AND date = '{date}'
        """)
        
        return {
            "kong_stats": kong_stats,
            "erp_corrections": erp_corrections,
            "alert_quality": alert_quality
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# skills/kpi_calculator.py
@register("kpi-calculator")
class KPICalculator(Skill):
    """根据原始数据计算 KPI 得分"""
    
    def calculate_agent_kpi(self, agent_id: str, agent_type: str, data: dict):
        if agent_type == "sales-assistant":
            # 业务跟单助理的 KPI 计算公式
            orders_total = data["total_orders"]
            orders_wrong = data["corrected_orders"]
            accuracy = (orders_total - orders_wrong) / orders_total * 100
            
            within_2h = data["orders_within_2h"]
            timeliness = within_2h / orders_total * 100
            
            alert_correct = data["correct_alerts"]
            alert_false = data["false_positive"]
            alert_missed = data["false_negative"]
            alert_accuracy = alert_correct / (alert_correct + alert_false + alert_missed) * 100
            
            success = data["successful_requests"]
            total = data["total_requests"]
            availability = success / total * 100
            
            # 加权综合分
            total_score = (
                min(accuracy, 100) * 0.40 +
                min(timeliness, 100) * 0.30 +
                min(alert_accuracy, 100) * 0.20 +
                min(availability, 100) * 0.10
            )
            
            return {
                "agent_id": agent_id,
                "date": data["date"],
                "kpi_details": {
                    "accuracy": round(accuracy, 2),
                    "timeliness": round(timeliness, 2),
                    "alert_accuracy": round(alert_accuracy, 2),
                    "availability": round(availability, 2)
                },
                "total_score": round(total_score, 2),
                "status": "pass" if total_score >= 97.5 else "fail",
                "below_threshold": [
                    k for k, v in {
                        "accuracy": accuracy, 
                        "timeliness": timeliness,
                        "alert_accuracy": alert_accuracy,
                        "availability": availability
                    }.items()
                    if (k == "accuracy" and v < 99.5) or
                       (k == "timeliness" and v < 98) or
                       (k == "alert_accuracy" and v < 95) or
                       (k == "availability" and v < 99)
                ]
            }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# skills/kpi_reporting.py
@register("kpi-reporting")
class KPIReporting(Skill):
    """生成 KPI 报告并推送到指定渠道"""
    
    def generate_daily_summary(self, all_kpis: list):
        """生成日简报"""
        results = []
        for kpi in all_kpis:
            status_emoji = "✅" if kpi["status"] == "pass" else "🚨"
            results.append(
                f"{status_emoji} **{kpi['agent_id']}** 综合: {kpi['total_score']}分\n"
                f"  准确率: {kpi['kpi_details']['accuracy']}% / "
                f"及时率: {kpi['kpi_details']['timeliness']}%\n"
                f"  告警准确率: {kpi['kpi_details']['alert_accuracy']}% / "
                f"可用性: {kpi['kpi_details']['availability']}%"
            )
        
        if any(kpi.get("below_threshold") for kpi in all_kpis):
            results.append("\n⚠️ **需要关注**：以下指标未达标—")
            for kpi in all_kpis:
                for item in kpi.get("below_threshold", []):
                    results.append(f"  · {kpi['agent_id']} → {item}")
        
        return "\n".join(results)
    
    async def send_to_wecom(self, report: str, report_type: str = "daily"):
        """推送到企业微信"""
        title = {
            "daily": "📊 数字员工日报",
            "weekly": "📈 数字员工周报",
            "monthly": "📋 数字员工月度考核报告"
        }
        await self.wecom.send_message(
            target="IT 管理群",
            title=title[report_type],
            content=report
        )