AI Agent 框架深度解析:nanobot vs bub

从 ReAct 理论到生产级实现的完整技术分析。


目录

  1. 理论基础:什么是 AI Agent
  2. ReAct:让 LLM 自主工作的核心模式
  3. Tool Call 的三层分工与训练证据
  4. nanobot:MessageBus 驱动的个人助理框架
  5. bub:Hook-first 的群聊协作框架
  6. 记忆管理的两种哲学
  7. 横向对比
  8. 延伸阅读

1. 理论基础

1.1 AI Agent 的定义

“An agent is anything that can be viewed as perceiving its environment through sensors and acting upon that environment through actuators.”

— Russell & Norvig, Artificial Intelligence: A Modern Approach (4th ed., 2020)

LLM-based Agent 是这一经典定义在大语言模型时代的具体实现:

  • 感知:接收用户消息、工具执行结果、记忆内容
  • 推理:LLM 处理上下文,决定下一步行动
  • 行动:调用工具(文件读写、代码执行、网络请求)

1.2 Chain-of-Thought 的奠基作用

论文Chain-of-Thought Prompting Elicits Reasoning in Large Language Models 作者:Jason Wei et al.(Google Research) ArXiv2201.11903(2022)

CoT 证明了通过在 prompt 中插入推理步骤,LLM 能够解决此前无法处理的复杂问题。这是 ReAct 模式的直接前驱——ReAct 将 CoT 的”内部推理”与”外部行动”结合。

CoT:   Thought → Thought → Thought → Answer
ReAct: Thought → Action → Observation → Thought → Action → Answer

2. ReAct 模式

论文ReAct: Synergizing Reasoning and Acting in Language Models 作者:Shunyu Yao, Jeffrey Zhao et al.(Princeton / Google Research) ArXiv2210.03629(ICLR 2023) 项目页react-lm.github.io

2.1 核心思想

ReAct 将两个能力交织在一起:

能力来源作用
Reasoning(推理)Chain-of-Thought让模型解释自己的决策,减少幻觉
Acting(行动)工具调用让模型获取外部信息,执行真实操作

2.2 标准轨迹格式

Question: 科罗拉多造山运动的海拔范围是多少?

Thought 1: 我需要搜索科罗拉多造山运动,找到东部延伸到哪里...
Action 1:  Search[Colorado orogeny]
Observation 1: 科罗拉多造山运动是一次山地形成事件...

Thought 2: 没有提到东部。我需要查找 eastern sector。
Action 2:  Lookup[eastern sector]
Observation 2: 东部延伸至高平原...

Thought 5: 高平原海拔从 1,800 到 7,000 英尺。
Action 5:  Finish[1,800 to 7,000 ft]

2.3 ReAct 在 nanobot 中的直接映射

ReAct 的轨迹格式对应 nanobot _run_agent_looploop.py:195)的 while 循环:

Thought N    → LLM 生成内容(response.content)
Action N     → LLM 决定调用工具(response.tool_calls)      loop.py:206
Observation N → 工具执行结果追加回 messages                 loop.py:230

2.4 ReAct 在 bub 中的映射

bub 的 _agent_loopagent.py)委托给 republic 库的 tape.run_tools_async()

output = await tape.run_tools_async(prompt, system_prompt, tools)
if output.kind == "text":    return output.text        # ReAct 终止
if output.kind == "tools":   next_prompt = "Continue the task."  # 继续

关键区别:nanobot 自己实现 while 循环;bub 把循环委托给 republic,自己只处理退出条件。


3. Tool Call 训练原理

3.1 工具能力不是涌现的

ToolLLM 论文直接指出:

“现有指令微调数据集(Alpaca、FLAN 等)几乎完全忽视了工具使用。”

— Qin et al., ToolLLM, arXiv:2307.16789(ICLR 2024)

这意味着:没有专门的工具调用训练数据,模型不会使用工具

3.2 Toolformer:自监督标注(Meta AI,2023)

ArXiv2302.04761

核心方法:在 Common Crawl 语料里自动插入 API 调用标记,筛选标准是插入后 next-token loss 降低。

