从NanoBot源码解析子智能体协作架构：轻量级多智能体系统设计

1. 项目缘起从OpenClaw到NanoBot的平替探索最近在社区里关于OpenClaw的讨论热度一直没降下来。这个项目确实厉害它提出的“子智能体”Subagents架构让一个主智能体能够动态地创建、管理和协调多个专门化的子智能体来协同完成复杂任务这个思路非常吸引人。但说实话对于很多想深入学习和二次开发的个人开发者或小团队来说OpenClaw的代码库规模庞大依赖复杂直接上手去啃源码学习曲线相当陡峭。这就好比你想学造一辆精巧的自行车结果面前摆着的是一台航空发动机的图纸虽然原理相通但细节的复杂程度完全不是一个量级。正是在这种背景下我开始关注NanoBot这个项目。它被很多人视为OpenClaw理念的一个轻量级、更易理解的“平替”实现。所谓“平替”核心不在于功能上的完全对等而在于用更简洁的代码、更清晰的架构把那个最核心、最迷人的思想——子智能体协作——给呈现出来。这就像有人用乐高积木搭出了一个能跑的小车模型虽然不能上路但传动、转向的基本原理一目了然对于学习者来说价值巨大。我花了些时间把NanoBot里关于子智能体的源码部分仔细梳理了一遍。这个过程不是简单的代码翻译而是试图理解设计者是如何在有限的代码行数内构建起一套可运行的子智能体协作框架的。这其中有哪些精妙的设计取舍哪些地方为了简洁性做了妥协又有哪些设计是直击子智能体协作本质的这篇解析就是把我阅读NanoBot子智能体模块源码的笔记和思考分享出来。目标读者是对智能体架构感兴趣想了解子智能体如何运作但又暂时被大型框架拒之门外的开发者。我们会深入到NanoBot的具体实现中看看这个“平替”是如何工作的。2. NanoBot子智能体模块的核心设计哲学在拆解具体代码之前我们必须先理解NanoBot在设计子智能体模块时的核心哲学。这决定了我们后面会看到什么样的代码结构。与OpenClaw可能追求的工业级鲁棒性和完备性不同NanoBot的哲学非常明确极简主义与概念验证优先。2.1 极简的抽象与明确的边界NanoBot没有试图去构建一个无所不包的“智能体宇宙”。它对于“子智能体”的抽象非常直接一个能够接收消息、进行处理、并返回消息的独立执行单元。这个单元内部的具体能力是通过提示词Prompt和大语言模型LLM的调用来实现的。这种设计将复杂性从框架层转移到了提示词工程层使得框架本身保持极度轻量。框架只负责三件事1子智能体的生命周期管理创建、销毁2消息的路由与传递3有限的执行状态跟踪。这种极简抽象带来的一个好处是边界异常清晰。作为框架使用者你很清楚框架保证消息能送到并调用你定义的逻辑但逻辑具体怎么实现、效果如何那是你通过提示词和LLM要负责的事情。这种责任分离让调试和问题定位变得相对简单。2.2 动态性与静态性的权衡OpenClaw的子智能体体系强调高度的动态性智能体可以根据任务实时创建和调整。NanoBot在这个点上做了显著的简化。在它的典型实现中子智能体的类型或者说“角色”往往是预定义好的。主智能体更像是一个“调度员”它根据当前上下文从一篮子预先定义好的子智能体“模板”中选择一个或多个来激活和执行任务。这种“静态注册动态调用”的模式大大降低了实现的复杂度。你不需要一个复杂的实时代码生成或加载机制。你只需要在系统初始化时像注册插件一样把各种可能的子智能体如“代码专家”、“文案写手”、“数据分析师”的定义主要是系统提示词和基础配置注册进去。运行时主智能体根据需求“点播”即可。这种权衡牺牲了完全动态生成的灵活性但换来了架构的稳定性和可预测性对于大多数已知领域的任务编排来说已经足够有效。2.3 通信模型的简化基于消息的同步调用在复杂的多智能体系统中通信可以是非常复杂的包括异步消息队列、发布订阅、黑板模型等。NanoBot选择了最简单、最直观的一种同步函数调用式的消息传递。主智能体调用子智能体传入输入消息然后等待子智能体返回输出消息。这个过程在代码层面看起来就像调用一个普通的函数。这种简化带来了巨大的认知便利。你不需要去理解复杂的并发原语或消息中间件。整个协作流程是线性的、易于跟踪的。当然它的局限性也很明显无法处理需要长时间运行或并行执行的子任务。但在NanoBot作为“平替”和概念验证的定位下这个选择是合理的。它先把“智能体间能对话、能协作”这个核心流程跑通更高级的并发模式可以留待后续迭代或由使用者基于此基础自行扩展。3. 源码逐层解析子智能体如何被定义与管理理解了设计哲学我们开始进入代码层面。NanoBot的子智能体模块通常集中在一个或几个核心文件中。我们按照从定义到执行的生命周期来拆解。3.1 子智能体的“蓝图”SubAgent类定义首先框架会定义一个基础的SubAgent类。这个类是所有具体子智能体的基类它规定了子智能体必须有的基本结构和行为。我们来看一个高度简化的示意class SubAgent: def __init__(self, name, system_prompt, llm_client, **kwargs): self.name name self.system_prompt system_prompt self.llm_client llm_client self.memory kwargs.get(memory, []) # 简单的对话记忆 self.config kwargs # 其他配置参数 async def invoke(self, input_message, contextNone): 核心调用方法。主智能体通过调用此方法来“使用”该子智能体。 # 1. 构建本次对话的提示词 messages self._construct_messages(input_message, context) # 2. 调用LLM response await self.llm_client.chat_completion( messagesmessages, **self.config.get(llm_params, {}) # 温度、最大token等参数 ) # 3. 解析和包装响应 output_message self._parse_response(response) # 4. (可选)更新记忆 self._update_memory(input_message, output_message) return output_message def _construct_messages(self, user_input, context): 构建发送给LLM的消息列表。 messages [] # 系统提示词定义子智能体的角色和能力 messages.append({role: system, content: self.system_prompt}) # 注入上下文信息例如主智能体提供的任务背景 if context: messages.append({role: user, content: f上下文信息{context}}) # 本次具体的用户输入 messages.append({role: user, content: user_input}) # 可能包含的历史对话记忆 for mem in self.memory[-5:]: # 只保留最近几轮记忆 messages.append(mem) return messages def _parse_response(self, llm_response): 从LLM的原始响应中提取出结构化的输出消息。 # 这里可能做一些清洗、格式化或结构化提取 # 例如确保输出是JSON或者提取出“思考过程”和“最终答案” content llm_response.choices[0].message.content return {role: assistant, content: content, agent: self.name} def _update_memory(self, input_msg, output_msg): 将本轮对话存入记忆。 self.memory.append(input_msg) self.memory.append(output_msg) # 简单的记忆截断防止无限增长 if len(self.memory) 20: self.memory self.memory[-20:]这个基础类非常清晰。name和system_prompt定义了这个子智能体是谁、擅长什么。llm_client是它思考的“大脑”。核心的invoke方法封装了从接收输入到返回输出的完整过程。_construct_messages是提示词工程发生的地方它决定了子智能体看到什么样的信息。_parse_response提供了对LLM输出进行后处理的钩子。memory实现了一个简易的会话记忆让子智能体在多次调用中能保持一定的连贯性。注意在实际的NanoBot源码中这个类可能更复杂一些会包含错误处理、超时控制、工具调用如果子智能体被赋予使用API或函数的能力等。但上面的简化版本已经抓住了最核心的骨架。3.2 子智能体的“花名册”注册表模式既然子智能体是预定义的就需要一个地方来管理它们。NanoBot通常采用一个注册表Registry模式。这是一个全局的、中心化的字典用来存放所有可用的子智能体“蓝图”或实例。class SubAgentRegistry: _agents {} # 类变量存储注册的智能体 classmethod def register(cls, name, agent_class_or_instance): 注册一个子智能体。可以是一个类也可以是一个实例。 cls._agents[name] agent_class_or_instance classmethod def get(cls, name): 根据名称获取子智能体。如果是类则实例化它。 agent cls._agents.get(name) if agent is None: raise KeyError(fSubAgent {name} not found in registry.) # 如果注册的是类则实例化可能需要传入配置 if isinstance(agent, type): # 这里通常会有从配置文件中读取对应配置的逻辑 config load_agent_config(name) return agent(**config) # 如果已经是实例直接返回 return agent classmethod def list_available(cls): 列出所有已注册的子智能体名称。 return list(cls._agents.keys())这个注册表是连接主智能体和具体子智能体的桥梁。在应用启动时各种子智能体被注册进来。当主智能体决定需要某个专家例如“PythonCodeReviewer”时它就去注册表里按名字“要人”。get方法负责交付一个准备好的、可用的子智能体实例。3.3 定义具体的子智能体以“代码审查员”为例现在我们看看如何利用上面的基础框架定义一个具体的子智能体。假设我们要创建一个专门做Python代码审查的子智能体。# 首先我们可能从基础类继承进行定制 class CodeReviewAgent(SubAgent): def __init__(self, llm_client): # 定义专属的系统提示词 system_prompt 你是一个资深的Python代码审查专家。你的任务是仔细检查用户提供的Python代码找出其中的bug、潜在的性能问题、不符合PEP 8规范的写法、不安全或不优雅的代码片段。 请按以下格式输出你的审查结果 1. **关键问题Critical**会导致程序崩溃、数据错误或安全漏洞的问题。 2. **改进建议Major**影响代码可读性、可维护性或性能的问题。 3. **风格问题Minor**PEP 8规范、命名约定等代码风格问题。 对于每个问题请指出具体的代码行如果可能并解释原因同时提供修改后的代码示例。 你的审查应当严格、专业且具有建设性。 super().__init__(namePythonCodeReviewer, system_promptsystem_prompt, llm_clientllm_client, llm_params{temperature: 0.1}) # 代码审查需要低随机性 # 我们可以重写 _parse_response 方法让输出更结构化 def _parse_response(self, llm_response): raw_content llm_response.choices[0].message.content # 这里可以尝试用正则或LLM再次解析将文本转换为结构化的JSON # 为了简化我们假设LLM已经按照我们要求的格式输出了 structured_output { agent: self.name, review_sections: { critical: [], major: [], minor: [] }, raw_feedback: raw_content } # 简单的行解析逻辑示例实际更复杂 lines raw_content.split(\n) current_section None for line in lines: if **关键问题** in line: current_section critical elif **改进建议** in line: current_section major elif **风格问题** in line: current_section minor elif line.strip().startswith(-) or line.strip().startswith(*): if current_section: structured_output[review_sections][current_section].append(line.strip()) return structured_output # 在应用初始化时注册这个智能体 def initialize_agents(): llm_client get_llm_client() # 获取配置好的LLM客户端 reviewer_agent CodeReviewAgent(llm_client) SubAgentRegistry.register(PythonCodeReviewer, reviewer_agent) # 可以继续注册其他智能体如“DataAnalyst”“CopyWriter”等通过这个例子我们可以看到创建一个具体的子智能体核心就是两件事1编写一个高度针对性的系统提示词定义其角色和输出格式2根据需要定制其输入输出的处理逻辑如重写_parse_response来获得结构化数据。注册之后这个智能体就随时待命了。4. 主智能体的调度逻辑如何指挥子智能体们工作子智能体定义好了注册好了那么谁来指挥它们呢这就是主智能体或称为“协调器”、“调度器”的工作。在NanoBot的架构里主智能体本身通常也是一个基于LLM的智能体但它被赋予了更高的权限理解全局任务并决定调用哪个子智能体。4.1 主智能体的核心循环与工具调用主智能体的核心是一个循环它接收用户的任务思考然后决定下一步动作。关键的一步是它需要具备“调用子智能体”这个能力。在LLM的Function Calling或Tool Calling框架下这通常被实现为一个“工具”Tool。class OrchestratorAgent: def __init__(self, llm_client): self.llm_client llm_client self.available_tools self._setup_tools() # 可用的工具列表包括调用子智能体的工具 def _setup_tools(self): 定义主智能体可以使用的工具其中最重要的就是调用子智能体。 tools [] # 工具1调用子智能体 call_subagent_tool { type: function, function: { name: call_subagent, description: 调用一个专业的子智能体来协助处理任务的某个特定部分。, parameters: { type: object, properties: { subagent_name: { type: string, description: 要调用的子智能体名称。, enum: SubAgentRegistry.list_available() # 动态枚举所有注册的智能体 }, task_description: { type: string, description: 需要子智能体完成的具体任务描述。 }, input_data: { type: string, description: 需要提供给子智能体的输入数据或上下文。 } }, required: [subagent_name, task_description] } } } tools.append(call_subagent_tool) # 可以添加其他工具如搜索网络、查询数据库等 return tools async def process_task(self, user_query): 主处理循环。 conversation_history [{role: user, content: user_query}] max_turns 5 # 防止无限循环 for turn in range(max_turns): # 1. 调用LLM并告诉它可用的工具 response await self.llm_client.chat_completion( messagesconversation_history, toolsself.available_tools, tool_choiceauto # 让LLM决定是否调用工具 ) message response.choices[0].message conversation_history.append(message) # 2. 检查LLM是否决定调用工具比如调用子智能体 if message.tool_calls: for tool_call in message.tool_calls: if tool_call.function.name call_subagent: # 解析工具调用参数 import json args json.loads(tool_call.function.arguments) subagent_name args[subagent_name] task_desc args[task_description] input_data args.get(input_data, ) # 3. 执行工具调用实际调用子智能体 tool_output await self._execute_subagent_call(subagent_name, task_desc, input_data) # 4. 将子智能体的返回结果作为工具执行结果反馈给主智能体LLM conversation_history.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(tool_output, ensure_asciiFalse) }) # 可以处理其他工具... else: # LLM没有调用工具直接生成了最终回复任务结束 final_answer message.content return final_answer return 任务处理达到最大轮数可能过于复杂。 async def _execute_subagent_call(self, name, task_desc, input_data): 实际调用子智能体的私有方法。 try: # 从注册表获取子智能体实例 agent_instance SubAgentRegistry.get(name) # 构建给子智能体的输入消息 # 通常会把主智能体的“任务描述”和“输入数据”合并或格式化后传入 agent_input f任务{task_desc}\n\n输入数据{input_data} if input_data else task_desc # 调用子智能体 agent_output await agent_instance.invoke(agent_input, contextuser_query) return { status: success, subagent: name, result: agent_output } except Exception as e: return { status: error, subagent: name, error: str(e) }这个OrchestratorAgent是大脑中的“指挥官”。它通过_setup_tools将“调用子智能体”这个能力暴露给内部的LLM。当LLM在思考过程中认为“这部分代码审查工作应该交给专家”它就会生成一个call_subagent的工具调用请求。主智能体的代码捕获这个请求解析出要调用谁subagent_name和具体任务task_description然后通过注册表找到对应的子智能体实例将任务派发出去。子智能体返回的结果再被包装成工具执行结果送回到主智能体的对话历史中让主LLM能基于这个结果继续思考或整合。4.2 提示词工程教会主智能体何时以及如何调度主智能体能否有效工作极大程度上取决于给它的系统提示词。你需要在这个提示词里清晰地定义它的角色和调度策略。你是一个智能任务协调大师。你的核心能力是理解复杂任务并将其拆解调用合适的专家子智能体来协同完成。 你拥有以下专家团队 - PythonCodeReviewerPython代码审查专家擅长发现代码缺陷、性能问题和风格问题。 - DataAnalyst数据分析专家擅长解读数据、生成图表描述和统计摘要。 - CopyWriter文案写作专家擅长撰写邮件、报告、营销文案等。 你的工作流程 1. 分析用户请求判断其是否是一个需要多个步骤或专业知识的复杂任务。 2. 如果是简单任务你可以直接回答。 3. 如果是复杂任务思考如何将其拆解。问自己任务的哪个部分最适合交给哪个专家处理 4. 使用你被赋予的call_subagent工具来调用专家。在调用时必须清晰说明 - subagent_name专家名称。 - task_description你需要该专家为你完成的**具体、明确**的子任务。 - input_data提供给专家的必要输入信息如需要审查的代码、需要分析的数据。 5. 等待专家返回结果然后基于所有专家的反馈整合成一个完整、连贯的最终答案回复给用户。 记住你是协调者不是所有领域的专家。信任你的专家团队将专业工作分配给它们。这样的提示词相当于给了主LLM一个明确的“操作规程”。它引导LLM去识别任务的复杂性主动进行任务分解并规范地使用工具。没有这个提示词主LLM可能只会尝试自己回答所有问题而不会想到去调用子智能体。实操心得编写主智能体的系统提示词时“角色定义”和“工作流程描述”至关重要。要用它来“塑造”LLM的行为模式。同时call_subagent工具的description和parameters的description也要写得非常清晰这相当于给LLM的“工具说明书”决定了它调用工具时的准确度。5. 运行流程与实战中的挑战当我们把定义、注册、调度这些部分串联起来一个完整的NanoBot子智能体协作流程就形成了。用户提出请求 - 主智能体分析并可能调用一个或多个子智能体 - 子智能体工作并返回结果 - 主智能体整合结果并回复用户。这个过程在代码上体现为一系列异步的函数调用和消息传递。5.1 一个完整的交互示例假设用户请求“帮我审查下面这段Python代码并写一份简单的分析报告。”code_to_review def calculate_average(numbers): sum 0 for i in range(len(numbers)): sum numbers[i] return sum / len(numbers) 主智能体分析主LLM读取提示词和用户请求。它判断这是一个涉及“代码审查”和“报告撰写”的复合任务。第一次工具调用主LLM决定先调用PythonCodeReviewer。它生成工具调用{ subagent_name: PythonCodeReviewer, task_description: 对提供的Python函数进行全面的代码审查找出所有问题。, input_data: def calculate_average(numbers):\n sum 0\n for i in range(len(numbers)):\n sum numbers[i]\n return sum / len(numbers) }子智能体工作OrchestratorAgent._execute_subagent_call被执行获取PythonCodeReviewer实例并调用其invoke方法。CodeReviewAgent使用其专业的系统提示词和LLM生成审查意见例如变量名sum与内置函数冲突、循环可改用for num in numbers、未处理除零错误等。结果返回审查结果以结构化格式返回给主智能体并作为工具执行结果插入对话历史。第二次工具调用主LLM看到代码审查结果后决定调用CopyWriter。{ subagent_name: CopyWriter, task_description: 根据提供的代码审查结果撰写一份给非技术同事看的、简洁明了的代码质量分析报告摘要。, input_data: [这里放入PythonCodeReviewer返回的结构化结果] }整合与回复CopyWriter返回一份书面报告。主LLM将代码审查要点和书面报告整合形成最终回复给用户“已为您完成代码审查。主要发现1. 变量命名冲突2. 循环可优化3. 缺少异常处理。详细报告如下[整合后的报告]”。5.2 实战中遇到的典型挑战与应对在真正运行这样的系统时你会发现一些在简单Demo中不会暴露的问题。挑战一上下文管理与信息衰减子智能体是独立调用的它默认只看到主智能体在调用工具时传给它的input_data。如果主智能体和用户之间有多轮对话或者子智能体需要了解更早的全局上下文信息就可能丢失或不足。应对技巧在_execute_subagent_call方法中除了传递具体的task_description和input_data我通常会通过context参数将原始的user_query甚至最近几轮的主对话历史摘要也传过去。这样能确保子智能体知道自己正在处理的“大任务”是什么做出的判断会更精准。在子智能体的_construct_messages方法里需要妥善地将这个context插入到提示词中。挑战二工具调用的不可预测性尽管有详细的工具描述和系统提示词LLM生成工具调用的行为仍然有一定随机性。它可能调用错误的子智能体或者给出的task_description模糊不清导致子智能体无法有效工作。应对技巧首先优化工具和提示词描述是关键。其次可以在_execute_subagent_call中加入一层“守卫逻辑”。例如在调用前检查subagent_name是否在注册表中甚至可以做一个简单的任务分类器如果发现任务描述与子智能体的专长明显不匹配比如让“文案写手”去审查代码可以拦截这次调用并返回一个错误信息给主LLM让它重新思考。这增加了系统的鲁棒性。挑战三错误处理与状态回滚子智能体在执行过程中可能出错LLM调用失败、解析错误等或者返回的结果质量很差。主智能体需要能处理这些情况而不是直接崩溃或将错误结果整合进去。应对技巧_execute_subagent_call方法必须要有完善的try...except包裹并将任何异常转化为结构化的错误信息返回如上面代码示例中的{status: error, ...}。在主智能体的提示词中需要明确告知它如何处理工具调用的错误“如果专家返回了错误请分析错误原因决定是重试、换一个专家还是向用户说明情况。” 这教会了主LLM进行简单的故障恢复。挑战四成本与延迟每个子智能体调用都是一次LLM API请求。在复杂任务中多次串行调用会导致总响应时间变长API成本也成倍增加。应对技巧对于无依赖关系的子任务可以考虑并行调用。但这需要修改主智能体的调度逻辑从简单的“思考-调用-等待”循环变为更复杂的“任务图”管理。在NanoBot的简约架构下更务实的做法是优化提示词让主智能体更精准地判断“是否真的需要调用子智能体”以及“能否在一次调用中合并多个问题”。此外为子智能体设置合理的超时和重试机制也能避免因单个调用卡住而导致的整体延迟。6. 从NanoBot看子智能体设计的本质通过对NanoBot源码的解析我们可以跳出具象的代码去思考子智能体架构设计的一些本质问题。NanoBot作为一个优秀的“平替”它清晰地展示了这种架构的核心价值与关键决策点。6.1 核心价值关注点分离与能力复用子智能体模式最根本的价值在于“关注点分离”。它将一个庞大、复杂的智能体系统按功能或领域分解为多个小型的、专注的智能体。CodeReviewAgent只需要关心如何审查代码CopyWriter只需要关心如何组织文字。这种分离使得每个单元的提示词可以设计得极其专业和深入而不必担心不同领域指令之间的相互干扰。同时这些专业化的子智能体成为了可复用的“能力模块”。一旦一个优秀的“代码审查专家”被训练通过提示词出来它可以在任何需要代码审查的任务中被主智能体调用。这避免了在每一个新任务中都需要在提示词里重新描述“如何做好代码审查”的问题。6.2 关键决策点耦合度与通信成本在设计子智能体系统时始终存在一个权衡耦合度与通信成本。NanoBot选择了紧耦合、低通信成本的路径。子智能体通过直接的函数调用与主智能体交互通信就是内存中的对象传递效率极高。但代价是子智能体与主智能体、子智能体之间是紧耦合的它们共享同一套框架、同一种通信协议简单的输入输出字典。这限制了系统的扩展性例如很难将一个用Java写的子智能体接入到这个Python框架中。另一种思路是松耦合、高通信成本的路径例如通过消息队列、REST API甚至专门的智能体通信语言如ACL来交互。这样智能体可以用任何语言编写部署在任何地方。但随之而来的是巨大的复杂性序列化、反序列化、网络延迟、错误处理、服务发现等。OpenClaw可能更倾向于后者而NanoBot明智地选择了前者来实现其“平替”目标——先让核心概念跑起来。6.3 平替的意义概念验证与快速迭代最后回到“平替”这个词。NanoBot的价值不在于它比OpenClaw更强而在于它用一个可理解的、可运行的代码实例为我们验证了子智能体协作这个概念的可行性。它像是一个“最小可行产品”MVP让你能用几百行代码就搭建起一个可演示、可调试的原型。这对于学习、教学、快速验证一个新想法来说是无价的。你可以基于NanoBot的骨架轻松地替换LLM后端、增加新的子智能体类型、或者实验不同的调度策略而不用一开始就陷入庞大框架的细节海洋中。在我自己的使用中我会先用NanoBot的模式快速搭建原型验证任务分解和智能体协作的逻辑是否通顺。当概念验证通过需要向生产环境演进时再去考虑引入OpenClaw那样更健壮、更分布式的基础设施或者基于NanoBot的核心思想自行构建更复杂的通信层。从这个角度看NanoBot不仅是OpenClaw的平替更是通往更复杂多智能体系统的一座非常实用的桥梁。它把那个看似高深的概念拉到了每一个开发者触手可及的地方。