培训需求分析

专栏：人工智能训练师（三级）备考全攻略
模块：卷三·知识体系 — 第六部分·培训与指导（第1篇）
难度：⭐⭐⭐☆☆
考试权重：中频（选择题 + 简答题）

---

一、为什么需要培训需求分析

没有需求分析的培训，就像没有诊断就开药方：

  场景A：团队AI能力参差
  ┌─────────────────────────────────────────────┐
  │ 领导：组织一个AI培训吧，大家都学点机器学习       │
  │                                              │
  │ 培训后：                                      │
  │  前端小哥：我学了5天神经网络，工作中还是写React  │
  │  算法工程师：讲的比我还浅，浪费时间              │
  │  产品经理：完全听不懂，不如告诉我怎么提需求       │
  │                                              │
  │ 结论：3天培训费2万，收获≈愤怒×3                 │
  └─────────────────────────────────────────────┘

  场景B：经过需求分析的培训
  ┌─────────────────────────────────────────────┐
  │ 需求分析 → 分层画像 → 定制内容 → 按需交付       │
  │                                              │
  │ 前端小哥 → 《如何调用模型API》2小时实战          │
  │ 算法工程师 → 《分布式训练优化》工作坊             │
  │ 产品经理 → 《AI产品从0到1》案例研讨              │
  │                                              │
  │ 结论：人均2小时有效投入，团队AI协作效率提升40%     │
  └─────────────────────────────────────────────┘

培训需求分析回答三个核心问题：
  ① 差距在哪？—— 现状 vs. 目标的能力差距
  ② 谁需要？  —— 不同角色需要不同的培训内容和深度
  ③ 怎么衡量？—— 培训后的效果如何评估

---

二、培训需求分析的三层模型

┌──────────────────────────────────────────────────────────────────┐
│                培训需求分析三层模型 (O-G-P 模型)                     │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  L1 — 组织层 (Organization Level)                                 │
│  ┌──────────────────────────────────────────────────────┐        │
│  │ 组织战略对AI能力的要求是什么？                           │        │
│  │ 当前AI人才储备能否支撑未来1-3年的业务发展？               │        │
│  │                                                       │        │
│  │ 分析工具：SWOT分析、人才盘点、战略解码                     │        │
│  │ 典型输出："公司计划在12个月内上线智能客服，需培养           │        │
│  │           3名NLP工程师 + 5名AI训练师"                   │        │
│  └──────────────────────────────────────────────────────┘        │
│  ⇩                                                                 │
│  L2 — 任务层 (Task/Job Level)                                     │
│  ┌──────────────────────────────────────────────────────┐        │
│  │ 具体岗位的工作任务需要哪些AI相关知识和技能？              │        │
│  │ 任务频率和重要性如何排序？                               │        │
│  │                                                       │        │
│  │ 分析工具：工作分析问卷、关键事件法、DACUM法              │        │
│  │ 典型输出："数据标注员核心任务Top5及对应技能矩阵"          │        │
│  └──────────────────────────────────────────────────────┘        │
│  ⇩                                                                 │
│  L3 — 个人层 (Individual Level)                                    │
│  ┌──────────────────────────────────────────────────────┐        │
│  │ 某个具体员工的当前能力水平与岗位要求之间的差距？         │        │
│  │ 个人的学习风格偏好是什么？                               │        │
│  │                                                       │        │
│  │ 分析工具：能力测评、360度反馈、个人发展计划(IDP)         │        │
│  │ 典型输出："张三：Python数据分析能力从L2提升至L3"         │        │
│  └──────────────────────────────────────────────────────┘        │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

---

三、AI岗位能力矩阵

3.1 典型AI团队角色与能力需求

