任务系统与依赖管理:让目标超越对话(构建自己的AI Agent)

来自程序员技术小站
跳转到导航 跳转到搜索

"大目标要拆成小任务,排好序,记在磁盘上" —— 文件持久化的任务图,为多 Agent 协作打基础。

前面的 TodoManager 是内存中的扁平清单:没有顺序、没有依赖、状态只有做完没做完。

真实的目标是有结构的——任务 B 依赖任务 A,任务 C 和 D 可以并行,任务 E 要等 C 和 D 都完成。

没有显式的关系,Agent 分不清什么能做、什么被卡住、什么能同时跑。

一、从清单到图

我们需要把扁平清单升级为持久化到磁盘的任务图

.tasks/
  task_1.json  {"id":1, "status":"completed"}
  task_2.json  {"id":2, "blockedBy":[1], "status":"pending"}
  task_3.json  {"id":3, "blockedBy":[1], "status":"pending"}
  task_4.json  {"id":4, "blockedBy":[2,3], "status":"pending"}

任务图 (DAG):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

顺序:   task 1 必须先完成, 才能开始 2 和 3
并行:   task 2 和 3 可以同时执行
依赖:   task 4 要等 2 和 3 都完成
状态:   pending -> in_progress -> completed

这个任务图是后续所有机制的协调骨架:后台执行、多 Agent 团队、工作区隔离都读写这同一个结构。

二、核心能力

任务图随时回答三个问题:

  1. 什么可以做? —— 状态为 pendingblockedBy 为空的任务
  2. 什么被卡住? —— 等待前置任务完成的任务
  3. 什么做完了? —— 状态为 completed 的任务,完成时自动解锁后续任务

三、核心机制

1. TaskManager

每个任务一个 JSON 文件,支持 CRUD + 依赖图:

class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        self._next_id = self._max_id() + 1

    def create(self, subject, description=""):
        task = {
            "id": self._next_id,
            "subject": subject,
            "status": "pending",
            "blockedBy": [],
            "blocks": [],
            "owner": ""
        }
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)

2. 依赖解除

完成任务时,自动将其 ID 从其他任务的 blockedBy 中移除:

def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)

3. 状态变更 + 依赖关联

update 处理状态转换和依赖边:

def update(self, task_id, status=None,
           add_blocked_by=None, add_blocks=None):
    task = self._load(task_id)
    if status:
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)
    self._save(task)

4. 工具集成

四个任务工具加入调度映射:

TOOL_HANDLERS = {
    # ...基础工具...
    "task_create": lambda **kw: TASKS.create(kw["subject"]),
    "task_update": lambda **kw: TASKS.update(
        kw["task_id"], kw.get("status")
    ),
    "task_list":   lambda **kw: TASKS.list_all(),
    "task_get":    lambda **kw: TASKS.get(kw["task_id"]),
}

四、实际工作流程

1. 创建任务
   task_create: "Setup project"
   task_create: "Write code"
   task_create: "Write tests"
   task_create: "Deploy"

2. 设置依赖
   task_update: task 2, add_blocked_by=[1]
   task_update: task 3, add_blocked_by=[1]
   task_update: task 4, add_blocked_by=[2,3]

3. 查看可执行任务
   task_list -> [task 1] (pending, no blockedBy)

4. 开始执行
   task_update: task 1, status="in_progress"
   # ... work ...
   task_update: task 1, status="completed"
   # 自动解锁 task 2 和 3

5. 并行执行
   task_update: task 2, status="in_progress"
   task_update: task 3, status="in_progress"
   # ... work ...
   task_update: task 2, status="completed"
   task_update: task 3, status="completed"
   # 自动解锁 task 4

6. 完成
   task_update: task 4, status="completed"

五、试一试

cd learn-claude-code
python agents/s07_task_system.py

试试这些 prompt:

  1. Create 3 tasks: "Setup project", "Write code", "Write tests". Make them depend on each other in order.
  2. List all tasks and show the dependency graph
  3. Complete task 1 and then list tasks to see task 2 unblocked
  4. Create a task board for refactoring: parse -> transform -> emit -> test, where transform and emit can run in parallel after parse

六、与 TodoManager 的区别

特性 TodoManager (s03) TaskManager (s07)
存储 内存 磁盘 (JSON 文件)
结构 扁平清单 依赖图 (DAG)
依赖 blockedBy + blocks
并行 不支持 支持
持久化 重启丢失 跨会话保存
适用场景 单次会话快速清单 复杂多步目标

从 s07 起,任务图是多步工作的默认选择。s03 的 Todo 仍可用于单次会话内的快速清单。

七、设计哲学

持久化任务比任何一次对话都长命。

这是 Harness 工程的核心原则:

  • 对话会结束,但目标可能延续
  • 上下文可能被压缩,但任务状态始终完整
  • 多个 Agent 可以协作同一个任务图
  • 重启后可以从断点继续

八、进阶应用

任务系统是以下高级功能的基础:

  • 后台执行 (s08):任务在后台运行,Agent 继续思考
  • Agent 团队 (s09+):多个 Agent 协作处理任务图
  • 自治 Agent (s11):Agent 自己扫描看板、认领任务
  • 工作区隔离 (s12):每个任务在自己的目录执行

没有任务系统,就没有真正的多 Agent 协作。


Bash 就够了。真正的 Agent 是宇宙所需要的全部。

延伸阅读:learn-claude-code 开源项目,12 个渐进式教程,完整教授 Harness 工程。