151_法规合规论文（论文中附有源码）：GDPR与AI Act - LLM应用的合规性检查框架与最佳实践

法规影响范围: GDPR(数据保护) → AI Act(风险分级) → 行业标准(实施指南)

合规组织架构:
管理层 → AI治理委员会 → 执行团队(技术/法律/隐私)

合规检查框架分层:
策略层(政策/标准) → 评估层(风险/审计) → 执行层(控制/监控)

# 风险评分计算逻辑示例
def calculate_risk_score(data_sensitivity, processing_risk, impact_score, mitigation_effectiveness):
    """
    计算LLM应用的综合风险评分
    
    参数:
    data_sensitivity: 数据敏感性评分 (1-5)
    processing_risk: 处理活动风险评分 (1-5)
    impact_score: 潜在影响评分 (1-5)
    mitigation_effectiveness: 缓解措施有效性 (1-5，5为最有效)
    
    返回:
    综合风险评分 (1-25)
    """
    ## 3. 数据合规管理

### 3.1 训练数据合规性

LLM训练数据的合规性是合规框架的基础。有效的数据合规管理包括：

- **数据来源审核**：验证训练数据的获取是否合法，是否符合许可要求
- **个人数据识别**：使用技术手段识别和处理训练数据中的个人信息
- **同意管理**：确保数据收集获得适当的用户同意，并维护同意记录
- **数据最小化**：只收集和使用实现特定目的所必需的数据

以下是一个训练数据合规性检查的示例实现：