角色	核心AI能力需求	深度要求	培训优先级	典型培训时长
AI产品经理	AI技术边界理解、数据思维、效果评估	L2-理解	⭐⭐⭐	3~5天
数据标注员	标注规范、质检标准、标注工具使用	L1-操作	⭐⭐⭐⭐⭐	1~2周
AI训练师	数据处理、模型评测、参数调优、Prompt工程	L3-熟练	⭐⭐⭐⭐⭐	2~4周
算法工程师	模型架构、分布式训练、推理优化	L4-精通	⭐⭐⭐⭐	持续学习
前端开发	API调用、推理结果展示、流式处理	L1-操作	⭐⭐	1~2天
运维工程师	模型部署、GPU集群管理、监控告警	L3-熟练	⭐⭐⭐	3~5天
业务方/管理层	AI价值理解、投入产出评估、风险认知	L1-了解	⭐⭐⭐	0.5~2天

3.2 能力水平分级标准

能力分级参考（类似于国家职业资格五级制）:

L0 — 无接触
  从未接触过AI相关知识或工具

L1 — 了解/操作
  知道基本概念，能按照SOP完成指定操作
  例：标注员按规范标注数据、前端调用模型API

L2 — 理解/应用
  能理解原理，独立完成常见任务，遇到简单问题能自行解决
  例：AI训练师独立完成一轮数据清洗和模型评测

L3 — 熟练/优化
  能优化现有流程，解决中等复杂问题，能指导L1-L2人员
  例：训练师优化标注规范，提升标注效率30%

L4 — 精通/创新
  能设计新方案，解决复杂问题，能带领团队
  例：算法工程师设计新模型架构

L5 — 专家/引领
  行业领先水平，能定义标准和方向

---

四、培训需求分析的核心方法

4.1 四种经典分析方法对比

方法	核心问题	数据来源	输出	适用场景
GAP分析	现状与目标差多少？	测评+岗位JD	能力差距矩阵	有明确的岗位能力标准
关键事件法	什么情况下会出问题？	事故复盘+案例	关键能力清单	安全/质量要求高的场景
DACUM法	这个工作具体做哪些事？	专家工作坊	职责-任务-技能树	新设立的AI岗位
问卷调研法	大家觉得需要学什么？	问卷+访谈	优先级排序	团队规模大、需求分散

4.2 GAP分析的完整流程

GAP分析五步法：

第1步：定义目标状态
  ┌─────────────────────────────────────┐
  │ 3个月内，团队需具备的能力：              │
  │  ✓ 能独立完成Prompt Engineering        │
  │  ✓ 能理解模型评估指标(AUC/F1/ACC)       │
  │  ✓ 能独立运行数据清洗脚本               │
  └─────────────────────────────────────┘

第2步：评估当前状态
  ┌─────────────────────────────────────┐
  │ 团队成员当前水平分布：                   │
  │  Prompt Eng:     L1 40% / L2 30% / L3 20% / L4 10%
  │  模型评估:       L1 60% / L2 25% / L3 10% / L4 5%
  │  数据清洗脚本:   L1 70% / L2 20% / L3 5%  / L4 5%
  └─────────────────────────────────────┘

第3步：量化差距
  目标：全团队达到L2以上
  差距 = 目标线以下的人数比例
  Prompt Eng:     目标L2+→缺口 40%（当前L1）
  模型评估:        目标L2+→缺口 60%（当前L1）
  数据清洗脚本:    目标L2+→缺口 70%（当前L1）← 最大缺口！

第4步：优先级排序
  数据清洗脚本 → 紧急（缺口大+基础能力）
  模型评估     → 重要（影响决策质量）
  Prompt Eng   → 一般（缺口较小）

第5步：制定培训计划
  针对数据清洗：全员实战培训（2天）
  针对模型评估：案例研讨 + 模拟题（1天）
  针对Prompt：  进阶工作坊（0.5天）

---

五、培训需求分析实战代码

5.1 能力差距分析器

"""
培训需求分析工具集 — 基于GAP分析的AI团队能力诊断系统。
"""

import json
import math
from typing import Dict, List, Optional, Tuple, Set
from dataclasses import dataclass, field
from enum import IntEnum
from collections import defaultdict


