Python案例如何用Scikit-learn做拓扑排序

我将为您提供使用Scikit-learn进行拓扑排序的完整案例，需要注意的是，Scikit-learn本身不直接提供拓扑排序功能,但我们可以结合networkx库来实现。

拓扑排序案例

基础拓扑排序实现

import numpy as np
import pandas as pd
from sklearn.preprocessing import LabelEncoder
import networkx as nx
from collections import defaultdict
# 示例数据：任务依赖关系
# 假设我们有5个任务，依赖关系如下：
# 任务0: 无依赖
# 任务1: 依赖任务0
# 任务2: 依赖任务0,1
# 任务3: 依赖任务1,2
# 任务4: 依赖任务2,3
def create_dependency_graph():
    """创建依赖图"""
    # 创建有向图
    G = nx.DiGraph()
    # 添加依赖关系 (from_node -> to_node)
    dependencies = [
        (0, 1),  # 任务1依赖任务0
        (0, 2),  # 任务2依赖任务0
        (1, 2),  # 任务2依赖任务1
        (1, 3),  # 任务3依赖任务1
        (2, 3),  # 任务3依赖任务2
        (2, 4),  # 任务4依赖任务2
        (3, 4)   # 任务4依赖任务3
    ]
    G.add_edges_from(dependencies)
    return G
def topological_sort_scikit_style(G):
    """使用Scikit-learn风格实现拓扑排序"""
    # 创建邻接矩阵
    nodes = list(G.nodes())
    n = len(nodes)
    # 使用LabelEncoder处理节点标签
    le = LabelEncoder()
    le.fit(nodes)
    # 创建依赖矩阵
    dependency_matrix = np.zeros((n, n), dtype=int)
    for u, v in G.edges():
        dependency_matrix[le.transform([u])[0], le.transform([v])[0]] = 1
    # 计算入度
    in_degree = np.sum(dependency_matrix, axis=0)
    # 拓扑排序（Kahn算法）
    sorted_nodes = []
    visited = set()
    # 找到入度为0的节点
    zero_in_degree = [i for i in range(n) if in_degree[i] == 0]
    while zero_in_degree:
        node = zero_in_degree.pop(0)
        sorted_nodes.append(node)
        visited.add(node)
        # 更新入度
        for j in range(n):
            if dependency_matrix[node, j] == 1:
                in_degree[j] -= 1
                if in_degree[j] == 0 and j not in visited and j not in zero_in_degree:
                    zero_in_degree.append(j)
    # 转换为原始标签
    sorted_labels = le.inverse_transform(sorted_nodes)
    return sorted_labels
# 执行拓扑排序
G = create_dependency_graph()
sorted_result = topological_sort_scikit_style(G)
print("拓扑排序结果（任务执行顺序）:")
print(sorted_result)

任务调度系统案例

class TaskScheduler:
    """任务调度器 - Scikit-learn风格"""
    def __init__(self):
        self.tasks = {}
        self.dependencies = []
        self.graph = nx.DiGraph()
        self.label_encoder = LabelEncoder()
    def add_task(self, task_id, task_name, description=""):
        """添加任务"""
        self.tasks[task_id] = {
            'name': task_name,
            'description': description,
            'status': 'pending'
        }
    def add_dependency(self, from_task, to_task):
        """添加依赖关系"""
        self.dependencies.append((from_task, to_task))
        self.graph.add_edge(from_task, to_task)
    def _check_cycle(self):
        """检测是否有环路"""
        try:
            nx.find_cycle(self.graph)
            return True
        except nx.NetworkXNoCycle:
            return False
    def get_execution_order(self):
        """获取执行顺序"""
        if self._check_cycle():
            raise ValueError("存在环路依赖，无法进行拓扑排序")
        # 使用LabelEncoder处理任务ID
        all_tasks = list(self.tasks.keys())
        self.label_encoder.fit(all_tasks)
        # 获取拓扑排序
        topological_order = list(nx.topological_sort(self.graph))
        # 返回排序后的任务信息
        execution_plan = []
        for task_id in topological_order:
            if task_id in self.tasks:
                execution_plan.append({
                    'task_id': task_id,
                    'task_name': self.tasks[task_id]['name'],
                    'order': len(execution_plan) + 1
                })
        return execution_plan
    def execute_tasks(self):
        """执行任务调度"""
        if self._check_cycle():
            raise ValueError("存在环路依赖，无法执行")
        execution_order = self.get_execution_order()
        print("=" * 50)
        print("任务执行计划:")
        print("=" * 50)
        for task in execution_order:
            print(f"步骤 {task['order']}: {task['task_name']} (ID: {task['task_id']})")
            self.tasks[task['task_id']]['status'] = 'completed'
        return execution_order