```python
class TrainingDataComplianceChecker:
    """
    LLM训练数据合规性检查器
    """
    
    def __init__(self):
        self.data_sources = []
        self.compliance_issues = []
    
    def register_data_source(self, source_name, source_type, license_info, contains_pii=False):
        """
        注册训练数据源
        
        参数:
        source_name: 数据源名称
        source_type: 数据源类型 (公开/私有/第三方)
        license_info: 许可信息
        contains_pii: 是否包含个人身份信息
        """
        source = {
            'name': source_name,
            'type': source_type,
            'license': license_info,
            'contains_pii': contains_pii
        }
        self.data_sources.append(source)
    
    def check_data_compliance(self):
        """
        执行数据合规性检查
        """
        self.compliance_issues = []
        
        for source in self.data_sources:
            # 检查许可合规性
            if not self._check_license_compliance(source['license']):
                self.compliance_issues.append({
                    'source': source['name'],
                    'issue_type': 'license',
                    'severity': 'high',
                    'description': f"数据源 {source['name']} 的许可条款可能不允许用于LLM训练"
                })
            
            # 检查个人数据处理
            if source['contains_pii']:
                self.compliance_issues.append({
                    'source': source['name'],
                    'issue_type': 'pii',
                    'severity': 'medium',
                    'description': f"数据源 {source['name']} 包含个人信息，需要额外的数据保护措施"
                })
        
        return self.compliance_issues
    
    def _check_license_compliance(self, license_info):
        """
        检查许可合规性
        """
        # 简化的许可检查逻辑
        permissive_licenses = ['MIT', 'Apache', 'BSD', 'CC0']
        return any(license in license_info for license in permissive_licenses)

class DataSubjectRightsManager:
    """
    数据主体权利管理器
    """
    
    def __init__(self, data_storage, model_registry):
        self.data_storage = data_storage
        self.model_registry = model_registry
    
    def request_data_access(self, user_id):
        """
        处理数据访问请求
        """
        user_data = self.data_storage.get_user_data(user_id)
        return user_data
    
    def request_data_deletion(self, user_id):
        """
        处理数据删除请求
        """
        # 1. 从活跃存储中删除数据
        deletion_result = self.data_storage.delete_user_data(user_id)
        
        # 2. 记录删除请求，用于未来模型更新
        self.model_registry.record_data_deletion(user_id)
        
        # 3. 提供删除证明
        return {
            'status': 'processed',
            'deletion_time': datetime.now(),
            'affected_systems': ['active_storage', 'model_registry']
        }

class DataProtectionImpactAssessment:
    """
    数据保护影响评估工具
    """
    
    def __init__(self):
        self.processing_activities = []
        self.risks = []
        self.mitigation_measures = []
    
    def add_processing_activity(self, activity_name, purpose, data_categories, individuals_affected):
        """
        添加数据处理活动
        """
        activity = {
            'name': activity_name,
            'purpose': purpose,
            'data_categories': data_categories,
            'individuals_affected': individuals_affected
        }
        self.processing_activities.append(activity)
    
    def assess_risks(self):
        """
        评估处理活动的风险
        """
        self.risks = []
        
        for activity in self.processing_activities:
            risk_score = self._calculate_risk_score(activity)
            risk_level = self._determine_risk_level(risk_score)
            
            risk = {
                'activity': activity['name'],
                'risk_score': risk_score,
                'risk_level': risk_level,
                'description': f"处理活动 {activity['name']} 的风险评估结果"
            }
            self.risks.append(risk)
        
        return self.risks
    
    def _calculate_risk_score(self, activity):
        """
        计算风险评分
        """
        # 简化的风险计算逻辑
        data_sensitivity = len(activity['data_categories']) * 2
        scale = min(activity['individuals_affected'] / 1000, 5)
        return data_sensitivity * scale

class ModelTransparencyManager:
    """
    模型透明度管理器
    """
    
    def __init__(self):
        self.documentation = {}
        self.explanation_methods = []
    
    def document_model(self, model_name, architecture, training_data, evaluation_metrics):
        """
        记录模型信息
        """
        self.documentation = {
            'model_name': model_name,
            'architecture': architecture,
            'training_data': training_data,
            'evaluation_metrics': evaluation_metrics,
            'documentation_date': datetime.now()
        }
    
    def generate_technical_document(self):
        """
        生成技术文档
        """
        document = f"""# {self.documentation['model_name']} 技术文档
        
## 架构概述
{self.documentation['architecture']}

## 训练数据
{self.documentation['training_data']}

## 评估结果
{self.documentation['evaluation_metrics']}

## 文档生成日期
{self.documentation['documentation_date']}
"""
        return document

class FairnessTester:
    """
    LLM公平性测试工具
    """
    
    def __init__(self, model):
        self.model = model
        self.test_datasets = {}
        self.fairness_results = {}
    
    def load_test_dataset(self, dataset_name, dataset, protected_attributes):
        """
        加载测试数据集
        """
        self.test_datasets[dataset_name] = {
            'data': dataset,
            'protected_attributes': protected_attributes
        }
    
    def _test_by_group(self, dataset, protected_attr):
        """
        按保护属性分组测试
        """
        groups = {}
        # 按保护属性值分组
        for item in dataset:
            group_key = item.get(protected_attr, 'unknown')
            if group_key not in groups:
                groups[group_key] = []
            groups[group_key].append(item)
        
        # 对每个组进行测试
        group_results = {}
        for group, items in groups.items():
            # 这里简化实现，实际应运行完整的模型评估
            responses = []
            for item in items:
                response = self.model.generate(item['prompt'])
                responses.append({
                    'input': item['prompt'],
                    'output': response,
                    'ground_truth': item.get('ground_truth', None)
                })
            group_results[group] = responses
        
        return group_results
    
    def _calculate_fairness_metrics(self, group_results):
        """
        计算公平性指标
        """
        metrics = {}
        
        # 计算各组响应分布
        for attr, groups in group_results.items():
            group_sizes = {group: len(responses) for group, responses in groups.items()}
            
            # 简化的公平性指标计算
            metrics[attr] = {
                'group_distribution': group_sizes,
                'disparity_index': self._calculate_disparity_index(group_sizes)
            }
        
        return metrics
    
    def _calculate_disparity_index(self, group_sizes):
        """
        计算分布差异指数
        """
        values = list(group_sizes.values())
        if len(values) <= 1:
            return 0
        
        max_size = max(values)
        min_size = min(values)
        return (max_size - min_size) / max_size

合规架构层次:
用户层 → API层 → 服务层 → 存储层 → 基础设施层

class DataSecurityManager:
    """
    数据安全管理器
    """
    
    def __init__(self):
        self.encryption_keys = {}
        self.access_control_list = {}
    
    def set_encryption(self, data_type, encryption_method):
        """
        设置数据加密方法
        """
        self.encryption_keys[data_type] = {
            'method': encryption_method,
            'last_rotated': datetime.now()
        }
    
    def define_access_control(self, role, resource, permissions):
        """
        定义访问控制规则
        """
        if role not in self.access_control_list:
            self.access_control_list[role] = {}
        
        self.access_control_list[role][resource] = permissions
    
    def check_access(self, user_role, resource, action):
        """
        检查访问权限
        """
        if user_role not in self.access_control_list:
            return False
        
        if resource not in self.access_control_list[user_role]:
            return False
        
        return action in self.access_control_list[user_role][resource]
    
    def log_access_attempt(self, user_id, user_role, resource, action, success):
        """
        记录访问尝试
        """
        log_entry = {
            'timestamp': datetime.now(),
            'user_id': user_id,
            'user_role': user_role,
            'resource': resource,
            'action': action,
            'success': success
        }
        # 实际应用中应存储到安全的日志系统
        print(f"ACCESS LOG: {log_entry}")

class IncidentResponseManager:
    """
    事件响应管理器
    """
    
    def __init__(self):
        self.incidents = []
        self.response_templates = {}
    
    def report_incident(self, incident_type, severity, affected_data, potential_impact):
        """
        报告安全事件
        """
        incident = {
            'incident_id': f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}",
            'type': incident_type,
            'severity': severity,
            'affected_data': affected_data,
            'potential_impact': potential_impact,
            'reported_time': datetime.now(),
            'status': 'open',
            'actions_taken': []
        }
        
        self.incidents.append(incident)
        
        # 自动触发初步响应
        if severity == 'high':
            self._trigger_urgent_response(incident)
        
        return incident['incident_id']
    
    def _trigger_urgent_response(self, incident):
        """
        触发紧急响应
        """
        # 记录响应动作
        incident['actions_taken'].append({
            'action': 'initiate_urgent_response',
            'timestamp': datetime.now(),
            'details': '启动了高严重性事件的紧急响应流程'
        })
        
        # 检查是否需要通知监管机构（72小时内）
        if self._requires_regulatory_notification(incident):
            self.prepare_regulatory_notification(incident)
    
    def prepare_regulatory_notification(self, incident):
        """
        准备监管通知
        """
        notification = {
            'incident_id': incident['incident_id'],
            'notification_date': datetime.now(),
            'regulatory_authorities': ['数据保护监管机构'],
            'reporting_deadline': (datetime.now() + timedelta(days=3)).strftime('%Y-%m-%d'),
            'content': f"发现数据泄露事件：{incident['incident_id']}，影响{incident['affected_data']}"
        }
        
        # 记录通知准备
        incident['actions_taken'].append({
            'action': 'prepare_regulatory_notification',
            'timestamp': datetime.now(),
            'notification': notification
        })
        
        return notification

MVP系统架构:
[用户界面] ←→ [API接口层] ←→ [合规检查引擎]
                          ←→ [数据管理模块]
                          ←→ [模型评估模块]
                          ←→ [报告生成器]

# 主应用类
class ComplianceCheckerApp:
    """
    LLM合规检查应用主类
    """
    
    def __init__(self):
        self.data_checker = TrainingDataComplianceChecker()
        self.rights_manager = DataSubjectRightsManager(
            data_storage=DataStorage(),
            model_registry=ModelRegistry()
        )
        self.dpia_tool = DataProtectionImpactAssessment()
        self.transparency_manager = ModelTransparencyManager()
        self.security_manager = DataSecurityManager()
        self.incident_manager = IncidentResponseManager()
        self.report_generator = ComplianceReportGenerator()
    
    def run_compliance_check(self, model_info, data_info):
        """
        执行完整的合规检查流程
        """
        # 1. 数据合规性检查
        data_compliance_results = self._check_data_compliance(data_info)
        
        # 2. 执行DPIA评估
        dpia_results = self._perform_dpia(model_info, data_info)
        
        # 3. 模型透明度检查
        transparency_results = self._check_transparency(model_info)
        
        # 4. 安全控制评估
        security_results = self._assess_security_controls()
        
        # 5. 生成综合报告
        report = self.report_generator.generate_comprehensive_report(
            data_compliance=data_compliance_results,
            dpia=dpia_results,
            transparency=transparency_results,
            security=security_results
        )
        
        return report
    
    def _check_data_compliance(self, data_info):
        """
        检查数据合规性
        """
        # 注册数据源
        for source in data_info['sources']:
            self.data_checker.register_data_source(
                source_name=source['name'],
                source_type=source['type'],
                license_info=source['license'],
                contains_pii=source.get('contains_pii', False)
            )
        
        # 执行合规检查
        compliance_issues = self.data_checker.check_data_compliance()
        
        return {
            'status': 'compliant' if len(compliance_issues) == 0 else 'non_compliant',
            'issues': compliance_issues,
            'check_date': datetime.now()
        }
    
    def _perform_dpia(self, model_info, data_info):
        """
        执行数据保护影响评估
        """
        # 添加处理活动
        self.dpia_tool.add_processing_activity(
            activity_name=model_info['name'],
            purpose=model_info['purpose'],
            data_categories=data_info['data_categories'],
            individuals_affected=data_info.get('individuals_affected', 0)
        )
        
        # 执行风险评估
        risks = self.dpia_tool.assess_risks()
        
        return {
            'risks': risks,
            'assessment_date': datetime.now()
        }
    
    def _check_transparency(self, model_info):
        """
        检查模型透明度
        """
        # 记录模型信息
        self.transparency_manager.document_model(
            model_name=model_info['name'],
            architecture=model_info['architecture'],
            training_data=model_info['training_data'],
            evaluation_metrics=model_info['evaluation_metrics']
        )
        
        # 生成技术文档
        technical_doc = self.transparency_manager.generate_technical_document()
        
        return {
            'technical_document_available': True,
            'documentation_date': self.transparency_manager.documentation['documentation_date']
        }
    
    def _assess_security_controls(self):
        """
        评估安全控制措施
        """
        # 检查安全控制措施
        # 这里简化实现
        return {
            'encryption_status': 'implemented',
            'access_control_status': 'implemented',
            'audit_logging_status': 'implemented'
        }

# 数据存储类
class DataStorage:
    """
    数据存储接口
    """
    
    def get_user_data(self, user_id):
        """
        获取用户数据
        """
        # 简化实现
        return {'user_id': user_id, 'data': 'sample_data'}
    
    def delete_user_data(self, user_id):
        """
        删除用户数据
        """
        # 简化实现
        return {'status': 'deleted', 'user_id': user_id}

# 模型注册表
class ModelRegistry:
    """
    模型注册表
    """
    
    def __init__(self):
        self.models = {}
        self.deletion_requests = []
    
    def register_model(self, model_id, model_info):
        """
        注册模型
        """
        self.models[model_id] = model_info
    
    def record_data_deletion(self, user_id):
        """
        记录数据删除请求
        """
        self.deletion_requests.append({
            'user_id': user_id,
            'request_time': datetime.now()
        })

# 合规报告生成器
class ComplianceReportGenerator:
    """
    合规报告生成器
    """
    
    def generate_comprehensive_report(self, data_compliance, dpia, transparency, security):
        """
        生成综合合规报告
        """
        report = {
            'report_id': f"COMP-{datetime.now().strftime('%Y%m%d%H%M%S')}",
            'generation_date': datetime.now(),
            'summary': self._generate_summary(data_compliance, dpia, transparency, security),
            'sections': {
                'data_compliance': data_compliance,
                'dpia': dpia,
                'transparency': transparency,
                'security': security
            },
            'recommendations': self._generate_recommendations(data_compliance, dpia, transparency, security)
        }
        
        return report
    
    def _generate_summary(self, data_compliance, dpia, transparency, security):
        """
        生成报告摘要
        """
        # 简化实现
        high_risks = [r for r in dpia['risks'] if r['risk_level'] == 'high']
        
        return {
            'overall_status': 'compliant' if data_compliance['status'] == 'compliant' and len(high_risks) == 0 else 'non_compliant',
            'high_risks_count': len(high_risks),
            'compliance_issues_count': len(data_compliance['issues'])
        }
    
    def _generate_recommendations(self, data_compliance, dpia, transparency, security):
        """
        生成改进建议
        """
        recommendations = []
        
        # 基于数据合规问题生成建议
        if data_compliance['status'] == 'non_compliant':
            for issue in data_compliance['issues']:
                recommendations.append({
                    'area': 'data_compliance',
                    'severity': issue['severity'],
                    'description': f"修复数据源 {issue['source']} 的 {issue['issue_type']} 问题: {issue['description']}"
                })
        
        # 基于DPIA风险生成建议
        high_risks = [r for r in dpia['risks'] if r['risk_level'] == 'high']
        for risk in high_risks:
            recommendations.append({
                'area': 'risk_mitigation',
                'severity': 'high',
                'description': f"实施缓解措施以降低 {risk['activity']} 的高风险"
            })
        
        return recommendations

# 使用示例
if __name__ == "__main__":
    # 初始化应用
    app = ComplianceCheckerApp()
    
    # 定义模型信息
    model_info = {
        'name': '客户支持LLM',
        'purpose': '提供客户服务自动化回复',
        'architecture': '基于开源LLaMA模型微调',
        'training_data': '公开对话数据集 + 脱敏客户对话',
        'evaluation_metrics': {
            'accuracy': 0.85,
            'f1_score': 0.82,
            'fairness_score': 0.90
        }
    }
    
    # 定义数据信息
    data_info = {
        'sources': [
            {
                'name': '公开对话数据集',
                'type': '公开',
                'license': 'CC0',
                'contains_pii': False
            },
            {
                'name': '脱敏客户对话',
                'type': '内部',
                'license': '公司所有',
                'contains_pii': True
            }
        ],
        'data_categories': ['客户查询', '个人信息', '对话历史'],
        'individuals_affected': 10000
    }
    
    # 运行合规检查
    report = app.run_compliance_check(model_info, data_info)
    
    # 输出报告摘要
    print("合规检查报告摘要:")
    print(f"报告ID: {report['report_id']}")
    print(f"生成日期: {report['generation_date']}")
    print(f"整体状态: {report['summary']['overall_status']}")
    print(f"高风险数量: {report['summary']['high_risks_count']}")
    print(f"合规问题数量: {report['summary']['compliance_issues_count']}")
    
    # 输出建议
    print("\n改进建议:")
    for i, rec in enumerate(report['recommendations'], 1):
        print(f"{i}. [{rec['severity']}] {rec['description']}")

class ComplianceAutomationSystem:
    """
    合规自动化系统
    """
    
    def __init__(self, compliance_app):
        self.compliance_app = compliance_app
        self.schedules = []
        self.monitoring_rules = []
    
    def schedule_compliance_check(self, name, frequency, model_info, data_info):
        """
        安排定期合规检查
        """
        schedule = {
            'name': name,
            'frequency': frequency,  # daily, weekly, monthly
            'model_info': model_info,
            'data_info': data_info,
            'last_run': None,
            'next_run': self._calculate_next_run(frequency)
        }
        
        self.schedules.append(schedule)
        return schedule
    
    def define_monitoring_rule(self, metric, threshold, action):
        """
        定义监控规则
        """
        rule = {
            'metric': metric,
            'threshold': threshold,
            'action': action
        }
        
        self.monitoring_rules.append(rule)
        return rule
    
    def run_scheduled_checks(self):
        """
        执行到期的计划检查
        """
        current_time = datetime.now()
        results = []
        
        for schedule in self.schedules:
            if schedule['next_run'] <= current_time:
                # 运行合规检查
                report = self.compliance_app.run_compliance_check(
                    schedule['model_info'],
                    schedule['data_info']
                )
                
                # 更新计划状态
                schedule['last_run'] = current_time
                schedule['next_run'] = self._calculate_next_run(schedule['frequency'])
                
                # 处理结果
                self._process_check_result(report)
                results.append(report)
        
        return results
    
    def _calculate_next_run(self, frequency):
        """
        计算下次运行时间
        """
        current = datetime.now()
        if frequency == 'daily':
            return current + timedelta(days=1)
        elif frequency == 'weekly':
            return current + timedelta(weeks=1)
        elif frequency == 'monthly':
            # 简化实现
            return current + timedelta(days=30)
        return current
    
    def _process_check_result(self, report):
        """
        处理检查结果
        """
        # 检查是否需要告警
        for rule in self.monitoring_rules:
            if self._check_rule_violation(report, rule):
                self._trigger_action(rule['action'], report)
    
    def _check_rule_violation(self, report, rule):
        """
        检查是否违反规则
        """
        if rule['metric'] == 'high_risks_count':
            return report['summary']['high_risks_count'] > rule['threshold']
        elif rule['metric'] == 'compliance_issues_count':
            return report['summary']['compliance_issues_count'] > rule['threshold']
        return False
    
### 7.2 合规检查自动化示例

```python
# 自动化示例
if __name__ == "__main__":
    # 初始化合规应用
    app = ComplianceCheckerApp()
    
    # 初始化自动化系统
    automation = ComplianceAutomationSystem(app)
    
    # 定义模型和数据信息
    model_info = {
        'name': '金融服务LLM',
        'purpose': '提供金融建议和分析',
        'architecture': '自定义Transformer架构',
        'training_data': '匿名化金融数据集 + 公开金融文本',
        'evaluation_metrics': {
            'accuracy': 0.88,
            'f1_score': 0.85,
            'fairness_score': 0.92
        }
    }
    
    data_info = {
        'sources': [
            {
                'name': '匿名化金融数据集',
                'type': '内部',
                'license': '公司所有',
                'contains_pii': False
            },
            {
                'name': '公开金融文本',
                'type': '公开',
                'license': 'MIT',
                'contains_pii': False
            }
        ],
        'data_categories': ['财务数据', '市场分析', '用户行为'],
        'individuals_affected': 50000
    }
    
    # 安排每周合规检查
    automation.schedule_compliance_check(
        name='金融服务LLM每周合规检查',
        frequency='weekly',
        model_info=model_info,
        data_info=data_info
    )
    
    # 定义监控规则
    # 高风险超过1个时发送告警
    automation.define_monitoring_rule(
        metric='high_risks_count',
        threshold=1,
        action='send_alert'
    )
    
    # 合规问题超过5个时生成详细报告
    automation.define_monitoring_rule(
        metric='compliance_issues_count',
        threshold=5,
        action='generate_report'
    )
    
    # 模拟运行计划检查
    results = automation.run_scheduled_checks()
    print(f"执行了 {len(results)} 次合规检查")

