从零构建子代理系统：提升AI智能体复杂任务处理能力

1. 项目概述：为什么我们需要“子代理”？

在构建智能体（Agent）系统的路上，很多朋友在实现了基础的任务规划、工具调用和记忆模块后，会感觉系统虽然能跑起来，但处理复杂、多步骤任务时，总显得有些力不从心。要么是主代理的思维链过长，容易“迷失”在细节里；要么是任务拆解得不够精细，导致执行效率低下。这正是我们进入“从零实现自己的Agent”系列第五期的核心驱动力——子代理（Sub-Agent）。

简单来说，子代理就是将一个大任务，分解成多个专业、专注的小任务，并交给不同的“专家”去执行。这就像你是一个项目经理，接到一个“开发一个网站”的大项目。你不会自己既写前端、又写后端、还做测试，而是会组建一个团队：前端工程师、后端工程师、测试工程师。这里的每个工程师，就是一个子代理。主代理（项目经理）负责理解总需求、拆解任务、协调资源并汇总结果；子代理们则在自己的专业领域内，高效、精准地完成被分配的具体工作。

实现子代理架构，能带来几个显著的提升：

1. 解耦与专注：每个子代理只需关注特定类型的任务（如数据查询、文本生成、代码执行），逻辑更清晰，内部状态更简单，也更容易调试和优化。
2. 并行与效率：对于可以独立执行的任务，子代理可以并行运行，大幅缩短整体任务的响应时间。
3. 可扩展性：当需要增加新能力时，你只需开发一个新的、具有特定功能的子代理并将其注册到系统中，而无需大规模修改主代理的核心逻辑。
4. 鲁棒性：单个子代理的失败或异常，可以被隔离和处理，避免导致整个系统崩溃。

本期，我们就来深入探讨如何从零开始，设计并实现一个灵活、高效的子代理系统。我们将超越简单的“函数调用”，构建一个具备自主规划、执行和反馈能力的真正子代理体系。

2. 核心架构设计：主从协作的智能体网络

在动手写代码之前，我们必须先把架构想清楚。一个粗糙的设计会导致后期无尽的修修补补。基于主流实践和我们的需求，我推荐一种“管理者-工作者” (Manager-Worker)的混合架构，它平衡了集中控制与分布式执行的优点。

2.1 系统组件与数据流

整个系统主要由以下几个核心组件构成，它们之间的协作关系构成了数据流的主干。

1. 主代理 (Manager Agent)：

   • 职责： 它是系统的大脑和总指挥。负责接收用户原始请求，进行第一层的任务理解和宏观规划。它不关心具体如何执行，而是决定“要做什么”以及“交给谁做”。
   • 核心能力： 意图识别、任务分解、子代理路由、结果综合与决策。

2. 子代理注册中心 (Sub-Agent Registry)：

   • 职责： 一个全局的目录或字典，保存所有可用子代理的“名片”。每张名片至少包含：子代理唯一ID、能力描述（自然语言）、触发条件/关键词、以及其实例的引用或工厂方法。
   • 作用： 主代理根据任务描述，在这里查询匹配的子代理。这实现了动态发现和松耦合。

3. 子代理 (Worker Agent)：

   • 职责： 领域的专家。接收来自主代理的明确子任务，利用自身内部的规划器、工具集和记忆模块，独立完成该任务，并将结果返回。
   • 核心能力： 专注于特定领域的任务规划、工具调用、状态管理。

4. 任务队列与协调器 (Task Queue & Coordinator)：

   • 职责： 负责管理子任务的执行生命周期。对于需要串行的任务，它维护一个执行顺序；对于可以并行的任务，它负责派发到不同的执行线程或进程。它还负责监控任务状态、处理超时和重试。
   • 作用： 这是实现并行执行和复杂工作流（如循环、条件分支）的关键。

5. 共享上下文与黑板 (Shared Context / Blackboard)：

   • 职责： 一个共享的存储区域，用于在主代理和子代理之间、以及子代理彼此之间传递信息。子任务的结果、中间状态、全局变量都可以放在这里。
   • 作用： 避免让主代理成为所有数据传递的瓶颈，也方便子代理之间进行协作（例如，子代理A的输出是子代理B的输入）。

典型数据流：

