Elephant 在「实现拓扑排序任务调度器」的评测结果

这是该 AI 模型在此测试用例上的详细评测结果。

基本信息

  • 模型名称：Elephant
  • 用例名称：实现拓扑排序任务调度器
  • 测试类型：文本生成
  • 评测维度：代码生成

系统提示词（System Prompt）

这是给 AI 模型的背景设定和角色指令：

你是一名资深后端算法工程师，专注于图算法与任务调度领域。 回答要求： 1. 代码实现清晰简洁，优先保证逻辑正确性，使用 Python 或其他主流语言均可 2. 必须包含完整可运行的代码，并附有简要的算法思路说明 3. 对关键步骤（如入度计算、队列处理、环路检测）添加注释 4. 提供至少两个测试用例验证：一个正常有序图，一个含环图 5. 输出结果需明确区分「成功排序」与「存在循环依赖」两种情况

用户提示词（User Prompt）

这是用户给 AI 模型的具体任务要求：

## 任务：实现基础拓扑排序算法 请实现一个拓扑排序函数，用于确定一组任务的合法执行顺序。 ### 背景说明 在项目管理中，某些任务必须在其他任务完成后才能开始（依赖关系）。 拓扑排序可以找到一个满足所有依赖关系的执行顺序。 若任务之间存在循环依赖（如 A 依赖 B，B 又依赖 A），则无法排序。 ### 输入格式 - `tasks`：任务 ID 列表，例如 `[「A」, 「B」, 「C」, 「D」]` - `dependencies`：依赖关系列表，每项为 `(先决任务, 后续任务)` 的元组， 表示「先决任务必须在后续任务之前完成」， 例如 `[(「A」, 「B」), (「A」, 「C」), (「B」, 「D」)]` 表示 A→B→D，A→C ### 输出格式 - 若排序成功：返回/打印一个合法的任务执行顺序列表，例如 `[「A」, 「B」, 「C」, 「D」]` - 若存在循环依赖：返回/打印明确的错误提示，例如 `「Error: 存在循环依赖，无法完成拓扑排序」` ### 功能要求 1. 使用 **Kahn 算法（BFS 入度法）** 或 **DFS 后序遍历法** 实现拓扑排序 2. 正确处理以下情况： - 标准有向无环图（DAG）：输出合法执行顺序 - 含环图：检测并报告循环依赖错误 - 孤立节点（无任何依赖关系的任务）：应包含在输出中 - 空任务列表：返回空列表 3. 函数签名建议：`topological_sort(tasks: list, dependencies: list) -> list` ### 测试用例 **用例 1（正常 DAG）**

评测结果

综合得分

该模型在此用例的得分为：94.0 分（满分 100 分），已通过

基础难度评测结果

  • 得分：94.0 分
  • 通过状态：已通过

模型生成产物

以下是该模型针对此任务的实际生成内容：

下面是使用 **Kahn 算法（BFS 入度法）** 实现的拓扑排序函数，满足所有要求，包括处理孤立节点、空输入和检测循环依赖。 --- ### ✅ 算法思路（Kahn 算法）： 1. 构建图的邻接表和入度字典。 2. 将所有入度为 0 的节点加入队列。 3. 不断从队列中取出节点，加入结果，并将其邻居的入度减 1；若邻居入度变为 0，则入队。 4. 若最终结果包含所有任务，则返回拓扑序；否则说明存在环。 --- ### ✅ Python 实现（完整可运行）： ```python from collections import deque, defaultdict def topological_sort(tasks, dependencies): """ 使用 Kahn 算法（BFS 入度法）进行拓扑排序 :param tasks: 任务 ID 列表，例如 ['A', 'B', 'C', 'D'] :param dependencies: 依赖关系列表，每个元素为 (pre, next) 表示 pre -> next :return: 合法拓扑序列表 或 错误提示字符串 """ # 处理空任务列表 if not tasks: return [] # 初始化邻接表和入度字典 graph = defaultdict(list) in_degree = {task: 0 for task in tasks} # 构建图和入度 for pre, nxt in dependencies: # 确保依赖中的任务都在 tasks 中（可选健壮性检查） if pre not in in_degree or nxt not in in_degree: raise ValueError(f"依赖中的任务 {pre} 或 {nxt} 不在任务列表中") graph[pre].append(nxt) in_degree[nxt] += 1 # nxt 的入度增加 # 将所有入度为 0 的任务加入队列 queue = deque([task for task in tasks if in_degree[task] == 0]) topo_order = [] # BFS 处理 while queue: node = queue.popleft() topo_order.append(node) # 遍历该节点的所有后继 for neighbor in graph[node]: in_degree[neighbor] -= 1 # 删除边 node -> neighbor if in_degree[neighbor] == 0: # 入度变为 0，可加入队列 queue.append(neighbor) # 检查是否所有任务都被排序（无环） if len(topo_order) == len(tasks): return topo_order else: return "Error: 存在循环依赖，无法完成拓扑排序" # ==================== 测试用例 ==================== if __name__ == "__main__": # 用例 1：正常 DAG tasks1 = ['A', 'B', 'C', 'D'] deps1 = [('A', 'B'), ('A', 'C'), ('B', 'D')] result1 = topological_sort(tasks1, deps1) print("用例 1 - 正常 DAG:") print("输入任务:", tasks1) print("依赖关系:", deps1) print("结果:", result1) print() # 用例 2：含环图 tasks2 = ['A', 'B', 'C'] deps2 = [('A', 'B'), ('B', 'C'), ('C', 'A')] # A -> B -> C -> A，形成环 result2 = topological_sort(tasks2, deps2) print("用例 2 - 含环图:") print("输入任务:", tasks2) print("依赖关系:", deps2) print("结果:", result2) print() # 用例 3：包含孤立节点 tasks3 = ['A', 'B', 'C', 'D'] deps3 = [('A', 'B'), ('B', 'C')] # D 是孤立节点 result3 = topological_sort(tasks3, deps3) print("用例 3 - 含孤立节点:") print("输入任务:", tasks3)...

