概述官网HKUDS开源GitHub42.1K Star7.4K Fork纳米级ClawdbotOpenClaw复刻Clawdbot几乎所有的核心智能体功能但代码量只有4000行。注NanoBot除HKUDS在开源维护外还有一个NanoBot-AIGitHub1.3K Star191 Fork是一个MCP Agents 构建工具。去掉一切学术装饰和工程冗余后剩下最小可用Agent内核保留一个成熟智能体必须具备的能力闭环网页搜索文件/代码操作定时任务记忆机制多场景Agent模板特点超轻型仅约4k行代码比OpenClaw小99%研究就绪代码易于理解、修改和扩展以进行研究闪电般的快速最小的占用空间意味着更快的启动、更低的资源使用和更快的迭代易于使用一键安装。核心价值在于可掌控性与极低的学习成本。内置四个模版24h实时行情分析师全栈开发助手随时随地执行开发任务私人日程管理可安排会议发送提醒个人知识库把PDF、笔记丢给它随时问答。斜杠命令命令功能/help显示帮助信息/ping测试机器人响应/skills列出可用技能/model切换AI模型/clear清除会话历史围绕Nanobot的生态扩展项目汇总项目描述nanobot-custom个人AI助手支持MiniMax、Gemini多模型切换nanobot-webuiWeb管理面板nanobot-desktopTauriReact桌面端NanoBot.netC#移植版2000行核心代码nanobot-tsTypeScript版本NanoBot-AndroidAndroid移植版专为移动设备优化awesome-nanobot精选资源、工具、Skills集合Agent-Loop相关文件nanobot/agent/loop.py主实现nanobot/agent/context.py上下文构建nanobot/agent/tools/registry.py工具注册和执行nanobot/session/manager.py会话管理AgentLoop是NanoBot的核心引擎负责消息处理的完整生命周期。架构┌────────────────────────────────┐ │ AgentLoop │ ├────────────────────────────────┤ │ - MessageBus (消息总线) │ │ - LLMProvider (LLM 提供商) │ │ - ContextBuilder (上下文构建器) │ │ - SessionManager (会话管理器) │ │ - ToolRegistry (工具注册表) │ │ - SubagentManager (子代理管理器)│ │ - MemoryStore (记忆存储) │ └────────────────────────────────┘ │ ▼ ┌─────────────┐ │ MessageBus │ │ (inbound) │ └──────┬──────┘ │ ▼ ┌─────────────┐ │ _dispatch │ ◄── 处理锁串行执行 │ (msg) │ └──────┬──────┘ │ ▼ ┌─────────────┐ │_process_msg │ └──────┬──────┘ │ ┌─────┴───┐ │ │ Slash命令 普通消息 │ ▼ ┌─────────────┐ │_run_agent_ │ ◄── 核心循环 │ loop │ └─────────────┘核心方法1.__init__初始化源码如下def__init__(self,bus:MessageBus,provider:LLMProvider,workspace:Path,model:str|NoneNone,max_iterations:int40,# 最大工具调用次数temperature:float0.1,max_tokens:int4096,memory_window:int100,# 会话历史窗口大小reasoning_effort:str|NoneNone,# Claude推理模式brave_api_key:str|NoneNone,# Web搜索APIweb_proxy:str|NoneNone,exec_config:ExecToolConfig|NoneNone,cron_service:CronService|NoneNone,restrict_to_workspace:boolFalse,# 限制工具在工作目录session_manager:SessionManager|NoneNone,mcp_servers:dict|NoneNone,# MCP服务器配置channels_config:ChannelsConfig|NoneNone,)初始化时会调用_register_default_tools()注册默认工具。2.run()主循环asyncdefrun(self)-None:运行 Agent 循环将消息作为任务分发以保持对 /stop 的响应。self._runningTrueawaitself._connect_mcp()whileself._running:try:msgawaitasyncio.wait_for(self.bus.consume_inbound(),timeout1.0)exceptasyncio.TimeoutError:continueifmsg.content.strip().lower()/stop:awaitself._handle_stop(msg)else:taskasyncio.create_task(self._dispatch(msg))self._active_tasks.setdefault(msg.session_key,[]).append(task)关键设计使用wait_for(timeout1.0)实现可中断的阻塞每条消息创建独立任务支持并发处理/stop特殊处理立即取消该会话的所有任务3._dispatch()消息分发asyncdef_dispatch(self,msg:InboundMessage)-None:在全局锁下处理消息。asyncwithself._processing_lock:try:responseawaitself._process_message(msg)ifresponseisnotNone:awaitself.bus.publish_outbound(response)exceptasyncio.CancelledError:raiseexceptException:logger.exception(Error processing message)awaitself.bus.publish_outbound(OutboundMessage(...))全局锁用途防止并发修改会话状态确保工具执行的原子性避免竞态条件4._process_message()消息处理核心处理逻辑处理三种消息类型系统消息ifmsg.channelsystem:# 从 chat_id 解析 origin (格式: channel:chat_id)channel,chat_idmsg.chat_id.split(:,1)sessionself.sessions.get_or_create(key)historysession.get_history(max_messagesself.memory_window)messagesself.context.build_messages(history,msg.content,...)final_content,_,all_msgsawaitself._run_agent_loop(messages)self._save_turn(session,all_msgs,...)Slash命令cmdmsg.content.strip().lower()ifcmd/new:# 触发记忆整合清空会话awaitself._consolidate_memory(session,archive_allTrue)session.clear()returnOutboundMessage(contentNew session started.)ifcmd/help:returnOutboundMessage(content命令帮助...)普通消息# 检查是否需要记忆整合unconsolidatedlen(session.messages)-session.last_consolidatedifunconsolidatedself.memory_window:awaitself._consolidate_memory(session)# 设置工具上下文channel, chat_idself._set_tool_context(msg.channel,msg.chat_id,msg_id)# 获取历史构建消息historysession.get_history(max_messagesself.memory_window)initial_messagesself.context.build_messages(history,msg.content,...)# 运行 Agent 循环final_content,_,all_msgsawaitself._run_agent_loop(initial_messages,on_progresson_progressor_bus_progress,)# 保存会话self._save_turn(session,all_msgs,1len(history))self.sessions.save(session)5._run_agent_loop()Agent迭代循环最核心的方法实现与LLM的交互循环asyncdef_run_agent_loop(self,initial_messages:list[dict],on_progress:Callable[...,Awaitable[None]]|NoneNone,)-tuple[str|None,list[str],list[dict]]: 运行 Agent 迭代循环 返回: (最终回复内容, 使用的工具列表, 完整消息列表) messagesinitial_messages iteration0tools_used:list[str][]whileiterationself.max_iterations:iteration1# 调用 LLMresponseawaitself.provider.chat(messagesmessages,toolsself.tools.get_definitions(),# 获取所有工具定义modelself.model,temperatureself.temperature,...)ifresponse.has_tool_calls:# 有工具调用ifon_progress:awaiton_progress(self._tool_hint(response.tool_calls),tool_hintTrue)# 添加assistant消息包含tool_callsmessagesself.context.add_assistant_message(messages,response.content,tool_call_dicts,reasoning_contentresponse.reasoning_content,)# 执行每个工具调用fortool_callinresponse.tool_calls:tools_used.append(tool_call.name)resultawaitself.tools.execute(tool_call.name,tool_call.arguments)messagesself.context.add_tool_result(messages,tool_call.id,tool_call.name,result)else:# 无工具调用返回结果cleanself._strip_think(response.content)messagesself.context.add_assistant_message(messages,clean,...)returnclean,tools_used,messages# 达到最大迭代次数returnfReached max iterations ({self.max_iterations})...,tools_used,messages流程说明循环条件最多max_iterations次默认40LLM调用每次循环调用provider.chat()传入当前消息列表和工具定义工具执行如果LLM返回tool_calls依次执行每个工具消息构建每次工具执行后将结果作为tool角色消息添加完成条件LLM不再返回工具调用时结束工具调用处理工具注册def_register_default_tools(self)-None:allowed_dirself.workspace ifself.restrict_to_workspace elseNone# 文件工具forclsin(ReadFileTool,WriteFileTool,EditFileTool,ListDirTool):self.tools.register(cls(workspaceself.workspace,allowed_dirallowed_dir))# 其他工具self.tools.register(ExecTool(...))self.tools.register(WebSearchTool(api_keyself.brave_api_key))self.tools.register(WebFetchTool())self.tools.register(MessageTool(send_callbackself.bus.publish_outbound))self.tools.register(SpawnTool(managerself.subagents))ifself.cron_service:self.tools.register(CronTool(self.cron_service))工具执行由ToolRegistry.execute()处理asyncdefexecute(self,name:str,params:dict[str,Any])-str:toolself._tools.get(name)ifnottool:returnfError: Tool {name} not found.try:# 参数类型转换paramstool.cast_params(params)# 参数验证errorstool.validate_params(params)iferrors:returnfError: Invalid parameters...# 执行工具resultawaittool.execute(params)returnresultexceptExceptionase:returnfError executing{name}:{str(e)}会话管理会话数据结构dataclassclassSession:key:str# channel:chat_idmessages:list[dict]# 消息列表created_at:datetime updated_at:datetime metadata:dictlast_consolidated:int# 已整合的消息数量保存对话轮次def_save_turn(self,session:Session,messages:list[dict],skip:int)-None:保存新轮次的消息到会话。forminmessages[skip:]:entrydict(m)# 跳过空的 assistant 消息ifroleassistantandnot content andnot entry.get(tool_calls):continue# 截断过长的工具结果ifroletoolandlen(content)self._TOOL_RESULT_MAX_CHARS:entry[content]content[:self._TOOL_RESULT_MAX_CHARS]\n... (truncated)# 去除运行时上下文前缀ifroleuser:ifcontent.startswith(ContextBuilder._RUNTIME_CONTEXT_TAG):partscontent.split(\n\n,1)iflen(parts)1:entry[content]parts[1]entry.setdefault(timestamp,datetime.now().isoformat())session.messages.append(entry)MCP服务器连接asyncdef_connect_mcp(self)-None:连接到配置的 MCP 服务器惰性初始化。ifself._mcp_connectedorself._mcp_connectingornotself._mcp_servers:returnself._mcp_connectingTruefromnanobot.agent.tools.mcpimportconnect_mcp_serverstry:self._mcp_stackAsyncExitStack()awaitself._mcp_stack.__aenter__()awaitconnect_mcp_servers(self._mcp_servers,self.tools,self._mcp_stack)self._mcp_connectedTrueexceptExceptionase:logger.error(Failed to connect MCP servers: {},e)进度回调支持流式输出执行进度asyncdef_bus_progress(content:str,*,tool_hint:boolFalse)-None:metadict(msg.metadataor{})meta[_progress]Truemeta[_tool_hint]tool_hintawaitself.bus.publish_outbound(OutboundMessage(channelmsg.channel,chat_idmsg.chat_id,contentcontent,metadatameta,))子代理通过SpawnTool支持启动子代理处理并行任务self.tools.register(SpawnTool(managerself.subagents))子代理共享相同LLM配置但拥有独立的会话和工具集。最佳实践调试Agent循环启用日志设置LOG_LEVELDEBUG查看详细日志检查消息列表在_run_agent_loop中打印messages验证工具定义确保工具schema正确LLM能正确调用扩展Agent能力添加新工具继承Tool基类实现execute()方法自定义提示词在工作目录创建AGENTS.md覆盖默认行为调整迭代次数复杂任务可能需要增加max_iterations实战安装pip install nanobot-ai源码gitclone https://github.com/HKUDS/nanobot.gitcdnanobot pipinstall-e.nanobot onboard nanobot gateway配置模型{providers:{openrouter:{apiKey:sk-or-v1-xxx}},agents:{defaults:{model:anthropic/claude-opus-4-5}},webSearch:{apiKey:BSA-xxx}}示例python main.py--adapterqq python main.py--configconfig.yaml skills /nanobot skills list# 加载skill /nanobot skillenableweather nanobot agent-mWhat is 22?自定义Skillsfromnanobot.pluginimportPluginclassMySkill(Plugin):namemy_skilldescription自定义技能asyncdefhandle(self,message,context):# 处理消息returnawaitself.reply(自定义回复!)# 注册技能pluginMySkill()