原创Agent 如何管理其他 Agent：四种Sub Agent 模式

在AI Agent系统中，单个Agent的能力往往存在边界，当面对复杂任务时，主Agent需要借助其他子Agent的能力协同完成目标。这种主Agent管理子Agent的模式，我们通常称为Sub Agent模式。不同的业务场景对协作效率、分工粒度、错误容忍度的要求不同，适用的Sub Agent模式也存在差异。本文将介绍四种常见的Sub Agent管理模式，帮助开发者根据需求选择合适的架构。

一、顺序执行模式

顺序执行模式是最基础的Sub Agent管理模式，主Agent将复杂任务拆分为多个有前后依赖关系的子任务，按照预设的顺序依次调用对应的子Agent，前一个子Agent的输出作为后一个子Agent的输入，直到所有子任务完成，主Agent整合最终结果返回给用户。

这种模式适合子任务之间强依赖、必须按序执行的场景，比如内容创作流程：先由资料收集Agent搜集素材，再由大纲生成Agent撰写框架，最后由内容撰写Agent完成完整文章。优点是逻辑清晰、实现简单，缺点是如果某个子Agent执行失败，后续流程会直接中断，容错性较低。

以下是顺序执行模式的简化代码示例：

# 主Agent：顺序调用子Agent完成内容创作任务
class MainAgent:
    def __init__(self):
        # 初始化三个子Agent实例
        self.collect_agent = CollectAgent()
        self.outline_agent = OutlineAgent()
        self.write_agent = WriteAgent()

    def execute_task(self, topic):
        # 第一步：调用资料收集子Agent
        print("开始执行资料收集任务...")
        collect_result = self.collect_agent.run(topic)
        if not collect_result["success"]:
            return {"success": False, "msg": "资料收集失败，流程终止"}

        # 第二步：调用大纲生成子Agent，传入上一步结果
        print("开始执行大纲生成任务...")
        outline_result = self.outline_agent.run(collect_result["data"])
        if not outline_result["success"]:
            return {"success": False, "msg": "大纲生成失败，流程终止"}

        # 第三步：调用内容撰写子Agent，传入上一步结果
        print("开始执行内容撰写任务...")
        write_result = self.write_agent.run(outline_result["data"])
        if not write_result["success"]:
            return {"success": False, "msg": "内容撰写失败，流程终止"}

        # 整合最终结果返回
        return {"success": True, "data": write_result["data"]}


# 子Agent：资料收集
class CollectAgent:
    def run(self, topic):
        # 模拟资料收集逻辑
        return {"success": True, "data": f"关于{topic}的相关资料已收集完成"}


# 子Agent：大纲生成
class OutlineAgent:
    def run(self, collect_data):
        # 模拟大纲生成逻辑
        return {"success": True, "data": f"基于收集的资料，已生成{topic}的内容大纲"}


# 子Agent：内容撰写
class WriteAgent:
    def run(self, outline_data):
        # 模拟内容撰写逻辑
        return {"success": True, "data": f"基于大纲，已完成完整内容撰写"}


# 执行示例
if __name__ == "__main__":
    main_agent = MainAgent()
    result = main_agent.execute_task("AI Agent发展趋势")
    print(result)

二、并行执行模式

并行执行模式中，主Agent将任务拆分为多个无依赖关系的子任务，同时调用多个子Agent并行执行，所有子Agent执行完成后，主Agent再整合所有结果输出。这种模式适合子任务之间相互独立、可以同步推进的场景，比如数据分析任务：同时调用用户行为分析Agent、交易数据分析Agent、流量来源分析Agent，最后整合多份分析报告给出综合结论。

相比顺序执行模式，并行执行可以大幅缩短任务整体耗时，提升响应效率，但需要注意子Agent之间的资源竞争问题，同时要保证所有子Agent都执行完成后再做结果整合，避免部分结果缺失。

以下是并行执行模式的简化代码示例：

import threading

# 主Agent：并行调用子Agent完成数据分析任务
class MainAgent:
    def __init__(self):
        self.user_agent = UserBehaviorAgent()
        self.trade_agent = TradeDataAgent()
        self.flow_agent = FlowSourceAgent()
        # 存储子Agent执行结果
        self.result_dict = {}

    def _run_agent(self, agent, task_params, result_key):
        # 子线程执行子Agent任务的方法
        result = agent.run(task_params)
        self.result_dict[result_key] = result

    def execute_task(self, params):
        # 创建三个线程分别执行三个子Agent
        t1 = threading.Thread(
            target=self._run_agent,
            args=(self.user_agent, params["user_params"], "user_result")
        )
        t2 = threading.Thread(
            target=self._run_agent,
            args=(self.trade_agent, params["trade_params"], "trade_result")
        )
        t3 = threading.Thread(
            target=self._run_agent,
            args=(self.flow_agent, params["flow_params"], "flow_result")
        )

        # 启动所有线程
        t1.start()
        t2.start()
        t3.start()

        # 等待所有线程执行完成
        t1.join()
        t2.join()
        t3.join()

        # 整合所有结果
        return {
            "success": True,
            "user_data": self.result_dict.get("user_result", {}),
            "trade_data": self.result_dict.get("trade_result", {}),
            "flow_data": self.result_dict.get("flow_result", {})
        }