AI 评审点评

以下是 AI 评审对该模型输出的点评：

【GEMINI】这是一份非常高质量的实现。模型不仅准确完成了核心算法，还充分考虑了边界条件（如孤立节点和空输入），并提供了完善的测试脚本和算法说明。输出格式完全符合题目要求，代码具有很强的可读性和健壮性。 【KIMI】整体而言，该实现满足了题目要求，正确实现了Kahn算法进行拓扑排序，并处理了各类边界情况。代码风格良好，关键步骤有注释说明，测试用例覆盖全面。但在输入有效性检查方面可以进一步加强，以确保依赖关系中的任务都在任务列表中。

进阶难度评测结果

  • 得分：92.1 分
  • 通过状态：已通过

模型生成产物

以下是该模型针对此任务的实际生成内容：

下面是一个支持并行执行的任务调度器，基于**拓扑排序**与**关键路径分析**设计，能够识别可并行执行的任务组并输出最优执行计划。 --- ## 一、类设计说明 ### 1. `Task` 类 表示单个任务节点。 - **属性**： - `id`: 任务唯一标识（字符串或整数） - `duration`: 执行耗时（数值类型） - **设计理由**：轻量级数据对象，用于封装任务元信息，便于序列化与调试。 ### 2. `TaskScheduler` 类 核心调度器，负责构建任务图、校验依赖、拓扑排序、并行分组与关键路径计算。 #### 核心数据结构： | 数据结构 | 类型 | 用途 | |--------|------|------| | `graph: Dict[str, List[str]]` | 邻接表 | 表示任务依赖关系（有向边 A→B 表示 A 依赖 B） | | `in_degree: Dict[str, int]` | 字典 | 记录每个任务的入度，用于拓扑排序 | | `durations: Dict[str, float]` | 字典 | 存储每个任务的执行时间 | | `reverse_graph: Dict[str, List[str]]` | 邻接表 | 反向图，用于计算最早开始时间（EST） | #### 算法选择理由： - **拓扑排序（Kahn 算法）**：适用于检测循环依赖并生成合法执行顺序，时间复杂度 O(V + E)。 - **关键路径法（CPM）**：通过前向/后向传递计算最早开始时间（EST）与最晚开始时间（LST），识别哪些任务在关键路径上（浮动时间为 0），从而指导并行调度与优化。 - **并行分组策略**：将同一“层级”（即入度为 0 且 EST 相同的任务）归为一组，可并行执行。 --- ## 二、完整代码实现 ```python from collections import deque, defaultdict from typing import Dict, List, Tuple, Set import json class Task: """表示单个任务节点""" def __init__(self, task_id: str, duration: float): if not isinstance(task_id, str) or not task_id.strip(): raise ValueError("Task ID must be a non-empty string.") if duration [dependent tasks] self.reverse_graph: Dict[str, List[str]] = defaultdict(list) # 反向图 self.in_degree: Dict[str, int] = {} self.durations: Dict[str, float] = {} self.tasks: Dict[str, Task] = {} def add_task(self, task: Task, dependencies: List[str] = None) -> None: """ 添加任务及其依赖关系。 :param task: Task 实例 :param dependencies: 该任务所依赖的任务 ID 列表（前置任务） :raises ValueError: 若存在未知依赖或循环依赖（检测在 schedule() 中） """ task_id = task.id if task_id in self.tasks: raise ValueError(f"Duplicate task ID: {task_id}") self.tasks[task_id] = task self.durations[task_id] = task.duration dependencies = dependencies or [] # 注册所有依赖节点（即使未显式添加任务，也允许作为依赖存在） for dep in dependencies: if dep not in self.tasks: self.tasks[dep] = None # 占位，后续需补全 duration self.durations[dep] = 0 # 占位 duration self.graph[dep].append(task_id) # dep -> task_id self.reverse_graph[task_id].append(dep) self.in_degree[task_id] =...

