nanobot 是由香港大学数据科学学院（HKUDS）开源的多渠道、可扩展的智能机器人框架。该框架核心目标是构建灵活、可定制的智能机器人系统，支持飞书、Telegram、WhatsApp 等多渠道接入，具备智能对话交互、技能扩展、定时任务、健康监控等核心能力，适用于构建企业级 / 个人级的智能助理、自动化运维机器人、多平台消息聚合机器人等场景。
仓库地址：https://github.com/HKUDS/nanobot.git
定义了两个子命令：channels和cron，分别用于管理频道和定时任务。
main.py文件是 项目启动入口，负责解析启动参数、初始化核心组件（如事件总线、agent、渠道）、启动机器人主服务。
以typer.Typer作为入口，定义了命令行工具的名称、帮助信息以及是否显示帮助信息。
简单的一次飞书访问的整体架构的执行流程为
channels模块负责管理机器人与不同渠道的连接，包括飞书、Telegram、WhatsApp等。它使用Python的asyncio库实现异步连接管理，支持连接监听和发布功能。
模块的基类的核心代码在 nanobot/channels/base.py 中，定义了两个子命令：send和_handle_message，分别用于发送消息和接受消息。
class BaseChannel(ABC):
"""
Abstract base class for chat channel implementations.
Each channel (Telegram, Discord, etc.) should implement this interface
to integrate with the nanobot message bus.
"""
name: str = "base"
def __init__(self, config: Any, bus: MessageBus):
"""
Initialize the channel.
Args:
config: Channel-specific configuration.
bus: The message bus for communication.
"""
self.config = config
self.bus = bus
self._running = False
MessageBus是一个异步消息总线，将聊天通道与代理核心解耦。它有两个以asyncio.Queue为基础的异步队列，inbound和outbound，分别用于接收和发送消息。
tips：asyncio.Queue：这是 Python 异步编程中的一个标准数据结构，它是一个线程安全的队列，专门为异步编程设计。它内部使用协友（coroutine）来确保在多任务环境下的安全访问。
其中 OutboundMessage 和 InboundMessage 是两个消息类，分别用于封装发送和接收的消息。二者都是@dataclass 装饰的类，用于简化对象的创建和操作。
tips: @dataclass 模块提供了一个装饰器和一些函数，用于自动为用户自定义的类添加生成的 special method 例如 init() 和 repr()
其他的feishu.py, telegram.py, whatsapp.py等文件，都是继承自BaseChannel，并实现了自己的send和_handle_message方法。为各具体渠道的实现类，基于base.py的抽象接口，适配对应平台的 API（如飞书开放平台、Telegram Bot API），处理消息编解码、事件回调、权限校验等渠道特有逻辑。
manager.py 是渠道管理器，统一注册 / 初始化 / 销毁所有渠道实例，负责消息路由（将不同渠道的用户消息分发到事件总线 /agent 核心），监控渠道在线状态。
核心代码是在 nanobot/channels/manager.py，它负责管理所有的聊天渠道，包括初始化、启动和停止渠道，以及路由消息。
class ChannelManager:
"""
Manages chat channels and coordinates message routing.
Responsibilities:
- Initialize enabled channels (Telegram, WhatsApp, etc.)
- Start/stop channels
- Route outbound messages
"""
def __init__(self, config: Config, bus: MessageBus):
self.config = config
self.bus = bus
self.channels: dict[str, BaseChannel] = {}
self._dispatch_task: asyncio.Task | None = None
self._init_channels()
还有一个核心函数就是启动函数，先启动一个协程，用于初始化分发识别器，然后再启动各个渠道的执行状态。
async def start_all(self) -> None:
"""Start WhatsApp channel and the outbound dispatcher."""
if not self.channels:
logger.warning("No channels enabled")
return
# Start outbound dispatcher
self._dispatch_task = asyncio.create_task(self._dispatch_outbound()) # 将外发消息分派到适当的渠道
# Start WhatsApp channel
tasks = []
for name, channel in self.channels.items():
logger.info(f"Starting {name} channel...")
tasks.append(asyncio.create_task(channel.start()))
# Wait for all to complete (they should run forever)
await asyncio.gather(*tasks, return_exceptions=True)
nanobot/config/ - 配置管理模块，负责读取和解析配置文件，提供配置项的访问和修改功能。
其中loader.py 是加载和保存配置文件的操作模块，
其中schema.py 是配置文件的结构定义模块，定义了AgentsConfig（调用Agent的API等），ChannelsConfig（渠道的配置信息，飞书，whatsapp的api信息），ProvidersConfig（LLM提供商的信息，api key和URL信息），GatewayConfig（网关的IP和PORT），ToolsConfig（包含网络搜索工具和命令行执行工具）四个配置项的结构。
nanobot/cli/ - 命令行交互模块，负责处理命令行输入，提供命令行接口，包括启动、停止、配置等命令。
基于 Click/Argparse 等库实现具体的 CLI 指令（如nanobot start启动机器人、nanobot skill add新增技能、nanobot channel list查看渠道状态）
nanobot/heartbeat/ - 健康监控服务，负责定期检查各个组件的健康状态，包括消息队列、数据库连接等。使用Python的asyncio库实现异步健康检查，支持健康检查结果的通知和记录。
service.py 心跳服务核心，定时检查渠道、agent、事件总线等核心组件的健康状态，支持状态上报（如推送到监控平台）、故障自动恢复（如重启离线渠道）。
class HeartbeatService:
"""
Periodic heartbeat service that wakes the agent to check for tasks.
The agent reads HEARTBEAT.md from the workspace and executes any
tasks listed there. If nothing needs attention, it replies HEARTBEAT_OK.
"""
def __init__(
self,
workspace: Path,
on_heartbeat: Callable[[str], Coroutine[Any, Any, str]] | None = None,
interval_s: int = DEFAULT_HEARTBEAT_INTERVAL_S,
enabled: bool = True,
):
self.workspace = workspace
self.on_heartbeat = on_heartbeat
self.interval_s = interval_s
self.enabled = enabled
self._running = False
self._task: asyncio.Task | None = None
每次心跳都会重复执行tick函数，tick函数的最重要组成是 self.on_heartbeat(HEARTBEAT_PROMPT)
该 HEARTBEAT_PROMPT 的定义是 """Read HEARTBEAT.md in your workspace (if it exists). Follow any instructions or tasks listed there. If nothing needs attention, reply with just: HEARTBEAT_OK"""
nanobot/agent/ - 核心职责：机器人 “智能” 的核心载体，处理用户意图解析、技能调度、对话上下文管理、响应生成等核心逻辑。是项目最核心的模块。
AgentLoop 脚本在 nanobot/agent/loop.py 文件下，是核心模块中的核心类。
核心职责：负责接收和处理消息，通过MessageBus和其他组件通信，使用LLMProvider提供LLM服务，集成了多种工具（文件操作，网络搜索，任务执行）
工作流程：
来看一下最重要的函数run函数。
async def run(self) -> None:
"""Run the agent loop, dispatching messages as tasks to stay responsive to /stop."""
self._running = True
await self._connect_mcp()
logger.info("Agent loop started")
while self._running:
try:
msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
except asyncio.TimeoutError:
continue
if msg.content.strip().lower() == "/stop":
await self._handle_stop(msg)
else:
task = asyncio.create_task(self._dispatch(msg))
self._active_tasks.setdefault(msg.session_key, []).append(task)
task.add_done_callback(lambda t, k=msg.session_key: self._active_tasks.get(k, []) and self._active_tasks[k].remove(t) if t in self._active_tasks.get(k, []) else None)
首先是建立mcp服务，然后进入一个while循环，不断从bus中获取消息，如果消息是“/stop”则调用_handle_stop函数，否则调用_dispatch函数，将消息分发给不同的处理器。
连接MCP服务
服务里面主要是调用connect_mcp_servers函数，这个函数会遍历self._mcp_servers，对每一个server调用connect_mcp_server函数，将返回的context manager添加到self._mcp_stack中。
self._mcp_stack = AsyncExitStack() # AsyncExitStack 是contextlib模块的一部分，专门用于管理异步上下文管理器的生命周期
await self._mcp_stack.__aenter__() # 在显式地调用 AsyncExitStack 实例的异步上下文管理器的进入方法
await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack)
self._mcp_connected = True
停止服务
停止服务主要是调用_handle_stop函数，这个函数会遍历self._active_tasks中的对应session_key的任务，将所有的任务取消，然后调用self._mcp_stack的__aexit__方法，将所有的上下文管理器退出。
async def _handle_stop(self, msg: InboundMessage) -> None:
"""Cancel all active tasks and subagents for the session."""
tasks = self._active_tasks.pop(msg.session_key, [])
cancelled = sum(1 for t in tasks if not t.done() and t.cancel()) # 任务未完成则取消，完成了则不计数
for t in tasks:
try:
await t
except (asyncio.CancelledError, Exception):
pass
sub_cancelled = await self.subagents.cancel_by_session(msg.session_key) # 通过子Agent进行取消任务
total = cancelled + sub_cancelled
content = f"⏹ Stopped {total} task(s)." if total else "No active task to stop."
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel, chat_id=msg.chat_id, content=content,
))
消息分发
消息分发主要是调用_dispatch函数，这个函数只封装了一个核心的_process_message函数，并用外部加原子锁。
async def _dispatch(self, msg: InboundMessage) -> None:
"""Process a message under the global lock."""
async with self._processing_lock: # 该对象是一个asyncio.Lock，用于保证消息处理的原子性
try:
response = await self._process_message(msg)
if response is not None:
await self.bus.publish_outbound(response)
elif msg.channel == "cli":
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel, chat_id=msg.chat_id,
content="", metadata=msg.metadata or {},
))
except asyncio.CancelledError:
logger.info("Task cancelled for session {}", msg.session_key)
raise
其中_process_message函数的整体流程是先对消息进行上下文的预处理，语句是
messages = self.context.build_messages(history=history, current_message=msg.content, channel=channel, chat_id=chat_id)
然后，调用final_content, _, all_msgs = await self._run_agent_loop(messages)来进一步处理带有上下文的信息。
其中上下文的预处理就是生成对应system，user，user的系统prompt，runtime context 和 用户信息。的prompt字典。
获得这些prompt信息后，_run_agent_loop函数做的就是调用provider的chat函数，将prompt信息作为参数传入，然后获得返回的response，并多次迭代。每次有工具调用，说明还在等待一些外部信息，则继续调用工具方法，直到没有工具调用的时候返回。
当_run_agent_loop函数中有工具调用时，需要增加add_assistant_message，和 add_tool_result，这两个函数分别向messages中添加工具调用和工具返回的信息。
很简单的模块，主要是记录一些prompt信息。这些信息包括了当前Agent具备的记忆信息和技能信息。
class ContextBuilder:
"""Builds the context (system prompt + messages) for the agent."""
BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md", "IDENTITY.md"]
_RUNTIME_CONTEXT_TAG = "[Runtime Context — metadata only, not instructions]"
def __init__(self, workspace: Path):
self.workspace = workspace
self.memory = MemoryStore(workspace)
self.skills = SkillsLoader(workspace)
该类包含了两个类，记忆模块和技能模块
记忆模块是准备了两个markdown文件，MEMORY.md 和 HISTORY.md 来记录执行情况和历史信息。
class MemoryStore:
"""Two-layer memory: MEMORY.md (long-term facts) + HISTORY.md (grep-searchable log)."""
def __init__(self, workspace: Path):
self.memory_dir = ensure_dir(workspace / "memory")
self.memory_file = self.memory_dir / "MEMORY.md"
self.history_file = self.memory_dir / "HISTORY.md"
技能模块是加载了workspace和skills目录下的SKILL.md文件，并返回一个技能列表。
包含了两个基础类，Tool和ToolRegistry
Tool是抽象类，定义了工具的接口，需要统一实现execute方法
class Tool(ABC):
"""
Abstract base class for agent tools.
Tools are capabilities that the agent can use to interact with
the environment, such as reading files, executing commands, etc.
"""
@abstractmethod
async def execute(self, **kwargs: Any) -> str:
"""
Execute the tool with given parameters.
Args:
**kwargs: Tool-specific parameters.
Returns:
String result of the tool execution.
"""
pass
工具注册类为
class ToolRegistry:
"""
Registry for agent tools.
Allows dynamic registration and execution of tools.
"""
def __init__(self):
self._tools: dict[str, Tool] = {}
主要是用于任务分解后的子任务在后台的运行。就是一个简单的AgentLoop。
命令模块主要在 nanobot/cli/commands.py 文件下，包括以下子模块：
gateway - 核心职责：基于Python的asyncio库实现异步事件处理，用于管理和协调多个组件，消息总线，Agent循环，channel管理器，会话管理器，定时任务服务器，心跳服务。
参数：port 是该服务的本地端口号， verbose 控制日志级别
agent - 核心职责：实现一个与AI Agent进行交互的命令行界面。
参数： messsage 需要发送给Agent的消息，session_id 会话ID，markdown是否将助手的输出渲染为Markdown格式，logs是否在聊天期间显示运行日志。
支持两种模式：
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。