nanobot 核心逻辑解析

一、整体架构

nanobot 是一个基于消息驱动的 AI 助手框架，核心思想是通过消息总线将各个组件解耦，实现模块化和可扩展性。

用户消息 → 通道适配器 → 消息总线 → Agent 循环 → LLM 调用 → 工具执行 → 响应返回

二、核心组件逻辑

1. 消息总线 (Message Bus)

位置： nanobot/bus/queue.py

这是整个系统的核心协调器，采用异步队列模式：

# 核心逻辑：
- 接收来自各个通道（Telegram、Discord等）的入站消息
- 将消息放入异步队列 (FIFO)
- Agent 循环从队列消费消息
- 处理完成后将响应作为出站消息发布

关键数据结构：

InboundMessage: 入站消息（包含 channel, sender_id, chat_id, content）
OutboundMessage: 出站消息（包含 channel, chat_id, content）

2. Agent 循环 (Agent Loop)

位置： nanobot/agent/loop.py

这是系统的大脑，负责处理每一条消息：

处理流程：
1. 从消息总线获取消息
2. 获取或创建会话 (Session)
3. 构建上下文（历史记录、工具定义、系统提示）
4. 调用 LLM 获取响应
5. 如果有工具调用 → 执行工具 → 将结果反馈给 LLM
6. 重复步骤 4-5 直到没有工具调用（最多 max_iterations 次）
7. 保存对话历史到会话
8. 将响应发送回消息总线

核心代码逻辑：

while iteration < max_iterations:
    # 1. 调用 LLM
    response = await provider.chat(
        messages=messages,           # 对话历史
        tools=tools.get_definitions(), # 可用工具定义
        model=model
    )

    # 2. 处理工具调用
    if response.has_tool_calls:
        # 添加助手消息（包含工具调用）
        messages = add_assistant_message(messages, response)

        # 执行每个工具
        for tool_call in response.tool_calls:
            result = await tools.execute(tool_call.name, tool_call.arguments)
            messages = add_tool_result(messages, result)
    else:
        # 没有工具调用，处理完成
        final_content = response.content
        break

3. 上下文构建器 (Context Builder)

位置： nanobot/agent/context.py

负责构建发送给 LLM 的完整提示：

上下文组成：
1. 系统提示词 (System Prompt)
   - 定义 AI 的角色和行为
   - 列出可用的工具
   - 包含技能文档

2. 对话历史 (Conversation History)
   - 从 Session 中获取
   - 包含之前的用户消息和助手响应
   - 限制历史长度避免 token 超限

3. 当前消息 (Current Message)
   - 用户新发送的消息
   - 可能包含媒体（图片、文件等）

4. 工具定义 (Tool Definitions)
   - 每个工具的名称、描述、参数 schema
   - 用于 Function Calling

4. 工具注册表 (Tool Registry)

位置： nanobot/agent/tools/registry.py

管理所有可用工具的核心逻辑：

工具注册流程：
1. 工具类继承 Tool 基类
2. 实现 name, description, properties 属性
3. 实现 execute() 方法
4. 注册到 ToolRegistry

工具执行流程：
1. 从 LLM 接收工具调用请求
2. 验证参数（基于 JSON Schema）
3. 执行工具的 execute() 方法
4. 返回结果（字符串格式）
5. 将结果反馈给 LLM

内置工具：

read_file: 读取文件
write_file: 写入文件
edit_file: 编辑文件
list_dir: 列出目录
exec: 执行 shell 命令
web_search: 网络搜索
web_fetch: 获取网页内容
message: 发送消息到通道
spawn: 创建后台子任务
cron: 调度定时任务

5. Provider 系统

位置： nanobot/providers/registry.py

统一的 LLM 提供商接口：

核心设计理念：单一真实来源 (Single Source of Truth)

Provider Registry 包含：
- 所有提供商的元数据
- API key 环境变量名称
- 模型名称前缀规则
- 自动检测逻辑（通过 key 前缀或 base URL）
- 模型参数覆盖配置

添加新提供商只需 2 步：
1. 在 PROVIDERS 元组中添加 ProviderSpec
2. 在 config/schema.py 中添加配置字段

自动匹配逻辑：

# 1. 优先使用配置中指定的提供商
# 2. 如果未指定，根据模型名称关键词匹配
# 3. 如果是网关（如 OpenRouter），通过 API key 前缀检测
# 4. 自动添加模型名称前缀（如 "claude-3" → "openrouter/claude-3"）

6. 通道管理器 (Channel Manager)

位置： nanobot/channels/

将不同聊天平台统一转换为核心消息格式：

通道适配器职责：
1. 连接到聊天平台（Telegram Bot API, Discord WebSocket 等）
2. 接收平台消息 → 转换为 InboundMessage
3. 从消息总线接收 OutboundMessage → 发送到平台
4. 处理平台特定的功能（如 Telegram 语音转文字）

