学习笔记nanobot 源码分析

1. 模块概述

AgentLoop 和 AgentRunner 是 nanobot 最核心的两个类，共同构成了 Agent 的处理引擎：

  • **AgentLoop**（nanobot/agent/loop.py）：产品层面的协调器，管理会话、状态机、消息路由、记忆巩固、MCP 连接等
  • **AgentRunner**（nanobot/agent/runner.py）：纯执行引擎，负责”消息→LLM→工具→LLM→…”的迭代循环，不包含产品层逻辑

📍 源码：nanobot/nanobot/agent/loop.py#L140、nanobot/nanobot/agent/runner.py#L132

1.1 职责边界

关注点 AgentLoop AgentRunner
会话管理 ✅ ❌
状态机驱动 ✅ ❌
LLM 调用循环 ❌ ✅
工具执行编排 ❌ ✅
上下文窗口治理 ❌ ✅
记忆巩固触发 ✅ ❌
MCP 连接管理 ✅ ❌
流式输出回调 ✅ ✅ (透传)
错误恢复/重试 ❌ ✅ (通过 Provider)

1.2 在系统中的位置

graph TB
    MB[MessageBus] -->|consume_inbound| AL[AgentLoop]
    AL -->|state machine| AL
    AL -->|_run_agent_loop| AR[AgentRunner]
    AR -->|chat_stream| PR[LLMProvider]
    AR -->|execute| TR[ToolRegistry]
    AL --> SM[SessionManager]
    AL --> CS[Consolidator]
    AL --> MS[MemoryStore]

2. 命名体系与易混淆函数对比

2.1 命名规律拆解

前缀 模块/对象 操作 含义
_state_ restore — 状态机处理器：恢复会话和检查点
_state_ compact — 状态机处理器：自动压缩检查
_state_ command — 状态机处理器：命令分发
_state_ build — 状态机处理器：构建上下文
_state_ run — 状态机处理器：执行 LLM 循环
_state_ save — 状态机处理器：持久化会话
_state_ respond — 状态机处理器：组装响应
_run_ agent_loop — 调用 AgentRunner 执行核心迭代
_run_ core — AgentRunner 的核心迭代循环
_process_ message — 处理单条消息的完整生命周期
_process_ system_message — 处理系统级消息（子 Agent 结果等）
_dispatch — — 消息分发入口，管理会话锁和并发
_save_ turn — 将本轮消息持久化到会话
_build_ initial_messages — 为 LLM 调用构建初始消息列表
_build_ bus_progress_callback — 构建进度回调发布到消息总线
_persist_ user_message_early — 在 turn 开始前提前持久化用户消息
_try_ drain_injections — AgentRunner：尝试排空中途注入消息
_try_ finalize_after_max_iterations — AgentRunner：迭代上限后的最终化尝试

命名规律总结：

  1. _state_xxx 方法对应状态机的 7 个处理阶段，每个返回事件字符串驱动状态转换
  2. _run_xxx 是实际执行 LLM 调用的方法，_run_agent_loop（AgentLoop）委托给 _run_core（AgentRunner）
  3. _try_xxx 方法表示可能失败/为空的尝试性操作，失败时降级而非崩溃

2.2 易混淆函数对比表

对比维度 _run_agent_loop (AgentLoop) _run_core (AgentRunner)
操作对象 TurnContext（含会话、历史等产品上下文） 纯消息列表 + AgentRunSpec
调用者 _state_run 和 _process_system_message AgentRunner.run()
调用阶段 状态机 RUN 阶段 AgentRunner 入口
附加逻辑 注入回调、检查点、工作区作用域绑定、流式回调 上下文治理（orphan/backfill/snip/microcompact）
一句话区分 _run_agent_loop 是产品层包装，绑定上下文和回调后委托给 AgentRunner _run_core 是纯执行核心，只管消息→LLM→工具的迭代
对比维度 _dispatch (AgentLoop) _process_message (AgentLoop)
操作对象 原始 InboundMessage 会话锁内的消息
调用者 run() 主循环 _dispatch
调用阶段 消息从总线消费后立即 获取会话锁后
附加逻辑 会话锁管理、并发门控、待处理队列路由、流式回调构建 状态机驱动、错误恢复
一句话区分 _dispatch 是并发控制层，负责锁和排队 _process_message 是业务处理层，负责状态机流转

