1. 项目概述为什么我们需要“子代理”在构建智能体Agent系统的路上很多朋友在实现了基础的任务规划、工具调用和记忆模块后会感觉系统虽然能跑起来但处理复杂、多步骤任务时总显得有些力不从心。要么是主代理的思维链过长容易“迷失”在细节里要么是任务拆解得不够精细导致执行效率低下。这正是我们进入“从零实现自己的Agent”系列第五期的核心驱动力——子代理Sub-Agent。简单来说子代理就是将一个大任务分解成多个专业、专注的小任务并交给不同的“专家”去执行。这就像你是一个项目经理接到一个“开发一个网站”的大项目。你不会自己既写前端、又写后端、还做测试而是会组建一个团队前端工程师、后端工程师、测试工程师。这里的每个工程师就是一个子代理。主代理项目经理负责理解总需求、拆解任务、协调资源并汇总结果子代理们则在自己的专业领域内高效、精准地完成被分配的具体工作。实现子代理架构能带来几个显著的提升解耦与专注每个子代理只需关注特定类型的任务如数据查询、文本生成、代码执行逻辑更清晰内部状态更简单也更容易调试和优化。并行与效率对于可以独立执行的任务子代理可以并行运行大幅缩短整体任务的响应时间。可扩展性当需要增加新能力时你只需开发一个新的、具有特定功能的子代理并将其注册到系统中而无需大规模修改主代理的核心逻辑。鲁棒性单个子代理的失败或异常可以被隔离和处理避免导致整个系统崩溃。本期我们就来深入探讨如何从零开始设计并实现一个灵活、高效的子代理系统。我们将超越简单的“函数调用”构建一个具备自主规划、执行和反馈能力的真正子代理体系。2. 核心架构设计主从协作的智能体网络在动手写代码之前我们必须先把架构想清楚。一个粗糙的设计会导致后期无尽的修修补补。基于主流实践和我们的需求我推荐一种“管理者-工作者” (Manager-Worker)的混合架构它平衡了集中控制与分布式执行的优点。2.1 系统组件与数据流整个系统主要由以下几个核心组件构成它们之间的协作关系构成了数据流的主干。主代理 (Manager Agent)职责 它是系统的大脑和总指挥。负责接收用户原始请求进行第一层的任务理解和宏观规划。它不关心具体如何执行而是决定“要做什么”以及“交给谁做”。核心能力 意图识别、任务分解、子代理路由、结果综合与决策。子代理注册中心 (Sub-Agent Registry)职责 一个全局的目录或字典保存所有可用子代理的“名片”。每张名片至少包含子代理唯一ID、能力描述自然语言、触发条件/关键词、以及其实例的引用或工厂方法。作用 主代理根据任务描述在这里查询匹配的子代理。这实现了动态发现和松耦合。子代理 (Worker Agent)职责 领域的专家。接收来自主代理的明确子任务利用自身内部的规划器、工具集和记忆模块独立完成该任务并将结果返回。核心能力 专注于特定领域的任务规划、工具调用、状态管理。任务队列与协调器 (Task Queue Coordinator)职责 负责管理子任务的执行生命周期。对于需要串行的任务它维护一个执行顺序对于可以并行的任务它负责派发到不同的执行线程或进程。它还负责监控任务状态、处理超时和重试。作用 这是实现并行执行和复杂工作流如循环、条件分支的关键。共享上下文与黑板 (Shared Context / Blackboard)职责 一个共享的存储区域用于在主代理和子代理之间、以及子代理彼此之间传递信息。子任务的结果、中间状态、全局变量都可以放在这里。作用 避免让主代理成为所有数据传递的瓶颈也方便子代理之间进行协作例如子代理A的输出是子代理B的输入。典型数据流用户输入 -主代理。主代理分析输入拆解出子任务列表。主代理查询注册中心为每个子任务分配合适的子代理。主代理将子任务和上下文提交给协调器。协调器根据依赖关系将任务放入队列并调度执行。子代理被唤醒从队列中领取任务从共享上下文中获取所需数据执行任务。子代理将结果写回共享上下文并通知协调器任务完成。协调器通知主代理某个子任务完成。主代理检查是否所有任务完成或根据中间结果动态调整计划。所有任务完成后主代理从共享上下文中汇总最终结果返回给用户。注意 并非所有场景都需要完整的上述组件。对于简单场景可以省略协调器和任务队列由主代理直接同步调用子代理。但理解这个完整架构有助于你在系统复杂时知道如何扩展。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, keycandidate_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, coordinatorNone): 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_jsonTrue) 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_idtask[‘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, defaultNone): 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的关键一步。它从架构上将一个庞杂的智能体拆解成一组协同工作的专业模块。开始实现时可以从一个简单的同步协调器和两三个子代理做起快速验证流程。随着需求复杂再逐步引入异步、更智能的路由、更健壮的错误处理。记住好的架构是演进而来的关键是保持模块间的清晰边界和良好定义的接口。当你看到一个个专业的“小助手”有条不紊地协作最终完成一个复杂任务时那种成就感会让你觉得所有的设计都是值得的。