1. 用户输入 ->主代理。
2. 主代理分析输入，拆解出子任务列表。
3. 主代理查询注册中心，为每个子任务分配合适的子代理。
4. 主代理将子任务和上下文提交给协调器。
5. 协调器根据依赖关系，将任务放入队列并调度执行。
6. 子代理被唤醒，从队列中领取任务，从共享上下文中获取所需数据，执行任务。
7. 子代理将结果写回共享上下文，并通知协调器任务完成。
8. 协调器通知主代理某个子任务完成。主代理检查是否所有任务完成，或根据中间结果动态调整计划。
9. 所有任务完成后，主代理从共享上下文中汇总最终结果，返回给用户。

注意： 并非所有场景都需要完整的上述组件。对于简单场景，可以省略协调器和任务队列，由主代理直接同步调用子代理。但理解这个完整架构，有助于你在系统复杂时知道如何扩展。

2.2 子代理的两种实现模式

在设计子代理本身时，我们有两种主要模式：

模式一：轻量级函数式代理这种模式下的子代理更像一个“智能函数”。它接收输入参数，内部可能有一个简单的决策逻辑（如基于规则或小型LLM判断），然后调用一个或多个工具完成工作，最后返回输出。它没有复杂的长期记忆或多轮规划能力。

• 适用场景： 执行单一、明确的操作，如“查询天气”、“计算数学公式”、“翻译文本”。
• 优点： 启动快、资源消耗小、执行效率高。
• 缺点： 处理复杂子任务的能力有限。

模式二：重量级自治代理这种子代理本身就是一个功能完整的Agent，拥有自己的思维链（Chain-of-Thought）、工具使用（Tool Use）、甚至短期记忆。主代理给它的可能是一个相对模糊的子目标（如“分析这份数据报告的趋势”），它需要自己规划步骤去完成。

• 适用场景： 子任务本身具有一定复杂性，需要多步推理和工具调用，如“进行竞品分析”、“编写一个功能模块的代码”、“总结会议纪要并提取行动项”。
• 优点： 能力强，能处理复杂子任务，减轻主代理的负担。
• 缺点： 设计复杂，资源消耗大，执行速度可能较慢。

在我们的实现中，我建议采用一种混合策略：系统支持这两种模式。为不同的能力注册不同类型的子代理。例如，CalculatorAgent是函数式的，而DataAnalysisAgent是自治式的。主代理根据任务复杂度来调用它们。

3. 关键实现细节与核心代码拆解

有了架构蓝图，我们开始填充代码。这里我会用Python和伪代码结合的方式，讲解最核心的几个模块的实现。我们假设你已经有了一个基础Agent类，它具备与LLM交互、调用工具的基本能力。

3.1 子代理注册中心：系统的服务发现层

注册中心的核心是一个字典，但我们需要让它更智能一些。它不仅要存储子代理，还要能根据自然语言描述进行匹配。

class SubAgentRegistry: def __init__(self): # 核心存储：agent_id -> {‘description‘， ‘keywords‘， ‘agent_class‘} self._agents = {} # 可以引入一个简单的嵌入模型或TF-IDF来进行语义匹配，这里先用关键词演示 self._keyword_index = {} # keyword -> list(agent_id) def register(self, agent_id, description, keywords, agent_class_or_instance): """注册一个子代理""" self._agents[agent_id] = { 'description': description, 'keywords': [k.lower() for k in keywords], 'handler': agent_class_or_instance # 可以是类，也可以是实例 } # 更新关键词索引 for keyword in self._agents[agent_id]['keywords']: self._keyword_index.setdefault(keyword, []).append(agent_id) def find_agent(self, task_description): """根据任务描述查找最匹配的子代理""" task_desc_lower = task_description.lower() candidate_scores = {} # 方法1：关键词匹配（简单快速） for agent_id, info in self._agents.items(): score = 0 for keyword in info['keywords']: if keyword in task_desc_lower: score += 1 if score > 0: # 可以加入描述相似度计算（如余弦相似度）作为加分项 candidate_scores[agent_id] = score # 方法2：使用LLM进行匹配（更精准但更慢） # 如果关键词匹配结果模糊，可以调用一个小型LLM，让它根据描述和注册信息选择最合适的agent_id # prompt = f“任务：{task_description}。可选代理：{self._agents}。请输出最合适的代理ID。” # agent_id = llm_call(prompt) if not candidate_scores: return None # 返回得分最高的代理ID return max(candidate_scores, key=candidate_scores.get) def get_agent(self, agent_id): """根据ID获取子代理处理器""" info = self._agents.get(agent_id) if not info: raise KeyError(f“Agent {agent_id} not registered”) handler = info['handler'] # 如果注册的是类，则实例化（可以传入共享的上下文等参数） if isinstance(handler, type): return handler() # 这里可以传入配置参数 # 如果注册的是实例，直接返回（单例模式） return handler

