四、Sub-Agent子智能体全拆解｜多Agent协作

拆解AI Agent当前最主流、最落地的形态——Sub-Agent（子智能体）！随着AI Agent需要处理的任务越来越复杂（比如企业内容中台搭建、多模态视频生成），单一的“主Agent+Skill”组合早已不够用，而Sub-Agent架构的出现，完美解决了Skill被动执行、无法自主协作的痛点。

本文全程结合可直接运行的实战Demo（自动生成销售周报场景），从核心区别、架构设计、源码解析到通信机制，手把手教你吃透Sub-Agent的核心逻辑，干货拉满，建议收藏备用，新手也能轻松理解～ （Demo源码可直接复用，文末有运行说明）

1. Sub-Agent (自主智能体)：AI Agent的当前终局形态

先一句话讲明白核心：Sub-Agent是Skill的高阶演进，本质是“具备自主思考能力的协作团队”——它以Skill为基础，将多个拥有不同专业能力的子智能体组合起来，实现“自主分工、独立决策、协同作战”，这也是阿里千问、Kimi等新一代大模型强化原生Agent能力的核心方向。

这里必须先厘清一个关键前提：在Level 3（Skill阶段），我们通过“Meta+Instruction+Tools”的三位一体架构，实现了专家能力的标准化封装，但Skill本质上还是“被动打工人”——只有被主Agent或人工调用时，才会按固定SOP干活，没有主动性，也不会灵活调整。

而Sub-Agent不一样，它相当于“升级后的独立员工”，核心构成可以总结为：Sub-Agent = Skill (专业能力) + Memory (记忆) + Planning (规划) + Persona (人设)，四大要素协同，让子智能体既能专业干活，又能主动思考、自主决策，这也是我们Demo中多Agent能自主协作的核心原因。

1.1 核心区别：Sub-Agent 与 Skill 的本质差异（新手避坑）

很多开发者容易混淆Skill和Sub-Agent，这里结合我们的实战Demo，从4个关键维度做对比，一看就懂（建议对照表格理解后续Demo角色）：

维度	Skill (技能)	Sub-Agent (智能体)
主动性	被动等待调用，不会主动发起任务（打工人）	主动思考、拆解任务、自我反思（独立员工）
状态管理	无状态，单次执行后不留记忆（做完就忘）	有状态，留存任务全程关键信息（记笔记）
生命周期	单次请求结束就终止（干完这单就走）	持续运行到任务完成，可动态调整策略（负责到底）
交互对象	只和人或主Agent交互，不会跨Skill协作	可与其他Agent、指挥官交互，协同作战

1.2 架构设计：实战Demo——多Agent协作系统拆解（可直接复用）

光说理论太抽象，我们直接上实战！围绕“自动生成销售周报”这个高频场景，搭建一个简单高效的多Agent协作系统，和VoltAgent等主流多智能体框架逻辑一致，遵循“单一职责原则”，每个角色各司其职，由指挥官统筹，新手也能快速上手。

整个系统由3个核心角色组成，分工明确，协同顺畅，具体设计如下（建议结合后续源码一起看）：

• Orchestrator (指挥官): 整个团队的“项目经理”，不干活只统筹，核心职责是拆解任务、调度Agent、审核结果，相当于多智能体系统的“大脑”。具体做这5件事，缺一不可：

• 接收用户模糊需求（比如“帮我写个销售周报”），自动解析核心要点（要数据趋势、核心结论、规范排版）；

• 拆解复杂任务为可落地的子任务（3步：分析数据→撰写周报→审核完整性）；

• 按需分配任务（数据分析给DataAgent，撰稿给WriterAgent）；

• 审核Agent输出结果，有问题（数据缺失、排版混乱）就让对应Agent返工；

• 汇总最终结果，反馈给用户，完成整个任务闭环。

• DataAgent (数据分析员): 专注数据处理的“专业员工”，核心依赖第三章封装的DataAnalyst Skill，不用重复开发，搭配记忆、规划能力和严谨人设，实现数据自主分析。具体配置（重点看，源码会用到）：

• Skill：复用第三章的DataAnalyst Skill，直接调用CSV读取、数据预览、异常处理等能力；

• Memory：记下来数据路径、异常信息（空值、格式错误）、核心分析结论，方便复盘和审核；

• Planning：自主规划分析流程（不用人工教），比如“先查数据结构→分析销售趋势→标注异常→生成摘要”；

• Persona：严谨细致，不隐瞒数据问题，发现异常及时记录反馈；

• 职责：接收指挥官任务，自主分析数据，输出带趋势、带异常提示的分析摘要。