class SkillLevel(IntEnum):
    """能力等级"""
    L0 = 0  # 无接触
    L1 = 1  # 了解/操作
    L2 = 2  # 理解/应用
    L3 = 3  # 熟练/优化
    L4 = 4  # 精通/创新
    L5 = 5  # 专家/引领


@dataclass
class CompetencyItem:
    """单个能力项"""
    name: str                    # 能力名称
    category: str                # 分类（如"数据处理"/"模型训练"/"系统运维"）
    description: str             # 描述
    target_level: SkillLevel     # 目标等级
    importance: int              # 重要度 1~5
    current_levels: Dict[str, SkillLevel] = field(default_factory=dict)  # 人员→当前等级


@dataclass
class PersonProfile:
    """个人能力画像"""
    name: str
    role: str                    # 角色
    department: str = ""
    assessment_date: str = ""
    competencies: Dict[str, SkillLevel] = field(default_factory=dict)  # 能力项→等级
    
    def get_gap(self, target_map: Dict[str, SkillLevel]) -> Dict[str, int]:
        """计算个人与目标的差距"""
        gaps = {}
        for comp_id, target_level in target_map.items():
            current = self.competencies.get(comp_id, SkillLevel.L0)
            gap = target_level.value - current.value
            if gap > 0:
                gaps[comp_id] = gap
        return gaps  # 只返回有差距的项
    
    def total_gap_score(self, target_map: Dict[str, SkillLevel]) -> int:
        """总差距分数（用于排序）"""
        return sum(self.get_gap(target_map).values())


class CompetencyFramework:
    """能力框架 — 定义整个团队的AI能力标准"""
    
    def __init__(self, name: str = "AI团队能力框架"):
        self.name = name
        self.competencies: Dict[str, CompetencyItem] = {}
    
    def add_competency(
        self,
        comp_id: str,
        name: str,
        category: str,
        description: str,
        target_level: SkillLevel,
        importance: int = 3,
    ):
        """添加能力项"""
        self.competencies[comp_id] = CompetencyItem(
            name=name,
            category=category,
            description=description,
            target_level=target_level,
            importance=importance,
        )
    
    def get_target_map(self) -> Dict[str, SkillLevel]:
        """获取目标等级映射（用于计算差距）"""
        return {cid: c.target_level for cid, c in self.competencies.items()}
    
    def get_category_summary(self) -> Dict[str, List[str]]:
        """按分类汇总能力项"""
        summary = defaultdict(list)
        for cid, comp in self.competencies.items():
            summary[comp.category].append(cid)
        return dict(summary)


