🤖 系列：Java工程师转AI Agent 3个月学习计划
👤 作者：宸丶一 | 28岁Java程序员，正在学习 AI Agent 开发中ing…
🎯 今日目标： 任务追踪、树状任务系统、与检查点系统的联动
💬 个人格言： 代码改不改变世界我不知道，但先让我准时下班。

---

📖 前言

在 Day 13 的学习中，我们掌握了持久化记忆，学会了如何让 Agent 在多次会话间保持对项目的理解。

今天，我们进入 Day 14 的学习：任务追踪。

如果说 Day 13 是"持久化记忆"，那么 Day 14 就是"任务追踪"。任务追踪让 Agent 拥有项目管理能力，能够管理复杂的任务分解、依赖关系和进度追踪，就像一个智能的项目经理。

特别的是，今天我们还结合了小米开源的 MiMo-Code 框架，学习了其中的任务追踪机制，并采用了深度思想的学习模式，通过开放性思考题来加深理解。

---

🎯 学习目标

1. 理解任务追踪的核心概念
2. 掌握树状任务系统的设计
3. 学习与检查点系统的联动
4. 结合 MiMo-Code 框架学习最佳实践

---

📚 核心概念

1. 什么是任务追踪？

定义： 记录任务的执行状态

Java 对标： 项目管理

理解： 任务追踪就像项目经理，负责管理任务的创建、分配、执行和完成。

本质： 任务追踪的本质是"控制机制"，让任务状态可见、可控、可预测。

为什么需要任务追踪：

1. 可见性 — 让任务状态可见，避免"黑盒"
2. 反馈 — 提供反馈循环，支持调整
3. 问责 — 明确责任，支持追溯

优势：

• 可控性 — 任务状态可控
• 可预测性 — 任务进度可预测
• 可优化性 — 基于数据优化流程

劣势：

• 开销 — 增加系统开销
• 复杂性 — 增加系统复杂性
• 依赖性 — 过度依赖追踪可能影响效率

---

2. 树状任务系统

定义： 层级结构的任务组织

Java 对标： WBS（工作分解结构）

理解： 树状任务系统就像项目的 WBS，将大任务分解为小任务，形成层级结构。

核心概念：

• 根任务：没有父任务的任务
• 子任务：有父任务的任务
• 任务树：展示任务的层级关系

示例：

T1: 查询北京天气 [completed] (100%)
  T1.1: 查询天气数据 [completed] (100%)
  T1.2: 格式化输出 [completed] (100%)

为什么需要树状结构：

1. 任务分解 — 将大任务分解为小任务
2. 进度聚合 — 子任务进度汇总到父任务
3. 可视化 — 支持任务可视化

---

3. 与检查点系统的联动

定义： 任务进度与检查点绑定

Java 对标： 状态持久化

理解： 任务进度与检查点联动，让任务状态能够持久化，支持会话恢复。

核心概念：

• 检查点：会话状态快照
• 任务状态：任务的当前状态
• 联动：任务进度与检查点绑定

为什么需要联动：

1. 持久化 — 任务进度能够持久化
2. 恢复 — 支持会话恢复
3. 一致性 — 保持任务状态一致

---

4. 任务状态

定义： 任务的当前状态

状态类型：

• pending — 待开始
• in_progress — 进行中
• completed — 已完成
• cancelled — 已取消

状态转换：

pending → in_progress → completed
pending → cancelled
in_progress → cancelled

---

🔄 任务追踪工作流

流程图：

执行流程：

1. 创建任务 — 创建任务记录
2. 处理任务 — 执行具体任务
3. 更新状态 — 更新任务状态和进度
4. 返回结果 — 返回处理结果

---

🆚 树状任务系统对比

特性	扁平任务	树状任务
结构	扁平	层级
分解	不支持	支持
进度	单一	聚合
可视化	简单	直观

简单说：

• 扁平任务 — 简单任务，不需要分解
• 树状任务 — 复杂任务，需要分解

---

💻 代码示例

示例 1：任务追踪基础

执行流程：

用户输入: 查询北京天气
    ↓
创建任务: task_001 - 查询北京天气
    ↓
处理任务: 北京今天晴，25度，微风
    ↓
更新状态: completed, 100%
    ↓
返回结果: 北京今天晴，25度，微风

代码：

from typing import TypedDict
from langgraph.graph import StateGraph, END
import time