实操要点：

• 延迟实例化： 注册时存储类而不是实例，在get_agent时再实例化。这有助于节省内存，特别是子代理很多但并非同时使用时。
• 匹配策略： 初期用关键词匹配足够。当子代理数量增多、能力描述复杂时，务必升级为语义匹配（如用sentence-transformers生成嵌入向量计算相似度）。
• 描述质量： 注册时的description和keywords至关重要。description应清晰说明“我能做什么”，keywords应包含任务场景、工具名称、数据类型的常见词汇。

3.2 主代理：任务分解与路由的核心

主代理的run方法需要重写，加入任务分解和子代理调度逻辑。

class ManagerAgent(BaseAgent): def __init__(self, registry, shared_context, coordinator=None): super().__init__() self.registry = registry self.shared_context = shared_context # 共享上下文对象 self.coordinator = coordinator or SimpleCoordinator() # 协调器，默认为简单同步协调器 self.current_plan = [] def run(self, user_input): # 1. 理解用户意图，生成初始计划 initial_plan = self._create_plan(user_input) self.current_plan = initial_plan print(f“经理代理生成计划： {initial_plan}”) # 2. 为计划中的每个步骤分配子代理 tasks = [] for step in initial_plan: agent_id = self.registry.find_agent(step['description']) if not agent_id: # 如果没有找到合适的代理，可以由主代理尝试处理，或报错 print(f“警告：未找到处理步骤‘{step['description']}’的代理”) continue task = { 'id': step['id'], 'description': step['description'], 'assigned_agent_id': agent_id, 'dependencies': step.get('dependencies', []), # 步骤依赖关系 'status': 'pending' } tasks.append(task) # 3. 将任务提交给协调器执行 final_result = self.coordinator.execute_tasks(tasks, self.shared_context, self.registry) # 4. 整合最终结果并返回 return self._synthesize_result(final_result) def _create_plan(self, user_input): """调用LLM进行任务分解。返回一个步骤列表。""" prompt = f“”" 你是一个经验丰富的项目经理。请将以下用户请求分解成一个有序的步骤列表。 每个步骤应该是一个独立的、可执行的任务。 请以JSON格式输出，包含一个‘steps’数组，每个步骤有‘id‘（数字）、‘description‘（任务描述）和可选的‘dependencies‘（依赖的步骤id列表）字段。 用户请求：{user_input} 例如，对于“帮我查一下北京今天的天气，然后翻译成英文”，输出： {{ “steps”: [ {{“id”: 1, “description”: “查询北京市今天的天气情况”， “dependencies”: []}}, {{“id”: 2, “description”: “将查询到的中文天气信息翻译成英文”， “dependencies”: [1]}} ] }} “”" # 假设self.llm_call返回解析后的JSON plan_json = self.llm_call(prompt, expect_json=True) return plan_json.get(“steps”, [])

注意事项：

• 计划的质量：_create_plan提示词（Prompt）的设计直接决定分解质量。你需要用大量例子来引导LLM产出结构良好、依赖关系清晰的计划。对于复杂领域，可以训练一个专门的规划模型。
• 错误处理： 必须考虑find_agent失败的情况。策略可以是：1) 主代理降级处理；2) 请求用户澄清；3) 尝试寻找功能近似的代理。
• 上下文传递：shared_context需要设计好数据结构，例如用一个字典，键可以是任务ID或数据名称，值存储结果。子代理读写时需遵循约定，避免冲突。

3.3 子代理基类与示例实现

我们定义一个子代理的基类，它继承自基础Agent，但增加了与主代理系统交互的接口。