# 子Agent：用户行为分析
class UserBehaviorAgent:
    def run(self, params):
        # 模拟用户行为分析逻辑
        return {"success": True, "data": "用户行为分析报告已生成"}


# 子Agent：交易数据分析
class TradeDataAgent:
    def run(self, params):
        # 模拟交易数据分析逻辑
        return {"success": True, "data": "交易数据分析报告已生成"}


# 子Agent：流量来源分析
class FlowSourceAgent:
    def run(self, params):
        # 模拟流量来源分析逻辑
        return {"success": True, "data": "流量来源分析报告已生成"}


# 执行示例
if __name__ == "__main__":
    main_agent = MainAgent()
    task_params = {
        "user_params": {"start_date": "2024-01-01", "end_date": "2024-01-31"},
        "trade_params": {"start_date": "2024-01-01", "end_date": "2024-01-31"},
        "flow_params": {"start_date": "2024-01-01", "end_date": "2024-01-31"}
    }
    result = main_agent.execute_task(task_params)
    print(result)

三、仲裁调度模式

仲裁调度模式是指主Agent不直接指定调用哪个子Agent，而是先对任务进行解析，判断任务类型，再从多个同类型的子Agent中选择最合适的那个执行。这种模式适合存在多个能力相似但侧重不同的子Agent场景，比如智能客服系统：针对用户问题，主Agent先判断问题属于售前咨询、售后维权还是技术支持，再分别调用对应的售前Agent、售后Agent、技术Agent处理。

这种模式的核心是主Agent的仲裁逻辑，需要提前定义清晰的任务分类规则和子Agent的能力边界，才能保证调度的准确性。优点是资源利用率高，避免调用不匹配的子Agent浪费算力，缺点是对主Agent的仲裁能力要求高，规则定义不恰当容易出现调度错误。

以下是仲裁调度模式的简化代码示例：

# 主Agent：仲裁调度子Agent处理用户问题
class MainAgent:
    def __init__(self):
        # 初始化三个不同领域的子Agent
        self.pre_sale_agent = PreSaleAgent()
        self.after_sale_agent = AfterSaleAgent()
        self.tech_agent = TechAgent()

    def _judge_task_type(self, user_question):
        # 仲裁逻辑：判断用户问题类型
        if "购买" in user_question or "价格" in user_question or "优惠" in user_question:
            return "pre_sale"
        elif "退款" in user_question or "换货" in user_question or "投诉" in user_question:
            return "after_sale"
        elif "无法使用" in user_question or "报错" in user_question or "安装" in user_question:
            return "tech"
        else:
            return "unknown"

    def execute_task(self, user_question):
        # 先判断任务类型
        task_type = self._judge_task_type(user_question)

        # 根据类型调用对应子Agent
        if task_type == "pre_sale":
            return self.pre_sale_agent.run(user_question)
        elif task_type == "after_sale":
            return self.after_sale_agent.run(user_question)
        elif task_type == "tech":
            return self.tech_agent.run(user_question)
        else:
            return {"success": False, "msg": "无法识别问题类型，暂无对应处理Agent"}


# 子Agent：售前咨询
class PreSaleAgent:
    def run(self, question):
        return {"success": True, "data": f"售前Agent已处理问题：{question}，回复：商品当前有8折优惠"}


# 子Agent：售后维权
class AfterSaleAgent:
    def run(self, question):
        return {"success": True, "data": f"售后Agent已处理问题：{question}，回复：7天内可无理由退款"}


# 子Agent：技术支持
class TechAgent:
    def run(self, question):
        return {"success": True, "data": f"技术Agent已处理问题：{question}，回复：可尝试重启应用解决该问题"}


# 执行示例
if __name__ == "__main__":
    main_agent = MainAgent()
    test_question = "我刚买的商品有质量问题，想要退款"
    result = main_agent.execute_task(test_question)
    print(result)

四、动态编排模式