class ComplianceMaturityModel:
    """
    合规成熟度评估模型
    """
    
    def __init__(self):
        self.maturity_levels = {
            1: '初始级',  # 被动响应合规要求
            2: '可重复级',  # 基本流程已建立
            3: '已定义级',  # 标准文档化流程
            4: '可管理级',  # 有监控和度量
            5: '优化级'     # 持续改进
        }
    
    def assess_maturity(self, organization_info):
        """
        评估组织合规成熟度
        """
        # 评估维度
        dimensions = {
            'data_management': self._assess_data_management(organization_info),
            'transparency': self._assess_transparency(organization_info),
            'security': self._assess_security(organization_info),
            'processes': self._assess_processes(organization_info),
            'governance': self._assess_governance(organization_info)
        }
        
        # 计算总体成熟度级别
        overall_level = sum(dimensions.values()) / len(dimensions)
        
        return {
            'overall_maturity': overall_level,
            'overall_level': self.maturity_levels[round(overall_level)],
            'dimension_maturity': dimensions,
            'recommendations': self._generate_recommendations(dimensions)
        }
    
    def _assess_data_management(self, info):
        """评估数据管理成熟度"""
        # 简化实现
        score = 1  # 初始级
        if info.get('data_catalog_exists', False):
            score += 1
        if info.get('data_minimization_policy', False):
            score += 1
        if info.get('automated_pii_detection', False):
            score += 1
        if info.get('continuous_improvement', False):
            score += 1
        return min(5, score)
    
    def _assess_transparency(self, info):
        """评估透明度成熟度"""
        # 简化实现
        score = 1
        if info.get('basic_documentation', False):
            score += 1
        if info.get('model_cards', False):
            score += 1
        if info.get('explainability_tools', False):
            score += 1
        if info.get('comprehensive_audit_trail', False):
            score += 1
        return min(5, score)
    
    def _assess_security(self, info):
        """评估安全性成熟度"""
        # 简化实现
        score = 1
        if info.get('basic_encryption', False):
            score += 1
        if info.get('access_control', False):
            score += 1
        if info.get('regular_security_audits', False):
            score += 1
        if info.get('advanced_threat_detection', False):
            score += 1
        return min(5, score)
    
    def _assess_processes(self, info):
        """评估流程成熟度"""
        # 简化实现
        score = 1
        if info.get('ad_hoc_processes', False):
            score += 1
        if info.get('documented_processes', False):
            score += 1
        if info.get('standardized_processes', False):
            score += 1
        if info.get('optimized_processes', False):
            score += 1
        return min(5, score)
    
    def _assess_governance(self, info):
        """评估治理成熟度"""
        # 简化实现
        score = 1
        if info.get('compliance_awareness', False):
            score += 1
        if info.get('designated_responsibility', False):
            score += 1
        if info.get('governance_framework', False):
            score += 1
        if info.get('board_involvement', False):
            score += 1
        return min(5, score)
    
    def _generate_recommendations(self, dimensions):
        """生成改进建议"""
        recommendations = []
        
        for dimension, score in dimensions.items():
            if score < 3:
                recommendations.append(f"改进 {dimension} 流程，建立标准化文档")
            elif score < 4:
                recommendations.append(f"增强 {dimension} 的监控和度量能力")
            elif score < 5:
                recommendations.append(f"优化 {dimension} 流程，实现持续改进")
        
        return recommendations