1. 项目概述
Nanobot是由香港大学数据科学实验室(HKUDS)开源的超轻量级个人AI助手框架,定位为"Ultra-Lightweight OpenClaw"。本文将通过解析Nanobot源码中的AgentLoop模块,深入理解AI Agent的核心运行机制。
AgentLoop是AI Agent运行的核心组件,负责处理消息接收、上下文构建、LLM调用、工具执行和响应发送的全流程。理解AgentLoop的工作原理,对于构建高效可靠的AI Agent系统至关重要。
2. Agent核心原理
2.1 Agent的基本概念
一个Agent代表一个完整的AI"大脑实例",每个Agent都拥有独立资源。Agent是"执行平面",解决"消息怎样变成模型调用、工具执行和最终回复"的问题。
Agent的最小循环包含以下步骤:
- 从总线接收消息
- 组装上下文
- 推理该做什么(LLM调用)
- 根据决定行动(调用工具、执行命令)
- 观察结果,保存状态
- 判断是否完成或需要继续循环
- 完成后回复
2.2 Pi-Agent框架设计
OpenClaw使用的Pi-Agent框架是一个精简高效的AI编程Agent框架,其设计理念可以总结为:
- 系统提示词简短明了(不到1000个token)
- 内置工具数量精简(仅4个核心工具)
- 没有复杂的规划模式和多代理通信协议
- 让LLM模型发挥核心作用
Pi-Agent的四个核心工具:
| 工具 | 主要功能 |
|---|---|
| read | 读取文件、审查代码、获取上下文信息 |
| write | 创建文件、写入内容 |
| edit | 修改代码、进行增量更新 |
| bash | 执行命令、操作环境、通过自我调用来拆分任务 |
这种精简设计带来了三大优势:
- 节省上下文空间,降低推理成本
- 行为更加灵活自主
- 更好的适应性
3. AgentLoop架构解析
3.1 整体架构
AgentLoop类的核心职责包括:
- 从总线接收消息
- 构建上下文(包含历史、记忆、技能)
- 调用LLM
- 执行工具调用
- 发送响应返回
AgentLoop的架构图如下:
code复制[Message Bus] → [AgentLoop] → [Message Bus]
↑ ↓
[LLM] [Tools]
3.2 核心流程
AgentLoop处理消息的完整流程:
- 消息到达(InboundMessage)
- AgentLoop.run() - 监听并接收消息
- AgentLoop._dispatch() - 分派处理
- AgentLoop._process_message() - 主要处理逻辑
- ContextBuilder.build_messages() - 构建上下文
- AgentLoop._run_agent_loop() - 核心代理循环
- Provider.chat() - LLM交互
- 判断是否有工具调用:
- 是:执行工具调用 → 添加工具结果 → 继续循环
- 否:返回最终内容
- AgentLoop._save_turn() - 保存交互记录
- 通过MessageBus发布OutboundMessage - 发送响应
3.3 初始化与工具注册
AgentLoop的初始化包含以下关键步骤:
python复制def __init__(
self,
bus: MessageBus, # 消息总线
provider: LLMProvider, # LLM提供者
workspace: Path, # 工作目录
model: str | None = None, # 模型名称
max_iterations: int = 40, # 最大迭代次数
temperature: float = 0.1, # 温度参数
max_tokens: int = 4096, # 最大Token数
memory_window: int = 100, # 记忆窗口大小
# ...其他参数
):
# 基础属性初始化
self.bus = bus
self.provider = provider
self.workspace = workspace
self.model = model or provider.get_default_model()
# 核心组件初始化
self.context = ContextBuilder(workspace) # 上下文构建器
self.sessions = SessionManager(workspace) # 会话管理器
self.tools = ToolRegistry() # 工具注册表
# 注册默认工具
self._register_default_tools()
工具注册方法_register_default_tools()会注册以下核心工具:
- 文件系统工具(读/写/编辑/列目录)
- 命令执行工具
- 网页相关工具(搜索/爬取)
- 消息发送工具
- 子Agent生成工具
- 定时任务工具(如果配置了定时服务)
4. AgentLoop核心方法实现
4.1 run方法
run方法是代理的主循环入口,负责持续消费消息总线的入站消息并异步分发处理。
python复制async def run(self) -> None:
"""运行代理循环,分发消息任务以保持对/stop指令的响应"""
self._running = True
await self._connect_mcp() # 连接MCP服务器(懒加载)
logger.info("Agent loop started")
while self._running:
try:
# 1秒超时消费消息,保证/stop指令响应性
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, []).remove(t) if t in self._active_tasks.get(k, []) else None)
关键设计点:
- 1秒超时消费消息:避免主线程阻塞,确保/stop能及时被处理
- 异步任务分发:非/stop消息通过_dispatch异步处理
- 任务追踪:通过_active_tasks记录各会话的活跃任务
4.2 _dispatch方法
_dispatch是消息分发的核心方法,在全局锁保护下执行消息处理。
python复制async def _dispatch(self, msg: InboundMessage) -> None:
"""在全局锁下处理消息"""
async with self._processing_lock: # 全局处理锁
try:
response = await self._process_message(msg)
if response is not None:
await self.bus.publish_outbound(response)
elif msg.channel == "cli":
# 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
except Exception:
logger.exception("Error processing message for session {}", msg.session_key)
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel, chat_id=msg.chat_id,
content="Sorry, I encountered an error.",
))
关键设计点:
- 全局锁_processing_lock:避免多任务并发处理消息导致的资源冲突
- 异常处理:区分任务取消异常和通用异常
- CLI渠道特殊处理:发布空消息避免命令行交互阻塞
4.3 _process_message方法
_process_message是单条消息处理的核心入口,支持系统消息、斜杠命令和普通对话三种场景。
python复制async def _process_message(
self,
msg: InboundMessage,
session_key: str | None = None,
on_progress: Callable[[str], Awaitable[None]] | None = None,
) -> OutboundMessage | None:
"""处理单个入站消息并返回响应"""
# 处理系统消息
if msg.channel == "system":
channel, chat_id = (msg.chat_id.split(":", 1) if ":" in msg.chat_id
else ("cli", msg.chat_id))
key = f"{channel}:{chat_id}"
session = self.sessions.get_or_create(key)
self._set_tool_context(channel, chat_id, msg.metadata.get("message_id"))
history = session.get_history(max_messages=self.memory_window)
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)
self._save_turn(session, all_msgs, 1 + len(history))
self.sessions.save(session)
return OutboundMessage(channel=channel, chat_id=chat_id,
content=final_content or "Background task completed.")
# 处理斜杠命令
cmd = msg.content.strip().lower()
if cmd == "/new":
# 处理新建会话逻辑
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
content="New session started.")
if cmd == "/help":
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
content="🐈 nanobot commands:\n/new — Start a new conversation\n/stop — Stop the current task\n/help — Show available commands")
# 处理普通消息
key = session_key or msg.session_key
session = self.sessions.get_or_create(key)
self._set_tool_context(msg.channel, msg.chat_id, msg.metadata.get("message_id"))
# 构建初始上下文
history = session.get_history(max_messages=self.memory_window)
initial_messages = self.context.build_messages(
history=history,
current_message=msg.content,
media=msg.media if msg.media else None,
channel=msg.channel, chat_id=msg.chat_id,
)
# 运行代理循环
final_content, _, all_msgs = await self._run_agent_loop(
initial_messages, on_progress=on_progress
)
# 保存会话状态
self._save_turn(session, all_msgs, 1 + len(history))
self.sessions.save(session)
return OutboundMessage(
channel=msg.channel, chat_id=msg.chat_id, content=final_content,
metadata=msg.metadata or {},
)
关键功能:
- 系统消息处理:独立构建会话和上下文
- 斜杠命令处理:/new合并记忆并清空会话,/help返回命令列表
- 记忆合并:未合并消息达阈值时异步执行
- 进度回调:实时推送处理进度
- 重复回复防护:消息工具已发送过消息则返回None
4.4 _run_agent_loop方法
_run_agent_loop是智能体的核心执行循环,通过不断调用大模型并根据响应决定是否调用工具。
python复制async def _run_agent_loop(
self,
initial_messages: list[dict],
on_progress: Callable[..., Awaitable[None]] | None = None,
) -> tuple[str | None, list[str], list[dict]]:
"""运行代理迭代循环。返回(final_content, tools_used, messages)"""
messages = initial_messages
iteration = 0
final_content = None
tools_used: list[str] = []
while iteration < self.max_iterations:
iteration += 1
# 调用LLM
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(),
model=self.model,
temperature=self.temperature,
max_tokens=self.max_tokens,
)
if response.has_tool_calls:
# 处理工具调用
if on_progress:
clean = self._strip_think(response.content)
if clean:
await on_progress(clean)
await on_progress(self._tool_hint(response.tool_calls), tool_hint=True)
# 添加助手消息到上下文
tool_call_dicts = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments, ensure_ascii=False)
}
}
for tc in response.tool_calls
]
messages = self.context.add_assistant_message(
messages, response.content, tool_call_dicts,
reasoning_content=response.reasoning_content,
)
# 执行工具调用
for tool_call in response.tool_calls:
tools_used.append(tool_call.name)
result = await self.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, reasoning_content=response.reasoning_content,
)
final_content = clean
break
# 处理达到最大迭代次数的情况
if final_content is None and iteration >= self.max_iterations:
logger.warning("Max iterations ({}) reached", self.max_iterations)
final_content = (
f"I reached the maximum number of tool call iterations ({self.max_iterations}) "
"without completing the task. You can try breaking the task into smaller steps."
)
return final_content, tools_used, messages
核心逻辑:
- 循环调用LLM并根据响应决定下一步
- 有工具调用时:
- 记录工具调用
- 执行工具
- 将工具结果加入上下文
- 继续循环
- 无工具调用时:
- 将模型回答作为最终结果
- 终止循环
- 边界处理:达到最大迭代次数时生成提示文本
5. 关键设计思考与实践经验
5.1 并发控制设计
AgentLoop采用了多层次的并发控制机制:
- 全局处理锁(_processing_lock):确保单条消息的完整处理流程是串行的
- 会话级任务管理(_active_tasks):跟踪每个会话的活跃任务,支持/stop指令的批量终止
- 记忆合并锁(_consolidation_locks):避免多个任务同时修改同一会话的记忆
这种设计既保证了关键操作的原子性,又保持了系统的响应能力。
5.2 错误处理与恢复
系统实现了全面的错误处理机制:
- 任务取消:通过asyncio.CancelledError处理/stop指令
- 工具调用失败:记录错误日志但不中断流程
- MCP连接失败:记录错误并在下次消息处理时重试
- 最大迭代次数:防止无限循环,给出明确提示
5.3 性能优化实践
- 懒加载连接:MCP服务器连接在首次消息处理时建立
- 异步记忆合并:记忆合并操作在后台异步执行,不阻塞主流程
- 上下文管理:严格控制记忆窗口大小,避免上下文膨胀
- 工具执行隔离:每个工具在工作目录内操作,避免冲突
5.4 扩展性设计
- 工具系统:通过ToolRegistry统一管理,支持动态注册
- 子Agent支持:通过SubagentManager生成子Agent处理子任务
- MCP集成:支持连接多个MCP服务器扩展功能
- 会话管理:支持多种会话类型(普通、系统、CLI)
6. 实际应用中的经验教训
在实现和使用AgentLoop过程中,我们积累了一些宝贵的经验:
-
工具调用频率控制:需要合理设置max_iterations参数,避免陷入工具调用循环。实践中发现40次是一个比较合理的默认值。
-
上下文管理技巧:
- 及时清理过期的会话历史
- 对长对话采用分段记忆合并策略
- 对关键信息进行摘要保存而非完整记录
-
工具设计原则:
- 工具接口应尽可能简单明确
- 工具执行应有超时机制
- 工具结果应包含足够的上下文信息
-
调试与监控:
- 记录完整的工具调用链
- 保存关键决策点的中间状态
- 实现进度回调机制实时监控处理状态
-
性能调优:
- 批量处理工具调用结果
- 异步执行非关键路径操作
- 合理设置LLM参数(temperature、max_tokens等)
