AI 驱动的项目风险预测：从历史数据到资源调度的智能决策

一、项目管理的"后视镜"困境：风险总是事后才看清

项目延期、预算超支、关键人员离职——这些风险在事后复盘时"显而易见"，但在事前却很难预判。传统风险管理依赖项目经理的经验判断，但人的认知偏差（乐观偏差、锚定效应）导致风险预估系统性偏低。更关键的是，风险因素之间相互关联——需求变更导致工期延长，工期延长导致人员疲劳，人员疲劳导致质量下降——这种连锁效应很难靠直觉预判。

AI 驱动的风险预测核心是"从历史数据中学习风险模式，用概率模型量化风险影响"。不是替代项目经理的判断，而是提供数据驱动的风险信号，让决策从"我觉得可能有问题"升级为"数据显示 72% 概率延期"。

二、风险预测模型架构

graph TB
    subgraph 数据采集
        A[项目历史数据<br/>工期/预算/变更记录]
        B[团队数据<br/>人员/技能/离职率]
        C[需求指标<br/>复杂度/变更频率]
    end

    subgraph 风险建模
        A --> D[特征工程<br/>提取风险因子]
        B --> D
        C --> D
        D --> E[概率模型<br/>贝叶斯网络/随机森林]
        E --> F[风险评分<br/>延期/超支/质量]
    end

    subgraph 决策支持
        F --> G[风险缓解建议<br/>资源调配/范围调整]
        F --> H[蒙特卡洛模拟<br/>工期概率分布]
    end

风险预测分三步：数据采集（项目历史、团队数据、需求指标）、风险建模（特征提取+概率模型）、决策支持（风险评分+缓解建议+蒙特卡洛模拟）。每一步都基于数据，而非直觉。

三、系统实现

3.1 风险因子提取

from dataclasses import dataclass
from typing import List, Dict

@dataclass
class ProjectFeatures:
    """项目风险特征"""
    # 需求维度
    requirement_count: int       # 需求数量
    requirement_change_rate: float  # 需求变更率
    cross_department_count: int  # 跨部门需求数

    # 团队维度
    team_size: int
    avg_tenure_months: float     # 平均在职时间
    key_person_ratio: float      # 关键人员占比
    new_member_ratio: float      # 新人占比

    # 技术维度
    tech_stack_novelty: float    # 技术栈新颖度 0-1
    integration_count: int       # 外部集成数
    legacy_code_ratio: float     # 遗留代码占比

    # 进度维度
    planned_duration_weeks: int
    parallel_task_ratio: float   # 并行任务占比
    milestone_count: int

class FeatureExtractor:
    """风险特征提取器"""

    def extract(self, project_data: dict) -> ProjectFeatures:
        """从原始项目数据中提取风险特征"""
        return ProjectFeatures(
            requirement_count=len(
                project_data.get('requirements', [])
            ),
            requirement_change_rate=self._calc_change_rate(
                project_data
            ),
            cross_department_count=self._count_cross_dept(
                project_data
            ),
            team_size=project_data.get('team_size', 0),
            avg_tenure_months=self._avg_tenure(project_data),
            key_person_ratio=self._key_person_ratio(
                project_data
            ),
            new_member_ratio=self._new_member_ratio(
                project_data
            ),
            tech_stack_novelty=self._tech_novelty(
                project_data
            ),
            integration_count=len(
                project_data.get('integrations', [])
            ),
            legacy_code_ratio=project_data.get(
                'legacy_ratio', 0.0
            ),
            planned_duration_weeks=project_data.get(
                'duration_weeks', 0
            ),
            parallel_task_ratio=self._parallel_ratio(
                project_data
            ),
            milestone_count=len(
                project_data.get('milestones', [])
            ),
        )

    def _calc_change_rate(self, data: dict) -> float:
        changes = data.get('requirement_changes', 0)
        total = len(data.get('requirements', []))
        return changes / total if total > 0 else 0.0

    def _count_cross_dept(self, data: dict) -> int:
        reqs = data.get('requirements', [])
        return sum(
            1 for r in reqs
            if len(r.get('departments', [])) > 1
        )

    def _avg_tenure(self, data: dict) -> float:
        members = data.get('team_members', [])
        if not members:
            return 0.0
        return sum(
            m.get('tenure_months', 0) for m in members
        ) / len(members)

    def _key_person_ratio(self, data: dict) -> float:
        members = data.get('team_members', [])
        if not members:
            return 0.0
        key_count = sum(
            1 for m in members if m.get('is_key', False)
        )
        return key_count / len(members)

    def _new_member_ratio(self, data: dict) -> float:
        members = data.get('team_members', [])
        if not members:
            return 0.0
        new_count = sum(
            1 for m in members
            if m.get('tenure_months', 0) < 6
        )
        return new_count / len(members)

    def _tech_novelty(self, data: dict) -> float:
        techs = data.get('tech_stack', [])
        if not techs:
            return 0.0
        novel = sum(1 for t in techs if t.get('is_new', False))
        return novel / len(techs)

    def _parallel_ratio(self, data: dict) -> float:
        tasks = data.get('tasks', [])
        if not tasks:
            return 0.0
        parallel = sum(
            1 for t in tasks if t.get('is_parallel', False)
        )
        return parallel / len(tasks)

3.2 风险评分模型