支持的通道：
- Telegram: Bot API，支持代理
- Discord: WebSocket，支持 intents
- WhatsApp: 通过 Node.js 桥接，本地 WebSocket
- Feishu: WebSocket 长连接（无需公网 IP）
- DingTalk: Stream 模式（无需公网 IP）

7. 会话管理器 (Session Manager)

位置： nanobot/session/manager.py

持久化对话状态：

会话逻辑：
1. 会话键格式："{channel}:{chat_id}"
2. 每个会话维护独立的对话历史
3. 对话历史保存为 JSON 文件：~/.nanobot/sessions/{session_key}.json
4. 支持跨重启保持上下文
5. 定期清理过期会话（可选）

8. 子代理系统 (Subagent System)

位置： nanobot/agent/subagent.py

支持后台并发任务执行：

子代理创建流程：
1. 用户请求创建子任务
2. SubagentManager 创建新的 asyncio.Task
3. 子任务运行独立的 agent 实例
4. 子任务完成后，将结果发送到原始通道
5. 原始用户可以继续对话，不受影响

使用场景：
- 长时间运行的任务
- 并行处理多个任务
- 不阻塞主对话流程

9. 定时任务 (Cron Service)

位置： nanobot/cron/service.py

基于 cron 表达式的任务调度：

定时任务逻辑：
1. 解析 cron 表达式（使用 croniter 库）
2. 计算下次执行时间
3. 到达执行时间时，发送消息到 Agent
4. Agent 处理消息并执行相应操作
5. 支持添加、删除、列出任务

三、核心工作流程

完整消息处理流程：

1. 用户在 Telegram 发送消息："帮我搜索最新的 AI 新闻"

2. Telegram 通道接收消息
   → 创建 InboundMessage(channel="telegram", sender_id="123", chat_id="456", content="...")

3. 消息发布到总线
   → await bus.publish_inbound(message)

4. Agent 循环从队列消费消息

5. 获取会话
   → session = sessions.get_or_create("telegram:456")

6. 构建上下文
   → messages = context.build_messages(history, current_message, tools)

7. 调用 LLM
   → response = await provider.chat(messages, tools)
   → LLM 返回：tool_calls=[{"name": "web_search", "arguments": {"query": "AI 新闻"}}]

8. 执行工具
   → result = await tools.execute("web_search", {"query": "AI 新闻"})
   → result: "最新的 AI 新闻包括..."

9. 将工具结果反馈给 LLM
   → messages.append({"role": "tool", "content": result})
   → response = await provider.chat(messages, tools)
   → LLM 返回：content="根据搜索结果，最新的 AI 新闻..."

10. 保存对话历史
    → session.add_message("user", "帮我搜索最新的 AI 新闻")
    → session.add_message("assistant", "根据搜索结果，...")
    → sessions.save(session)

11. 发送响应
    → await bus.publish_outbound(OutboundMessage(channel="telegram", chat_id="456", content="..."))

12. Telegram 通道接收并发送消息到用户

四、关键技术点

1. 异步编程 (Async/Await)

整个系统基于 Python 的 asyncio 实现：

所有 I/O 操作都是异步的（网络请求、文件读写）
使用 asyncio.Queue 实现消息传递
支持高并发处理

2. Function Calling

nanobot 充分利用 LLM 的 Function Calling 能力：

工具定义转换为 OpenAI Function 格式
LLM 决定何时调用哪个工具
工具执行结果反馈给 LLM 继续处理
支持多轮工具调用

3. 模块化设计

每个组件都有清晰的职责：

通道只负责消息适配
Agent 只负责逻辑处理
Provider 只负责 LLM 调用
工具只负责特定功能

4. 插件化扩展

添加新通道：继承 Channel 基类
添加新工具：继承 Tool 基类
添加新 Provider：添加 ProviderSpec
添加新技能：创建技能目录和 SKILL.md

五、安全机制

1. 工作区限制

restrict_to_workspace = True  # 限制所有文件操作在工作区内

2. 访问控制

{
  "channels": {
    "telegram": {
      "allowFrom": ["123456789"]  // 只允许特定用户
    }
  }
}

3. 命令过滤

阻止危险命令（如 rm -rf /）
防止 fork 炸弹
阻止文件系统格式化

4. 参数验证

每个工具都有 JSON Schema 定义，自动验证参数类型和范围。

六、总结

nanobot 的核心逻辑可以概括为：

消息驱动：通过异步消息总线解耦各个组件
工具编排：利用 LLM 的 Function Calling 能力自动调用工具
模块化：清晰的组件划分，易于扩展
异步处理：基于 asyncio 的高性能架构
会话持久化：保持对话上下文
多平台支持：统一的通道适配器接口

这种设计使得 nanobot 既轻量（仅 3000+ 行核心代码），又功能完整，非常适合作为 AI 助手的研究和开发框架。

文档生成时间：2026-02-09

一路狂飙

nanobot-核心逻辑解析