• WriterAgent (撰稿人): 专注文本撰写的“专业员工”，不用额外工具，依托LLM自身能力，搭配记忆、规划能力和专业人设，自动写规范的销售周报。具体配置：

• Skill：无需额外工具，复用LLM的文案撰写能力，重点做排版、润色；

• Memory：记住DataAgent的分析结果、周报排版规范、指挥官的修改意见；

• Planning：自主规划撰写流程（梳理要点→确定结构→填充内容→润色检查）；

• Persona：专业简洁，贴合企业周报规范，不冗余、不捏造数据；

• 职责：结合分析摘要，自主撰写周报，收到修改意见及时优化。

1.3 运行流程：从需求到交付的完整闭环（无需人工干预）

整个Demo的运行逻辑特别简单，全程无需人工干预，闭环流程可简化为一句话：

User（用户）→ Orchestrator（指挥官拆解任务）→ DataAgent（分析数据）→ WriterAgent（撰写周报）→ Orchestrator（审核）→ User（接收结果）

这种模式就是“中心化编排”，指挥官主导所有信息流转，确保每个Agent的输出都能精准衔接，不脱节、不重复，具体拆解为6个步骤（对照后续源码看，更容易理解）：

1. 需求发起：用户输入模糊需求（比如“帮我写个销售周报”），不用明确具体步骤，系统自动承接；

2. 任务拆解：指挥官解析需求，拆解为3个子任务（读取分析数据、撰写周报、审核完整性）；

3. 数据处理：指挥官调度DataAgent，DataAgent自主调用Skill分析数据，输出分析摘要，反馈给指挥官；

4. 报告撰写：指挥官审核分析摘要无误后，调度WriterAgent，WriterAgent结合摘要撰写周报，反馈给指挥官；

5. 结果审核与整合：指挥官审核周报，有问题（排版乱、要点缺）就让WriterAgent返工，没问题就整合最终版本；

6. 结果交付：指挥官将最终周报反馈给用户，任务完成；同时各Agent留存任务记忆，方便后续复盘。

1.3.1 源码解析：Sub-Agent协作Demo（可直接复制运行）

重点来了！结合我们上传的3个核心文件（agent.py、core.py、team_runner.py），逐模块解析源码，所有代码可直接复制到本地运行，依赖第三章的DataAnalyst Skill和sales_data.csv数据文件（不用额外修改）。

1.3.1.1 基础封装：BaseAgent核心类（core.py解析）

BaseAgent是所有Sub-Agent的“父类”，相当于“员工模板”，封装了Sub-Agent必备的四大核心能力（Skill加载、Memory管理、Planning思考、Persona定义），后续所有子智能体都继承这个类，不用重复开发核心逻辑（重点看注释，理清关键代码作用）。

import json
import os
import sys
from typing import List, Dict, Any, Optional
import importlib.util

# 关键：导入第三章的SkillLoader，复用之前的Skill封装能力，不用重复造轮子
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import get_client, MODEL_NAME
skill_runner_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "03_skill", "skill_runner.py")
spec = importlib.util.spec_from_file_location("skill_runner", skill_runner_path)
skill_runner_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(skill_runner_module)
SkillLoader = skill_runner_module.SkillLoader  # 关联第三章的Skill加载逻辑

class Message:
    """消息类：Agent间通信的“纸条”，标准化传递信息，记录发送者、角色和内容"""
    def __init__(self, role: str, content: str, sender: str = None):
        self.role = role          # 角色（user/assistant/tool）
        self.content = content    # 消息核心内容
        self.sender = sender      # 发送者（比如Orchestrator、DataAgent），实现身份识别

    def to_dict(self):
        """转换为LLM可识别的格式，方便拼接上下文"""
        d = {"role": self.role, "content": self.content}
        if self.sender:
            d["name"] = self.sender
        return d

