源码
阅读源码---了解背后的实现方式，以及为什么这么设计
--二次开发和应用，定制自己需要的功能
--了解上下文的背景内容，增强对其了解程度
--应用
源码结构图
源码结构图
nanobot/
├── agent/ # [核心] 智能体大脑
│ ├── loop.py # ReAct 主循环 (引擎心脏)
│ ├── context.py # 上下文组装 (Prompt 构建)
│ ├── memory.py # 记忆系统 (三层存储)
│ ├── skills.py # 技能管理
│ ├── subagent.py # 子代理）则处理更复杂的编程任
│ └── tools/ # 工具箱 (Shell, Web, File 等)
│ ├── filesystem.py # 工具箱 (Shell, Web, File 等)
│ ├── registry.py # 工具箱 (Shell, Web, File 等)
│ ├── mcp.py # 工具箱 (Shell, Web, File 等) from mcp.server.fastmcp import FastMC
│ └── base.py # 工具箱 (Shell, Web, File 等)
├── bus/ # [通信] 消息总线
│ ├── queue.py # 异步消息队列 (核心解耦机制)
│ └── events.py # 事件定义
├── channels/ # [触角] 多平台接入
│ ├── base.py # 标准接口定义
│ ├── manager.py # 渠道管理器
│ └── feishu/ # 具体实现 (如飞书、微信等)
├── config/ # [配置] Pydantic 配置管理
├── session/ # [存储] 会话持久化 (JSONL)
├── skills.py # 技能管理
├── providers/ # [LLM] LLM提供商
└── cli/ # [入口] 命令行启动器
内置工具集
内置工具集，涵盖了文件操作、搜索、命令执行、项目管理等多个方面
provider model
workspace allowed_dir
loop.py
内置工具集：
文件系统的命令： ListDirTool ReadFileTool, WriteFileTool, EditFileTool,
命令行的shell、spawn ：ExecTool SpawnTool
Web的工具： WebSearchTool WebFetchTool
定时任务的 CronTool
消息： MessageTool
工具注册： ToolRegistry
基本调用
base-context
spawn-subagent
agent-runtime==loop
产品设计思路： 接入IM系统，软件的交互方式-连接旧世界和新时代的大门，
历史的B端采用客户端后者board的方式，现在可以通过已有的已安装的交互方式，B端和C端产品的交互方式，从这一次开始了变革
base.py
name-description-parameters-schema
validate_paramters
tool.to_schema()---> tools.get_definitions
{
"type":"function"
"function":{
"name",self.name.
"description":self.description,
"parameters": self.parameters,
},
}
大模型的function call
2023 年 7 月， OpenAI 就为其 GPT 模型引入了函数调用功能
FunctionCall 操作本质上是一种指令控制机制，使得 LLM 能够在生成文本的过程中调用特定函数，以便执行计算、查询外部数据源，或与其他系统交互
过程：
用户向应用程序发出提示词（prompt）
LLM起到决策的作用，告知业务系统应该调用什么函数，以及入参是什么。
业务系统负责实现对应的函数（比如本地实现，或者调用其他系统提供的服务），并且将函数的响应结果再次抛给LLM。
LLM根据响应结果，组织自然语言，继续与业务系统进行交互
LLM仅仅做决策，而实际的调用是由业务系统完成的。
现阶段，function-calling能力的实现有两种主流方式：
LLM本身支持。
利用Prompt模板实现，典型如ReAct模板。
自主决策（Autonomous decision making） ：模型能够智能地选择所需工具来回答问题。
可靠地解析过程（Reliable parsing） ：响应一般以 JSON 格式呈现，而非更典型的对话式响应（dialogue-like response
JSON 格式呈现，可以让agent稳定的解析LLM的意图，而不需要复杂的文件本解析逻辑
Function call 让工具调用变得可预测和可靠的核心逻辑
LLM的功能之一： 将非结构化的用户需求转换为结构化的函数调用，与已有的外部系统进行交互
（***）AI-agent的基础： LLM实现了 非结构化--结构化的转换
（***）判断的本质： LLM根据上下文做决策
MCP： Function-call--MCP client--》 json +Http请求---》 MCP server--MCP json响应--》MCP client--结果--LLM
Skills： 提供了一个方式，让用户用文字定义指令脚本等，形成可以服用的任务流程--让LLM去查找和加载固定的skills文档
用户用文字来写流程，替代MCP调用函数中的各种API的逻辑流程，将决策权完全给到模型和prompt
React
Reasoning +Acting-=-一边想一边做
观察-思考-行动-继续观察-思考-行动，每一步根据历史上下文和当前环境信息，动态决定下一步该做什么
根据实际情况灵活应对
History --Current Environment Info -- LLM Thinking --Toolcall--Toolresult
###工作方式
传统编码--Workflow工作方式--Agent工作方式
传统编码： 要提前想好---遇到新情况，异常处理--修改成本--程序员做决策
Workflow： 模块化--条件判断 --- 产品或开发做决策
Agent ： 边走边看，动态调整 --AI在做决策
工具注册
ToolRegistry
ToolRegistry
get_definitions
# 定义一个并不存在的工具，仅用于约束输出格式
架构：
通信层处理特定平台的集成，
Agent 核心层处理消息并管理状态，
执行层提供工具能力和 LLM 连接。消息总线作为连接这些层的中央事件分发机制
总线
channel + chat_id + content
输入： nanobot/channels/base.py
异步消息队列 This should be a long-running async task that:队列异步化--asyncio.Queue 非阻塞消息传递
1. Connects to the chat platform 连接聊天平台
2. Listens for incoming messages 收到数据信息
3. Forwards messages to the bus via _handle_message() 通过总线转发消息
Channel--Bus -- chat_id
Bus--Agent
Bus--Channel
Session_key =={channel_chat_id}
channels 频道
nanobot/channels/base.py
nanobot/channels/manager.py
"BaseChannel", "ChannelManager"
Bus:总线设计
events.py InboundMessage OutboundMessage
queue.py MessageBus (publish_inbound publish_outbound consume_inbound consume_outbound)
nanobot 的总线设计极度精简，却实现了完美的异步解耦。
Inbound Queue： 所有 Channel 接收到的消息，经过标准化封装后，扔进这个队列。
Outbound Queue： Agent 思考产生的回复，扔进这个队列，由 Channel Manager 派发回对应的渠道。
_handle_message
session 会话---处理 MEMORY.md/HISTORY.md 记忆碎片
维护对话历史-- workspace/sessions/{channel_chat_id}.jsonl
from nanobot.session.manager import Session, SessionManager
add_message get_history clear
memory: 调用LLM
两层记忆架构
read_long_term() write_long_term
append_history()
获取记忆上下文-记忆整合 get_memory_context consolidate()
记忆整合： 当会话消息超过 memory_window (默认100条)
--提取待整合的消息，调用LLM使用save_memory
--写入Memory.md 和 history.md
--更新 session.lat_consolidated 标记
Context 上下文构建
构建系统提示词 构建完整的消息队列 构建运行时元数据 build_message build_system_prompt
处理用户消息
添加工具结果到消息 添加助手消息
self.memory = MemoryStore(workspace)
self.skills = SkillsLoader(workspace)
build_system_prompt
_get_identity(self) ## Workspace {platform_policy} ## nanobot Guidelines
bootstrap = self._load_bootstrap_files()
memory = self.memory.get_memory_context()
skills_summary = self.skills.build_skills_summary()
build_messages
_build_user_content
SOUL.md、USER.md和 AGENTS.md是定义智能体行为、个性和工作流的三个核心配置文件
SOUL.md = 价值观与性格（AI 它是谁，它信奉什么） 定义 AI 的核心人格、语气风格、道德底线和思维模式
USER.md = 用户画像（它服务于谁，用户就是你的习惯和偏好） 它记录了用户的背景、偏好、工作环境、常用工具和禁忌
AGENTS.md = 行为准则与工作流（AI 它具体怎么干活，遵守什么规则）定义 AI 如何执行任务的具体规则和流程
IDENTITY.md IDENTITY——人格
Tools.md
context.py 存储与内存--运行时--每条消息处理
Provider
LLMProvider(LLM适配器)
hearbeat ===心跳检测
_HEARTBEAT_TOOL
response = await self.provider.chat_with_retry(
messages=[
{"role": "system", "content": "You are a heartbeat agent. Call the heartbeat tool to report your decision."},
{"role": "user", "content": (
"Review the following HEARTBEAT.md and decide whether there are active tasks.\n\n"
f"{content}"
)},
],
tools=_HEARTBEAT_TOOL,
model=self.model,
)
AgentLoop（主Agent循环）-- subagentManager(子Agent) -- HeartbeatService 心跳服务-- MemoryStore(记忆整合)
配置系统
配置系统 (config)
schema.py — 基于 Pydantic + pydantic-settings 的完整配置模型，支持 camelCase 和 snake_case 双向兼容
loader.py — 配置加载/保存（~/.nanobot/config.json）
工具函数 (helpers.py)
通用工具：目录确保创建、工作区路径获取、安全文件名转换、时间戳、字符串截断等
###cli
nanobot gateway 启动网关服务（Agent + 所有渠道 + Cron + 心跳）
nanobot 的命令行接口模块，提供以下功能：
- onboard: 初始化配置和工作区
- agent: 与智能体对话（单条消息或交互模式）
- gateway: 启动网关服务器，连接多个聊天渠道
- channels: 管理聊天渠道（状态查看、登录等）
- cron: 管理定时任务
- status: 显示系统状态
prompt_toolkit 是一个强大的 Python 输入库
Skills
渐进式技能加载 — always 技能全文进提示词，其余按需 read_file，节省 token
运行
项目启动入口，负责解析启动参数、初始化核心组件（如事件总线、agent、渠道）、启动机器人主服务。
nanobot/__main__.py
from nanobot.cli.commands import app
nanobot/cli/commands.py
app = typer.Typer(
name="nanobot",
help=f"{__logo__} nanobot - Personal AI Assistant",
no_args_is_help=True,
)
typer.Typer作为入口，定义了命令行工具的名称、帮助信息以及是否显示帮助信息
def onboard():
"""初始化 nanobot 配置和工作区。
执行步骤：
1. 检查配置文件是否已存在，如果存在则询问是否覆盖
2. 创建默认配置文件 (~/.nanobot/config.json)
3. 创建工作区目录 (~/.nanobot/workspace)
4. 创建默认引导文件（AGENTS.md、SOUL.md、USER.md）
5. 创建记忆目录和 MEMORY.md 文件
6. 创建 skills 目录用于自定义技能
@app.command()
def gateway(
port: int = typer.Option(18790, "--port", "-p", help="网关端口"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
):
"""启动 nanobot 网关服务器 port 是该服务的本地端口号， verbose 控制日志级别
执行流程：
1. 加载配置
2. 创建消息总线（MessageBus）
3. 创建 LLM 提供商
4. 创建会话管理器（SessionManager）
5. 创建定时任务服务（CronService）
6. 创建智能体循环（AgentLoop）
7. 创建心跳服务（HeartbeatService）
8. 创建渠道管理器（ChannelManager）
9. 启动所有服务并保持运行
引导文件说明：
- AGENTS.md: 智能体行为规范和指导原则
- SOUL.md: 个性定义和价值观
- USER.md: 用户偏好设置
- memory/MEMORY.md: 长期记忆存储
base
name description parameters schema execute
registry:
register unregister execute
return f"Error: Invalid parameters for tool '{name}': " + "; ".join(errors) + _HINT
_HINT = "\n\n[Analyze the error above and try a different approach.]"
filesystem
workspace allowed_dir
shell:
asyncio.subprocess.PIPE
cron
_channel _chat_id
spawn
_origin_channel _origin_chat_id _session_key
message
default_channel default_chat_id default_message_id OutboundMessage
mcp:
connect_mcp_servers
wrapper = MCPToolWrapper(session, name, tool_def, tool_timeout=cfg.tool_timeout)
registry.register(wrapper)
md文件和jsonl文件
self.memory_file = self.memory_dir / "MEMORY.md"
self.history_file = self.memory_dir / "HISTORY.md"
skill_file = skill_dir / "SKILL.md"
content
List the contents of a directory.
Read the contents of a file at the given path.
Write content to a file at the given path. Creates parent directories if needed
Edit a file by replacing old_text with new_text. The old_text must exist exactly in the file.
"Execute a shell command and return its output. Use with caution."
"Search the web. Returns titles, URLs, and snippets."
"Fetch URL and extract readable content (HTML → markdown/text)."
"Schedule reminders and recurring tasks. Actions: add, list, remove."
"Spawn a subagent to handle a task in the background. "
"Use this for complex or time-consuming tasks that can run independently. "
"The subagent will complete the task and report back when done."
"Send a message to the user. Use this when you want to communicate something."
f"""# Subagent
You are a subagent spawned by the main agent to complete a specific task.
Stay focused on the assigned task. Your final response will be reported back to the main agent. {self.workspace}"""
f"""# Skills
The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.{skills_summary}"""
f"""# nanobot 🐈
You are nanobot, a helpful AI assistant.
## Runtime
{runtime}
## Workspace
Your workspace is at: {workspace_path}
- Long-term memory: {workspace_path}/memory/MEMORY.md (write important facts here)
- History log: {workspace_path}/memory/HISTORY.md (grep-searchable). Each entry starts with [YYYY-MM-DD HH:MM].
- Custom skills: {workspace_path}/skills/{{skill-name}}/SKILL.md
## nanobot Guidelines
- State intent before tool calls, but NEVER predict or claim results before receiving them.
- Before modifying a file, read it first. Do not assume files or directories exist.
- After writing or editing a file, re-read it if accuracy matters.
- If a tool call fails, analyze the error before retrying with a different approach.
- Ask for clarification when the request is ambiguous.
Reply directly with text for conversations. Only use the 'message' tool to send to a specific chat channel."""
函数调用
loop
tool_call_dicts = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments, ensure_ascii=False)
}
}
memory
_SAVE_MEMORY_TOOL = [
{
"type": "function",
"function": {
"name": "save_memory",
"description": "Save the memory consolidation result to persistent storage.",
"parameters": {
"type": "object",
"properties": {
"history_entry": {
"type": "string",
"description": "A paragraph (2-5 sentences) summarizing key events/decisions/topics. "
"Start with [YYYY-MM-DD HH:MM]. Include detail useful for grep search.",
},
"memory_update": {
"type": "string",
"description": "Full updated long-term memory as markdown. Include all existing "
"facts plus new ones. Return unchanged if nothing new.",
},
},
"required": ["history_entry", "memory_update"],
},
},
}
]
"""Consolidate old messages into MEMORY.md + HISTORY.md via LLM tool call.
prompt = f"""Process this conversation and call the save_memory tool with your consolidation.
provider.chat(
messages=[
{"role": "system", "content": "You are a memory consolidation agent. Call the save_memory tool with your consolidation of the conversation."},
{"role": "user", "content": prompt},
],
tools=_SAVE_MEMORY_TOOL,
model=model,
)
_HEARTBEAT_TOOL = [
{
"type": "function",
"function": {
"name": "heartbeat",
"description": "Report heartbeat decision after reviewing tasks.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["skip", "run"],
"description": "skip = nothing to do, run = has active tasks",
},
"tasks": {
"type": "string",
"description": "Natural-language summary of active tasks (required for run)",
},
},
"required": ["action"],
},
},
}
]
claude code
####记忆机制
新版本还会额外维护一个全局的 ~/.claude/history.jsonl 文件，把所有项目的会话汇总在一起，便于做全局历史浏览
claude -c
# 等价于
claude --continue
~/.claude/projects/*/chat_*.jsonl
参考
nanobot（openclaw轻量化）代码详解 https://zhuanlan.zhihu.com/p/2009334676117012665
深入剖析 nanobot：轻量级 AI Agent 框架的架构之道 https://chuna2.787528.xyz/accordion/p/19659822
nanobot 学习笔记2 --- 核心代码解读 cli/command.py https://zhuanlan.zhihu.com/p/2006136993969620563
浙公网安备 33010602011771号