class TrainingNeedsAnalyzer:
    """培训需求分析器"""
    
    def __init__(self, framework: CompetencyFramework):
        self.framework = framework
        self.people: Dict[str, PersonProfile] = {}
    
    def add_person(self, profile: PersonProfile):
        """添加人员画像"""
        self.people[profile.name] = profile
    
    def bulk_add_from_assessment(
        self, 
        assessment_data: List[Dict]
    ):
        """
        批量导入测评数据。
        
        assessment_data格式：
        [
            {
                "name": "张三",
                "role": "AI训练师",
                "scores": {"data_cleaning": 2, "model_eval": 1, "prompt_eng": 3}
            }
        ]
        """
        for item in assessment_data:
            profile = PersonProfile(
                name=item["name"],
                role=item.get("role", ""),
                department=item.get("department", ""),
                assessment_date=item.get("date", ""),
            )
            for comp_id, level in item.get("scores", {}).items():
                if isinstance(level, int):
                    profile.competencies[comp_id] = SkillLevel(level)
                elif isinstance(level, SkillLevel):
                    profile.competencies[comp_id] = level
            
            self.add_person(profile)
    
    def compute_individual_gaps(self) -> Dict[str, Dict]:
        """
        计算每个人的能力差距。
        
        Returns:
            {name: {gap_details, total_gap, top_gaps}}
        """
        target_map = self.framework.get_target_map()
        results = {}
        
        for name, person in self.people.items():
            gaps = person.get_gap(target_map)
            total = person.total_gap_score(target_map)
            
            # Top-3最大差距
            sorted_gaps = sorted(gaps.items(), key=lambda x: x[1], reverse=True)
            top_gaps = sorted_gaps[:3]
            
            # 翻译为可读格式
            top_gap_details = []
            for comp_id, gap_levels in top_gaps:
                comp = self.framework.competencies.get(comp_id)
                comp_name = comp.name if comp else comp_id
                current = person.competencies.get(comp_id, SkillLevel.L0)
                target = target_map[comp_id]
                top_gap_details.append({
                    "competency": comp_name,
                    "current": current.name,
                    "target": target.name,
                    "gap": gap_levels,
                })
            
            results[name] = {
                "role": person.role,
                "total_gap": total,
                "gap_count": len(gaps),
                "top_gaps": top_gap_details,
            }
        
        return results
    
    def compute_competency_heatmap(self) -> Dict:
        """
        计算团队能力热力图。
        
        Returns:
            {competency_id: {avg_level, target, coverage, risk_level}}
        """
        target_map = self.framework.get_target_map()
        heatmap = {}
        
        for comp_id, comp in self.framework.competencies.items():
            levels = []
            for person in self.people.values():
                levels.append(
                    person.competencies.get(comp_id, SkillLevel.L0).value
                )
            
            if levels:
                avg = sum(levels) / len(levels)
                target = target_map[comp_id].value
                gap = target - avg
                coverage = sum(1 for l in levels if l >= target) / len(levels)
                
                # 风险等级
                if gap > 2:
                    risk = "🔴 高风险"
                elif gap > 1:
                    risk = "🟡 中风险"
                elif gap > 0:
                    risk = "🟢 低风险"
                else:
                    risk = "✅ 达标"
                
                heatmap[comp_id] = {
                    "name": comp.name,
                    "category": comp.category,
                    "importance": comp.importance,
                    "avg_level": round(avg, 1),
                    "target_level": target,
                    "gap": round(gap, 1),
                    "coverage": f"{coverage*100:.0f}%",
                    "risk": risk,
                }
        
        return heatmap
    
    def generate_training_plan(self) -> Dict:
        """
        基于差距分析结果生成培训计划建议。
        """
        heatmap = self.compute_competency_heatmap()
        individual_gaps = self.compute_individual_gaps()
        
        # 按重要度加权排序能力项的紧迫性
        urgency_scores = []
        for comp_id, info in heatmap.items():
            score = info["gap"] * info["importance"]  # 加权分
            urgency_scores.append((comp_id, score, info))
        
        urgency_scores.sort(key=lambda x: x[1], reverse=True)
        
        # 生成课程建议
        courses = []
        for comp_id, score, info in urgency_scores[:8]:  # Top-8
            # 按gap推荐培训方式
            if info["gap"] > 2:
                method = "集中培训 + 实战演练"
                duration = "2~5天"
            elif info["gap"] > 1:
                method = "专题工作坊"
                duration = "1~2天"
            elif info["gap"] > 0:
                method = "自学 + 分享会"
                duration = "0.5天"
            else:
                continue  # 已达标，跳过
            
            # 需要参与的人员
            participants = []
            for name, gap_info in individual_gaps.items():
                for g in gap_info["top_gaps"]:
                    if g["competency"] == info["name"]:
                        participants.append({
                            "name": name,
                            "role": gap_info["role"],
                            "gap": g["gap"],
                        })
                        break
            
            courses.append({
                "priority": len(courses) + 1,
                "competency": info["name"],
                "category": info["category"],
                "urgency_score": score,
                "method": method,
                "duration": duration,
                "participants": participants,
                "risk": info["risk"],
            })
        
        # 生成分类汇总
        by_category = defaultdict(list)
        for c in courses:
            by_category[c["category"]].append(c["competency"])
        
        return {
            "framework_name": self.framework.name,
            "total_people": len(self.people),
            "overall_avg_gap": round(
                sum(info["gap"] for info in heatmap.values()) / max(len(heatmap), 1), 1
            ),
            "high_risk_count": sum(
                1 for info in heatmap.values() if "高风险" in info["risk"]
            ),
            "courses": courses,
            "by_category": dict(by_category),
        }
    
    def export_report_markdown(self) -> str:
        """导出Markdown格式的分析报告"""
        plan = self.generate_training_plan()
        heatmap = self.compute_competency_heatmap()
        individual = self.compute_individual_gaps()
        
        md = f"# AI团队培训需求分析报告\n\n"
        md += f"**分析框架**: {plan['framework_name']}\n"
        md += f"**覆盖人数**: {plan['total_people']}人\n"
        md += f"**整体平均差距**: {plan['overall_avg_gap']}级\n"
        md += f"**高风险能力项**: {plan['high_risk_count']}个\n\n"
        
        md += "## 1. 团队能力热度\n\n"
        md += "| 能力项 | 分类 | 平均等级 | 目标等级 | 差距 | 达标率 | 风险 |\n"
        md += "|-------|------|---------|---------|------|-------|------|\n"
        for comp_id, info in sorted(
            heatmap.items(), 
            key=lambda x: x[1]["importance"] * x[1]["gap"], 
            reverse=True
        ):
            md += (
                f"| {info['name']} | {info['category']} | "
                f"L{info['avg_level']} | L{info['target_level']} | "
                f"{info['gap']}级 | {info['coverage']} | {info['risk']} |\n"
            )
        
        md += "\n## 2. 个人差距Top-3\n\n"
        for name, info in individual.items():
            md += f"### {name} [{info['role']}] — 总差距{info['total_gap']}级\n\n"
            for g in info['top_gaps']:
                md += (
                    f"- **{g['competency']}**: "
                    f"{g['current']} → {g['target']} "
                    f"(差距{g['gap']}级)\n"
                )
            md += "\n"
        
        md += "## 3. 建议培训课程\n\n"
        md += "| 优先级 | 能力项 | 紧急分 | 培训方式 | 时长 | 风险 |\n"
        md += "|-------|-------|-------|---------|------|------|\n"
        for c in plan['courses']:
            md += (
                f"| {c['priority']} | {c['competency']} | "
                f"{c['urgency_score']:.0f} | {c['method']} | "
                f"{c['duration']} | {c['risk']} |\n"
            )
        
        return md