class SubAgentBase(BaseAgent): def __init__(self, agent_id, capabilities): super().__init__() self.agent_id = agent_id self.capabilities = capabilities # 自身能力描述，可用于注册 def execute(self, task_description, shared_context, **kwargs): """ 子代理执行入口。 :param task_description: 主代理分配的任务描述 :param shared_context: 共享上下文对象，用于读取输入和写入输出 :param kwargs: 其他执行参数 :return: 执行结果（通常也会写入shared_context） """ # 1. 从共享上下文中提取本任务需要的输入数据 # 这可能需要解析task_description或依赖约定 input_data = self._extract_input(task_description, shared_context) # 2. 内部规划与执行（可能涉及多轮LLM调用和工具使用） result = self._internal_execution(input_data, **kwargs) # 3. 将结果写回共享上下文 output_key = f“task_{kwargs.get('task_id', 'unknown')}_result” shared_context.set(output_key, result) # 4. 返回结果 return result def _extract_input(self, task_description, shared_context): """子类需重写：如何根据任务描述从上下文中获取数据""" # 示例：简单地从描述中提取关键词，然后去上下文中找 # 更复杂的可以用LLM解析描述，生成查询指令 raise NotImplementedError def _internal_execution(self, input_data, **kwargs): """子类需重写：子代理核心的执行逻辑""" raise NotImplementedError

一个具体的函数式子代理示例：计算器代理

class CalculatorAgent(SubAgentBase): def __init__(self): # 注册时的描述和关键词 super().__init__(“calculator”, [“计算”， “算术”， “加减乘除”， “数学”， “等于多少”]) # 可以配置一个专门用于数学的LLM，或者直接使用规则/库 def _extract_input(self, task_description, shared_context): # 对于计算器，输入就是任务描述本身中的数学表达式 # 例如任务描述是“计算一下 125 + 37 * 2 的结果” # 我们需要从中提取出 “125 + 37 * 2” # 这里简化处理，实际可能需要用正则或小模型提取 return task_description def _internal_execution(self, input_data, **kwargs): # **安全警告**：直接eval极其危险！切勿在生产环境使用！ # 这里仅作演示。正确做法是使用安全的数学表达式解析库（如`asteval`、`numexpr`） # 或者调用一个被严格限制的LLM/API来完成计算。 try: # 演示：简单提取数字和运算符（非常简陋的演示） expression = input_data.replace(“计算”， “”).replace(“一下”， “”).replace(“的结果”， “”).strip() # 强烈建议使用安全的方法，例如： # result = safe_eval(expression) print(f“计算器代理正在计算表达式： {expression}”) # 此处应替换为安全计算逻辑 # 仅为示例： if “+” in expression: a, b = expression.split(“+”) result = float(a.strip()) + float(b.strip()) else: result = “无法解析的表达式” return {“expression”: expression, “result”: result} except Exception as e: return {“error”: f“计算失败： {str(e)}”}

一个具体的自治式子代理示例：数据分析代理

class DataAnalysisAgent(SubAgentBase): def __init__(self): super().__init__(“data_analyst”, [“分析”， “趋势”， “统计”， “图表”， “数据报告”， “可视化”]) # 这个代理可能需要访问数据库、pandas、绘图库等工具 self.tools.extend([SQLQueryTool(), PandasAnalysisTool(), PlotGenerationTool()]) def _extract_input(self, task_description, shared_context): # 分析任务可能需要从上下文中获取数据集 # 假设数据集在上下文中以‘dataset’为键存储 dataset = shared_context.get(“dataset”) # 也可能需要从描述中解析分析维度，这里简化返回描述和数据集 return {“instruction”: task_description, “data”: dataset} def _internal_execution(self, input_data, **kwargs): instruction = input_data[“instruction”] data = input_data[“data”] # 自治代理的核心：自己规划分析步骤 # 1. 用LLM理解指令，生成分析计划 analysis_plan_prompt = f“”" 你是一个数据分析师。面对数据集和以下请求，请列出具体的分析步骤。 数据集已就绪。请求：{instruction} 请输出步骤列表。 “”" steps = self.llm_call(analysis_plan_prompt) # 假设返回文本步骤 # 2. 为每个步骤选择合适的工具并执行 results = [] for step in steps.split(‘\n’): if “查询” in step: tool = self._get_tool(“SQLQueryTool”) result = tool.use(step, data) elif “统计” in step: tool = self._get_tool(“PandasAnalysisTool”) result = tool.use(step, data) elif “画图” in step or “可视化” in step: tool = self._get_tool(“PlotGenerationTool”) result = tool.use(step, data) else: result = f“未处理步骤： {step}” results.append(result) # 3. 综合所有结果，生成最终分析报告 summary_prompt = f“”" 以下是对数据执行‘{instruction}’请求后得到的各项结果： {results} 请整合成一份简洁、清晰的分析报告。 “”" final_report = self.llm_call(summary_prompt) return {“analysis_report”: final_report, “raw_steps”: results}