3. API Signatures

graph TB
    subgraph "AgentLoop 公共接口"
        run["run()"]
        stop["stop()"]
        process_direct["process_direct()"]
        submit_cron_turn["submit_cron_turn()"]
        set_model_preset["set_model_preset()"]
        from_config["from_config() [classmethod]"]
    end

    subgraph "AgentLoop 内部调用链"
        dispatch["_dispatch()"]
        process_message["_process_message()"]
        process_system["_process_system_message()"]
        run_agent_loop["_run_agent_loop()"]
        state_handlers["_state_restore/compact/command/build/run/save/respond()"]
    end

    subgraph "AgentRunner 公共接口"
        runner_run["run(spec) -> AgentRunResult"]
    end

    subgraph "AgentRunner 内部调用链"
        run_core["_run_core()"]
        request_model["_request_model()"]
        execute_tools["_execute_tools()"]
        snip_history["_snip_history()"]
        microcompact["_microcompact()"]
        drop_orphan["_drop_orphan_tool_results()"]
        backfill["_backfill_missing_tool_results()"]
    end

    run --> dispatch
    dispatch --> process_message
    process_message --> state_handlers
    state_handlers --> run_agent_loop
    run_agent_loop --> runner_run
    runner_run --> run_core
    run_core --> request_model
    run_core --> execute_tools
    run_core --> snip_history

AgentLoop

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class AgentLoop:
    # 构造函数 — 注入所有依赖
    def __init__(
        self,
        bus: MessageBus,                      # 消息总线
        provider: LLMProvider,                 # LLM 提供商
        workspace: Path,                       # 工作区路径
        model: str | None = None,             # 模型名称
        max_iterations: int | None = None,    # 最大工具迭代次数
        max_concurrent_subagents: int | None = None,  # 最大并发子Agent数
        context_window_tokens: int | None = None,     # 上下文窗口Token数
        context_block_limit: int | None = None,       # 上下文块限制
        max_tool_result_chars: int | None = None,     # 工具结果最大字符数
        provider_retry_mode: str = "standard",        # Provider重试模式
        tool_hint_max_length: int | None = None,      # 工具提示最大长度
        cron_service: CronService | None = None,      # 定时任务服务
        restrict_to_workspace: bool = False,          # 限制工作区访问
        session_manager: SessionManager | None = None,# 会话管理器
        mcp_servers: dict | None = None,              # MCP服务器配置
        channels_config: ChannelsConfig | None = None,# 渠道配置
        timezone: str | None = None,                  # 时区
        session_ttl_minutes: int = 0,                 # 会话TTL(分钟)
        consolidation_ratio: float = 0.5,             # 巩固比例
        max_messages: int = 120,                      # 最大消息数
        hooks: list[AgentHook] | None = None,         # Agent钩子列表
        unified_session: bool = False,                # 统一会话模式
        disabled_skills: list[str] | None = None,     # 禁用的技能
        tools_config: ToolsConfig | None = None,      # 工具配置
        model_presets: dict[str, ModelPresetConfig] | None = None,  # 模型预设
        model_preset: str | None = None,              # 当前模型预设名
        runtime_events: RuntimeEventBus | None = None,# 运行时事件总线
    ):
        ...
    # 场景：由 from_config() 工厂方法或测试代码创建

    @classmethod
    def from_config(cls, config, bus=None, **extra) -> AgentLoop:
        ...
    # 从 Config 对象创建 AgentLoop，自动解析 Provider、模型、预设等
    # 场景：CLI 入口 (nanobot/cli/commands.py) 和 API 服务器启动时调用

    async def run(self) -> None:
        ...
    # 主循环：从 MessageBus 消费消息，创建 asyncio.Task 分发给 _dispatch()
    # 场景：网关启动后作为长期运行任务调用

    def stop(self) -> None:
        ...
    # 设置 _running = False，主循环将在下次迭代时退出
    # 场景：网关关闭、/stop 命令

    async def process_direct(
        self, content, session_key="cli:direct", channel="cli",
        chat_id="direct", media=None, on_progress=None,
        on_stream=None, on_stream_end=None, ephemeral=False,
        tools=None, persist_user_message=True,
    ) -> OutboundMessage | None:
        ...
    # 直接处理消息（不走 MessageBus），返回响应
    # 场景：CLI 直接调用、API 服务器、Cron 任务、Dream 巩固

    async def submit_cron_turn(self, msg) -> OutboundMessage | None:
        ...
    # 提交一个 Cron 触发的 turn
    # 场景：CronService 定时触发 Dream 巩固等任务

    def set_model_preset(self, name, *, publish_update=True) -> None:
        ...
    # 运行时切换模型预设
    # 场景：用户通过 /model 命令或 WebUI 切换模型

    def llm_runtime(self) -> LLMRuntime:
        ...
    # 返回当前 provider/model 对
    # 场景：WebUI 展示当前运行时模型信息

    @property
    def model_preset(self) -> str | None:
        ...
    # 获取当前激活的模型预设名称