# ── 使用示例 ──

if __name__ == "__main__":
    # Step 1: 定义AI训练师团队能力框架（对应国标六大模块）
    framework = CompetencyFramework("AI训练师团队能力标准v2.0")
    
    framework.add_competency(
        "data_cleaning", "数据清洗与预处理", "数据处理",
        "能使用Python进行数据清洗、缺失值处理、异常值检测",
        SkillLevel.L3, importance=5,
    )
    framework.add_competency(
        "data_labeling", "数据标注规范", "数据处理",
        "能制定标注规范、质检标准、协调标注团队",
        SkillLevel.L3, importance=4,
    )
    framework.add_competency(
        "model_eval", "模型评估与测试", "模型训练",
        "能独立设计评估方案、理解AUC/F1等指标、撰写测试报告",
        SkillLevel.L3, importance=5,
    )
    framework.add_competency(
        "prompt_eng", "Prompt工程", "模型训练",
        "能编写高质量Prompt模板、理解Few-shot/Chain-of-Thought技术",
        SkillLevel.L2, importance=3,
    )
    framework.add_competency(
        "model_deploy", "模型部署", "系统运维",
        "能使用Docker部署模型、配置推理服务、基本性能监控",
        SkillLevel.L2, importance=4,
    )
    framework.add_competency(
        "troubleshoot", "故障排查", "系统运维",
        "能定位常见AI系统故障、执行五层排查流程",
        SkillLevel.L2, importance=3,
    )
    framework.add_competency(
        "doc_writing", "技术文档撰写", "培训指导",
        "能撰写清晰的技术文档、SOP、培训材料",
        SkillLevel.L2, importance=3,
    )
    framework.add_competency(
        "team_train", "团队培训", "培训指导",
        "能分析培训需求、设计培训方案、评估培训效果",
        SkillLevel.L3, importance=3,
    )
    
    # Step 2: 导入团队成员能力测评数据
    assess_data = [
        {
            "name": "张三", "role": "AI训练师", "date": "2026-06-15",
            "scores": {
                "data_cleaning": 3, "data_labeling": 2,
                "model_eval": 1, "prompt_eng": 3,
                "model_deploy": 1, "troubleshoot": 1,
                "doc_writing": 2, "team_train": 0,
            }
        },
        {
            "name": "李四", "role": "AI训练师", "date": "2026-06-15",
            "scores": {
                "data_cleaning": 2, "data_labeling": 3,
                "model_eval": 2, "prompt_eng": 2,
                "model_deploy": 0, "troubleshoot": 0,
                "doc_writing": 1, "team_train": 1,
            }
        },
        {
            "name": "王五", "role": "初级AI训练师", "date": "2026-06-15",
            "scores": {
                "data_cleaning": 1, "data_labeling": 1,
                "model_eval": 0, "prompt_eng": 1,
                "model_deploy": 0, "troubleshoot": 0,
                "doc_writing": 0, "team_train": 0,
            }
        },
        {
            "name": "赵六", "role": "AI产品经理", "date": "2026-06-15",
            "scores": {
                "data_cleaning": 1, "data_labeling": 2,
                "model_eval": 1, "prompt_eng": 2,
                "model_deploy": 1, "troubleshoot": 0,
                "doc_writing": 3, "team_train": 2,
            }
        },
    ]
    
    # Step 3: 运行分析
    analyzer = TrainingNeedsAnalyzer(framework)
    analyzer.bulk_add_from_assessment(assess_data)
    
    # 生成报告
    report = analyzer.export_report_markdown()
    print(report)