3.4 协调器与共享上下文

简单同步协调器：

class SimpleCoordinator: """一个简单的同步协调器，按顺序执行任务""" def execute_tasks(self, tasks, shared_context, registry): results = {} # 拓扑排序处理依赖，这里简化按顺序执行 for task in tasks: print(f“开始执行任务 {task['id']}: {task['description']}”) agent = registry.get_agent(task[‘assigned_agent_id’]) # 执行子代理，传入任务ID和共享上下文 result = agent.execute(task[‘description’], shared_context, task_id=task[‘id’]) results[task[‘id’]] = result task[‘status’] = ‘completed’ return results

共享上下文：

class SharedContext: def __init__(self): self._store = {} self._lock = threading.Lock() # 如果涉及多线程，需要加锁 def set(self, key, value): with self._lock: self._store[key] = value def get(self, key, default=None): with self._lock: return self._store.get(key, default) def update(self, data_dict): with self._lock: self._store.update(data_dict)

4. 实战演练：构建一个智能内容创作流水线

让我们用一个综合例子把上面的零件组装起来。我们要构建一个能自动完成“数据获取->分析->生成报告->制作简报PPT大纲”的智能流水线。

第一步：定义子代理并注册

# 1. 网络搜索代理 class WebSearchAgent(SubAgentBase): # ... 实现，使用SerpAPI或类似工具 pass # 2. 数据分析代理 (复用上面的DataAnalysisAgent) # 3. 报告撰写代理 class ReportWritingAgent(SubAgentBase): def __init__(self): super().__init__(“report_writer”, [“撰写”， “总结”， “报告”， “文章”， “文档”]) self.tools.extend([TemplateTool(), GrammarCheckTool()]) def _internal_execution(self, input_data, **kwargs): # input_data 应包含分析结果和报告要求 analysis = input_data.get(“analysis”) outline = self._generate_outline(analysis) draft = self._write_draft(outline) final_report = self._polish(draft) return final_report # ... 其他内部方法 # 4. PPT大纲生成代理 class PPTOutlineAgent(SubAgentBase): # ... 实现，将报告提炼成PPT结构 pass # 注册 registry = SubAgentRegistry() registry.register(“web_searcher”, “从互联网搜索最新信息和数据”， [“搜索”， “查询”， “获取信息”], WebSearchAgent) registry.register(“data_analyst”, “对数据进行统计分析、趋势挖掘和可视化”， [“分析”， “统计”， “图表”], DataAnalysisAgent) registry.register(“report_writer”, “根据素材撰写结构完整、语言流畅的报告”， [“写作”， “报告”， “总结”], ReportWritingAgent) registry.register(“ppt_maker”, “根据报告内容生成PPT演示文稿的大纲”， [“PPT”， “演示”， “大纲”， “幻灯片”], PPTOutlineAgent)

第二步：主代理处理用户请求用户输入：“帮我调研一下2024年人工智能在医疗领域的最新应用趋势，并整理成一份报告和PPT大纲。”

主代理的_create_plan可能会生成如下计划：

{ “steps”: [ {“id”: 1, “description”: “搜索‘2024年 人工智能 医疗 应用 趋势’的最新中英文资料”， “dependencies”: []}, {“id”: 2, “description”: “对搜索到的资料进行整理、去重和关键信息提取”， “dependencies”: [1]}, {“id”: 3, “description”: “分析提取的信息，总结出主要趋势、应用场景和典型案例”， “dependencies”: [2]}, {“id”: 4, “description”: “基于分析结果，撰写一篇关于‘AI在医疗领域应用趋势’的详细报告”， “dependencies”: [3]}, {“id”: 5, “description”: “根据撰写的报告，生成一份用于演示的PPT大纲”， “dependencies”: [4]} ] }

