查看“︁任务系统与依赖管理:让目标超越对话(构建自己的AI Agent)”︁的源代码
←
任务系统与依赖管理:让目标超越对话(构建自己的AI Agent)
跳转到导航
跳转到搜索
因为以下原因,您没有权限编辑该页面:
您请求的操作仅限属于该用户组的用户执行:
用户
您可以查看和复制此页面的源代码。
<blockquote>"大目标要拆成小任务,排好序,记在磁盘上" —— 文件持久化的任务图,为多 Agent 协作打基础。</blockquote>前面的 TodoManager 是内存中的扁平清单:没有顺序、没有依赖、状态只有做完没做完。 真实的目标是有结构的——任务 B 依赖任务 A,任务 C 和 D 可以并行,任务 E 要等 C 和 D 都完成。 没有显式的关系,Agent 分不清什么能做、什么被卡住、什么能同时跑。 == 一、从清单到图 == 我们需要把扁平清单升级为'''持久化到磁盘的任务图'''。 <code>.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</code> 这个任务图是后续所有机制的协调骨架:后台执行、多 Agent 团队、工作区隔离都读写这同一个结构。 == 二、核心能力 == 任务图随时回答三个问题: # '''什么可以做?''' —— 状态为 <code>pending</code> 且 <code>blockedBy</code> 为空的任务 # '''什么被卡住?''' —— 等待前置任务完成的任务 # '''什么做完了?''' —— 状态为 <code>completed</code> 的任务,完成时自动解锁后续任务 == 三、核心机制 == === 1. TaskManager === 每个任务一个 JSON 文件,支持 CRUD + 依赖图: <code>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)</code> === 2. 依赖解除 === 完成任务时,自动将其 ID 从其他任务的 <code>blockedBy</code> 中移除: <code>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)</code> === 3. 状态变更 + 依赖关联 === <code>update</code> 处理状态转换和依赖边: <code>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)</code> === 4. 工具集成 === 四个任务工具加入调度映射: <code>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"]), }</code> == 四、实际工作流程 == <code>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"</code> == 五、试一试 == <code>cd learn-claude-code python agents/s07_task_system.py</code> 试试这些 prompt: # <code>Create 3 tasks: "Setup project", "Write code", "Write tests". Make them depend on each other in order.</code> # <code>List all tasks and show the dependency graph</code> # <code>Complete task 1 and then list tasks to see task 2 unblocked</code> # <code>Create a task board for refactoring: parse -> transform -> emit -> test, where transform and emit can run in parallel after parse</code> == 六、与 TodoManager 的区别 == {| class="wikitable" !特性 !TodoManager (s03) !TaskManager (s07) |- |存储 |内存 |磁盘 (JSON 文件) |- |结构 |扁平清单 |依赖图 (DAG) |- |依赖 |无 |<code>blockedBy</code> + <code>blocks</code> |- |并行 |不支持 |支持 |- |持久化 |重启丢失 |跨会话保存 |- |适用场景 |单次会话快速清单 |复杂多步目标 |} 从 s07 起,任务图是多步工作的默认选择。s03 的 Todo 仍可用于单次会话内的快速清单。 == 七、设计哲学 == '''持久化任务比任何一次对话都长命。''' 这是 Harness 工程的核心原则: * 对话会结束,但目标可能延续 * 上下文可能被压缩,但任务状态始终完整 * 多个 Agent 可以协作同一个任务图 * 重启后可以从断点继续 == 八、进阶应用 == 任务系统是以下高级功能的基础: * '''后台执行 (s08)''':任务在后台运行,Agent 继续思考 * '''Agent 团队 (s09+)''':多个 Agent 协作处理任务图 * '''自治 Agent (s11)''':Agent 自己扫描看板、认领任务 * '''工作区隔离 (s12)''':每个任务在自己的目录执行 '''没有任务系统,就没有真正的多 Agent 协作。''' ----''Bash 就够了。真正的 Agent 是宇宙所需要的全部。'' '''延伸阅读''':learn-claude-code 开源项目,12 个渐进式教程,完整教授 Harness 工程。
返回
任务系统与依赖管理:让目标超越对话(构建自己的AI Agent)
。
导航菜单
个人工具
登录
命名空间
页面
讨论
大陆简体
查看
阅读
查看源代码
查看历史
更多
搜索
导航
首页
最近更改
随机页面
特殊页面
工具
链入页面
相关更改
页面信息