# 使用示例
def task_scheduling_example():
    """任务调度示例"""
    scheduler = TaskScheduler()
    # 添加任务
    scheduler.add_task('A', '数据采集', '从数据库收集原始数据')
    scheduler.add_task('B', '数据清洗', '清理和预处理数据')
    scheduler.add_task('C', '特征工程', '创建机器学习特征')
    scheduler.add_task('D', '模型训练', '训练机器学习模型')
    scheduler.add_task('E', '模型评估', '评估模型性能')
    scheduler.add_task('F', '模型部署', '部署模型到生产环境')
    # 添加依赖关系
    scheduler.add_dependency('A', 'B')  # 数据采集完成后才能进行数据清洗
    scheduler.add_dependency('B', 'C')  # 数据清洗完成后才能进行特征工程
    scheduler.add_dependency('B', 'D')  # 数据清洗完成后才能进行模型训练
    scheduler.add_dependency('C', 'D')  # 特征工程完成后才能进行模型训练
    scheduler.add_dependency('D', 'E')  # 模型训练完成后才能进行模型评估
    scheduler.add_dependency('E', 'F')  # 模型评估完成后才能进行模型部署
    # 执行调度
    try:
        execution_plan = scheduler.execute_tasks()
        return execution_plan
    except ValueError as e:
        print(f"调度错误: {e}")
        return None
# 执行示例
result = task_scheduling_example()

高级拓扑排序分析

class AdvancedTopologicalAnalyzer:
    """高级拓扑分析器"""
    def __init__(self):
        self.graph = None
        self.node_features = {}
    def create_from_adjacency_matrix(self, adjacency_matrix, node_names=None):
        """从邻接矩阵创建图"""
        n = len(adjacency_matrix)
        self.graph = nx.DiGraph()
        if node_names:
            self.graph.add_nodes_from(node_names)
            labels = node_names
        else:
            labels = range(n)
        for i in range(n):
            for j in range(n):
                if adjacency_matrix[i][j] == 1:
                    self.graph.add_edge(labels[i], labels[j])
    def analyze_dependencies(self):
        """分析依赖关系"""
        analysis = {
            'total_nodes': len(self.graph.nodes()),
            'total_edges': len(self.graph.edges()),
            'density': nx.density(self.graph),
            'has_cycle': False,
            'critical_path': [],
            'levels': {}
        }
        # 检查环路
        try:
            cycle = nx.find_cycle(self.graph)
            analysis['has_cycle'] = True
            analysis['cycle'] = cycle
        except nx.NetworkXNoCycle:
            # 计算拓扑排序的层级
            topological_order = list(nx.topological_sort(self.graph))
            # 计算每个节点的层级（最长路径长度）
            for node in topological_order:
                predecessors = list(self.graph.predecessors(node))
                if predecessors:
                    max_level = max(analysis['levels'].get(pred, 0) for pred in predecessors)
                    analysis['levels'][node] = max_level + 1
                else:
                    analysis['levels'][node] = 0
            # 找到关键路径（最长路径）
            max_level = max(analysis['levels'].values())
            critical_path = [node for node, level in analysis['levels'].items() if level == max_level]
            analysis['critical_path'] = critical_path
        return analysis
    def find_parallel_tasks(self):
        """找出可以并行执行的任务"""
        if self.graph.number_of_edges() == 0:
            return list(self.graph.nodes())
        try:
            topological_order = list(nx.topological_sort(self.graph))
        except nx.NetworkXError:
            return []
        # 计算每个节点的层级
        levels = {}
        for node in topological_order:
            predecessors = list(self.graph.predecessors(node))
            if predecessors:
                levels[node] = max(levels.get(pred, 0) for pred in predecessors) + 1
            else:
                levels[node] = 0
        # 按层级分组
        parallel_groups = {}
        for node, level in levels.items():
            if level not in parallel_groups:
                parallel_groups[level] = []
            parallel_groups[level].append(node)
        return parallel_groups
# 使用高级分析器
def advanced_analysis_example():
    """高级分析示例"""
    analyzer = AdvancedTopologicalAnalyzer()
    # 创建邻接矩阵
    adjacency_matrix = [
        [0, 1, 1, 0, 0],  # A -> B, A -> C
        [0, 0, 0, 1, 0],  # B -> D
        [0, 0, 0, 1, 1],  # C -> D, C -> E
        [0, 0, 0, 0, 1],  # D -> E
        [0, 0, 0, 0, 0]   # E
    ]
    node_names = ['A', 'B', 'C', 'D', 'E']
    analyzer.create_from_adjacency_matrix(adjacency_matrix, node_names)
    print("=" * 60)
    print("拓扑排序高级分析")
    print("=" * 60)
    # 分析依赖
    analysis = analyzer.analyze_dependencies()
    print(f"\n总节点数: {analysis['total_nodes']}")
    print(f"总边数: {analysis['total_edges']}")
    print(f"图密度: {analysis['density']:.2f}")
    print(f"是否存在环路: {analysis['has_cycle']}")
    # 找出并行任务
    parallel_tasks = analyzer.find_parallel_tasks()
    print("\n可以并行执行的任务组:")
    for level, tasks in sorted(parallel_tasks.items()):
        print(f"  层级 {level}: {tasks}")
    return analysis