第三步：协调执行协调器会依次（因为存在依赖）调用：WebSearchAgent->DataAnalysisAgent(用于信息提取和整理) ->DataAnalysisAgent(用于趋势分析) ->ReportWritingAgent->PPTOutlineAgent。每个代理都将自己的产出写入SharedContext，供后续代理使用。

第四步：结果交付最终，主代理从上下文中取出PPT大纲和报告，组合后返回给用户。

5. 避坑指南与性能优化

在实际开发和运行中，你会遇到不少挑战。以下是我踩过坑后总结的经验：

1. 子代理的“边界”与“接口”定义不清

• 问题： 子代理之间职责重叠，或者输入输出格式不统一，导致主代理协调困难。
• 解决： 在设计和注册子代理时，严格定义其“能力契约”。使用清晰的接口规范，例如，规定所有子代理的execute方法必须返回一个包含status、data、error字段的字典。使用共享上下文时，规定好数据键的命名规范（如task_<id>_result）。

2. 任务分解的不可控性

• 问题： 完全依赖LLM进行任务分解，可能产生不切实际、循环依赖或过于琐碎的计划。
• 解决：
  • 提供模板和约束： 在给LLM的提示词中，提供分解模板和规则，如“最多分解为5个步骤”、“每个步骤必须对应一个已注册子代理的能力”。
  • 后置校验与修复： 生成计划后，增加一个校验步骤，检查步骤可行性、依赖闭环等，并尝试自动修复或提示用户。
  • 采用分层规划： 先让LLM进行高层目标分解，然后对每个高层目标，再用更具体的提示词进行细化。

3. 共享上下文的爆炸与混乱

• 问题： 所有中间结果都堆在共享上下文里，难以查找和管理，还可能被意外覆盖。
• 解决：
  • 结构化存储： 不要只用一个大字典。可以设计成类似文件系统的结构，例如/tasks/<task_id>/output，/global/variables等。
  • 版本控制： 对于关键数据，可以保存多个版本。
  • 生命周期管理： 引入数据过期或清理机制，对于临时中间结果，在使用后可以标记为可清理。

4. 错误处理与系统韧性

• 问题： 一个子代理失败导致整个流程卡住。
• 解决：
  • 超时与重试： 为每个子代理任务设置超时，并允许有限次数的重试。
  • 降级策略： 当某个专业子代理失败时，主代理可以尝试用一个更通用但能力稍弱的子代理，或者自己尝试处理核心部分。
  • 隔离与熔断： 监控子代理的健康状态，如果某个子代理频繁失败，暂时将其从注册中心“熔断”，避免拖垮整个系统。

5. 性能瓶颈

• 问题： 大量串行任务导致响应慢；LLM调用是主要耗时点。
• 解决：
  • 异步并行： 实现一个真正的异步协调器（如使用asyncio），让没有依赖关系的任务并行执行。
  • LLM调用优化： 对于轻量子代理，考虑使用小型、快速的本地模型。缓存LLM的常见响应。批量处理可能并行的LLM请求。
  • 资源池： 对于重量级子代理（如加载了大模型的），使用连接池或实例池来管理，避免频繁创建销毁的开销。

6. 调试与监控困难

• 问题： 系统黑盒，出了问题不知道是哪个环节、哪个代理的问题。
• 解决：
  • 全链路日志： 为每个任务和子代理调用生成唯一的trace_id，记录详细的输入、输出、耗时和错误信息。
  • 可视化面板： 开发一个简单的面板，实时显示任务执行状态、依赖图、各个代理的负载情况。
  • 可观测性： 集成像Prometheus这样的监控工具，收集关键指标（如任务成功率、平均耗时、LLM调用次数等）。

实现子代理系统是构建强大Agent的关键一步。它从架构上将一个庞杂的智能体，拆解成一组协同工作的专业模块。开始实现时，可以从一个简单的同步协调器和两三个子代理做起，快速验证流程。随着需求复杂，再逐步引入异步、更智能的路由、更健壮的错误处理。记住，好的架构是演进而来的，关键是保持模块间的清晰边界和良好定义的接口。当你看到一个个专业的“小助手”有条不紊地协作，最终完成一个复杂任务时，那种成就感会让你觉得所有的设计都是值得的。