class BaseAgent:
    def __init__(self, name: str, instructions: str, skills: List[str] = []):
        self.name = name                  # Agent名称（对应Persona人设，比如“DataAgent”）
        self.instructions = instructions  # 人设+工作规范，定义Agent该怎么干活
        self.client = get_client()        # LLM客户端（OpenAI/DeepSeek均可，配置在config.py）
        self.history: List[Message] = []  # 消息历史=Memory，留存交互记录和任务信息
        self.skills: Dict[str, SkillLoader] = {}  # 加载的Skill，复用第三章能力
        
        # 自动加载指定Skill，不用手动调用
        for skill_name in skills:
            print(f"[{self.name}] 正在加载 Skill: {skill_name}...")
            skills_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "03_skill", "skills")
            self.skills[skill_name] = SkillLoader(skill_name, skills_dir=skills_dir)

    def _get_system_prompt(self) -> Dict[str, str]:
        """生成系统提示词：告诉LLM，这个Agent是谁、该干什么、有什么技能"""
        prompt = f"你是 {self.name}。\n\n指令:\n{self.instructions}\n"
        if self.skills:
            prompt += "\n可用 Skill:\n"
            for s_name, s_loader in self.skills.items():
                prompt += f"- {s_name}: {s_loader.meta.get('description', '')}\n"
                prompt += f"  SOP: {s_loader.instruction[:100]}...\n"
        return {"role": "system", "content": prompt}

    def _get_tools_schema(self):
        """获取Skill的工具Schema，让LLM知道自己能调用哪些工具"""
        tools = []
        for s_loader in self.skills.values():
            tools.extend(s_loader.tools_schema)
        return tools if tools else None

    def receive_message(self, message: Message):
        """接收消息，并存入Memory（历史记录），相当于“记笔记”"""
        self.history.append(message)

    def run(self) -> str:
        """核心思考循环=Planning能力：让Agent自主决策、自主调用工具，直到完成任务"""
        messages = [self._get_system_prompt()] + [m.to_dict() for m in self.history]
        tools = self._get_tools_schema()
        
        print(f"[{self.name}] 思考中...")
        max_turns = 5  # 最大思考轮数，避免无限循环（可根据需求调整）
        current_turn = 0
        
        while current_turn < max_turns:
            current_turn += 1
            # 1. 调用LLM，让Agent自主决策：要不要调用工具、调用哪个工具
            kwargs = {"model": MODEL_NAME, "messages": messages}
            if tools:
                kwargs["tools"] = tools
                kwargs["tool_choice"] = "auto"  # 让LLM自动决定是否调用工具
            response = self.client.chat.completions.create(**kwargs)
            response_msg = response.choices[0].message
            
            # 2. 如果需要调用工具，就执行工具并获取结果
            if response_msg.tool_calls:
                print(f"[{self.name}] 第 {current_turn} 轮: 决定使用 {len(response_msg.tool_calls)} 个工具。")
                messages.append(response_msg)
                for tool_call in response_msg.tool_calls:
                    func_name = tool_call.function.name
                    func_args = json.loads(tool_call.function.arguments)
                    # 执行Skill中的工具函数
                    result = None
                    for s_loader in self.skills.values():
                        if hasattr(s_loader.tools_module, func_name):
                            print(f"[{self.name}] 正在执行 {func_name}")
                            try:
                                result = s_loader.execute_tool(func_name, func_args)
                            except Exception as e:
                                result = f"Error: {str(e)}"  # 捕获异常，避免程序崩溃
                            break
                    if result is None:
                        result = f"Error: Tool {func_name} not found."
                    # 把工具执行结果存入Memory，供后续思考使用
                    messages.append({
                        "tool_call_id": tool_call.id,
                        "role": "tool",
                        "name": func_name,
                        "content": str(result)
                    })
            else:
                # 3. 不需要调用工具，直接返回最终结果，思考循环结束
                final_content = response_msg.content
                print(f"[{self.name}] 思考结束。")
                self.history.append(Message("assistant", final_content, sender=self.name))
                return final_content
        return "错误：Agent 超过最大对话轮数。"

核心解析（必看）：BaseAgent的核心价值是“标准化、可复用”，不管是DataAgent还是WriterAgent，只要继承这个类，修改name（人设名称）、instructions（工作规范）和skills（需要的技能），就能快速创建一个新的子智能体，不用重复写记忆、规划相关的代码。

1.3.1.2 子智能体定义：DataAgent与WriterAgent（agent.py解析）

基于上面的BaseAgent父类，我们快速定义两个具体的子智能体，代码非常简洁，重点关注instructions（工作规范）的配置——这部分直接决定了Agent的“人设”和工作方式，贴合我们之前设计的角色分工。

from core import BaseAgent

class DataAgent(BaseAgent):
    def __init__(self):
        # 定义DataAgent的人设和工作规范：严谨的数据分析师，明确工作规则
        instructions="""
        你是一名严谨的数据分析师。
        你的工作是接收数据分析请求，使用 DataAnalyst 技能检查和分析数据，并报告发现。
        关键规则（必须严格遵守）：
        1. 总是先检查文件结构（调用inspect_csv工具），不盲目分析；
        2. 重点关注Sales列，准确总结统计数据（平均值、趋势、异常值）；
        3. 不假设任何数据结论，所有结论必须用工具验证；
        4. 分析结果简洁明了，包含核心统计信息和异常提示，方便撰稿人使用。
        """
        # 继承BaseAgent，加载DataAnalyst Skill，实现数据分析能力
        super().__init__(name="DataAgent", instructions=instructions, skills=["DataAnalyst"])

