AI Agent Engineer

AI Agent Engineer

项目：依赖任务调度器

8 min read

项目：依赖任务调度器

项目目标

从内存中的 Python 对象构建一个确定性任务顺序：依赖全部完成的任务才进入 ready 最小堆；数值越小优先级越高，同优先级按 task_id 排序。项目组合 dict、set、邻接表、入度、堆和输出 list，并用 24 项单元测试验证契约。

它不读取 JSON、不执行外部工具，也不使用线程、网络、文件写入或密钥。调度“顺序”不等于业务执行成功；生产工作流的持久化、重试、租约、权限和恢复不在本项目中。

输入契约

输入任务集合必须是内建 list 或 tuple。这里刻意不接受 generator 等一次性 iterator：调度器若消费它，调用方就无法再次遍历，容易与“输入不变”产生歧义。集合中的每个任务是恰好含三个字段的映射：

{
    "id": "validate",
    "priority": 1,
    "depends_on": ["extract"],
}
字段规则
id内建 str；非空、无首尾空白、全局唯一
priority必填且 type(value) is int；显式拒绝 bool
depends_onlist；元素是内建 str，非空、无首尾空白；内部无重复

所有依赖 ID 必须存在；直接自依赖立即拒绝，间接环在拓扑推进后报告。未知/缺失字段、非法任务对象和脏依赖都转换为 ScheduleInputError，不泄漏 AttributeError 或不可哈希值导致的裸 TypeError。

选择与不变量

by_id: id → 不可变 Task          # 唯一真相源
remaining: id → 未完成前驱数      # 入度工作副本
dependents: 前驱 id → 后继 id 列表 # 反向邻接表
ready: (priority, task_id) 最小堆 # 当前可执行视图
order: 已选择的 task_id           # 结果

必须保持：

  1. remaining 与 dependents 只引用 by_id 中的 ID；
  2. ready 中的任务入度为 0；
  3. 每条依赖边只让对应后继入度减少一次；
  4. 每个任务最多入堆一次、最多输出一次；
  5. priority 只在当前已就绪任务中比较；
  6. 调度不修改调用方的 list/tuple、任务映射或依赖列表；
  7. 相同输入重复运行得到相同顺序。

算法步骤

  1. parse_task 逐项把不可信映射转成冻结 Task。
  2. _parse_tasks 建立 by_id，检查 ID 唯一和依赖存在。
  3. 为每个任务计算 remaining，同时建立 dependents。
  4. 将所有入度 0 的任务压入最小堆。
  5. 弹出当前 (priority, task_id) 最小项，将后继入度减 1。
  6. 后继入度变 0 时入堆，直到 ready 为空。
  7. 若输出数小于任务数，剩余任务被一个或多个环阻塞。

构图和入度更新是 ；每个任务最多入堆/出堆一次，增加 。总时间可写为 ，辅助空间为 。这不是对真实分布式调度器吞吐的性能声明。

运行

从 vault 根目录运行演示：

python -B '.\Knowledge\AI Agent Engineer\docs\数据结构基础\examples\task_scheduler.py'

预期：

execution order: extract -> validate -> summarize -> review

运行测试：

python -B -m unittest discover `
  -s '.\Knowledge\AI Agent Engineer\docs\数据结构基础\examples' `
  -p 'test_*.py' `
  -v

-B 避免生成 .pyc；示例不安装依赖、不写文件。若 python 不是预期解释器，先用 Get-Command python、python --version 和 sys.executable 核对。

测试矩阵

类别已覆盖
基本集合根类型、拒绝且不消费一次性 iterator、空、单任务、链、菱形、多独立分量
优先级仅 ready 内比较；同级按 ID，而非输入顺序
标识空/首尾空白/非内建字符串（含不可哈希子类）、重复 ID
字段缺失、未知、非映射对象、不可哈希字段名的错误归一化
优先级类型bool、float、str、None 均拒绝
依赖非 list、脏元素、重复、缺失、直接自依赖
环间接环 + 被环阻塞下游，不误称所有 blocked 都是环成员
性质输入不变、重复运行确定性

单元测试数量是当前实现的证据，不是“越多越好”的目标。改变契约时先改文档与失败测试，再改实现。

必做练习

  1. 不看成品，从数据契约写出 Task 与解析器，让非法输入测试先失败。
  2. 画出菱形依赖的前驱表、后继表和每一步入度变化。
  3. 新增两个同级任务，把平局策略改为输入 FIFO；引入 sequence 后更新契约和测试。
  4. 让 schedule 额外返回每轮 ready 快照，但不要暴露内部可变堆。
  5. 写一个只读 payload 扩展，证明调度过程不修改 payload；说明浅拷贝的限制。

可选进阶

  • 使用 graphlib.TopologicalSorter 重做“只求任一合法拓扑序”的版本，比较其同层顺序。
  • 为错误返回稳定代码而非依赖整句中文；保留对人可读消息。
  • 单独实现精确环路径/强连通分量分析，再区分环成员和下游 blocked；不要靠剩余集合猜。

不要在此脚本中加入真实 API 调用、持久化队列或异步 worker。它们会把数据结构正确性与副作用恢复混成一个难以验收的问题。

常见错误

  • 把 bool 当合法整数优先级；
  • 在解析所有输入前开始调度，让脏优先级被环错误掩盖；
  • 每轮重扫所有任务并重建依赖集合，造成不必要的高阶增长；
  • 同优先级比较不可比较 payload；
  • 把“出堆”当作外部任务已经成功；
  • 把所有剩余 blocked 节点都称为环成员；
  • 用 assert 校验不可信输入；优化模式可能移除断言。

项目验收

  • Python 3.11+ 环境中演示和 24 项测试通过。
  • 能解释五个内部结构的职责与不变量。
  • 能手算链、菱形和带环图的入度变化。
  • 能证明优先级只在 ready 中生效，同级按 ID。
  • 能解释 的每一项从何而来。
  • 能区分“环成员”“被环阻塞”“无关可执行分量”。
  • 能指出未覆盖的生产语义，不虚假声称已验证外部执行。

完成后进入 全库自测与掌握检查。

参考资料

获取日期：2026-07-14。

Backlinks