AI 评审点评

以下是 AI 评审对该模型输出的点评：

【GEMINI】这是一份高质量的 AI 生成结果。它不仅正确实现了基于拓扑排序的任务调度，还结合了工业界常用的关键路径分析（CPM）来优化并行调度。代码编写规范，文档说明详尽，异常处理机制健全，能够很好地解决复杂的任务依赖调度问题。 【KIMI】整体来说，这是一个优秀的任务调度系统实现。代码实现了拓扑排序和关键路径法，能够正确识别并行任务组和关键路径。功能覆盖全面，异常处理也比较到位。代码结构清晰，可读性较好。但是在一些细节上还有改进空间，如变量命名、异常处理等。

困难难度评测结果

  • 得分：84.0 分
  • 通过状态：已通过

模型生成产物

以下是该模型针对此任务的实际生成内容：

下面是一个完整、可运行的生产级 DAG 调度引擎实现，包含动态 DAG 管理、条件分支调度、资源感知调度、失败重试与状态机、执行计划与监控。 ```python from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, Set, Tuple import heapq import json import time from collections import deque, defaultdict # ====================== # 枚举与数据类定义 # ====================== class TaskStatus(Enum): PENDING = "PENDING" READY = "READY" RUNNING = "RUNNING" SUCCESS = "SUCCESS" FAILED = "FAILED" SKIPPED = "SKIPPED" class DependencyType(Enum): ON_SUCCESS = "on_success" ON_FAILURE = "on_failure" ON_COMPLETE = "on_complete" # 默认行为 @dataclass class Task: """ 任务定义，包含执行所需资源、依赖关系与重试策略。 """ task_id: str duration_s: float # 模拟执行时长（秒） cpu_cores: int = 1 memory_mb: int = 512 max_retries: int = 0 retry_delay_s: float = 1.0 dependencies: Dict[str, DependencyType] = field(default_factory=dict) # task_id -> DependencyType def __post_init__(self): if self.dependencies is None: self.dependencies = {} # ====================== # 状态存储模块（StateStore） # ====================== @dataclass class TaskInstance: """ 任务运行时实例，记录执行状态与重试信息。 """ task: Task status: TaskStatus = TaskStatus.PENDING retry_count: int = 0 last_error: Optional[str] = None start_time: Optional[float] = None end_time: Optional[float] = None class StateStore: """ 存储所有任务实例的状态，提供原子化状态更新与查询。 """ def __init__(self): self._tasks: Dict[str, TaskInstance] = {} def add_task(self, task: Task) -> None: if task.task_id in self._tasks: raise ValueError(f"Task {task.task_id} already exists") self._tasks[task.task_id] = TaskInstance(task=task) def remove_task(self, task_id: str) -> Optional[TaskInstance]: return self._tasks.pop(task_id, None) def get_task(self, task_id: str) -> Optional[TaskInstance]: return self._tasks.get(task_id) def update_status(self, task_id: str, status: TaskStatus, error: Optional[str] = None) -> None: inst = self._tasks.get(task_id) if not inst: raise KeyError(f"Task {task_id} not found") inst.status = status if error:...

AI 评审点评

以下是 AI 评审对该模型输出的点评：

【GEMINI】这是一个架构设计良好的任务调度引擎实现。作者展示了深厚的后端设计功底，特别是在 DAG 管理、资源调度算法和监控指标计算（如关键路径）方面表现出色。虽然在任务状态机的某些特定转换逻辑（SKIPPED 状态处理）和重试触发条件上存在小瑕疵，但整体代码的工业级规范性和模块化程度很高，是一个非常优秀的参考实现。 【KIMI】该实现覆盖了题目要求的核心功能点，代码结构清晰，模块化设计合理。但在任务依赖关系处理、资源利用率计算等细节上还有改进空间，代码的健壮性和边界情况处理也需要加强。总体来说是一个比较完整的DAG调度引擎实现，达到了较高的正确性和完整性，但在代码质量方面还有提升空间。

相关链接

您可以通过以下链接查看更多相关内容：

加载中...
100%