5.2 培训需求调查问卷数据处理器

"""
培训需求调查问卷数据分析 — 从问卷原始数据到培训优先级排序。
"""

import json
from typing import Dict, List, Optional
from dataclasses import dataclass
from collections import Counter


@dataclass
class SurveyResponse:
    """单份问卷回复"""
    respondent_id: str
    role: str
    department: str
    # 各能力项的自评（1-5，1=完全不会，5=专家级）
    self_ratings: Dict[str, int]
    # 希望学习的内容（多选）
    learning_interests: List[str]
    # 偏好的学习方式
    preferred_method: str  # online/offline/hybrid/self_study
    # 每周可投入时间（小时）
    available_hours: float
    # 培训期望/建议
    expectations: str = ""


class SurveyAnalyzer:
    """问卷分析器"""
    
    def __init__(self):
        self.responses: List[SurveyResponse] = []
    
    def load_from_json(self, file_path: str):
        """从JSON文件加载问卷数据"""
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        
        for item in data:
            self.responses.append(SurveyResponse(
                respondent_id=item.get("id", ""),
                role=item.get("role", ""),
                department=item.get("department", ""),
                self_ratings=item.get("ratings", {}),
                learning_interests=item.get("interests", []),
                preferred_method=item.get("method", "online"),
                available_hours=item.get("hours", 4.0),
                expectations=item.get("expectations", ""),
            ))
        
        return len(self.responses)
    
    def analyze_skill_gaps(self, min_required: int = 3) -> Dict:
        """
        分析技能缺口 — 自评低于阈值的项目。
        
        Args:
            min_required: 最低要求等级（默认3=熟练）
        
        Returns:
            包含gap_summary和by_role的字典
        """
        # 汇总所有能力项
        all_skills = set()
        for r in self.responses:
            all_skills.update(r.self_ratings.keys())
        
        # 统计每个能力的缺口
        skill_gaps = {}
        for skill in sorted(all_skills):
            ratings = [
                r.self_ratings.get(skill, 0) 
                for r in self.responses
            ]
            avg = sum(ratings) / len(ratings) if ratings else 0
            below = sum(1 for r in ratings if r < min_required)
            pct_below = below / len(ratings) * 100 if ratings else 0
            
            skill_gaps[skill] = {
                "avg_rating": round(avg, 2),
                "min": min(ratings),
                "max": max(ratings),
                "below_threshold_pct": f"{pct_below:.0f}%",
                "below_count": below,
            }
        
        # 按角色分组
        by_role = {}
        for r in self.responses:
            if r.role not in by_role:
                by_role[r.role] = {}
            
            for skill in all_skills:
                if skill not in by_role[r.role]:
                    by_role[r.role][skill] = []
                by_role[r.role][skill].append(
                    r.self_ratings.get(skill, 0)
                )
        
        role_summary = {}
        for role, skills in by_role.items():
            role_summary[role] = {}
            for skill, ratings in skills.items():
                avg = sum(ratings) / len(ratings) if ratings else 0
                role_summary[role][skill] = {
                    "avg": round(avg, 2),
                    "count": len(ratings),
                }
        
        return {
            "total_respondents": len(self.responses),
            "skills_assessed": len(all_skills),
            "min_required": min_required,
            "skill_gaps": skill_gaps,
            "by_role": role_summary,
        }
    
    def analyze_learning_interests(self) -> Dict:
        """分析学习兴趣分布"""
        interest_counter = Counter()
        role_interests = defaultdict(Counter)
        
        for r in self.responses:
            interest_counter.update(r.learning_interests)
            role_interests[r.role].update(r.learning_interests)
        
        total = len(self.responses)
        
        # 兴趣排序
        ranked = []
        for topic, count in interest_counter.most_common():
            ranked.append({
                "topic": topic,
                "count": count,
                "pct": f"{count/total*100:.0f}%",
            })
        
        return {
            "ranked_interests": ranked,
            "top3": ranked[:3],
        }
    
    def analyze_learning_preferences(self) -> Dict:
        """分析学习偏好"""
        method_counter = Counter()
        for r in self.responses:
            method_counter[r.preferred_method] += 1
        
        total = len(self.responses)
        preferences = {}
        for method, count in method_counter.most_common():
            preferences[method] = {
                "count": count,
                "pct": f"{count/total*100:.0f}%",
            }
        
        avg_hours = sum(r.available_hours for r in self.responses) / total
        
        return {
            "methods": preferences,
            "avg_available_hours_per_week": round(avg_hours, 1),
        }
    
    def generate_priority_matrix(self) -> List[Dict]:
        """
        生成培训优先级矩阵。
        
        优先级 = 缺口(高) × 兴趣(高) → 最紧急培训
        """
        gaps = self.analyze_skill_gaps()
        interests = self.analyze_learning_interests()
        
        # 构建兴趣映射
        interest_counts = {}
        for item in interests["ranked_interests"]:
            interest_counts[item["topic"]] = item["count"]
        
        # 计算优先级
        priority_items = []
        max_count = max(interest_counts.values()) if interest_counts else 1
        
        for skill, gap_info in gaps["skill_gaps"].items():
            interest_count = interest_counts.get(skill, 0)
            
            # 标准化分数 (0~1)
            gap_score = gap_info["below_count"] / max(1, len(self.responses))
            interest_score = interest_count / max_count
            
            # 综合得分 = 缺口40% + 兴趣30% + 重要度30%（此处简化）
            composite = gap_score * 0.5 + interest_score * 0.5
            
            priority_items.append({
                "skill": skill,
                "gap_score": round(gap_score, 2),
                "interest_score": round(interest_score, 2),
                "composite": round(composite, 2),
                "below_pct": gap_info["below_threshold_pct"],
                "avg_rating": gap_info["avg_rating"],
            })
        
        # 排序
        priority_items.sort(key=lambda x: x["composite"], reverse=True)
        return priority_items


