290 lines
11 KiB
Python
290 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
分析模板系统 - 提供预定义的分析场景
|
||
"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
from typing import List, Dict, Any
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass
|
||
class AnalysisStep:
|
||
"""分析步骤"""
|
||
name: str
|
||
description: str
|
||
analysis_type: str # explore, visualize, calculate, report
|
||
prompt: str
|
||
|
||
|
||
class AnalysisTemplate(ABC):
|
||
"""分析模板基类"""
|
||
|
||
def __init__(self, name: str, description: str):
|
||
self.name = name
|
||
self.description = description
|
||
self.steps: List[AnalysisStep] = []
|
||
|
||
@abstractmethod
|
||
def build_steps(self, **kwargs) -> List[AnalysisStep]:
|
||
"""构建分析步骤"""
|
||
pass
|
||
|
||
def get_full_prompt(self, **kwargs) -> str:
|
||
"""获取完整的分析提示词"""
|
||
steps = self.build_steps(**kwargs)
|
||
|
||
prompt = f"# {self.name}\n\n{self.description}\n\n"
|
||
prompt += "## 分析步骤：\n\n"
|
||
|
||
for i, step in enumerate(steps, 1):
|
||
prompt += f"### {i}. {step.name}\n"
|
||
prompt += f"{step.description}\n\n"
|
||
prompt += f"```\n{step.prompt}\n```\n\n"
|
||
|
||
return prompt
|
||
|
||
|
||
class HealthReportTemplate(AnalysisTemplate):
|
||
"""健康度报告模板 - 专门用于车联网工单健康度分析"""
|
||
|
||
def __init__(self):
|
||
super().__init__(
|
||
name="车联网工单健康度报告",
|
||
description="全面分析车联网技术支持工单的健康状况，从多个维度评估工单处理效率和质量"
|
||
)
|
||
|
||
def build_steps(self, **kwargs) -> List[AnalysisStep]:
|
||
"""构建健康度报告的分析步骤"""
|
||
return [
|
||
AnalysisStep(
|
||
name="数据概览与质量检查",
|
||
description="检查数据完整性、缺失值、异常值等",
|
||
analysis_type="explore",
|
||
prompt="加载数据并进行质量检查，输出数据概况和潜在问题"
|
||
),
|
||
AnalysisStep(
|
||
name="工单总量分析",
|
||
description="统计总工单数、时间分布、趋势变化",
|
||
analysis_type="calculate",
|
||
prompt="计算总工单数，按时间维度统计工单量，绘制时间序列趋势图"
|
||
),
|
||
AnalysisStep(
|
||
name="车型维度分析",
|
||
description="分析不同车型的工单分布和问题特征",
|
||
analysis_type="visualize",
|
||
prompt="统计各车型工单数量，绘制车型分布饼图和柱状图，识别高风险车型"
|
||
),
|
||
AnalysisStep(
|
||
name="模块维度分析",
|
||
description="分析工单涉及的技术模块分布",
|
||
analysis_type="visualize",
|
||
prompt="统计各技术模块的工单量，绘制模块分布图，识别高频问题模块"
|
||
),
|
||
AnalysisStep(
|
||
name="功能维度分析",
|
||
description="分析具体功能点的问题分布",
|
||
analysis_type="visualize",
|
||
prompt="统计各功能的工单量，绘制TOP功能问题排行，分析功能稳定性"
|
||
),
|
||
AnalysisStep(
|
||
name="问题严重程度分析",
|
||
description="分析工单的严重程度分布",
|
||
analysis_type="visualize",
|
||
prompt="统计不同严重程度的工单比例，绘制严重程度分布图"
|
||
),
|
||
AnalysisStep(
|
||
name="处理时长分析",
|
||
description="分析工单处理时效性",
|
||
analysis_type="calculate",
|
||
prompt="计算平均处理时长、SLA达成率，识别超时工单，绘制时长分布图"
|
||
),
|
||
AnalysisStep(
|
||
name="责任人工作负载分析",
|
||
description="分析各责任人的工单负载和处理效率",
|
||
analysis_type="visualize",
|
||
prompt="统计各责任人的工单数和处理效率，绘制负载分布图，识别超负荷人员"
|
||
),
|
||
AnalysisStep(
|
||
name="来源渠道分析",
|
||
description="分析工单来源渠道分布",
|
||
analysis_type="visualize",
|
||
prompt="统计各来源渠道的工单量，绘制渠道分布图"
|
||
),
|
||
AnalysisStep(
|
||
name="高频问题深度分析",
|
||
description="识别并深入分析高频问题",
|
||
analysis_type="explore",
|
||
prompt="提取TOP10高频问题，分析问题原因、影响范围和解决方案"
|
||
),
|
||
AnalysisStep(
|
||
name="综合健康度评分",
|
||
description="基于多个维度计算综合健康度评分",
|
||
analysis_type="calculate",
|
||
prompt="综合考虑工单量、处理时长、问题严重度等指标，计算健康度评分"
|
||
),
|
||
AnalysisStep(
|
||
name="生成最终报告",
|
||
description="整合所有分析结果，生成完整报告",
|
||
analysis_type="report",
|
||
prompt="整合所有图表和分析结论，生成一份完整的车联网工单健康度报告"
|
||
)
|
||
]
|
||
|
||
|
||
class TrendAnalysisTemplate(AnalysisTemplate):
|
||
"""趋势分析模板"""
|
||
|
||
def __init__(self):
|
||
super().__init__(
|
||
name="时间序列趋势分析",
|
||
description="分析数据的时间趋势、季节性和周期性特征"
|
||
)
|
||
|
||
def build_steps(self, time_column: str = "日期", value_column: str = "数值", **kwargs) -> List[AnalysisStep]:
|
||
return [
|
||
AnalysisStep(
|
||
name="时间序列数据准备",
|
||
description="将数据转换为时间序列格式",
|
||
analysis_type="explore",
|
||
prompt=f"将 '{time_column}' 列转换为日期格式，按时间排序数据"
|
||
),
|
||
AnalysisStep(
|
||
name="趋势可视化",
|
||
description="绘制时间序列图",
|
||
analysis_type="visualize",
|
||
prompt=f"绘制 '{value_column}' 随 '{time_column}' 的变化趋势图，添加移动平均线"
|
||
),
|
||
AnalysisStep(
|
||
name="趋势分析",
|
||
description="识别上升、下降或平稳趋势",
|
||
analysis_type="calculate",
|
||
prompt="计算趋势线斜率，判断整体趋势方向和变化速率"
|
||
),
|
||
AnalysisStep(
|
||
name="季节性分析",
|
||
description="检测季节性模式",
|
||
analysis_type="visualize",
|
||
prompt="分析月度、季度等周期性模式，绘制季节性分解图"
|
||
),
|
||
AnalysisStep(
|
||
name="异常点检测",
|
||
description="识别时间序列中的异常点",
|
||
analysis_type="calculate",
|
||
prompt="使用统计方法检测时间序列中的异常值，标注在图表上"
|
||
)
|
||
]
|
||
|
||
|
||
class AnomalyDetectionTemplate(AnalysisTemplate):
|
||
"""异常检测模板"""
|
||
|
||
def __init__(self):
|
||
super().__init__(
|
||
name="异常值检测分析",
|
||
description="识别数据中的异常值和离群点"
|
||
)
|
||
|
||
def build_steps(self, **kwargs) -> List[AnalysisStep]:
|
||
return [
|
||
AnalysisStep(
|
||
name="数值列统计分析",
|
||
description="计算数值列的统计特征",
|
||
analysis_type="calculate",
|
||
prompt="计算所有数值列的均值、标准差、四分位数等统计量"
|
||
),
|
||
AnalysisStep(
|
||
name="箱线图可视化",
|
||
description="使用箱线图识别异常值",
|
||
analysis_type="visualize",
|
||
prompt="为每个数值列绘制箱线图，直观展示异常值分布"
|
||
),
|
||
AnalysisStep(
|
||
name="Z-Score异常检测",
|
||
description="使用Z-Score方法检测异常值",
|
||
analysis_type="calculate",
|
||
prompt="计算每个数值的Z-Score，标记|Z|>3的异常值"
|
||
),
|
||
AnalysisStep(
|
||
name="IQR异常检测",
|
||
description="使用四分位距方法检测异常值",
|
||
analysis_type="calculate",
|
||
prompt="使用IQR方法(Q1-1.5*IQR, Q3+1.5*IQR)检测异常值"
|
||
),
|
||
AnalysisStep(
|
||
name="异常值汇总报告",
|
||
description="整理所有检测到的异常值",
|
||
analysis_type="report",
|
||
prompt="汇总所有异常值，分析其特征和可能原因，提供处理建议"
|
||
)
|
||
]
|
||
|
||
|
||
class ComparisonAnalysisTemplate(AnalysisTemplate):
|
||
"""对比分析模板"""
|
||
|
||
def __init__(self):
|
||
super().__init__(
|
||
name="分组对比分析",
|
||
description="对比不同分组之间的差异和特征"
|
||
)
|
||
|
||
def build_steps(self, group_column: str = "分组", value_column: str = "数值", **kwargs) -> List[AnalysisStep]:
|
||
return [
|
||
AnalysisStep(
|
||
name="分组统计",
|
||
description="计算各组的统计指标",
|
||
analysis_type="calculate",
|
||
prompt=f"按 '{group_column}' 分组，计算 '{value_column}' 的均值、中位数、标准差"
|
||
),
|
||
AnalysisStep(
|
||
name="分组可视化对比",
|
||
description="绘制对比图表",
|
||
analysis_type="visualize",
|
||
prompt=f"绘制各组的柱状图和箱线图，直观对比差异"
|
||
),
|
||
AnalysisStep(
|
||
name="差异显著性检验",
|
||
description="统计检验组间差异",
|
||
analysis_type="calculate",
|
||
prompt="进行t检验或方差分析，判断组间差异是否显著"
|
||
),
|
||
AnalysisStep(
|
||
name="对比结论",
|
||
description="总结对比结果",
|
||
analysis_type="report",
|
||
prompt="总结各组特征、主要差异和业务洞察"
|
||
)
|
||
]
|
||
|
||
|
||
# 模板注册表
|
||
TEMPLATE_REGISTRY = {
|
||
"health_report": HealthReportTemplate,
|
||
"trend_analysis": TrendAnalysisTemplate,
|
||
"anomaly_detection": AnomalyDetectionTemplate,
|
||
"comparison": ComparisonAnalysisTemplate
|
||
}
|
||
|
||
|
||
def get_template(template_name: str) -> AnalysisTemplate:
|
||
"""获取分析模板"""
|
||
template_class = TEMPLATE_REGISTRY.get(template_name)
|
||
if template_class:
|
||
return template_class()
|
||
else:
|
||
raise ValueError(f"未找到模板: {template_name}。可用模板: {list(TEMPLATE_REGISTRY.keys())}")
|
||
|
||
|
||
def list_templates() -> List[Dict[str, str]]:
|
||
"""列出所有可用模板"""
|
||
templates = []
|
||
for name, template_class in TEMPLATE_REGISTRY.items():
|
||
template = template_class()
|
||
templates.append({
|
||
"name": name,
|
||
"display_name": template.name,
|
||
"description": template.description
|
||
})
|
||
return templates
|