AgentRunner

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class AgentRunner:
    def __init__(self, provider: LLMProvider):
        ...
    # 构造器只接收 Provider，其他参数通过 AgentRunSpec 传入
    # 场景：由 AgentLoop.__init__ 创建

    async def run(self, spec: AgentRunSpec) -> AgentRunResult:
        ...
    # 执行一次完整的 Agent 运行，返回 AgentRunResult
    # 场景：AgentLoop._run_agent_loop 调用
    # 内部流程：before_run hook → _run_core → after_run/on_error/on_finally hooks

@dataclass(slots=True)
class AgentRunSpec:
    # Agent 运行配置
    initial_messages: list[dict[str, Any]]        # 初始消息列表
    tools: ToolRegistry                            # 工具注册表
    model: str                                     # 模型名
    max_iterations: int                            # 最大迭代次数
    max_tool_result_chars: int                     # 工具结果最大字符
    temperature: float | None = None               # 温度
    max_tokens: int | None = None                  # 最大输出Token
    reasoning_effort: str | None = None            # 推理力度
    hook: AgentHook | None = None                  # Agent钩子
    error_message: str | None = ...                # 错误消息
    concurrent_tools: bool = False                 # 并发工具执行
    workspace: Path | None = None                  # 工作区
    session_key: str | None = None                 # 会话键
    context_window_tokens: int | None = None       # 上下文窗口Token
    context_block_limit: int | None = None         # 上下文块限制
    provider_retry_mode: str = "standard"          # Provider重试模式
    progress_callback: Any | None = None           # 进度回调
    stream_progress_deltas: bool = True            # 是否流式进度
    retry_wait_callback: Any | None = None         # 重试等待回调
    checkpoint_callback: Any | None = None         # 检查点回调
    injection_callback: Any | None = None          # 中途注入回调
    llm_timeout_s: float | None = None             # LLM超时(秒)
    goal_active_predicate: Callable | None = None  # 持续目标活跃判断
    goal_continue_message: GoalContinueMessage | None = None  # 持续目标续推消息
    finalize_on_max_iterations: bool = True        # 达上限时是否最终化

@dataclass(slots=True)
class AgentRunResult:
    # Agent 运行结果
    final_content: str | None                      # 最终回复内容
    messages: list[dict[str, Any]]                 # 完整消息列表
    tools_used: list[str]                          # 使用的工具列表
    usage: dict[str, int]                          # Token 用量
    stop_reason: str = "completed"                 # 停止原因
    error: str | None = None                       # 错误信息
    tool_events: list[dict[str, str]]              # 工具事件
    had_injections: bool = False                   # 是否有中途注入

