引子

最近用了很多Agent工具，也做了一些AI-coding。突然对Agent底层感兴趣(其实已经看完了)。

原本很有动力看OpenClaw、HermesAgent、jiuwenclaw，毕竟边看边用、边用边看，比较有用。不过总的来说一眼望去繁花锦簇，比较分散注意力，挑了个最简单的NanoBot，号称总代码4000多行，优点就是比较完整。

个人看代码习惯是以前C++阶段养成的，有个工具叫insights(主要是基于symbol的索引能力比较强)：

先确定最小可运行单元，看明白。

然后根据需要或者兴趣，从explore或exploit两个方向，往上叠积木(必须是个完整的结构块，比如看chromium项目，adress_bar -> autocomplete -> search_engine -> history -> db -> cloud_sync……)；

直到拓展完全，尤其是以最近1～5个版本的变更做收尾。

AgentLoop的UML

残念：不支持mermaid

class AgentLoop:
    """
    The agent loop is the core processing engine.

    It:
    1. Receives messages from the bus
    2. Builds context with history, memory, skills
    3. Calls the LLM
    4. Executes tool calls
    5. Sends responses back
    """

代码中AgentLoop的类定义注释基本开宗明义了，它就是最核心的类：收发消息、构建上下文、调用LLM、处理工具调用。

由此可以通过这个Loop看看这个Agent都有哪些东西：

1、MessageBus：消息总线的方式驱动，工程师应该不陌生。支持多消息的并发处理。

2、LLMProvider：封装了LLM API，算是一坨啰嗦细节，自己想用的时候，没必要从0开始撸，谁写的好就参考谁的。

3、ContextBuilder：上下文构建器。核心的三部曲Prompt、Context、Harness之一，后面可以细看。

4、SessionManager：会话管理器。大多数人习惯单人连续对话，不考虑跨天、Context使用率等的话，Agent多半没啥压力。如果是多人连续对话，Agent还能精准区分，哪个人的画像是什么、哪几个人连续讨论的关注点/话题边界，并且做到互不混淆、不污染memory、不大幅膨胀token消耗，工程能力肯定很强。

5、ToolRegistry：工具 = How to do something。

6、AgentRunner：先不深究。

7、SubagentManager：支持子Agent。基本有 并发任务、前台后台、定时任务、Context隔离、一定的Context膨胀的防治能力。

8、Consolidator：按字面意思，巩固器。应该有在Memory维护方面有加强；可能也有长程任务中，对Context中的常驻信息构成有一定的策略管理能力。不管做到什么程度，肯定对长时间连续运行的效能，是有优化的。

9、AutoCompact：(目前看到的)所有的Agent都有。尤其连续对话过程、长程任务过程中，突然感觉Agent开始降智、失忆等，跟这个都有点关系。最核心的其实是Memory子系统是怎么设计的，总体思路上对什么东西做了取舍，自动压缩只是冰山一角。

10、Dream：好家伙，肯定会有做梦变强(类似反省、Casestudy)的能力了。

11、CommandRouter：先不管具体Command抽象的是什么东西，大概率要么对内类似Windows File System的IPR、要么对外提供cli模式可以支持的命令这种。暂时没必要深究。

12、Session：与SessionManager对应，信息管理有多强，都能cover哪些场景，基本看Session的数据结构设计、runtime的维护策略和持久化策略，就能知道了。

13、InboundMessage & OutboundMessage：不深究。

有的人更喜欢看关系图

关系图其实意义不大，coding经验丰富的情况下，只要代码不是a1、a2...这种超级命名规范，组件的用途、相互之间的关系是self-explain的，甚至你看到第一个class，它的整个system会写成啥样都会有所预见。

AgentLoop的生命周期回调

一般体现的是程序员对整个系统的生命周期理解和状态机把控。

