HKUDS开源NanoBot

概述

官网，HKUDS开源（GitHub，42.1K Star，7.4K Fork）纳米级Clawdbot（OpenClaw），复刻Clawdbot几乎所有的核心智能体功能，但代码量只有4000行。

注：NanoBot除HKUDS在开源维护外，还有一个NanoBot-AI，GitHub，1.3K Star，191 Fork，是一个MCP Agents 构建工具。

去掉一切学术装饰和工程冗余后，剩下最小可用Agent内核，保留一个成熟智能体必须具备的能力闭环：

• 网页搜索
• 文件/代码操作
• 定时任务
• 记忆机制
• 多场景Agent模板

特点：

• 超轻型：仅约4k行代码，比OpenClaw小99%；
• 研究就绪：代码易于理解、修改和扩展以进行研究；
• 闪电般的快速：最小的占用空间意味着更快的启动、更低的资源使用和更快的迭代；
• 易于使用：一键安装。

核心价值在于可掌控性，与极低的学习成本。

内置四个模版：

• 24h实时行情分析师
• 全栈开发助手：随时随地执行开发任务；
• 私人日程管理：可安排会议，发送提醒；
• 个人知识库：把PDF、笔记丢给它，随时问答。

斜杠命令

围绕Nanobot的生态扩展项目汇总

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|None=None,max_iterations:int=40,# 最大工具调用次数temperature:float=0.1,max_tokens:int=4096,memory_window:int=100,# 会话历史窗口大小reasoning_effort:str|None=None,# Claude推理模式brave_api_key:str|None=None,# Web搜索APIweb_proxy:str|None=None,exec_config:ExecToolConfig|None=None,cron_service:CronService|None=None,restrict_to_workspace:bool=False,# 限制工具在工作目录session_manager:SessionManager|None=None,mcp_servers:dict|None=None,# MCP服务器配置channels_config:ChannelsConfig|None=None,)

初始化时会调用_register_default_tools()注册默认工具。

2.run()

主循环：

asyncdefrun(self)->None:"""运行 Agent 循环，将消息作为任务分发以保持对 /stop 的响应。"""self._running=Trueawaitself._connect_mcp()whileself._running:try:msg=awaitasyncio.wait_for(self.bus.consume_inbound(),timeout=1.0)exceptasyncio.TimeoutError:continueifmsg.content.strip().lower()=="/stop":awaitself._handle_stop(msg)else:task=asyncio.create_task(self._dispatch(msg))self._active_tasks.setdefault(msg.session_key,[]).append(task)

关键设计：

• 使用wait_for(timeout=1.0)实现可中断的阻塞
• 每条消息创建独立任务，支持并发处理
• /stop特殊处理，立即取消该会话的所有任务

3._dispatch()

消息分发：

asyncdef_dispatch(self,msg:InboundMessage)->None:"""在全局锁下处理消息。"""asyncwithself._processing_lock:try:response=awaitself._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.channel=="system":# 从 chat_id 解析 origin (格式: "channel:chat_id")channel,chat_id=msg.chat_id.split(":",1)session=self.sessions.get_or_create(key)history=session.get_history(max_messages=self.memory_window)messages=self.context.build_messages(history,msg.content,...)final_content,_,all_msgs=awaitself._run_agent_loop(messages)self._save_turn(session,all_msgs,...)

• Slash命令

cmd=msg.content.strip().lower()ifcmd=="/new":# 触发记忆整合，清空会话awaitself._consolidate_memory(session,archive_all=True)session.clear()returnOutboundMessage(content="New session started.")ifcmd=="/help":returnOutboundMessage(content="命令帮助...")

• 普通消息