4. 数据结构深度解析

4.1 TurnContext

4.a 结构体存在的理由

TurnContext 是 AgentLoop 状态机各阶段之间共享状态的载体。如果没有它，每个状态处理器需要独立的参数传递和返回值约定，状态之间的数据流转将变得脆弱且难以扩展。它将一个消息处理的完整生命周期中的所有临时数据集中管理，使状态机各阶段只需读取/写入 context 字段即可协作。

4.b 结构定义

📍 源码：nanobot/nanobot/agent/loop.py#L99-L137

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@dataclass
class TurnContext:
    msg: InboundMessage                         # 当前处理的消息
    session_key: str                            # 会话键
    state: TurnState                            # 当前状态
    turn_id: str                                # 本轮唯一标识
    session: Session | None = None              # 会话对象

    history: list[dict[str, Any]] = field(default_factory=list)        # 历史消息
    initial_messages: list[dict[str, Any]] = field(default_factory=list) # LLM初始消息

    final_content: str | None = None            # 最终回复
    tools_used: list[str] = field(default_factory=list)               # 使用的工具
    all_messages: list[dict[str, Any]] = field(default_factory=list)  # 全部消息
    stop_reason: str = ""                       # 停止原因
    had_injections: bool = False                # 是否有中途注入

    user_persisted_early: bool = False          # 是否提前持久化用户消息
    save_skip: int = 0                          # 保存时跳过的消息数

    outbound: OutboundMessage | None = None     # 最终的出站消息
    suppress_response: bool = False             # 是否抑制响应

    # 流式回调
    on_progress: Callable | None = None
    on_stream: Callable[[str], Awaitable[None]] | None = None
    on_stream_end: Callable | None = None
    on_retry_wait: Callable[[str], Awaitable[None]] | None = None

    pending_queue: asyncio.Queue | None = None  # 待处理消息队列
    pending_summary: str | None = None          # 待注入的摘要

    ephemeral: bool = False                     # 是否临时执行
    tools: ToolRegistry | None = None           # 覆盖的工具注册表

    turn_wall_started_at: float                 # Turn 开始时间
    visible_run_started_at: float | None = None # 可见运行开始时间
    turn_latency_ms: int | None = None          # Turn 延迟(毫秒)

    trace: list[StateTraceEntry] = field(default_factory=list)  # 状态追踪

4.c 字段三层分析表

字段 设计动机 反事实 替代方案
msg 状态机各阶段都需要访问原始消息内容 各阶段需要独立传参，重复且易出错 可以用全局变量，但破坏了无状态设计
session_key 唯一标识会话，用于锁、队列、持久化 每次需要时从 msg 重新计算，浪费且有 bug 风险 可以存在 session 对象上，但 session 可能为 None
state 跟踪当前状态机位置 无法知道当前处于哪个阶段 可以用方法调用栈隐式表示，但失去了显式追踪
turn_id 日志追踪和调试的唯一标识 日志中无法关联同一 turn 的各阶段 可以用 (session_key, timestamp)，但不保证唯一
trace 记录每个状态的耗时，用于性能分析 无法定位性能瓶颈 可以用外部 profiler，但侵入性更强
ephemeral 标记临时执行（如 Dream），不持久化会话 临时任务污染正常会话历史 可以用单独的 AgentLoop 实例，但资源开销大
pending_queue 支持中途消息注入（子Agent完成通知等） 子Agent完成后无法通知主Agent继续 可以用回调，但队列模式与消息总线一致
suppress_response 某些工具（如 message 工具）需要抑制自动回复 用户看到重复或多余的回复 可以在上层判断，但需要传递额外状态

4.d 生命周期状态图