class _LoopHook(AgentHook):
    """Core hook for the main loop."""

    def __init__(
        self,
        agent_loop: AgentLoop,
        on_progress: Callable[..., Awaitable[None]] | None = None,
        on_stream: Callable[[str], Awaitable[None]] | None = None,
        on_stream_end: Callable[..., Awaitable[None]] | None = None,
        *,
        channel: str = "cli",
        chat_id: str = "direct",
        message_id: str | None = None,

emmm，此处无责任猜测多进程多线程的处理多半不咋优雅。

初始化：_init()

比较简单

主循环：run()

重新手绘了：主消息循环

主消息循环相对比较干净，很多人喜欢在一个function里攒超大坨的代码，主线支线逻辑交织混杂，第一次写简单，第二眼看上去就不认识了，没啥harness的话Code Agent也喜欢这么干。主消息循环把最简洁的逻辑过程主线交代清楚就行：

1、announced：我开始运行了

2、mcp处理啥先不管，这里猜测应该也是消息来源之一

3、看了源码，这里开始

logger.info("Agent loop started")

所以1和2都应该是不耗时的同步处理或异步处理触发。相对而言，更喜欢C++的构造函数和析构函数必然配对触发机制，在构造函数里埋log：xxx willStart，在析构函数里埋log：xxx hasDone(timecost，result)，行数第一行就写CLogger logger，很爽。

4.1、获取消息总线里的消息：

时序图用AI画的，它应该没读懂，只取了感兴趣的关键词线索，有个明确的针对在本代码文件内的组件的function call。

源码处理比较清晰，就是获取消息和SEH。

try:
    msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
except asyncio.TimeoutError:
    self.auto_compact.check_expired(
        self._schedule_background,
        active_session_keys=self._pending_queues.keys(),
    )
    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

4.2 检查是不是带权重的命令，是就处理下。先不深究细节。

4.3 检查【会话】是不是已经在活跃了，如果已经在活跃就带着这条【消息】接上该会话；如果会话的并发队列满了，重启【会话任务】(看上去，这里可能存在“Agent忘记此前对话”现象，看并发队列容量了)。

4.4 启动新的会话任务。

运行消息循环：_run_agent_loop()

AI对代码的理解完全不知所谓

"""Run the agent iteration loop.

*on_stream*: called with each content delta during streaming.
*on_stream_end(resuming)*: called when a streaming session finishes.
``resuming=True`` means tool calls follow (spinner should restart);
``resuming=False`` means this is the final response.

Returns (final_content, tools_used, messages, stop_reason, had_injections).
"""

消息分发：_dispatch

也没有很复杂，主要是丰富的SEH保障所有消息都能被处理完。

消息处理：_processmessage()

这里AI画时序图基本没画明白过，手画了下，代码写得比较混乱(结构化做得差)。

总体而言，一个小小处理函数里，同时考虑了：对话的初始消息、对话的中间轮次消息、工具引发的消息、其他Agent引发的消息、系统消息、反斜线命令引发的消息等逻辑线，同时兼顾下会话的保存、Consolidator、媒体文件摘出等，都放在一大段过程式代码里面处理。

改进的话，是可以考虑首先分清消息类型和输入输出场景，用类工厂模式创建对应的处理函数(应该都是被主线程调用)，把共有的处理部分拆出独立的函数(调用者负责线程安全)，上面这个可读性和可维护性就会好很多。另外一种思路，就看对 真·面向对象 的理解了。

为什么看这么细

对未知事物，要么极端恐惧，要么极端自负。

所以这里 祛魅。只通过口头语言，大家脑袋中没有画面，有“简单”直白的代码在眼前，理解起来更简单。

系统分层上来说，假设把行业的总token量看作是一个池子：

1. 如果work for increasing Tokens，是一种命题，他们大概率不擅长这篇文章对应的命题；

2. 如果work for decreasing Tokens，人人都是程序员，

   1. 一部分算浅水层开发，比如vibe-coding，对着各种工具“讲话”它就能实现你要的东西，甚至能写自己的skill。出问题谁能查出根因并修复？至少目前只有程序员；

   2. 一部分算深水层开发，不论把Agent当Application还是当OS，或者给vibe-coding提供深度定制的更高质量更易用的平台，都跟你最擅长的通用开发深度交织。

它带来的最大影响，可能是以前coding能力10和20的差距，从(20-10)扩大到(20-10)*100；21、22、23和20的差距有限。强者恒强，差距更大。

最后，行业走势终会完美：increase = decrease。你看到的increasing有多快，decreasing也会有多厚。