训练数据格式:

During ArXiv Dives on Friday,
<API>Calendar() → 02/09/2024</API> February 9th...

推理机制:模型生成到 就暂停 → 真正调用 API → 插入结果 → 继续生成。 这和 nanobot/bub 的 agent loop 结构完全相同,只是外层循环在 Python 代码里。

3.3 OpenAI Function Calling:Fine-tune 的直接声明

官方公告(2023年6月13日):openai.com/index/function-calling

“These models have been fine-tuned to both detect when a function needs to be called and to respond with JSON that adheres to the function signature.”

OpenAI Cookbook 公开了训练数据格式:

{
  "messages": [
    {"role": "user", "content": "天气怎么样?"},
    {
      "role": "assistant",
      "tool_calls": [{"function": {"name": "get_weather", "arguments": "{\"city\": \"Beijing\"}"}}]
    },
    {"role": "tool", "tool_call_id": "...", "content": "晴天,26°C"}
  ],
  "tools": [{"name": "get_weather", "description": "...", "parameters": {...}}]
}

tool_calls 就是监督学习的 ground-truth 标签。训练数据格式 = 运行时格式,这不是巧合,而是设计原则。

3.4 Anthropic Claude Model Card 声明

“Claude models are thoroughly trained to utilize them (search tools and other databases).”

Claude 3 Model Card

3.5 description 为何有效

LLM 在训练时见过数万条 (tool description → 正确调用) 样本对,因此:

{
  "name": "edit_file",
  "description": "Edit a file by replacing old_text with new_text. Supports minor whitespace differences.",
  "parameters": {"path": "...", "old_text": "...", "new_text": "..."}
}

模型能读懂这段 JSON,并在合适时机调用它——本质上和写好函数文档没有区别


4. nanobot 架构

4.1 整体架构

Chat Channels(Telegram / Discord / Slack 等)
        │
        ▼
  MessageBus(asyncio.Queue pair: inbound / outbound)   bus/queue.py
        │         ← 唯一耦合点,channel 永不直接调 agent
        ▼
  AgentLoop                                               agent/loop.py
  ├─ ContextBuilder   — system prompt + message history   agent/context.py
  ├─ ToolRegistry     — 工具注册与执行分发                 agent/tools/registry.py
  ├─ SessionManager   — JSONL 会话持久化                  session/manager.py
  ├─ MemoryConsolidator — token 触发的历史压缩             agent/memory.py
  ├─ SubagentManager  — 后台子 agent                       agent/subagent.py
  └─ LLMProvider      — LiteLLM / Azure / Codex           providers/

并发服务(gateway 模式):
  ├─ CronService      — croniter 定时任务                  cron/
  └─ HeartbeatService — 每 30min 读 HEARTBEAT.md           heartbeat/

4.2 核心:ReAct while 循环

# agent/loop.py:195
while iteration < self.max_iterations:   # 默认 40 轮
 
    response = await provider.chat_with_retry(
        messages=messages,           # 累积的完整上下文
        tools=tool_defs,             # 所有工具的 JSON Schema
        model=self.model,
    )                                                    # loop.py:200
 
    if response.has_tool_calls:                          # loop.py:206
        messages = context.add_assistant_message(...)    # loop.py:219
        for tool_call in response.tool_calls:
            result = await tools.execute(               # loop.py:229
                tool_call.name, tool_call.arguments)
            messages = context.add_tool_result(         # loop.py:230
                messages, tool_call.id, result)
    else:
        final_content = response.content                # loop.py:245
        break

messages 始终是完整上下文,每轮 LLM 都能看到:

[system: SOUL.md + MEMORY.md + 工具列表]
[user: 用户消息]
[assistant: tool_call(read_file)]
[tool: 文件内容...]
[assistant: tool_call(exec)]
[tool: 执行结果...]
[assistant: 最终回答]  ← 触发 break

4.3 工具系统

每个工具继承 Tool 基类(agent/tools/base.py):