stateDiagram-v2
    [*] --> Created: _process_message() 创建
    Created --> RESTORE: 状态机启动
    RESTORE --> COMPACT: event="ok"
    COMPACT --> COMMAND: event="ok"
    COMMAND --> BUILD: event="dispatch"
    COMMAND --> Destroyed: event="shortcut"
    BUILD --> RUN: event="ok"
    RUN --> SAVE: event="ok"
    SAVE --> RESPOND: event="ok"
    RESPOND --> Destroyed: event="ok"
    Destroyed --> [*]: ctx.outbound 已设置

4.2 AgentRunSpec

4.a 结构体存在的理由

AgentRunSpec 将一次 Agent 运行的所有配置参数集中在一个 dataclass 中，避免了 AgentRunner.run() 拥有 20+ 个参数。它也使得添加新配置项不需要修改方法签名，只需在 dataclass 中增加字段即可。

4.b 字段三层分析表

字段 设计动机 反事实 替代方案
injection_callback 支持 LLM 循环中途注入新消息（子Agent结果、目标续推） 子Agent完成后无法通知主循环 可以用 polling，但延迟高
checkpoint_callback 在工具执行间隙持久化进度，支持 /stop 后恢复 /stop 后丢失所有进度 可以每次工具完成后全量保存，但 I/O 开销大
goal_active_predicate 判断持续目标是否活跃，影响超时策略 持续目标被普通超时杀死 可以用全局标志，但无法按 session 区分
finalize_on_max_iterations 控制达到迭代上限时是否尝试无工具最终化 迭代上限后只能返回固定错误消息 可以始终最终化，但某些场景不需要
context_block_limit 替代基于 token 的上下文窗口限制 无法适应不同模型的 block 限制 可以只用 context_window_tokens，但某些模型有 block 数限制

5. 函数逐行精讲

5.1 AgentLoop.run() — 主循环

5.a 场景卡片

函数：AgentLoop.run()

  • 调用时机：网关启动后，作为长期运行的 asyncio 任务调用
  • 典型调用者：nanobot/cli/commands.py 的 gateway 命令
  • 前置条件：MessageBus 已创建、Provider 已初始化、MCP 已连接
  • 目的：持续从 MessageBus 消费消息，为每条消息创建独立的 asyncio.Task

5.b 逐行注释式精讲

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
async def run(self) -> None:
    # → 主循环入口，不接收参数，所有依赖已在 __init__ 注入
    self._running = True
    # → 设置运行标志，stop() 将其设为 False 即可优雅退出
    await self._connect_mcp()
    # → 连接所有配置的 MCP 服务器，失败不阻塞启动
    logger.info("Agent loop started")
    # → 记录启动日志

    while self._running:
        # → 主循环：_running 为 False 时退出
        try:
            msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
            # → 从 MessageBus 消费消息，1 秒超时
            # → 超时不是错误，而是给 auto_compact 执行机会
        except asyncio.TimeoutError:
            # → 1 秒内没有消息到达
            self.auto_compact.check_expired(
                self._schedule_background,
                active_session_keys=self._pending_queues.keys(),
            )
            # → 检查是否有过期会话需要压缩（不阻塞活跃会话）
            continue
            # → 继续等待下一条消息
        except asyncio.CancelledError:
            # → 任务被取消（通常是关闭信号）
            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()
        # → 获取消息文本（去除首尾空白）
        effective_key = self._effective_session_key(msg)
        # → 计算有效的会话键（统一会话模式下可能返回 UNIFIED_SESSION_KEY）

        if await agent_context.handle_runtime_control(self, msg, self.tools):
            continue
            # → MCP 运行时控制命令（如 /mcp-connect），不进入正常处理

        if self.commands.is_priority(raw):
            # → 优先命令（如 /stop），不排队直接执行
            await self._dispatch_command_inline(msg, effective_key, raw, self.commands.dispatch_priority)
            continue

        if self._cron_turns.defer_if_active(msg, session_key=effective_key, ...):
            # → 如果该会话有活跃任务，cron turn 需要延迟
            continue

        if effective_key in self._pending_queues:
            # → 该会话已有活跃任务，将消息路由到待处理队列供中途注入
            ...
            continue

        task = asyncio.create_task(self._dispatch(msg))
        # → 创建异步任务处理消息，不阻塞主循环
        self._active_tasks.setdefault(effective_key, []).append(task)
        # → 记录活跃任务，供 /stop 取消

