🤖 系列：Java工程师转AI Agent 3个月学习计划
👤 作者：宸丶一 | 28岁Java程序员，正在学习 AI Agent 开发中ing…
🎯 今日目标： 子智能体系统、共享上下文、生命周期追踪
💬 个人格言： 代码改不改变世界我不知道，但先让我准时下班。

---

📖 前言

在 Day 14 的学习中，我们掌握了任务追踪，学会了如何让 Agent 拥有项目管理能力。

今天，我们进入 Day 15 的学习：子智能体系统。

如果说 Day 14 是"任务追踪"，那么 Day 15 就是"子智能体系统"。子智能体系统让 Agent 拥有团队协作能力，能够按需生成子智能体，共享上下文并行工作，就像一个高效的开发团队。

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

---

🎯 学习目标

1. 理解子智能体系统的核心概念
2. 掌握主智能体按需生成子智能体
3. 学习共享上下文并行工作
4. 学习生命周期追踪

---

📚 核心概念

1. 什么是子智能体系统？

定义： 主智能体按需生成子智能体

Java 对标： 子线程

理解： 子智能体系统就像项目经理带领团队，主智能体负责分配任务，子智能体负责执行任务。

本质： 子智能体系统的本质是"分工机制"，将复杂问题拆分处理。

为什么需要子智能体系统：

1. 分工 — 将复杂任务分解为简单任务
2. 并行 — 多个子智能体同时执行
3. 扩展 — 支持水平扩展

优势：

• 可以同时执行多个简单问题
• 可以处理多模态问题
• 支持水平扩展

劣势：

• 单个简单问题处理效率下降
• 通信开销增加
• 协调成本增加

---

2. 子智能体的类型

参考 MiMo-Code 的设计：

子智能体类型	职责	示例
工作者子智能体	执行具体任务	查询天气、计算数学
分析子智能体	分析数据和信息	数据分析、趋势预测
协调子智能体	协调多个子智能体	任务分配、结果汇总

工作者子智能体：

• 职责：执行具体任务
• 示例：查询天气、计算数学、搜索信息
• 特点：专注、高效

分析子智能体：

• 职责：分析数据和信息
• 示例：数据分析、趋势预测、模式识别
• 特点：深入、专业

协调子智能体：

• 职责：协调多个子智能体
• 示例：任务分配、结果汇总、冲突解决
• 特点：全局视角、决策能力

---

3. 共享上下文

定义： 子智能体共享的信息

Java 对标： 共享内存

理解： 共享上下文就像团队的共享文档，所有成员都能访问和更新。

核心概念：

• 共享数据：子智能体共享的数据
• 同步机制：如何保证数据一致性
• 访问控制：如何控制数据访问

为什么需要共享上下文：

1. 信息共享 — 子智能体能够共享用户信息、任务状态等
2. 一致性 — 保证所有子智能体看到的数据是一致的
3. 效率 — 避免重复查询

实现方式：

• 内存共享：直接共享内存（简单，但不持久）
• 数据库共享：通过数据库共享（持久，但有开销）
• 消息队列：通过消息队列共享（异步，但复杂）

---

4. 生命周期追踪

定义： 追踪子智能体的状态变化

Java 对标： 线程状态

理解： 生命周期追踪就像项目管理中的进度追踪，记录每个任务的状态变化。

状态类型：

• created — 已创建
• running — 运行中
• completed — 已完成
• failed — 失败
• cancelled — 已取消

状态转换：

created → running → completed
created → running → failed
created → running → cancelled

为什么需要生命周期追踪：

1. 可见性 — 让任务状态可见
2. 调试 — 方便调试和排查问题
3. 监控 — 支持系统监控

---

🔄 子智能体系统工作流

流程图：

执行流程：

1. 主智能体 — 接收任务，分析需求
2. 创建子智能体 — 按需生成子智能体
3. 共享上下文 — 设置共享信息
4. 并行执行 — 多个子智能体同时执行
5. 结果汇总 — 主智能体汇总结果
6. 返回结果 — 返回最终结果