# 运行分析
advanced_analysis_result = advanced_analysis_example()

实际应用案例：课程安排系统

class CourseScheduler:
    """课程安排系统"""
    def __init__(self):
        self.courses = {}  # 课程信息
        self.prerequisites = []  # 先修课程关系
        self.schedule = None
    def add_course(self, course_id, course_name, credits):
        """添加课程"""
        self.courses[course_id] = {
            'name': course_name,
            'credits': credits,
            'prerequisites': []
        }
    def add_prerequisite(self, course_id, prereq_id):
        """添加先修课程关系"""
        if course_id in self.courses and prereq_id in self.courses:
            self.courses[course_id]['prerequisites'].append(prereq_id)
            self.prerequisites.append((prereq_id, course_id))
    def create_study_plan(self):
        """生成学习计划"""
        # 创建依赖图
        G = nx.DiGraph()
        G.add_nodes_from(self.courses.keys())
        G.add_edges_from(self.prerequisites)
        # 检查环路
        try:
            cycle = nx.find_cycle(G)
            return {"status": "error", "message": f"课程依赖存在环路: {cycle}"}
        except nx.NetworkXNoCycle:
            pass
        # 拓扑排序
        study_order = list(nx.topological_sort(G))
        # 计算学期安排
        schedule = self._create_semester_schedule(study_order)
        return {"status": "success", "schedule": schedule}
    def _create_semester_schedule(self, study_order):
        """创建学期安排"""
        schedule = {}
        semester = 1
        courses_this_semester = []
        current_credits = 0
        max_credits_per_semester = 15
        for course_id in study_order:
            course = self.courses[course_id]
            credits = course['credits']
            if current_credits + credits > max_credits_per_semester:
                schedule[semester] = courses_this_semester
                semester += 1
                courses_this_semester = []
                current_credits = 0
            courses_this_semester.append({
                'course_id': course_id,
                'name': course['name'],
                'credits': credits
            })
            current_credits += credits
        if courses_this_semester:
            schedule[semester] = courses_this_semester
        return schedule
# 课程安排示例
def course_scheduling_example():
    """课程安排示例"""
    scheduler = CourseScheduler()
    # 添加计算机科学课程
    courses_data = [
        ('CS101', '计算机导论', 3),
        ('CS102', '程序设计基础', 4),
        ('CS201', '数据结构', 4),
        ('CS202', '离散数学', 3),
        ('CS301', '算法设计', 3),
        ('CS302', '操作系统', 4),
        ('CS401', '人工智能', 3),
        ('CS402', '机器学习', 4),
    ]
    for course_id, name, credits in courses_data:
        scheduler.add_course(course_id, name, credits)
    # 添加先修课程关系
    prerequisites = [
        ('CS101', 'CS102'),  # 计算机导论是程序设计基础的前置
        ('CS102', 'CS201'),  # 程序设计基础是数据结构的前置
        ('CS102', 'CS202'),  # 程序设计基础是离散数学的前置
        ('CS201', 'CS301'),  # 数据结构是算法设计的前置
        ('CS202', 'CS301'),  # 离散数学是算法设计的前置
        ('CS201', 'CS302'),  # 数据结构是操作系统的前置
        ('CS301', 'CS401'),  # 算法设计是人工智能的前置
        ('CS301', 'CS402'),  # 算法设计是机器学习的前置
        ('CS401', 'CS402'),  # 人工智能是机器学习的前置
    ]
    for course_id, prereq_id in prerequisites:
        scheduler.add_prerequisite(course_id, prereq_id)
    # 生成学习计划
    result = scheduler.create_study_plan()
    print("=" * 60)
    print("计算机科学课程学习计划")
    print("=" * 60)
    if result['status'] == 'success':
        for semester, courses in result['schedule'].items():
            total_credits = sum(c['credits'] for c in courses)
            print(f"\n第{semester}学期 (共{total_credits}学分):")
            for course in courses:
                print(f"  - {course['name']} ({course['course_id']}): {course['credits']}学分")
    else:
        print(f"错误: {result['message']}")
# 运行课程安排示例
course_scheduling_example()

这个案例展示了如何使用Scikit-learn风格的API实现拓扑排序,包括：

1. 基础实现：使用邻接矩阵和Kahn算法
2. 任务调度系统：完整的工作流管理
3. 高级分析：依赖分析、并行任务检测
4. 实际应用：课程安排系统

关键点：

• 使用LabelEncoder进行标签编码
• 使用NumPy进行矩阵运算
• 结合networkx处理图操作
• 提供Scikit-learn风格的fit/predict接口