5.2 AgentLoop._dispatch() — 消息分发

5.a 场景卡片

函数：AgentLoop._dispatch()

  • 调用时机：主循环为每条非注入消息创建 asyncio.Task 时
  • 典型调用者：run() 主循环
  • 前置条件：消息已从总线消费，已通过优先命令和路由检查
  • 目的：管理会话锁和并发门控，然后调用 _process_message 执行实际处理

关键逻辑：

  • 获取会话级别的 asyncio.Lock，确保同一会话的消息串行处理
  • 获取全局并发信号量（默认 3），限制同时处理的会话数
  • 为需要流式输出的消息构建 on_stream / on_stream_end 回调
  • 处理完成后清理待处理队列、发布 turn_completed 事件

5.3 AgentRunner._run_core() — 核心迭代循环

5.a 场景卡片

函数：AgentRunner._run_core()

  • 调用时机：AgentRunner.run() 在 before_run hook 之后
  • 典型调用者：AgentRunner.run()
  • 前置条件：AgentRunSpec 已完整填充，hook 已就绪
  • 目的：执行”LLM调用→工具执行→LLM调用→…”的迭代循环，直到获得最终回复或达到迭代上限

5.b 逐行注释式精讲

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def _run_core(self, spec, hook, messages) -> AgentRunResult:
    # → 核心迭代循环，接收配置、钩子和初始消息
    final_content: str | None = None
    # → 最终回复内容，循环结束前保持 None
    tools_used: list[str] = []
    usage: dict[str, int] = {"prompt_tokens": 0, "completion_tokens": 0}
    # → Token 用量累加器
    stop_reason = "completed"
    # → 默认为正常完成
    empty_content_retries = 0
    # → 空回复重试计数（最多 _MAX_EMPTY_RETRIES=2 次）
    length_recovery_count = 0
    # → 输出截断恢复计数（最多 _MAX_LENGTH_RECOVERIES=3 次）
    injection_cycles = 0
    # → 中途注入循环计数（最多 _MAX_INJECTION_CYCLES=5 次）

    for iteration in range(spec.max_iterations):
        # → 主迭代循环，每次迭代是一次完整的"LLM调用→工具执行"或"LLM调用→回复"

        # --- 上下文治理（每次迭代前执行）---
        messages_for_model = self._drop_orphan_tool_results(messages)
        # → 移除孤儿工具结果：没有对应 assistant tool_call 的 tool 消息
        # → 如果不去除，某些 Provider（如 Anthropic）会拒绝请求
        messages_for_model = self._backfill_missing_tool_results(messages_for_model)
        # → 回填缺失的工具结果：有 assistant tool_call 但没有对应 tool 结果的
        # → 插入合成错误消息 "[Tool result unavailable — call was interrupted or lost]"
        messages_for_model = self._microcompact(messages_for_model)
        # → 微压缩：将旧的 compactable 工具结果替换为 "[xxx result omitted from context]"
        # → 只保留最近 _MICROCOMPACT_KEEP_RECENT=10 条可压缩结果
        messages_for_model = self._apply_tool_result_budget(spec, messages_for_model)
        # → 对历史中的工具结果应用字符数限制
        messages_for_model = self._snip_history(spec, messages_for_model)
        # → 如果消息总 token 数超过上下文窗口预算，从头部裁剪历史
        # → 确保裁剪后以 user 消息开头（满足 Provider 角色交替要求）

        # --- LLM 调用 ---
        response = await self._request_model(spec, messages_for_model, hook, context)
        # → 调用 Provider，支持流式/非流式/进度流三种模式

        # --- 工具执行路径 ---
        if response.should_execute_tools:
            # → finish_reason 为 tool_calls/function_call/stop 且有工具调用
            assistant_message = build_assistant_message(...)
            messages.append(assistant_message)
            # → 将 assistant 消息（含 tool_calls）追加到消息列表

            results, new_events, fatal_error = await self._execute_tools(...)
            # → 执行工具，支持并发/串行两种模式
            # → 分区策略：concurrency_safe 的工具可并发，exclusive 的工具必须单独执行

            # 将工具结果追加到消息列表
            for tool_call, result in zip(response.tool_calls, results):
                tool_message = {"role": "tool", "tool_call_id": ..., "content": ...}
                messages.append(tool_message)

            # --- 检查注入 ---
            drained, injection_cycles = await self._try_drain_injections(...)
            # → 排空中途注入消息（子Agent完成、目标续推等）
            continue
            # → 继续下一轮迭代

        # --- 最终回复路径 ---
        if response.finish_reason != "error" and is_blank_text(clean):
            empty_content_retries += 1
            if empty_content_retries < _MAX_EMPTY_RETRIES:
                continue  # 重试
            # → 空回复重试耗尽，尝试 finalization retry

        # --- 长度截断恢复 ---
        if response.finish_reason == "length" and not is_blank_text(clean):
            length_recovery_count += 1
            if length_recovery_count <= _MAX_LENGTH_RECOVERIES:
                messages.append(build_assistant_message(clean))
                messages.append(build_length_recovery_message())
                # → 追加 "请继续" 消息，让 LLM 继续生成
                continue

        # --- 中途注入检查 ---
        should_continue, injection_cycles = await self._try_drain_injections(...)
        if should_continue:
            had_injections = True
            continue

        # --- 最终回复 ---
        final_content = clean
        break
        # → 获得最终回复，退出循环

    else:
        # → for 循环正常结束（达到 max_iterations 上限）
        stop_reason = "max_iterations"
        if spec.finalize_on_max_iterations:
            final_content = await self._try_finalize_after_max_iterations(...)
            # → 移除工具定义后再次请求 LLM 生成最终回复

    return AgentRunResult(...)