---

🆚 子智能体系统对比

特性	单智能体	子智能体系统
复杂度	简单	复杂
并行	不支持	支持
扩展性	有限	非常灵活
适用场景	简单任务	复杂任务

简单说：

• 单智能体 — 一个人干所有事
• 子智能体系统 — 团队协作，各司其职

---

💻 代码示例

示例 1：子智能体基础

执行流程：

用户输入: 查询北京天气
    ↓
主智能体: 创建子智能体
    ↓
子智能体: 执行任务
    ↓
子智能体: 返回结果
    ↓
主智能体: 汇总结果
    ↓
返回结果: 北京今天晴，25度，微风

代码：

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

# 定义子智能体管理器
class SubAgentManager:
    def __init__(self):
        self.subagents = {}
        self.subagent_counter = 0
    
    def create_subagent(self, subagent_id: str, task: str, parent_id: str = "main") -> dict:
        """创建子智能体"""
        subagent = {
            "id": subagent_id,
            "task": task,
            "status": "created",
            "result": "",
            "parent_id": parent_id,
            "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "updated_at": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        self.subagents[subagent_id] = subagent
        return subagent
    
    def execute_subagent(self, subagent_id: str) -> str:
        """执行子智能体"""
        if subagent_id not in self.subagents:
            return f"子智能体不存在: {subagent_id}"
        
        subagent = self.subagents[subagent_id]
        subagent["status"] = "running"
        subagent["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
        
        # 模拟执行任务
        task = subagent["task"]
        if "天气" in task:
            result = "北京今天晴，25度，微风"
        elif "计算" in task:
            result = "计算结果: 126"
        else:
            result = f"已处理：{task}"
        
        # 更新状态
        subagent["status"] = "completed"
        subagent["result"] = result
        subagent["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
        
        return result
    
    def get_subagent(self, subagent_id: str) -> dict:
        """获取子智能体"""
        return self.subagents.get(subagent_id, {})
    
    def get_all_subagents(self) -> dict:
        """获取所有子智能体"""
        return self.subagents

# 定义主智能体
def main_agent(state: dict, subagent_manager: SubAgentManager) -> dict:
    """主智能体：管理子智能体"""
    current_task = state.get("current_task", "")
    
    # 创建子智能体
    subagent_id = f"subagent_{uuid.uuid4().hex[:8]}"
    subagent_manager.create_subagent(subagent_id, current_task, "main")
    
    # 执行子智能体
    result = subagent_manager.execute_subagent(subagent_id)
    
    return {
        **state,
        "task_result": result,
        "subagent_id": subagent_id,
        "subagent_status": "completed",
        "subagent_result": result,
        "parent_agent_id": "main"
    }

# 创建工作流图
def create_subagent_graph(subagent_manager: SubAgentManager) -> StateGraph:
    graph = StateGraph(dict)
    graph.add_node("main", lambda state: main_agent(state, subagent_manager))
    graph.set_entry_point("main")
    graph.add_edge("main", END)
    return graph

# 编译并运行
subagent_manager = SubAgentManager()
graph = create_subagent_graph(subagent_manager)
app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "查询北京天气"}],
    "current_task": "查询北京天气",
    "task_result": "",
    "subagent_id": "",
    "subagent_status": "",
    "subagent_result": "",
    "parent_agent_id": ""
})
print(result["task_result"])

---

示例 2：共享上下文并行工作

执行流程：

用户输入: 查询北京天气
    ↓
主智能体: 设置共享上下文
    ↓
创建子智能体 1: 查询天气
创建子智能体 2: 穿衣建议
    ↓
并行执行
    ↓
结果汇总
    ↓
返回结果: 北京今天晴，25度，微风 | 穿衣建议

代码：

import time
import concurrent.futures

# 定义共享上下文
class SharedContext:
    def __init__(self):
        self.data = {}
    
    def set(self, key: str, value: any):
        self.data[key] = value
    
    def get(self, key: str, default=None) -> any:
        return self.data.get(key, default)
    
    def get_all(self) -> dict:
        return self.data.copy()