# 检查是否需要记忆整合unconsolidated=len(session.messages)-session.last_consolidatedifunconsolidated>=self.memory_window:awaitself._consolidate_memory(session)# 设置工具上下文（channel, chat_id）self._set_tool_context(msg.channel,msg.chat_id,msg_id)# 获取历史，构建消息history=session.get_history(max_messages=self.memory_window)initial_messages=self.context.build_messages(history,msg.content,...)# 运行 Agent 循环final_content,_,all_msgs=awaitself._run_agent_loop(initial_messages,on_progress=on_progressor_bus_progress,)# 保存会话self._save_turn(session,all_msgs,1+len(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]]|None=None,)->tuple[str|None,list[str],list[dict]]:""" 运行 Agent 迭代循环 返回: (最终回复内容, 使用的工具列表, 完整消息列表) """messages=initial_messages iteration=0tools_used:list[str]=[]whileiteration<self.max_iterations:iteration+=1# 调用 LLMresponse=awaitself.provider.chat(messages=messages,tools=self.tools.get_definitions(),# 获取所有工具定义model=self.model,temperature=self.temperature,...)ifresponse.has_tool_calls:# 有工具调用ifon_progress:awaiton_progress(self._tool_hint(response.tool_calls),tool_hint=True)# 添加assistant消息（包含tool_calls）messages=self.context.add_assistant_message(messages,response.content,tool_call_dicts,reasoning_content=response.reasoning_content,)# 执行每个工具调用fortool_callinresponse.tool_calls:tools_used.append(tool_call.name)result=awaitself.tools.execute(tool_call.name,tool_call.arguments)messages=self.context.add_tool_result(messages,tool_call.id,tool_call.name,result)else:# 无工具调用，返回结果clean=self._strip_think(response.content)messages=self.context.add_assistant_message(messages,clean,...)returnclean,tools_used,messages# 达到最大迭代次数returnf"Reached max iterations ({self.max_iterations})...",tools_used,messages

流程说明：

1. 循环条件：最多max_iterations次（默认40）
2. LLM调用：每次循环调用provider.chat()，传入当前消息列表和工具定义
3. 工具执行：如果LLM返回tool_calls，依次执行每个工具
4. 消息构建：每次工具执行后，将结果作为tool角色消息添加
5. 完成条件：LLM不再返回工具调用时结束

工具调用处理

工具注册：

def_register_default_tools(self)->None:allowed_dir=self.workspace ifself.restrict_to_workspace elseNone# 文件工具forclsin(ReadFileTool,WriteFileTool,EditFileTool,ListDirTool):self.tools.register(cls(workspace=self.workspace,allowed_dir=allowed_dir))# 其他工具self.tools.register(ExecTool(...))self.tools.register(WebSearchTool(api_key=self.brave_api_key))self.tools.register(WebFetchTool())self.tools.register(MessageTool(send_callback=self.bus.publish_outbound))self.tools.register(SpawnTool(manager=self.subagents))ifself.cron_service:self.tools.register(CronTool(self.cron_service))

工具执行由ToolRegistry.execute()处理：

asyncdefexecute(self,name:str,params:dict[str,Any])->str:tool=self._tools.get(name)ifnottool:returnf"Error: Tool '{name}' not found."try:# 参数类型转换params=tool.cast_params(params)# 参数验证errors=tool.validate_params(params)iferrors:returnf"Error: Invalid parameters..."# 执行工具result=awaittool.execute(params)returnresultexceptExceptionase:returnf"Error executing{name}:{str(e)}"

会话管理

会话数据结构

@dataclassclassSession:key:str# "channel:chat_id"messages: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:]:entry=dict(m)# 跳过空的 assistant 消息ifrole=="assistant"andnot content andnot entry.get("tool_calls"):continue# 截断过长的工具结果ifrole=="tool"andlen(content)>self._TOOL_RESULT_MAX_CHARS:entry["content"]=content[:self._TOOL_RESULT_MAX_CHARS]+"\n... (truncated)"# 去除运行时上下文前缀ifrole=="user":ifcontent.startswith(ContextBuilder._RUNTIME_CONTEXT_TAG):parts=content.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_connecting=Truefromnanobot.agent.tools.mcpimportconnect_mcp_serverstry:self._mcp_stack=AsyncExitStack()awaitself._mcp_stack.__aenter__()awaitconnect_mcp_servers(self._mcp_servers,self.tools,self._mcp_stack)self._mcp_connected=TrueexceptExceptionase:logger.error("Failed to connect MCP servers: {}",e)

进度回调

支持流式输出执行进度：

asyncdef_bus_progress(content:str,*,tool_hint:bool=False)->None:meta=dict(msg.metadataor{})meta["_progress"]=Truemeta["_tool_hint"]=tool_hintawaitself.bus.publish_outbound(OutboundMessage(channel=msg.channel,chat_id=msg.chat_id,content=content,metadata=meta,))

子代理

通过SpawnTool支持启动子代理处理并行任务：

self.tools.register(SpawnTool(manager=self.subagents))

子代理共享相同LLM配置，但拥有独立的会话和工具集。

最佳实践

调试Agent循环

• 启用日志：设置LOG_LEVEL=DEBUG查看详细日志
• 检查消息列表：在_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-m"What is 2+2?"

自定义Skills：

fromnanobot.pluginimportPluginclassMySkill(Plugin):name="my_skill"description="自定义技能"asyncdefhandle(self,message,context):# 处理消息returnawaitself.reply("自定义回复!")# 注册技能plugin=MySkill()