class WriterAgent(BaseAgent):
    def __init__(self):
        # 定义WriterAgent的人设和工作规范：专业的商业报告撰写人，明确排版和语气要求
        instructions="""
        你是一名专业的商业报告撰写员，擅长撰写企业销售周报。
        你的工作是将原始数据分析结果，转化为润色过、易读、规范的商业报告。
        关键规则（必须严格遵守）：
        1. 语气专业、简洁，避免冗余表述，贴合企业周报风格；
        2. 用清晰的标题构建报告结构（如“一、数据概况”“二、核心发现”“三、总结建议”）；
        3. 重点突出Sales列的趋势和异常，完全基于提供的数据分析结果，不捏造数字；
        4. 报告结尾必须添加简要总结和合理建议，提升报告价值；
        5. 排版工整，段落清晰，不用复杂格式，适合企业内部传阅。
        """
        # 继承BaseAgent，无需额外Skill，依赖LLM自身的撰写能力
        super().__init__(name="WriterAgent", instructions=instructions, skills=[])

核心解析：这部分代码最能体现Sub-Agent的“模块化优势”——新增子智能体时，不用修改BaseAgent，也不用重新开发核心逻辑，只要继承BaseAgent，配置好instructions和skills，就能快速实现一个新角色（比如新增MailAgent，负责发送周报，只需新增一个类即可）。

1.3.1.3 调度逻辑：Orchestrator指挥官（team_runner.py解析）

Orchestrator的核心逻辑都在这个文件里，相当于“项目经理的工作流程”，代码不长，但串联了整个多Agent协作的全流程，重点看“任务拆解→调度Agent→接收结果→汇总交付”的逻辑（可直接复制运行）。

import os
import sys
from core import Message
from agents import DataAgent, WriterAgent

def orchestrator():
    # 0. 初始化配置：指定销售数据路径，关联第三章的sales_data.csv（不用修改路径，按实际调整）
    csv_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "03_skill", "sales_data.csv")
    
    # 1. 初始化Sub-Agent团队：创建DataAgent和WriterAgent实例
    print("🚀 正在初始化 Agent 团队...")
    data_agent = DataAgent()    # 数据分析子智能体
    writer_agent = WriterAgent()# 报告撰写子智能体
    
    # 2. 接收用户需求（模拟真实场景的模糊需求，用户不用明确具体步骤）
    user_request = f"请分析位于 {csv_path} 的销售数据，并写一份简要商业报告，总结关键发现，特别是关于 Sales 列的情况。"
    print(f"\n用户请求: {user_request}")
    print("-" * 50)
    
    # 3. 核心逻辑：任务拆解与Agent调度（Orchestrator的核心工作）
    # 步骤1：调度DataAgent，执行数据分析任务
    print(f"\n[Orchestrator] 正在委派任务给 DataAgent...")
    # 给DataAgent发任务消息，存入其Memory，明确任务要求
    data_task = f"Analyze the data at {csv_path}. Focus on 'Sales' column stats."
    data_agent.receive_message(Message("user", data_task, sender="Orchestrator"))
    # 启动DataAgent，获取数据分析结果（等待DataAgent完成工作）
    data_result = data_agent.run()
    print(f"\n[Orchestrator] 收到 DataAgent 分析结果:\n{data_result}")
    print("-" * 50)
    
    # 步骤2：调度WriterAgent，执行报告撰写任务（传递DataAgent的分析结果）
    print(f"\n[Orchestrator] 正在委派任务给 WriterAgent...")
    # 给WriterAgent发任务，附带DataAgent的分析结果，明确撰写要求
    writer_task = f"Here is the data analysis result:\n{data_result}\n\nPlease write a business report for the user based on this."
    writer_agent.receive_message(Message("user", writer_task, sender="Orchestrator"))
    # 启动WriterAgent，获取最终报告
    final_report = writer_agent.run()
    
    # 4. 结果汇总与交付：将最终报告反馈给用户，完成闭环
    print("\n[Orchestrator] 最终销售周报:")
    print("=" * 50)
    print(final_report)
    print("=" * 50)

# 启动Orchestrator，执行整个协作流程（运行这个文件即可）
if __name__ == "__main__":
    orchestrator()

核心解析（关键必看）：Orchestrator的核心是“显式调度”——它不干活，只负责“分配任务、传递信息、审核结果”，而且是“阻塞式调度”：必须等DataAgent完成分析，才能调度WriterAgent，确保任务有序执行，不会出现“撰稿人等数据”“分析师白干活”的情况。