6. 关键算法剖析

6.1 上下文窗口治理（Context Governance）

每次 LLM 调用前，AgentRunner 对消息列表执行 5 步治理流水线：

1
2
3
4
5
6
7
原始消息列表
  → _drop_orphan_tool_results()    # 移除孤儿工具结果
  → _backfill_missing_tool_results() # 回填缺失工具结果
  → _microcompact()                 # 微压缩旧工具结果
  → _apply_tool_result_budget()     # 应用工具结果字符限制
  → _snip_history()                 # Token 预算裁剪
  → 清理后的消息列表

时间复杂度：O(n) 每步，总体 O(5n)，n 为消息数。

边界处理：

  • 裁剪后确保以 user 消息开头（Provider 要求）
  • 如果找不到 user 消息，插入合成 "(conversation continued)" 消息
  • 保留 system 消息不受裁剪影响

6.2 工具分区并发执行

_partition_tool_batches() 将工具调用列表分为可并发组和必须串行组：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def _partition_tool_batches(self, spec, tool_calls):
    batches = []
    current = []
    for tool_call in tool_calls:
        tool = spec.tools.get(tool_call.name)
        can_batch = bool(tool and tool.concurrency_safe)
        # → concurrency_safe = read_only AND NOT exclusive
        if can_batch:
            current.append(tool_call)
            continue
        if current:
            batches.append(current)  # 提交当前并发组
            current = []
        batches.append([tool_call])  # 单独执行
    if current:
        batches.append(current)
    return batches

7. 设计决策分析

7.1 为什么 AgentLoop 和 AgentRunner 是两个类？

决策：将产品层关注点（会话、记忆、状态机、MCP）与纯执行逻辑（LLM 调用、工具执行、上下文治理）分离。

权衡：

  • ✅ AgentRunner 可独立测试，不需要完整的 MessageBus/Channel 环境
  • ✅ 产品层逻辑变更（如增加新的状态机阶段）不影响执行核心
  • ✅ 执行核心的优化（如上下文治理策略）不影响产品层
  • ❌ 增加了一层间接调用，调试时需要跨类追踪

