背景

在之前的文章中我们分析了nanobot onboardnanobot gateway命令的实现,
这次我们分析 AgentLoop.run方法怎么驱动整个Agent的运行,
从整理上来看,该 nanobot用到了Typer,Rich,Questionary,prompt_toolkit这种现代、美观且交互式命令行界面 (CLI) 的强大工具组合。
Typer 用于定义 CLI 结构和参数;Rich 负责文本样式、表格、面板和 Markdown 渲染;Questionary 用于创建交互式问答界面
其中 Rich中的Console,Markdown,Table,Text用来进行渲染,支持颜色、表格、面板、语法高亮和 Markdown ,以更好的进行个性化的展示。
其中最主要都是用协程实现,其中也会调用线程来做其他工作。
关于协程和线程的区别:

协程是用户态的轻量级“微线程”,切换由用户程序控制,没有内核态切换,开销极小;线程是内核态的资源单元,切换由操作系统调度,开销较大.
线程切换需要涉及用户态到内核态的转换,上下文包括内核栈、硬件寄存器等,保存和恢复资源较多

Agent.run命令

这个是所有Agent驱动的主循环方法,也是消息传递的主入口,它从 MessageBus 拉入站消息,再异步分发到真正的处理逻辑,
最终把处理结果发送到 outbound队列中,从而发送给对应的 Channel
于此同时,能够优先处理 /stop , /restart, /status 这种命令

1.设置启动标志并且链接MCP

self._running = True
await self._connect_mcp()

首先用一个标志位来标记并且保持主循环持续运行,再次连接远程的MCP服务器,并把远程Tool注册进当前 ToolRegistry中。
_connect_mcp的实现中使用了AsyncExitStack来手动管理上下文,
这里主要用 enter_async_contextaclose 两个方法注册和释放异步资源。如果在注册的过程中发生异常的话,则调用close方法。
connect_mcp_servers方法中根据MCP的类型(是stdio或者sse)来建立不同的连接(使用了httpx模块)。最后以MCPToolWrapper类的方式注册到ToolRegistry中,
可以看到其实MCP也是最终通过Tool方式调用的。

2.主循环,消费入站队列消息
这里是个死循环,以最大超时1秒的间隔去从入站队列中消费消息。

        while self._running:
            try:
                msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
            except asyncio.TimeoutError:
                continue
            except asyncio.CancelledError:
                # Preserve real task cancellation so shutdown can complete cleanly.
                # Only ignore non-task CancelledError signals that may leak from integrations.
                if not self._running or asyncio.current_task().cancelling():
                    raise
                continue
            except Exception as e:
                logger.warning("Error consuming inbound message: {}, continuing...", e)
                continue

            raw = msg.content.strip()
            if self.commands.is_priority(raw):
                ctx = CommandContext(msg=msg, session=None, key=msg.session_key, raw=raw, loop=self)
                result = await self.commands.dispatch_priority(ctx)
                if result:
                    await self.bus.publish_outbound(result)
                continue
            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)
  • 如果是 /stop , /restart, /status 则优先处理该命令,如果有结果返回,并把该命令的处理结果发送给出站队列
  • 否则使用 asyncio.create_task来创建一个协程任务来异步执行,并按照 session_key记录活跃任务,同时使用add_done_callback方法来增加回调(任务完成后进行活跃任务的清理)

_dispatch方法

1. 并发控制

      lock = self._session_locks.setdefault(msg.session_key, asyncio.Lock())
      gate = self._concurrency_gate or nullcontext()

这里会按照每个session_key 一把锁,保证同一个session的并发资源控制;
全局 asyncio.Semaphore 控制可以同时并发的运行的Session会话(由环境变量 NANOBOT_MAX_CONCURRENT_REQUESTS 控制,默认 3;≤0 为 nullcontext,不限制)

2. 流式输出(可选)
默认 on_stream / on_stream_end 为 None。
若 msg.metadata.get(“_wants_stream”) 为真(例如某频道希望流式),也就是channel配置为 "streaming": true
on_stream(delta):把模型增量以 OutboundMessage 发出,metadata 带 _stream_delta: True。
on_stream_end(resuming=…):发空内容,metadata 带 _stream_end、_resuming(供 UI 区分是彻底结束还是工具轮次中间暂停)。