# ── 使用示例 ──

if __name__ == "__main__":
    analyzer = SurveyAnalyzer()
    
    # 模拟问卷调查数据
    survey_responses = [
        {
            "id": "R001", "role": "AI训练师", "department": "AI平台部",
            "ratings": {"Python编程": 3, "数据清洗": 4, "模型评估": 2, "部署运维": 1},
            "interests": ["模型评估", "Prompt工程", "部署运维"],
            "method": "offline", "hours": 5,
            "expectations": "希望有实战项目练习",
        },
        {
            "id": "R002", "role": "AI训练师", "department": "AI平台部",
            "ratings": {"Python编程": 2, "数据清洗": 3, "模型评估": 3, "部署运维": 1},
            "interests": ["部署运维", "数据清洗进阶", "团队培训方法"],
            "method": "hybrid", "hours": 3,
            "expectations": "希望学习MLOps完整流程",
        },
        {
            "id": "R003", "role": "初级训练师", "department": "AI平台部",
            "ratings": {"Python编程": 1, "数据清洗": 1, "模型评估": 0, "部署运维": 0},
            "interests": ["Python编程", "数据清洗", "模型评估"],
            "method": "offline", "hours": 8,
            "expectations": "从零开始系统学习",
        },
    ]
    
    # 分析
    for r in survey_responses:
        analyzer.responses.append(SurveyResponse(**r))
    
    gaps = analyzer.analyze_skill_gaps(min_required=3)
    interests = analyzer.analyze_learning_interests()
    preferences = analyzer.analyze_learning_preferences()
    priorities = analyzer.generate_priority_matrix()
    
    print("=== 技能缺口 ===")
    for skill, info in gaps["skill_gaps"].items():
        print(f"  {skill}: 平均{info['avg_rating']}/5, {info['below_threshold_pct']}不达标")
    
    print("\n=== 学习兴趣Top3 ===")
    for item in interests["top3"]:
        print(f"  {item['topic']}: {item['count']}人({item['pct']})")
    
    print("\n=== 培训优先级 ===")
    for item in priorities:
        flag = "🔴" if item["composite"] > 0.6 else "🟡" if item["composite"] > 0.3 else "🟢"
        print(f"  {flag} {item['skill']}: 综合={item['composite']} "
              f"(缺口={item['gap_score']}, 兴趣={item['interest_score']})")