class Tool(ABC):
    @property def name(self) -> str: ...          # LLM 用这个名字调用
    @property def description(self) -> str: ...   # LLM 靠此决定是否调用
    @property def parameters(self) -> dict: ...   # JSON Schema,LLM 按此填参
    async def execute(self, **kwargs) -> Any: ... # 真正的执行逻辑

内置工具一览:

工具文件位置说明
read_filefilesystem.py:67带分页的文件读取
write_filefilesystem.py:160写文件,自动创建目录
edit_filefilesystem.py:224精确字符串替换
list_dirfilesystem.py:323列目录(支持递归)
execshell.py:43shell 命令执行
web_searchweb.py:78联网搜索
web_fetchweb.py:221URL 抓取(HTML→Markdown)
messagemessage.py:41主动向用户推送(不等循环结束)
spawnspawn.py:28派生后台子 agent
croncron.py:36定时任务管理

4.4 安全设计

restrict_to_workspace=True 时(config.json: tools.restrict_to_workspace):

  • filesystem.py:23_resolve_path 对 workspace 外路径抛 PermissionError
  • shell.py:161:拦截 ../..\ 路径穿越

始终启用的 exec 安全卫士(shell.py:26-36):rm -rfmkfsdd if=shutdown、fork bomb。

SSRF 防护:security/network.py 在 web 工具和 exec 工具里均检查内网 URL。


5. bub 架构

5.1 设计哲学

“a common shape for agents that live alongside people”

bub 专为多人群聊、多 agent 协作场景设计,核心理念:

  1. Hook-first:框架本身 ~200 行,所有行为都是 hookspec,可被外部插件覆盖
  2. 无特权核心BuiltinImpl 只是第一个注册的普通插件
  3. Tape 即事实:上下文不是压缩的历史摘要,而是 append-only 的事件日志

5.2 整体架构

Chat Channels(Telegram / CLI)
        │
        ▼
  ChannelManager                                      channels/manager.py
  ├─ BufferedMessageHandler(防抖 + 批处理)          channels/handler.py
  └─ asyncio.Queue
        │
        ▼
  BubFramework.process_inbound()                      framework.py
        │
  HookRuntime(pluggy)                               hook_runtime.py
        │
  Turn Pipeline(全为 hookspec,可替换):
    resolve_session → load_state → build_prompt
    → run_model → save_state → render_outbound → dispatch_outbound

5.3 Hook 执行模式

HookRuntimehook_runtime.py)实现两种 pluggy 执行模式:

模式适用 Hook语义
call_firstresolve_session, run_model, build_prompt第一个非 None 获胜,后注册插件优先
call_manyload_state, render_outbound, save_state所有插件都执行,结果合并

5.4 Agent 内部循环

# agent.py: _agent_loop
for step in range(1, self.settings.max_steps + 1):   # 默认 50 步
    output = await tape.run_tools_async(
        prompt=next_prompt,
        system_prompt=self._system_prompt(...),
        tools=model_tools(tools),
    )
    outcome = _resolve_tool_auto_result(output)
 
    if outcome.kind == "text":     return outcome.text          # 终止
    if outcome.kind == "continue": next_prompt = "Continue the task."
    else:                          raise RuntimeError(outcome.error)

5.5 ForkTapeStore:事务性 Turn