1.3.1.4 Demo运行说明

很多开发者复制代码后运行失败，这里明确运行条件和步骤，按要求来，一次性成功：

1. 依赖准备：先安装必备依赖（pandas、mcp、openai/DeepSeek），同时确保第三章的DataAnalyst Skill（skills/DataAnalyst目录）配置正确，没有缺失文件；

2. 运行命令：在终端进入项目根目录，执行以下命令，自动启动Orchestrator，调度Agent完成任务：

python 04_sub_agent/team_runner.py

3. 关键特性：运行过程中，终端会打印各Agent的思考日志、工具调用记录（比如DataAgent调用inspect_csv工具），最终输出完整的销售周报；

• 若运行失败，优先检查：数据路径是否正确、Skill目录是否存在、LLM配置（config.py）是否正确；

• 代码可直接复用，只需替换csv_path，就能适配自己的数据集。

1.4 核心优势与总结（开发者重点关注）

结合我们的实战Demo，总结Sub-Agent架构的3个核心优势，这也是它能成为当前AI Agent主流形态的原因，尤其适合开发者落地：

• 自主化程度高，大幅省人力：无需人工拆解任务、调度Agent，指挥官统筹全局，子智能体自主思考、自主执行，甚至能自我反思修正错误，不用全程盯着；

• 扩展性极强，复用性高：新增场景（比如新增“邮件发送”“报告归档”），不用修改原有Agent，只需新增对应Sub-Agent，各Agent独立运行，互不干扰，降低开发成本；

• 容错率高，落地更稳定：指挥官有审核机制，子智能体有记忆能力，能及时发现数据异常、报告不规范等问题，避免单一环节出错导致整个任务失败，适合生产环境落地。

最后用一句话梳理AI Agent的演进主线，帮大家建立完整认知：

Function Call（能动手）→ MCP（标准化动手）→ Skill（专业动手）→ Sub-Agent（自主协同动手）

对于开发者而言，Sub-Agent的落地门槛极低——可基于我们之前实现的Skill，快速搭建子智能体，通过指挥官实现协同，不用从零开发复杂的决策、记忆逻辑，是当前AI Agent工程化落地的最优形态之一。

1.5 通信机制解密 (Communication Mechanism)（进阶知识点）

多Agent协作的核心前提是“高效通信”，我们的Demo中实现了一种“朴实且易落地”的通信模式——显式传递（Explicit Handoff），属于显式路由的一种，适合新手入门，不用复杂配置，就能实现精准通信，避免混乱。

用生活化的“办公室隐喻”，快速理解这种通信逻辑，特别好记：

1.5.1 办公室隐喻

• DataAgent（数据分析师）：闷头干活型，只专注数据分析，不直接和撰稿人沟通；

• WriterAgent（撰稿人）：专注撰写，不直接问分析师要数据，只看指挥官传递的结果；

• Orchestrator（项目经理）：专门传递“工作纸条”，负责在两人之间传递信息，确保衔接顺畅——这就是Demo中通信不脱节的核心。

1.5.2 通信媒介：Message 对象

Agent之间不直接调用函数，而是通过传递Message对象实现交互——这个对象就相当于“工作纸条”，是标准化的通信载体，确保信息完整、可追溯，核心代码我们之前已经解析过，这里再提炼关键：

class Message:
    def __init__(self, role: str, content: str, sender: str = None):
        self.role = role       # 角色 (user/assistant)，告诉接收方“这是谁发的消息”
        self.content = content # 消息核心内容（任务、结果），相当于“纸条上的字”
        self.sender = sender   # 发送者标识（比如“Orchestrator”），实现Agent间身份识别

1.5.3 通信流程：中心化编排（核心重点）

Demo中通信的核心特点是“指挥官主导、阻塞式、线性流转”，简单说就是“先做A，再做B，不做完A，不开始B”，确保任务有序，核心代码片段如下（提炼关键，一看就懂）：

# 第一步：指挥官给分析师派活，传递“纸条”（Message）
data_task = "Analyze the data..."
data_agent.receive_message(Message("user", data_task, sender="Orchestrator"))
data_result = data_agent.run()  # 等待分析师完成，接收结果

# 第二步：指挥官把分析师的结果，转交给撰稿人，再派活
writer_task = f"Here is the data analysis result:\n{data_result}\n请写报告"
writer_agent.receive_message(Message("user", writer_task, sender="Orchestrator"))
final_report = writer_agent.run()  # 等待撰稿人完成，接收最终报告

1.5.4 流程图解（直观理解）

用时序图拆解整个通信和执行流程，清晰看到各Agent的交互逻辑，不用死记硬背代码：