import numpy as np

class RiskScorer:
    """风险评分模型"""

    # 基于历史数据校准的权重
    RISK_WEIGHTS = {
        'delay': {  # 延期风险
            'requirement_change_rate': 0.25,
            'cross_department_count': 0.15,
            'key_person_ratio': 0.15,
            'new_member_ratio': 0.10,
            'tech_stack_novelty': 0.15,
            'parallel_task_ratio': 0.10,
            'legacy_code_ratio': 0.10,
        },
        'budget': {  # 超支风险
            'requirement_change_rate': 0.20,
            'integration_count': 0.20,
            'tech_stack_novelty': 0.20,
            'team_size': 0.15,
            'planned_duration_weeks': 0.15,
            'legacy_code_ratio': 0.10,
        },
        'quality': {  # 质量风险
            'new_member_ratio': 0.25,
            'requirement_change_rate': 0.20,
            'legacy_code_ratio': 0.20,
            'parallel_task_ratio': 0.15,
            'tech_stack_novelty': 0.10,
            'key_person_ratio': 0.10,
        },
    }

    def score(
        self, features: ProjectFeatures
    ) -> Dict[str, float]:
        """计算多维度风险评分"""
        feature_values = {
            'requirement_change_rate': features.requirement_change_rate,
            'cross_department_count': min(features.cross_department_count / 10, 1.0),
            'key_person_ratio': features.key_person_ratio,
            'new_member_ratio': features.new_member_ratio,
            'tech_stack_novelty': features.tech_stack_novelty,
            'parallel_task_ratio': features.parallel_task_ratio,
            'legacy_code_ratio': features.legacy_code_ratio,
            'integration_count': min(features.integration_count / 5, 1.0),
            'team_size': min(features.team_size / 20, 1.0),
            'planned_duration_weeks': min(features.planned_duration_weeks / 26, 1.0),
        }

        scores = {}
        for risk_type, weights in self.RISK_WEIGHTS.items():
            score = sum(
                weights.get(k, 0) * feature_values.get(k, 0)
                for k in weights
            )
            scores[risk_type] = min(score, 1.0)

        return scores

    def risk_level(self, score: float) -> str:
        """风险等级"""
        if score > 0.7:
            return '高'
        elif score > 0.4:
            return '中'
        else:
            return '低'

3.3 蒙特卡洛工期模拟

class MonteCarloScheduler:
    """蒙特卡洛工期模拟"""

    def simulate(
        self, tasks: List[dict], iterations: int = 1000
    ) -> dict:
        """模拟项目工期分布"""
        durations = []

        for _ in range(iterations):
            total = 0
            for task in tasks:
                # 三点估算法：乐观、最可能、悲观
                optimistic = task['optimistic_weeks']
                most_likely = task['most_likely_weeks']
                pessimistic = task['pessimistic_weeks']

                # PERT 分布采样
                sample = self._pert_sample(
                    optimistic, most_likely, pessimistic
                )
                total += sample

            durations.append(total)

        durations = sorted(durations)

        return {
            'p50': durations[int(len(durations) * 0.5)],
            'p70': durations[int(len(durations) * 0.7)],
            'p90': durations[int(len(durations) * 0.9)],
            'p95': durations[int(len(durations) * 0.95)],
            'mean': np.mean(durations),
            'std': np.std(durations),
        }

    def _pert_sample(
        self, o: float, m: float, p: float
    ) -> float:
        """PERT 分布采样"""
        mean = (o + 4 * m + p) / 6
        std = (p - o) / 6
        return max(o, np.random.normal(mean, std))

四、风险预测的 Trade-offs 分析

数据质量依赖：风险模型的准确性高度依赖历史数据的质量和数量。数据不足时（创业公司项目少），模型容易过拟合。建议至少积累 20+ 个项目数据后再启用预测模型，数据不足时使用规则引擎（阈值告警）替代。

可解释性 vs. 准确性：复杂模型（随机森林、神经网络）准确率高但可解释性差，项目经理难以理解"为什么模型认为延期概率 72%"。贝叶斯网络的可解释性更好（因果图直观），但准确率略低。建议优先使用可解释模型，让决策者信任并采纳预测结果。

风险缓解的成本：每个风险信号都对应一个缓解措施，但缓解措施本身有成本（增加测试、增加人手、缩减范围）。不是所有风险都需要缓解——低概率低影响的风险可以接受，高概率高影响的风险必须缓解。风险矩阵（概率×影响）是决策的辅助工具。

预测的自我实现：如果团队知道"模型预测延期概率 72%"，可能产生两种效应——积极效应（提前行动避免延期）或消极效应（既然大概率延期，不如躺平）。需要配合激励机制，确保预测促进行动而非消极等待。

五、总结

AI 驱动的项目风险预测，核心是"从历史数据中学习风险模式，用概率模型量化风险影响"。特征提取将项目数据转化为风险因子，评分模型计算多维度风险概率，蒙特卡洛模拟给出工期分布。数据驱动的风险信号比直觉判断更客观、更可量化。

落地建议：先积累项目历史数据（至少 20 个项目），然后实现规则引擎（阈值告警）作为基线，再逐步引入机器学习模型。风险预测结果配合风险矩阵使用，高概率高影响的风险优先缓解。全程监控预测准确率，持续优化特征和模型。