替代方案：合并为一个类。这会使类超过 2000 行，难以维护和测试。

7.2 为什么使用状态机而不是简单的顺序调用？

决策：使用 8 状态状态机 (TurnState 枚举 + _TRANSITIONS 表) 驱动消息处理。

权衡：

  • ✅ 每个状态是独立方法，职责单一
  • ✅ 状态转换显式声明，易于追踪和调试
  • ✅ 可以独立测试每个状态处理器
  • ✅ StateTraceEntry 记录每个状态的耗时，支持性能分析
  • ❌ 简单场景（如纯文本回复）也要走完整状态链

替代方案：线性调用链。但这会使错误处理和条件分支难以管理。

7.3 为什么使用 checkpoint 机制而不是每次工具调用后全量保存？

决策：在工具执行间隙通过 checkpoint_callback 保存最小状态（assistant_message + tool_results），而不是每次全量保存会话。

权衡：

  • ✅ 减少 I/O 开销（只保存增量）
  • ✅ /stop 取消后可以恢复部分进度
  • ✅ checkpoint 保存在 session metadata 中，不污染消息历史
  • ❌ 恢复时需要合并逻辑（_restore_runtime_checkpoint）
  • ❌ 如果 checkpoint 与实际消息历史不同步，需要 overlap 检测

8. 学习检查点

📝 本章小结

  1. AgentLoop 是产品协调器，AgentRunner 是纯执行引擎，两者职责分明
  2. 8 状态状态机覆盖了从消息恢复到最终响应的完整生命周期
  3. 上下文治理流水线（5 步）确保每次 LLM 调用前的消息列表干净合法
  4. 工具分区并发执行根据 concurrency_safe 属性自动分组
  5. checkpoint + injection 机制支持中途恢复和中途消息注入

🤔 思考题

  1. 如果 _snip_history 裁剪后第一条非 system 消息是 assistant（不含 tool_calls），会发生什么？为什么需要 _enforce_role_alternation 中的 safety net？

    参考答案

    某些 Provider（如 GLM/Zhipu）会拒绝 system → assistant 的消息序列（错误 1214）。_enforce_role_alternation 在 nanobot/nanobot/providers/base.py#L548-L552 检测到这种情况时，会插入一条合成 user 消息 "(conversation continued)"，确保消息序列合法。这是 Provider 兼容性的最后防线。

  2. 为什么 _try_drain_injections 需要在工具执行后 和 最终回复后两个位置调用？只在一个位置调用会有什么问题？

    参考答案

    工具执行后的注入（nanobot/nanobot/agent/runner.py#L505-L508）处理子Agent完成等异步事件——这些事件可能在 LLM 思考期间到达，需要在下一轮 LLM 调用前注入。最终回复后的注入（nanobot/nanobot/agent/runner.py#L586-L591）处理持续目标续推——即使 LLM 已经生成了回复，如果有活跃的 sustained goal，应该继续工作而不是返回给用户。如果只在工具执行后检查，最终回复后会丢失目标续推；如果只在最终回复后检查，工具执行期间到达的子Agent结果会被延迟一轮。

  3. _microcompact 为什么只保留最近 10 条 compactable 工具结果？这个数字（_MICROCOMPACT_KEEP_RECENT=10）的选择有什么考量？

    参考答案

    10 条是平衡上下文质量和 LLM 记忆的折中选择（nanobot/nanobot/agent/runner.py#L70）。太少的保留（如 3 条）会导致 LLM 快速”忘记”最近的操作，可能在后续迭代中重复读取同一文件。太多的保留（如 50 条）则压缩效果有限，上下文窗口仍会被大量旧工具结果占据。10 条确保 LLM 能看到最近约 2-3 轮工具调用的完整结果，同时将更早的结果压缩为一行摘要，释放上下文空间给新的推理。