动态编排模式是一种更灵活的Sub Agent管理模式，主Agent没有预设的执行流程，而是根据当前任务的执行状态和中间结果，动态决定下一步调用哪个子Agent，甚至可以在执行过程中拆分新任务、调整执行顺序。这种模式适合任务流程不确定、需要根据实时情况调整的场景，比如复杂的项目策划任务：主Agent先调研项目需求，根据需求发现缺少市场数据，就调用市场调研Agent收集数据，再根据数据发现预算不足，就调用预算调整Agent优化方案，循环迭代直到完成符合要求的策划案。

这种模式对主Agent的决策能力要求最高，通常需要结合大模型的推理能力实现动态判断，优点是适应性极强，可以应对各种复杂多变的任务，缺点是流程不可控，排查问题难度较高，对子Agent的反馈格式也有统一要求，方便主Agent解析结果做下一步决策。

以下是动态编排模式的简化代码示例：

# 主Agent：动态编排子Agent完成项目策划任务
class MainAgent:
    def __init__(self):
        self.demand_agent = DemandAgent()
        self.market_agent = MarketAgent()
        self.budget_agent = BudgetAgent()
        self.plan_agent = PlanAgent()
        # 记录任务执行状态
        self.task_status = {"demand_done": False, "market_done": False, "budget_done": False, "plan_done": False}

    def execute_task(self, project_name):
        # 初始步骤：先调研需求
        current_step = "demand"
        result_cache = {}

        # 循环执行直到策划完成
        while not self.task_status["plan_done"]:
            if current_step == "demand":
                # 调用需求调研子Agent
                demand_result = self.demand_agent.run(project_name)
                result_cache["demand"] = demand_result["data"]
                self.task_status["demand_done"] = True
                # 判断下一步：需求完成后需要市场数据
                current_step = "market"

            elif current_step == "market":
                # 调用市场调研子Agent
                market_result = self.market_agent.run(result_cache["demand"])
                result_cache["market"] = market_result["data"]
                self.task_status["market_done"] = True
                # 判断下一步：市场数据返回后检查预算是否充足
                if market_result.get("budget_enough", True):
                    current_step = "plan"
                else:
                    current_step = "budget"

            elif current_step == "budget":
                # 调用预算调整子Agent
                budget_result = self.budget_agent.run(result_cache["market"])
                result_cache["budget"] = budget_result["data"]
                self.task_status["budget_done"] = True
                # 预算调整完成后进入策划步骤
                current_step = "plan"

            elif current_step == "plan":
                # 调用策划生成子Agent
                plan_result = self.plan_agent.run({
                    "demand": result_cache["demand"],
                    "market": result_cache["market"],
                    "budget": result_cache.get("budget", "预算充足")
                })
                result_cache["plan"] = plan_result["data"]
                self.task_status["plan_done"] = True
                # 策划完成，返回最终结果
                return {"success": True, "data": plan_result["data"]}

        return {"success": False, "msg": "项目策划未完成"}


# 子Agent：需求调研
class DemandAgent:
    def run(self, project_name):
        return {"success": True, "data": f"{project_name}的需求为：面向中小企业提供AI工具服务"}


# 子Agent：市场调研
class MarketAgent:
    def run(self, demand_data):
        # 模拟返回市场数据，假设预算不足
        return {"success": True, "data": "市场调研显示目标用户10万，预计投入成本50万", "budget_enough": False}


# 子Agent：预算调整
class BudgetAgent:
    def run(self, market_data):
        return {"success": True, "data": "调整后预算方案：优先开发核心功能，首期投入30万"}


# 子Agent：策划生成
class PlanAgent:
    def run(self, params):
        return {"success": True, "data": f"基于需求、市场、预算信息，已完成项目策划案：{params}"}


# 执行示例
if __name__ == "__main__":
    main_agent = MainAgent()
    result = main_agent.execute_task("中小企业AI工具项目")
    print(result)

四种模式的选择建议

在实际开发中，我们可以根据任务特性选择合适的Sub Agent模式：

• 如果子任务严格依赖顺序，必须逐个完成，优先选择顺序执行模式
• 如果子任务相互独立，希望缩短整体耗时，优先选择并行执行模式
• 如果存在多个同能力不同侧重点的子Agent，希望按需调度，优先选择仲裁调度模式
• 如果任务流程不确定，需要根据执行情况动态调整，优先选择动态编排模式

当然，实际复杂的Agent系统往往会混合使用多种模式，比如外层用动态编排做整体流程控制，内层对无依赖的子任务采用并行执行提升效率，开发者可以根据业务需求灵活组合，搭建更高效的Agent协作体系。

Agent协作Sub_Agent模式AI系统架构多Agent调度智能体管理 本作品最后修改时间：2026-05-22 05:14:09