3.核心处理_process_message

  • 当是普通用户消息时

    1. 加载对应的session,从目录 workspace/sessions加载当前会话所属的历史消息。

    2. 如果是 /new,/status,/help命令,则直接返回对应的处理结果

    3. prompt token 合并,

      1. 首先是获取对应session的独享锁
      2. 计算 bucket = self.context_window_tokens(65536) - self.max_completion_tokens(4096) - self._SAFETY_BUFFER(1024) = 60416 。
        计算 prompt占用token estimated:
        • 调用 session.get_history方法, 该方法从seesion所有中消息取 第一条 user 裁齐+去掉孤儿 tool 前缀+ 拷贝成只含 role / content / tool 相关键 的列表,
          供 ContextBuilder.build_messages 等拼真实请求使用
        • 调用 _build_messages 拼出一份和真实请求结构相近的 probe_messages(当前消息占位为 [token-probe])
        • 调用 estimate_prompt_tokens_chain 方法, 该方法先调用模型自带的estimate_prompt_tokens方法,如果模型没有对应的方法,则使用 tiktoken中的cl100k_base近似估算
          把各条消息里字符串 content、text 块、tool_calls/reasoning/name/tool_call_id 和 整份 tools JSON 拼成大字串,用 cl100k_base 计数 token,
          再加 4 * 消息条数 的粗开销,得到整段 prompt 的 tiktoken 估计
      3. 如果 estimated < budget:认为不需要合并,打 debug「idle」日志,直接返回,否则进行如下步骤
      4. 调用 pick_consolidation_boundary 获取对应每轮次要进行合并的 messege 边界,最多经过5轮,并调用consolidate_messages使用 LLM + save_memory 工具把这段对话摘要进 HISTORY.md / MEMORY.md文件中:
        这里巧妙的使用了 TOOL参数的方式来更新对应 HISTORY.md / MEMORY.md 文件的内容。
            entry = args["history_entry"]
            update = args["memory_update"]
            ...
            self.append_history(entry)
            ...
            self.write_long_term(update)
        
    4. 给带有路由信息的工具设置上下文

         for name in ("message", "spawn", "cron"):
             if tool := self.tools.get(name):
                 if hasattr(tool, "set_context"):
                     tool.set_context(channel, chat_id, *([message_id] if name == "message" else []))
      

      设置 MessageTool,SpawnTool,CronTool工具的目标channelchat_id 为入站信息的channelchat_id。让它们发消息、起子代理、建 cron 时路由到当前会话,避免多会话并发时上下文信息错位

    5. 拼发给模型的消息列表

         history = session.get_history(max_messages=0)
         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,
         )
      

      这里的history只包括未合并的信息,因为合并后的信息在 build_messages中添加进来,这里的build_messages 会包括:

      1. 一条 role: "system":大段 system_prompt(见下文「系统提示的组成」)。
      2. *history:由调用方传入的会话历史(通常为 Session.get_history() 等得到的 user / assistant / tool 消息列表)。
      3. 一条末尾消息rolecurrent_role(默认 "user";子代理回灌等场景可为 "assistant"),contentmerged(运行时元数据 + 当前用户正文/图片)。
        system_promptbuild_system_prompt(skill_names) 生成,各块之间用 \n\n---\n\n 连接,主要包括:
      部分 来源方法 内容概要
      身份与规范 _get_identity() nanobot 身份、运行环境(OS、架构、Python)、工作区路径、MEMORY/HISTORY/skills 路径、Windows/POSIX 策略、工具与通过 message 发文件等指引。
      Bootstrap 文件 _load_bootstrap_files() 工作区根目录下若存在则加载:AGENTS.mdSOUL.mdUSER.mdTOOLS.md
      长期记忆 MemoryStore.get_memory_context() 读取 memory/MEMORY.md,有内容则放在 # Memory 下。
      常驻技能 get_always_skills + load_skills_for_context 有则追加 # Active Skills
      技能目录 build_skills_summary() 有则追加 # Skills 及如何阅读 SKILL.md 等说明。
      其中 ~/.nanobot/workspace/
            ├── AGENTS.md       # 代理调度规则与标准作业程序,是代理的工作指南
            ├── HEARTBEAT.md    # 定时执行逻辑与主动任务状态自检,让 代理 具备“自主意识”
            ├── SOUL.md         # 响应语气、行为特征及输出格式配置,是代理的性格、核心价值观和长期指令
            ├── TOOLS.md        # 工具授权注册表及调用参数规范,是代理的技能配置清单,定义了工具准则
            ├── USER.md         # 用户画像数据,包含特定偏好与交互限制配置, 定义了 代理 如何服务你
      

      user_content 为由 "type": "text"或者"type": "image_url"等多模态输入组成,具体参考OpenAI Chat Completions APIChatCompletionContentPart
      具体的内容可以见 Nanobot的 system_prompt 示例

    6. 跑 Agent 主循环
      这里主要是 调 LLM +(有工具则并发执行并追加结果)+ 再问模型」 直到得到无工具调用的最终文本或错误/步数耗尽,并返回 最终正文、工具名列表、完整 messages 列表
      若 final_content is None 则 置为固定英文一句,避免对外返回空串。

    7. 持久化与后台合并

         self._save_turn(session, all_msgs, 1 + len(history))
         self.sessions.save(session)
      

      对新增的messages进行清洗 user、截断 tool、丢弃有害空 assistant,打上时间戳后 append 到 session.messages,而后续的 sessions.save 直接写到jsonl文件中。

    8. 如果 MessageTool 已发消息 ,则不再返回主回复 ,避免频道里再贴一条重复总结(由 message 工具已发过),否则返回正常的消息

  • 当是由SpawnTool工具执行产生的子代理的消息时,
    总体上处理逻辑和普通用户消息处理逻辑一样, 区别点如下:

    1. current_role = "assistant" if msg.sender_id == "subagent"如果是由 subagent 发送来的消息的话,则 设置 消息role 为 “assistant”,
    2. 没有对 MessageTool 的额外处理
    3. 没有对 /new 命令的处理
      4. 出站处理
   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 {},
       ))

如果主流程返回的内容不为空,则直接进行出站处理,否则如果channel是cli(主要是本地CLI命令nanobot agent),则把出站内容置为空。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