# 定义任务管理器
class TaskManager:
    def __init__(self):
        self.tasks = {}
        self.task_counter = 0
    
    def create_task(self, task_id: str, title: str, description: str = "", parent_id: str = "") -> dict:
        """创建任务"""
        task = {
            "id": task_id,
            "title": title,
            "description": description,
            "status": "pending",
            "progress": 0,
            "parent_id": parent_id,
            "subtasks": [],
            "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "updated_at": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        self.tasks[task_id] = task
        
        # 如果有父任务，添加到父任务的子任务列表
        if parent_id and parent_id in self.tasks:
            self.tasks[parent_id]["subtasks"].append(task_id)
        
        return task
    
    def update_task_status(self, task_id: str, status: str, progress: float = None):
        """更新任务状态"""
        if task_id in self.tasks:
            self.tasks[task_id]["status"] = status
            if progress is not None:
                self.tasks[task_id]["progress"] = progress
            self.tasks[task_id]["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
    
    def get_task(self, task_id: str) -> dict:
        """获取任务"""
        return self.tasks.get(task_id, {})
    
    def get_all_tasks(self) -> dict:
        """获取所有任务"""
        return self.tasks

# 定义任务 Agent
def task_agent(state: dict, task_manager: TaskManager) -> dict:
    """任务 Agent：管理任务追踪"""
    current_task = state.get("current_task", "")
    task_id = state.get("task_id", "task_001")
    
    # 创建任务
    task_manager.create_task(
        task_id=task_id,
        title=current_task,
        description=f"处理任务: {current_task}"
    )
    
    # 处理任务
    if "天气" in current_task:
        result = "北京今天晴，25度，微风"
    else:
        result = f"已处理：{current_task}"
    
    # 更新任务状态
    task_manager.update_task_status(task_id, "completed", 100)
    
    return {**state, "task_result": result, "task_status": "completed", "task_progress": 100}

# 创建工作流图
def create_task_graph(task_manager: TaskManager) -> StateGraph:
    graph = StateGraph(dict)
    graph.add_node("task", lambda state: task_agent(state, task_manager))
    graph.set_entry_point("task")
    graph.add_edge("task", END)
    return graph

# 编译并运行
task_manager = TaskManager()
graph = create_task_graph(task_manager)
app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "查询北京天气"}],
    "current_task": "查询北京天气",
    "task_result": "",
    "task_id": "task_001",
    "task_status": "",
    "task_progress": 0
})
print(result["task_result"])

---

示例 2：树状任务系统

执行流程：

用户输入: 查询北京天气
    ↓
创建根任务: T1 - 查询北京天气
    ↓
分解子任务: T1.1 - 查询天气数据, T1.2 - 格式化输出
    ↓
添加依赖: T1.2 依赖 T1.1
    ↓
执行任务: T1.1 completed → T1.2 completed
    ↓
自动更新: T1 progress = 100%
    ↓
返回结果: 北京今天晴，25度，微风

代码：

import time

