1. 从单体智能到群体协作为什么需要“Agent 生 Agent”在深入 Hermes Agent 的子代理系统源码之前我们得先聊聊一个根本问题为什么一个 Agent 需要、并且能够“生”出另一个 Agent这听起来有点科幻但在当前 AI 应用架构的演进中这正成为一个越来越关键的设计模式。回想一下我们早期接触的 AI 助手无论是简单的命令行工具还是初代的聊天机器人它们大多是一个“单体智能”。你向它提出一个复杂任务比如“帮我分析一下这个项目的代码仓库找出潜在的安全漏洞并生成一份修复报告”。这个单体 Agent 需要自己完成所有事情理解你的自然语言指令、克隆代码仓库、遍历文件结构、调用不同的代码分析工具可能是静态分析、依赖检查、理解每个工具的输出、汇总成人类可读的报告最后再格式化输出。整个过程冗长、脆弱且任何一个环节出错比如某个分析工具崩溃整个任务链就中断了。“Agent 生 Agent”模式的核心思想就是将这个庞大的、多步骤的复杂任务拆解成一系列更小、更专注的子任务并为每个子任务动态地创建一个专门的“子代理”去执行。这个负责拆解和调度的“母体”就是主代理Main Agent。它不再事必躬亲而是转型为一个“管理者”或“协调者”。它的核心能力变成了任务规划Planning、工具调用Tool Calling以及对子代理的“生命周期管理”——包括创建、监控、通信和回收。这样做带来了几个显而易见的好处1. 模块化与可维护性每个子代理职责单一。比如一个专门负责文件系统操作的FileOperatorAgent一个精通BanditPython 安全扫描工具的SecurityScannerAgent一个擅长生成Markdown报告的ReportGeneratorAgent。当需要更新安全扫描逻辑时你只需要修改SecurityScannerAgent不会影响到报告生成或文件操作的逻辑。系统的耦合度大大降低。2. 专业化与性能优化不同的子代理可以拥有不同的“技能包”即工具集和系统提示词System Prompt。CodeAnalyzerAgent的提示词里会充满代码规范和最佳实践而WebScraperAgent的提示词则更关注 HTML 解析和反爬策略。这种专业化能让每个 Agent 在其领域内表现更出色、更可靠。3. 并发与效率提升这是最诱人的一点。一旦任务被拆解成独立的子任务许多子任务是可以并行执行的。主代理在分析完任务后可以同时创建多个子代理。例如在分析项目时可以同时让DependencyCheckAgent去检查requirements.txt让CodeComplexityAgent去计算圈复杂度让LicenseCheckerAgent去扫描开源协议。这比串行执行要快得多。4. 动态性与适应性主代理可以根据任务的实时上下文动态决定需要创建什么样的子代理。它就像一个经验丰富的项目经理面对不同的项目任务会组建不同的临时团队子代理集合。这种动态组装能力使得整个系统极其灵活能够应对千变万化的复杂需求。所以当我们在 Hermes Agent 的源码中看到“子代理系统”时我们本质上是在研究一个多智能体协作框架的实现。它要解决的核心工程问题包括如何定义和描述一个子代理主代理如何创建和初始化它们子代理之间如何通信和共享状态任务结果如何汇总以及如何优雅地处理子代理的失败和超时接下来我们就带着这些问题潜入代码深处。2. 子代理的“基因蓝图”AgentSpec 与 Agent 类的设计在 Hermes 的架构里一个 Agent无论是主代理还是子代理的核心构成是什么我们可以将其类比为一个人的“基因蓝图”它决定了这个 Agent 的“先天能力”和“初始性格”。在代码中这主要由两个核心类来定义AgentSpec和Agent。2.1 AgentSpec静态的能力定义清单AgentSpec是一个数据类很可能使用了Pydantic或dataclasses它描述了一个 Agent 类型的静态、可复用的配置模板。你可以把它想象成工厂里生产机器人的设计图纸。一份AgentSpec通常包含以下关键字段name: 代理类型的名称如“CodeReviewer”,“DataFetcher”。这是它的唯一标识。description: 对该代理职责的详细描述。这个描述至关重要因为它会被主代理的 LLM 用来理解“什么时候该创建这个类型的子代理”。例如“一个专门用于审查 Python 代码风格和潜在错误的代理”。system_prompt: 系统提示词。这是子代理的“核心人格”和“专业知识库”。它定义了子代理应该如何思考、遵循什么规则、具备哪些领域知识。一个CodeReviewer的system_prompt里可能会写入 PEP 8 规范、常见的安全漏洞模式等。tools: 该类型代理可以使用的工具列表。每个工具都是一个可调用的函数例如read_file,run_pytest,search_web。AgentSpec里定义的是工具的名称和引用具体的实现可能注册在一个全局的工具库中。config: 其他配置项例如默认的 LLM 模型、温度参数、是否支持流式响应、会话记忆长度等。在 Hermes 的源码中你可能会看到一个AgentSpecRegistry或类似的注册中心所有预定义的AgentSpec都在这里注册。主代理在需要创建子代理时会查询这个注册中心通过name找到对应的AgentSpec作为模板。# 示例代码结构非真实源码基于常见模式推断 from pydantic import BaseModel from typing import List, Callable, Optional class Tool(BaseModel): name: str func: Callable description: str class AgentSpec(BaseModel): name: str description: str system_prompt: str tools: List[str] # 工具名称列表 model_config: dict {model: gpt-4, temperature: 0.1} # ... 其他字段 # 注册中心示例 class AgentSpecRegistry: _registry: Dict[str, AgentSpec] {} classmethod def register(cls, spec: AgentSpec): cls._registry[spec.name] spec classmethod def get(cls, name: str) - AgentSpec: return cls._registry.get(name) # 预定义一个代码审查代理的蓝图 code_reviewer_spec AgentSpec( nameCodeReviewer, descriptionExpert in reviewing Python code for style, bugs, and security issues., system_promptYou are a senior Python engineer. Review the given code..., tools[read_file, static_analysis, search_documentation], model_config{model: gpt-4, temperature: 0.0} ) AgentSpecRegistry.register(code_reviewer_spec)2.2 Agent 类动态的运行实例有了蓝图AgentSpec就可以实例化出一个个活的、可以交互的Agent对象。Agent类是运行时对象它封装了一次任务执行所需的所有动态上下文。一个Agent实例通常会包含spec: 指向创建它的那个AgentSpec模板。agent_id: 一个唯一的运行时标识符如 UUID用于在多个并发的子代理中区分彼此。llm_client: 实际调用大语言模型的客户端对象如 OpenAI, Anthropic 的客户端。它根据spec中的配置进行初始化。tools: 根据spec.tools中的名称从全局工具库中解析并绑定好的具体可调用函数对象。memory/history: 会话历史记录保存当前代理与用户可能是主代理的对话消息。这对于需要上下文连续性的任务很重要。state: 代理的当前状态如“idle”,“running”,“finished”,“error”。Agent类的核心方法通常是run(task: str) - str或step()。它接收一个任务描述来自主代理结合自己的系统提示词、可用工具和历史记录与 LLM 进行交互决定是直接回复还是调用某个工具直到任务完成或达到停止条件。这里有一个关键的设计抉择子代理是否知道自己是“子代理”在 Hermes 的设计中很可能子代理并不知情。从子代理的视角看它只是一个接收任务、执行任务、返回结果的普通 Agent。它不知道任务来自主代理还是最终用户也不知道还有其他兄弟代理在并行工作。这种设计简化了子代理的逻辑使其保持纯粹和可复用所有协调的复杂性都封装在主代理和子代理管理系统中。3. 主代理的“指挥艺术”任务分解与子代理调度机制现在我们来到了最核心的部分主代理是如何工作的它是如何理解一个复杂任务并将其拆解、分派给子代理的这个过程可以粗略地分为三个环节规划Planning、调度Dispatching和聚合Aggregation。3.1 规划阶段LLM 作为“架构师”当主代理收到一个复杂任务时例如“分析项目X的代码质量”它首先会进入规划阶段。这个阶段主要依靠主代理自身的 LLM通常是一个能力更强的模型如 GPT-4进行“思考”。主代理的system_prompt会被精心设计赋予它项目管理和任务分解的能力。提示词中会明确告诉它“你是一个协调者可以将复杂任务分解为子任务并调用专门的子代理来完成。” 同时它会拿到一份当前可用的AgentSpec清单包括名称和描述。然后主代理的 LLM 会根据用户任务和可用的子代理类型生成一个任务分解计划。这个计划可能是一个 JSON 数组其中每个元素描述了一个子任务[ { “subtask_id”: “1”, “description”: “克隆代码仓库 https://github.com/xxx/projectX 到本地临时目录”, “required_agent”: “GitOperator”, “inputs”: {“repo_url”: “https://github.com/xxx/projectX”} }, { “subtask_id”: “2”, “description”: “使用 Bandit 对项目中的 Python 文件进行静态安全扫描”, “required_agent”: “SecurityScanner”, “dependencies”: [“1”], // 依赖于任务1完成 “inputs”: {“project_path”: “output_of_task_1”} }, { “subtask_id”: “3”, “description”: “计算主要模块的圈复杂度和代码行数”, “required_agent”: “CodeMetricsCalculator”, “dependencies”: [“1”], “inputs”: {“project_path”: “output_of_task_1”} }, { “subtask_id”: “4”, “description”: “整合任务2和任务3的结果生成一份详细的 Markdown 格式报告”, “required_agent”: “ReportGenerator”, “dependencies”: [“2”, “3”], “inputs”: {“security_issues”: “output_of_task_2”, “complexity_metrics”: “output_of_task_3”} } ]这个规划过程本身可能就是一个链式或思维树Tree of Thoughts推理过程。主代理的 LLM 需要理解任务目标、识别子任务间的依赖关系哪些可以并行哪些必须串行并为每个子任务分配合适的代理类型。实操心得规划阶段的稳定性让 LLM 生成结构化的规划如 JSON比让它生成自由文本更可靠。使用 Pydantic 模型来定义SubTask的结构然后让 LLM 以函数调用的方式输出这个模型可以极大地提高输出格式的稳定性。此外规划不一定一步到位。可以设计一个“规划-验证”循环让主代理先生成一个计划然后让另一个“验证代理”或者主代理自己换一个提示词来检查这个计划的合理性和完整性必要时进行修正。这能有效避免因初始规划错误导致整个任务跑偏。3.2 调度与执行阶段子代理管理器的诞生拿到任务计划后主代理自己并不会去执行。它会将计划交给一个核心的子代理管理器SubAgentManager或Orchestrator。这个管理器是子代理系统的“引擎”负责具体的创建工作。实例化子代理管理器根据计划中每个子任务的required_agent字段去AgentSpecRegistry查找对应的AgentSpec。然后以该spec为模板实例化一个新的Agent对象。这个过程包括生成唯一的agent_id。根据spec.model_config初始化 LLM 客户端。根据spec.tools绑定工具函数。设置初始的system_prompt和空的历史记录。管理依赖与执行流管理器会解析子任务之间的dependencies。它会构建一个任务依赖图DAG。没有依赖的任务如上述的任务1可以立即开始执行。有依赖的任务如任务2、3依赖于任务1会等待其依赖任务成功完成后才被放入执行队列。执行与通信对于每个就绪的子任务管理器会调用对应子代理的run()方法并将inputs数据可能包含前序任务的输出作为任务描述的一部分传递进去。子代理独立运行与它自己的 LLM 交互调用工具最终产生一个结果。这里的关键是通信机制。子代理如何将结果返回给管理器通常有两种模式回调函数在创建子代理时管理器传入一个回调函数。子代理完成任务后通过这个回调函数将结果传回。消息队列/事件总线更解耦的方式。子代理将完成事件和结果发布到一个内部消息队列管理器订阅这些事件。这种方式更容易扩展到分布式环境。状态监控与超时处理管理器需要监控所有子代理的运行状态。它需要设置超时机制防止某个子代理卡死例如LLM 调用长时间无响应。如果一个子代理失败或超时管理器需要决定整个任务的处置策略是重试、跳过、还是让整个主任务失败3.3 结果聚合阶段从碎片到整体所有子任务或成功完成的部分都执行完毕后管理器会收集到一堆分散的结果。此时主代理需要再次登场扮演“整合者”的角色。管理器将收集到的所有子任务结果subtask_id和output整理好连同最初的用户请求再次提交给主代理的 LLM。主代理的提示词此时会被调整为“你是一个报告整合者。以下是各个专家子代理的分析结果请将它们综合成一份连贯、完整的最终答案回复给用户。”于是主代理的 LLM 会消化这些中间结果去除冗余理清逻辑生成一份格式优美、结论清晰的最终报告。这个过程可能不止一轮如果中间结果过于复杂主代理可能会先进行一轮摘要再进行最终整合。踩坑实录子代理结果的“信息损耗”子代理返回的可能是纯文本、JSON、甚至是一个文件路径。在聚合时最大的坑是“信息损耗”。例如一个安全扫描子代理返回了10个漏洞每个漏洞有类型、位置、危险等级等结构化信息。如果简单地用字符串拼接把这些结果扔给主代理主代理在总结时可能会漏掉关键细节。最佳实践是强制子代理以结构化的数据格式如 JSON Schema输出结果。管理器在聚合时不是拼接字符串而是维护一个结构化的结果字典。这样在最终整合阶段主代理可以通过函数调用直接获取这些结构化数据生成更准确的报告。我在早期实现时忽略了这点导致最终报告总是丢失具体行号、漏洞CVE编号等关键信息调试起来非常痛苦。4. 深入源码追踪一次“Agent 生 Agent”的完整调用链让我们结合可能的 Hermes 源码结构脑补一次完整的调用流程。假设我们有一个ProjectAnalyzer主代理它注册了GitCloner,SecurityScanner,ReportGen三个子代理规格。步骤1用户发起请求# 用户代码 from hermes import HermesClient client HermesClient() final_result client.run_agent( agent_nameProjectAnalyzer, task请全面分析仓库 https://github.com/example/ai-project 的代码质量和安全性。 )步骤2主代理规划在hermes/agents/main/project_analyzer.py中ProjectAnalyzer.run()方法被触发。它内部的_plan方法会调用 LLM生成我们之前看到的那个 JSON 任务计划。步骤3创建子代理管理器并执行主代理的run方法会实例化一个SubAgentOrchestrator并将计划传递给它。# hermes/core/orchestrator.py class SubAgentOrchestrator: def execute_plan(self, plan: List[SubTask]): task_graph build_dependency_graph(plan) # 构建依赖图 completed_results {} while not all_tasks_done(task_graph): ready_tasks get_ready_tasks(task_graph, completed_results) for task in ready_tasks: # 1. 创建子代理 agent_spec AgentSpecRegistry.get(task.required_agent) sub_agent AgentFactory.create_from_spec( specagent_spec, parent_agent_idself.main_agent_id, task_idtask.subtask_id ) # 2. 准备输入替换依赖项的实际结果 prepared_input resolve_inputs(task.inputs, completed_results) # 3. 异步执行子任务 future self.executor.submit(self._run_single_task, sub_agent, task.description, prepared_input) self.futures[task.subtask_id] future # 4. 等待部分完成更新状态和结果 completed wait_for_any_completion(self.futures) for task_id, result in completed.items(): completed_results[task_id] result mark_task_as_done(task_graph, task_id) # 5. 如果失败根据策略处理重试/失败主任务 if result.status failed: handle_task_failure(task_id, result.error) return completed_results步骤4子代理运行_run_single_task方法会调用子代理的run方法。子代理内部的运行逻辑是标准 ReAct 或类似模式# hermes/agents/base.py class Agent: async def run(self, task_input: str): self.state running messages [ {role: system, content: self.spec.system_prompt}, {role: user, content: task_input} ] while not self._is_task_complete(): llm_response await self.llm_client.chat_completion(messages, toolsself.tools) # ... 解析 LLM 响应判断是调用工具还是直接回复 # ... 执行工具调用将结果追加到 messages # ... 直到 LLM 返回最终答案 self.state finished return llm_response.final_answer步骤5结果聚合所有子任务结果返回给Orchestrator后它调用主代理的_aggregate方法。# 在主代理类中 def _aggregate(self, subtask_results: Dict[str, Any]) - str: aggregation_prompt f 你是一个高级技术分析师。用户最初的要求是{self.original_task} 以下是各个专项分析的结果 {json.dumps(subtask_results, indent2)} 请整合以上所有信息形成一份全面、清晰、有洞察力的最终分析报告。 final_response self.llm_client.chat_completion(...) return final_response步骤6返回最终结果final_result被返回给用户一次完整的“Agent 生 Agent”协作任务结束。5. 高级话题与实战中的“坑”看完了主干流程我们再来探讨几个更深层的问题和实践中必然遇到的挑战。5.1 子代理间的通信超越简单的输入输出在上述流程中子代理之间唯一的联系是通过主代理管理器的依赖关系传递数据。但有些场景需要更灵活的子代理间直接通信。例如Agent A在分析代码时发现一个模糊的库函数它可能需要实时询问专门研究文档的Agent B。Hermes 可能通过以下几种方式支持共享黑板Blackboard一个全局的、结构化的共享内存空间。子代理可以将自己的发现如“在文件X第Y行发现疑似SQL注入”写入黑板也可以从黑板读取其他代理的发现来辅助自己的判断。直接消息传递子代理被赋予一个send_message(to_agent_id, message)的工具。管理器负责路由这些消息。这实现了更动态的协作但复杂度也急剧上升需要处理循环依赖和死锁。发布/订阅事件子代理可以发布特定类型的事件如“code_issue_found”其他代理可以订阅它们关心的事件类型。这种方式非常解耦适合构建事件驱动的智能体系统。在源码中你可能在Agent基类或上下文对象中找到context或blackboard属性这就是实现共享状态的关键。5.2 错误处理与鲁棒性设计多代理系统的错误处理是噩梦也是艺术。错误可能发生在任何层面LLM 调用失败网络超时、API 限额、模型内部错误。必须有重试机制和后备模型如 GPT-4 失败后降级到 GPT-3.5。工具调用失败工具执行异常如文件不存在、命令执行错误。子代理需要能捕获这些异常并将其作为上下文信息反馈给 LLM让 LLM 决定是重试、换种方式还是报错。子代理逻辑失败子代理的 LLM 可能陷入死循环或生成无法解析的内容。需要设置最大轮次限制和超时控制。任务级失败一个关键子任务如“克隆代码”失败会导致后续所有依赖任务无法进行。管理器需要有任务失败策略配置是“fail_fast”立即终止所有“continue_on_error”跳过失败任务继续执行其他独立任务还是“retry”重试 N 次在 Hermes 的Orchestrator代码中你会看到大量的try...except块、状态枚举TaskStatus、以及策略模式的应用。一个健壮的系统其错误处理的代码量可能不亚于核心业务逻辑。5.3 资源管理与性能优化“Agent 生 Agent”听起来很强大但如果不加控制很容易引发“智能体爆炸”——瞬间创建成百上千个子代理耗尽内存和 API 额度。连接池与限流LLM 客户端必须有连接池和严格的速率限制RPM, TPM。Orchestrator在创建子代理时不应立即发起 LLM 调用而应将任务放入一个受控队列。代理实例复用对于无状态的子代理完成一个任务后是否可以重置其会话历史并放入池中供下一个同类任务使用这类似于数据库连接池能避免频繁创建和销毁对象的开销。异步与并发控制Python 的asyncio是处理大量 I/O 操作LLM 调用的利器。整个Orchestrator很可能完全基于异步编程构建。同时必须设置全局的并发子代理数量上限。成本监控每个子代理的每次 LLM 调用都应该记录其使用的模型、Token 数量。管理器需要实时汇总并在接近预算阈值时发出警告或停止创建新任务。5.4 调试与可观测性当十几个子代理同时运行时如何知道系统在干什么哪个环节卡住了结构化日志每个 Agent 的每次动作创建、接收输入、调用工具、LLM 请求/响应、完成、错误都必须打上唯一的agent_id和task_id以结构化格式如 JSON记录。这样可以通过日志聚合工具如 ELK清晰地追踪一个用户请求的完整生命周期。分布式追踪集成 OpenTelemetry 这样的标准。为每个用户请求生成一个trace_id贯穿主代理和所有子代理。每个 LLM 调用、工具调用都作为一个span。这样你可以在 Jaeger 或 Zipkin 上直观地看到一个请求的完整调用链和时间消耗快速定位瓶颈。可视化看板一个简单的 Web 看板实时显示当前运行中的主代理、子代理数量、状态、以及任务队列情况对于运维和调试有巨大帮助。在 Hermes 源码中你可能会在Agent基类的_call_llm,_call_tool等方法中找到装饰器或切面AOP代码它们统一负责注入日志记录和追踪逻辑。6. 从源码到实践构建你自己的子代理系统阅读 Hermes 的源码是为了理解其设计哲学和实现技巧但最终目的是为了能设计出适合自己业务场景的智能体系统。如果你打算从头构建以下是一些接地气的建议1. 起步要小验证核心链路。不要一开始就想着做完整的图形化任务规划和动态通信。先实现最核心的“主代理规划 - 创建单个子代理 - 执行 - 返回结果”闭环。用一个具体的、简单的场景如“让主代理调用一个专门做翻译的子代理”跑通整个流程。确保基础通信、生命周期管理、错误处理是稳固的。2. 定义清晰的 AgentSpec 接口。这是系统的契约。仔细思考一个AgentSpec需要哪些字段。除了name,description,system_prompt,tools你可能还需要version用于灰度升级、timeout、max_turns最大对话轮次、input_schema/output_schema用于结构化输入输出验证等。3. 工具管理是基石。子代理的能力来源于工具。建立一个全局的、安全的工具注册中心。工具函数要有清晰的文档字符串这会被自动转换成 LLM 可理解的描述要有统一的错误返回格式。考虑工具的执行权限控制比如文件写入工具只能写入特定沙箱目录。4. 测试策略至关重要。多智能体系统的测试极其复杂。 *单元测试测试单个Agent的run方法使用 Mock LLM 和 Mock 工具。 *集成测试测试Orchestrator执行一个简单任务计划的全流程同样使用 Mock。 *端到端测试用真实的、但额度很小的 LLM API如 GPT-3.5针对几个核心业务场景进行测试主要验证流程而非结果绝对正确。 *混沌测试故意模拟工具失败、网络超时、LLM返回乱码看系统的恢复和降级能力。5. 性能与成本始终是悬顶之剑。在架构设计早期就要考虑 * 能否用更便宜的小模型如 Claude Haiku, GPT-3.5承担一些简单的子代理角色 * 任务规划主代理思考是否过于频繁和昂贵能否缓存一些常见的任务规划模板 * 子代理的会话历史是否必要对于一次性任务完成任务后立即清空历史可以节省大量 Token。回过头看 Hermes Agent 的子代理系统它本质上是一个基于 LLM 的分布式任务调度与执行框架。它巧妙地将 LLM 的规划能力、专业能力和工具使用能力通过多智能体协作的模式进行了放大。理解它的源码不仅让我们学会如何使用它更让我们洞见了未来 AI 应用架构的一种重要范式从追求单个“全能巨人”转向培育一支分工明确、协作高效的“特种部队”。