FileTapeStore(磁盘 ~/.bub/tapes/*.jsonl)
    └─ ForkTapeStore(contextvars in-memory 分叉)
           turn 开始 → 写入分叉(store.py:fork)
           turn 成功 → merge_back 到磁盘
           turn 失败 → 丢弃分叉,磁盘无副作用

这使得每次 turn 具有原子性:要么完整记录,要么完全不记录。

5.6 Anchor 机制(上下文窗口管理)

tape: [e1][e2][anchor_A][e3][e4][anchor_B][e5][e6]
                                     ↑
                         TapeContext 只选这里之后的条目 → [e5][e6]

tape.handoff 工具插入 anchor,显式标记 phase 边界。这是 bub 管理上下文长度的唯一机制——手动、显式、由模型或用户触发

5.7 群聊设计

防抖BufferedMessageHandler):

[msg1 at 0ms][msg2 at 200ms][msg3 at 400ms]
        → debounce 1s → merge → [batch_message]

群聊过滤

@property
def is_active(self) -> bool:
    # 只有被 @mention 或直接回复 bot 才触发 agent
    # 其他消息在 active_time_window (60s) 内静默接收,超时丢弃

subagent 三种模式

模式sessiontape 归宿
inherit同一 session合并回当前 tape
temptemp/xxx不合并,临时隔离
指定 session ID任意合并到目标 tape

5.8 Comma-command

,fs.read path=README.md  →  直接执行工具,绕过 LLM,不消耗 token

6. 记忆管理

6.1 nanobot:自动压缩 + 双文件摘要

触发:prompt token 数 ≥ context_window_tokens(默认 65536)。

估算 prompt token 数(memory.py:276)
  │
  ├─ < 上限 → 退出
  └─ ≥ 上限 → 压缩循环(最多 5 轮)
       ├─ pick_consolidation_boundary()   memory.py:254
       │    找到能移除足够 token 的 user-turn 边界
       ├─ chunk = messages[last_consolidated:boundary]
       ├─ LLM forced tool_choice=save_memory(memory.py:139)
       │    ├─ history_entry → HISTORY.md(append)
       │    └─ memory_update → MEMORY.md(全量覆写)
       ├─ session.last_consolidated = boundary
       └─ 继续估算,直到 ≤ 上限 // 2
文件格式用途
MEMORY.mdMarkdown 事实列表每次注入 system prompt
HISTORY.md[YYYY-MM-DD HH:MM] 段落grep 搜索历史

降级保护:连续 3 次失败后 _raw_archive,直接 dump 原始 messages 文本,确保数据不丢。

理论支持:MemGPT 分层记忆设计(arXiv:2310.08560)。

6.2 bub:显式 Anchor + Tape 搜索

bub 没有自动压缩

  1. 模型调用 tape.handoff 插入 anchor,上下文窗口重置
  2. tape.search 用 rapidfuzz 对历史做全文/模糊搜索
  3. tape.reset 清空 tape(可归档)

6.3 哲学差异

维度nanobotbub
触发方式自动(token 计数)手动(显式调用)
压缩方式LLM 语义摘要不压缩,anchor 截断
跨会话记忆✅ MEMORY.md 持续累积❌ tape 各自独立
失败保护raw archive 降级ForkTapeStore 原子性
理论对应MemGPT 分层记忆Event Sourcing 模式

7. 横向对比

7.1 架构维度

维度nanobotbub
核心抽象MessageBus(Queue)pluggy HookSpec
扩展方式entry_points(channel)+ 继承(tool)任意 hookspec 均可覆盖
agent loop自己实现 while(loop.py:195)委托给 republic
并发模型asyncio task per sessionasyncio task + contextvars fork
最大迭代40(max_iterations50(BUB_MAX_STEPS

7.2 功能维度

功能nanobotbub
自动记忆压缩✅ LLM 语义摘要❌ 手动 anchor
跨会话记忆✅ MEMORY.md
群聊防抖✅ BufferedMessageHandler
群聊过滤✅ is_active + time_window
定时任务✅ CronService
Comma-command✅ 绕过 LLM 直接执行
Workspace 安全✅ restrict_to_workspace
SSRF 防护✅ network.py
Turn 原子性✅ ForkTapeStore
后台 shell 轮询✅ bash.output

7.3 适用场景

选 nanobot:个人助理 · 一对一 IM · 需要长期记忆 · 需要定时任务 · 需要安全边界

选 bub:多人群聊 · 多 agent 协作 · 高度可定制 pipeline · 需要 turn 原子性


8. 延伸阅读

论文ArXiv关联
ReAct2210.03629agent loop 理论基础
Chain-of-Thought2201.11903推理能力奠基
Toolformer2302.04761tool call 训练数据格式
ToolLLM2307.1678916k+ API 微调
Gorilla2305.15334RAT + API 微调
MemGPT2310.08560分层记忆管理
HuggingGPT2303.17580多模型协作 agent