# 定义树状任务系统
class TreeTaskSystem:
    def __init__(self):
        self.tasks = {}
    
    def create_task(self, task_id: str, title: str, description: str = "", parent_id: str = "") -> dict:
        """创建任务"""
        task = {
            "id": task_id,
            "title": title,
            "description": description,
            "status": "pending",
            "progress": 0,
            "parent_id": parent_id,
            "subtasks": [],
            "dependencies": [],
            "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "updated_at": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        self.tasks[task_id] = task
        
        if parent_id and parent_id in self.tasks:
            self.tasks[parent_id]["subtasks"].append(task_id)
        
        return task
    
    def add_dependency(self, task_id: str, dependency_id: str):
        """添加依赖关系"""
        if task_id in self.tasks:
            self.tasks[task_id]["dependencies"].append(dependency_id)
    
    def update_task_status(self, task_id: str, status: str, progress: float = None):
        """更新任务状态"""
        if task_id in self.tasks:
            self.tasks[task_id]["status"] = status
            if progress is not None:
                self.tasks[task_id]["progress"] = progress
            self.tasks[task_id]["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
            
            # 自动更新父任务进度
            self._update_parent_progress(task_id)
    
    def _update_parent_progress(self, task_id: str):
        """自动更新父任务进度"""
        task = self.tasks.get(task_id, {})
        parent_id = task.get("parent_id", "")
        
        if parent_id and parent_id in self.tasks:
            parent = self.tasks[parent_id]
            subtasks = parent.get("subtasks", [])
            
            if subtasks:
                total_progress = 0
                for subtask_id in subtasks:
                    subtask = self.tasks.get(subtask_id, {})
                    total_progress += subtask.get("progress", 0)
                
                avg_progress = total_progress / len(subtasks)
                parent["progress"] = avg_progress
                
                all_completed = all(
                    self.tasks.get(sid, {}).get("status") == "completed"
                    for sid in subtasks
                )
                if all_completed:
                    parent["status"] = "completed"
    
    def get_task_tree(self, task_id: str, level: int = 0) -> str:
        """获取任务树"""
        task = self.tasks.get(task_id, {})
        if not task:
            return ""
        
        indent = "  " * level
        result = f"{indent}{task['id']}: {task['title']} [{task['status']}] ({task['progress']:.1f}%)\n"
        
        for subtask_id in task.get("subtasks", []):
            result += self.get_task_tree(subtask_id, level + 1)
        
        return result

# 使用示例
task_system = TreeTaskSystem()

# 创建根任务
task_system.create_task("T1", "查询北京天气")

# 创建子任务
task_system.create_task("T1.1", "查询天气数据", parent_id="T1")
task_system.create_task("T1.2", "格式化输出", parent_id="T1")

# 添加依赖
task_system.add_dependency("T1.2", "T1.1")

# 执行任务
task_system.update_task_status("T1.1", "completed", 100)
task_system.update_task_status("T1.2", "completed", 100)

# 获取任务树
print(task_system.get_task_tree("T1"))

---

🎓 学习收获

1. 核心概念理解

• 任务追踪 = 项目管理
• 树状任务系统 = WBS（工作分解结构）
• 与检查点系统的联动 = 状态持久化
• 任务状态 = pending/in_progress/completed/cancelled

2. 树状任务系统对比

特性	扁平任务	树状任务
结构	扁平	层级
分解	不支持	支持
进度	单一	聚合
可视化	简单	直观

3. 技术实现

• 任务管理器 — 管理任务的创建、更新、查询
• 树状结构 — 支持父子任务关系
• 依赖关系 — 支持任务之间的依赖
• 进度聚合 — 子任务进度汇总到父任务

---

🧠 深度思想学习模式

为什么采用深度思想学习模式？

传统学习模式：

• 代码填空 — 重复性高，容易枯燥
• 固定答案 — 限制思维，缺乏创新
• 被动学习 — 缺乏主动思考

深度思想学习模式：

• 开放性设计 — 激发创造力
• 深度分析 — 培养分析能力
• 创新设计 — 培养创新思维
• 主动思考 — 提升学习效果

深度思想学习模式的特点

特点	说明	效果
开放性	没有标准答案	激发创造力
深度性	需要深入分析	培养分析能力
创新性	需要提出创新想法	培养创新思维
主动性	需要主动思考	提升学习效果

---

🎲 随机题型考核

为什么采用随机题型考核？

传统考核模式：

• 固定题型 — 容易预测，缺乏挑战
• 重复练习 — 容易枯燥，缺乏兴趣
• 被动应对 — 缺乏主动思考

随机题型考核：

• 随机题型 — 增加挑战性
• 多样化练习 — 增加趣味性
• 主动应对 — 提升思考能力

随机题型考核的特点

特点	说明	效果
随机性	随机选择题型	增加挑战性
多样性	多种题型	增加趣味性
主动性	需要主动应对	提升思考能力

---

❓ 思考题

第一部分：开放性设计题

1. 设计一个 “项目管理系统”

第二部分：深度分析题

1. 分析任务追踪的本质
2. 分析树状任务系统的作用

第三部分：创新设计题

1. 创新的任务分配方式

---

📝 总结

Day 14 的学习让我们从"持久化记忆"升级到"任务追踪"。

核心收获：

1. 任务追踪的核心概念
2. 树状任务系统的设计
3. 与检查点系统的联动
4. 深度思想学习模式的实践
5. 随机题型考核的实践

学习路径：

Day 1-8: Python 基础 + API 调用
    ↓
Day 9: Agent 源码解析
    ↓
Day 10: LangGraph
    ↓
Day 11: 多 Agent 协作
    ↓
Day 12: 人机协作 + Goal/停止条件
    ↓
Day 13: 持久化记忆
    ↓
Day 14: 任务追踪 (本文)
    ↓
Day 15: 子智能体系统

下一步：

• Day 15：子智能体系统
• Day 16：智能上下文管理
• Day 17：Dream & Distill

---

📚 参考资料

1. MiMo-Code GitHub
2. MiMo-Code 官网
3. LangGraph 官方文档

---

系列文章：

• Day 13：持久化记忆
• Day 14：任务追踪 (本文)
• Day 15：子智能体系统

---

作者简介： 宸一，Java 工程师，正在学习 AI Agent 开发中ing…