# 定义子智能体
class SubAgent:
    def __init__(self, agent_id: str, task: str, shared_context: SharedContext):
        self.agent_id = agent_id
        self.task = task
        self.shared_context = shared_context
        self.status = "created"
        self.result = ""
    
    def execute(self) -> str:
        self.status = "running"
        
        # 从共享上下文获取数据
        user_name = self.shared_context.get("user_name", "用户")
        
        # 模拟执行任务
        if "天气" in self.task:
            result = f"{user_name}，北京今天晴，25度，微风"
        elif "穿衣" in self.task:
            result = f"{user_name}，建议穿薄外套"
        else:
            result = f"{user_name}，已处理：{self.task}"
        
        # 更新共享上下文
        self.shared_context.set(f"result_{self.agent_id}", result)
        
        self.status = "completed"
        self.result = result
        
        return result

# 定义主智能体
def main_agent(state: dict, shared_context: SharedContext) -> dict:
    current_task = state.get("current_task", "")
    
    # 设置共享上下文
    shared_context.set("user_name", "宸一")
    shared_context.set("current_task", current_task)
    
    # 创建多个子智能体
    subagents = []
    subagent1 = SubAgent("weather", "查询天气", shared_context)
    subagent2 = SubAgent("clothing", "穿衣建议", shared_context)
    subagents.append(subagent1)
    subagents.append(subagent2)
    
    # 并行执行子智能体
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(subagents)) as executor:
        futures = {executor.submit(subagent.execute): subagent for subagent in subagents}
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            results.append(result)
    
    # 汇总结果
    final_result = " | ".join(results)
    
    return {**state, "task_result": final_result}

# 创建工作流图
def create_shared_context_graph(shared_context: SharedContext) -> StateGraph:
    graph = StateGraph(dict)
    graph.add_node("main", lambda state: main_agent(state, shared_context))
    graph.set_entry_point("main")
    graph.add_edge("main", END)
    return graph

# 编译并运行
shared_context = SharedContext()
graph = create_shared_context_graph(shared_context)
app = graph.compile()

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

---

🎓 学习收获

1. 核心概念理解

• 子智能体系统 = 分工机制
• 共享上下文 = 共享内存
• 生命周期追踪 = 线程状态
• 并行工作 = 多线程并行

2. 子智能体类型对比

子智能体类型	职责	示例
工作者子智能体	执行具体任务	查询天气、计算数学
分析子智能体	分析数据和信息	数据分析、趋势预测
协调子智能体	协调多个子智能体	任务分配、结果汇总

3. 技术实现

• 子智能体管理器 — 管理子智能体的创建、执行、销毁
• 共享上下文 — 子智能体共享信息
• 生命周期追踪 — 追踪状态变化
• 并行执行 — 多个子智能体同时执行

---

🧠 深度思想学习模式

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

传统学习模式：

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

深度思想学习模式：

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

深度思想学习模式的特点

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

---

🎲 随机题型考核

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

传统考核模式：

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

随机题型考核：

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

随机题型考核的特点

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

---

❓ 思考题

第一部分：开放性设计题

1. 设计一个 “多智能体协作系统”

第二部分：深度分析题

1. 分析子智能体系统的本质
2. 分析共享上下文的作用

第三部分：创新设计题

1. 创新的子智能体协作方式

---

📝 总结

Day 15 的学习让我们从"任务追踪"升级到"子智能体系统"。

核心收获：

1. 子智能体系统的核心概念
2. 主智能体按需生成子智能体
3. 共享上下文并行工作
4. 生命周期追踪
5. 深度思想学习模式的实践
6. 随机题型考核的实践

学习路径：

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

下一步：

• Day 16：智能上下文管理
• Day 17：Dream & Distill
• Day 18：综合项目

---

📚 参考资料

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

---

系列文章：

• Day 14：任务追踪
• Day 15：子智能体系统 (本文)
• Day 16：智能上下文管理

---

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