---

六、培训需求分析的常见误区

❌ 误区1：只看自评，不做他评
  自评L3，工作中实际表现L1 — 自我认知偏差普遍存在
  ✅ 修正：自评 + 上级评价 + 实际工作产出验证（三角验证）

❌ 误区2：全员一刀切培训
  所有人都学同样的内容，有人太简单有人跟不上
  ✅ 修正：先分层（初级/中级/高级），再按需分配课程

❌ 误区3：只关注技术能力，忽略软技能
  AI训练师还需要沟通能力、文档能力、培训指导能力
  ✅ 修正：在能力框架中纳入软技能维度

❌ 误区4：培训完成后不评估
  学完就完了，不知道效果如何
  ✅ 修正：Kirkpatrick四级评估（反应→学习→行为→结果）

❌ 误区5：需求分析一做管一年
  业务变化快，AI技术迭代更快
  ✅ 修正：每季度更新一次需求分析，与项目周期对齐

---

七、考试要点速记

考点	内容	出题形式
OGP三层模型	组织层→任务层→个人层	选择题/排序题
GAP分析五步	目标→现状→差距→排序→计划	简答题
能力等级L0-L5	无接触→专家，每级定义	判断题
三角验证	自评+他评+产出验证	选择题
四种分析方法	GAP/关键事件/DACUM/问卷	选择题（场景匹配）
培训需求分析误区	一刀切/只看技术/不评估/不更新	判断题

---

八、思维导图