生命周期事件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

// Agent 生命周期
agent_start
agent_end(messages)

// Turn 生命周期（一轮 = 一次助手响应 + 工具调用/结果）
turn_start
turn_end(message, toolResults)

// 消息生命周期（user / assistant / toolResult 消息都会发）
message_start(message)
message_update(message)      // 仅异步流式
message_end(message)

// 工具执行生命周期（ToolExecutionStart/End 可被 hook 干预）
tool_execution_start(toolCallId, toolName, args)
tool_execution_update(toolCallId, toolName, args, partialResult)  // 仅异步流式
tool_execution_end(toolCallId, toolName, result, isError)

事件集对齐 pi 的生命周期模型（Agent/Turn/Message/Tool 四组，正常执行 start/end 成对；被拦截/畸形参数的调用不发射 End）。
每个事件自动带 timestamp（Unix 秒，实例化时刻）。

事件的作用

一、事件的作用：把「可观察」变成「可扩展」

一句话：事件机制让 Agent.run() 成为一个「可被插桩的黑盒」——循环本身写死了，但循环的每一步都广播出去，让任何新功能都能「监听而不改框架」。

先看现在 run() 里实际发生了什么：

1
2
3
4
5
6
7
8
9
10
11
12
13

Agent.run("东京天气")
│
├─ AgentStart
├─ MessageStart(user) → MessageEnd(user)          # 你的问题进 transcript
├─ TurnStart(1)
│   ├─ MessageStart(assistant) → MessageEnd(assistant)   # 模型想调工具
│   ├─ ToolExecutionStart → ToolExecutionEnd      # 执行 get_weather
│   ├─ MessageStart(tool) → MessageEnd(tool)      # 结果写回
├─ TurnEnd(message=assistant, tool_results=[...])
├─ TurnStart(2)
│   ├─ MessageStart(assistant) → MessageEnd(assistant)   # 模型直接回答
│   └─ TurnEnd(message=assistant, tool_results=[])       # 本轮工作闭环结算
├─ AgentEnd(messages=[...], ...)

这 8 个事件（同步阶段）就是「循环的完整心电图」——每一段代码执行都对应一个事件。而事件的作用 = 把「这段代码发生了」告诉任何想知道的人。

二、核心：为什么事件 = 扩展点

关键洞察在于——你的所有后续功能，几乎都是「想观察循环在干什么」，而不是「想改循环本身」。事件机制让这些功能都变成「加一个 hook」，而不是「改 run()」。

看三个后续阶段怎么用：

扩展 1：session 落盘（阶段 3）—— Agent 显式双写（不是 hook 消费事件流）

1
2
3
4

# 实际实现：Agent.run() 里每加一条消息，同时写内存 + 落盘
#   （不走 hook——session 是核心持久化，Agent 显式调用）
self.messages.append(user_msg)                  # 内存 transcript
self.session.add_message("user", user_input)    # 树 + 原子落盘（session 必填，无条件）

（注：早期设计设想过”用 hook 消费事件流写 session”，但实际实现是 Agent 双写——session 是核心功能，显式调用比 hook 更直接、不依赖事件顺序。）

扩展 2：context 压缩（阶段 4）—— prepare 每轮调用（不是监听事件）

1
2
3
4
5
6

# 实际实现：prepare() 在每次 llm.chat 前调用，内部判断是否超阈
#   （不走"监听 MessageEnd 计数"——压缩是发送前视图变换，时机在发消息前）
while ...:
    view = self._ctx.prepare(self.messages)     # 四层管线（超阈 → 摘要）
    resp = self.llm.chat(messages=view, tools=...)
    ...

ContextCompacted 事件早在设计时就预留了（阶段 4 用）——这就是「事件先定义、后发射」的意义：接口提前定好，实现到时接上。

扩展 3：coding agent UI（未来）—— 渲染过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 新功能：TUI 渲染 —— 监听事件更新界面
def ui_render(event):
    if isinstance(event, TurnStart):
        ui.show_round(event.iteration)
    elif isinstance(event, ToolExecutionStart):
        ui.show_tool_call(event.tool_name, event.args)
    elif isinstance(event, MessageEnd):
        ui.show_message(event.message.content)
    return None

```python
# 观察者只读订阅（UI 渲染、日志打点）
unsubscribe = agent.subscribe(ui_render)

# 决策拦截门禁（权限检查、参数改写）
agent = Agent(llm=llm, tools=tools, hooks=[
    (UserInputHook, input_sanitizer),
    (ToolCallHook, permission_guard),
    (ToolResultHook, output_redactor),
])
```

hook系统

架构解耦重塑（对标 Pi & Tau）：在最新架构中，只读事件（Event）与决策门禁（*Hook）实现彻底的物理与概念解耦：

• 只读事件（events.py）：AgentStart, TurnEnd, MessageEnd, ToolExecutionStart 等事实通知，统一通过 agent.subscribe(listener) 进行只读旁路监听；
• 决策门禁（hooks.py）：UserInputHook, AgentStartHook, BeforeModelCallHook, ToolCallHook, ToolResultHook 5 大专职拦截载荷，在 Agent(hooks=[(HookClass, callback), ...]) 中批量注册，负责阻断、参数改写与结果篡改。

多 hook 注册表

核心两个函数：注册（register）和触发（emit）。

注册：把 hook 回调挂到 Hook 目标名下

1
2

def register(self, target: type, callback: Callable):
    self._hooks.setdefault(target, []).append(callback)

• target = Hook 类型（如 ToolCallHook）
• callback = 门禁函数（(*Hook) -> HookResult | None）
• 同一个 Hook 类型可挂载多个回调，按注册顺序保序存储

触发：到点时异步逐个调用（支持 async/sync 混合回调）

1
2
3
4
5
6
7
8
9
10
11
12

async def emit(self, event: Any) -> HookResult | None:
    """异步触发 Hook，支持协程与普通函数，返回第一个非 None 结果（短路）。"""
    for cb in self._hooks.get(type(event), []):
        try:
            res = cb(event)
            if inspect.isawaitable(res):
                res = await res
            if res is not None:
                return res  # 短路：第一个干预结果即刻生效返回
        except Exception as exc:
            logger.error(f"Hook callback {cb} error: {exc}", exc_info=True)
    return None

HookResult 干预模型与五大决策拦截点

Hook 回调返回 None = 放行；返回 HookResult = 强力干预。

1
2
3
4
5
6
7
8
9
10

@dataclass(frozen=True)
class HookResult:
    block: bool = False                        # 是否阻断当前执行
    reason: str | None = None                  # block 阻断时的原因说明
    terminate: bool | None = None              # 三态：是否请求微内核提前退出当前批次
    updated_input: str | None = None           # 改写用户原始输入 (UserInputHook 用)
    updated_system_prompt: str | None = None   # 改写系统提示词 (AgentStartHook 用)
    updated_messages: list | None = None       # 临时改写送入模型的上下文 (BeforeModelCallHook 用)
    updated_args: dict | None = None           # 改写工具入参 (ToolCallHook 用)
    updated_result: str | None = None          # 改写工具执行结果 (ToolResultHook 用)

五大独立生命周期决策拦截点（hooks.py）

1. UserInputHook：截获用户原始输入，支持 block 阻断或 updated_input 前置清洗与改写；
2. AgentStartHook：启动前拦截，支持 updated_system_prompt 动态更新首条 system 消息；
3. BeforeModelCallHook：调 LLM 前拦截，支持 updated_messages 临时改写视图（临时 View 改写 vs 真实 Session 零污染）；
4. ToolCallHook：工具执行前拦截，支持 block 拦截高危命令或 updated_args 动态修补参数；
5. ToolResultHook：工具执行后拦截，支持 updated_result 篡改出参，以及 terminate 提前退出整批 ReAct 循环。

用起来的样子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 注册阶段（应用层，构造 Agent 后）
def guard_hook(event) -> HookResult | None:
    """权限门控（挂到 ToolExecutionStart，可拦截/改参数）"""
    if isinstance(event, ToolExecutionStart):
        if event.tool_name == "bash" and "rm -rf" in event.args.get("command", ""):
            return HookResult(block=True, reason="危险命令被拦截")
        if event.tool_name == "write":
            return HookResult(updated_args={**event.args, "path": abs(event.args["path"])})
    return None

def log_hook(event) -> HookResult | None:
    """日志（挂到 ToolExecutionStart，纯观察）"""
    if isinstance(event, ToolExecutionStart):
        print(f"[hook] {event.tool_name}({event.args})")
    return None

agent = Agent(llm=llm, tools=tools, hooks=[
    (ToolExecutionStart, guard_hook),   # 同一事件挂多个
    (ToolExecutionStart, log_hook),
])

机制一句话：注册表 = 事件类 → 回调列表；触发 = 到点时按序调用，短路返回。 你说「在某个事件时调用注册的 hook」——完全正确，_emit 就是那个「到点」的调用点。

Agent类（轻量 Harness 架构演进）

在经历 Tau 对齐重塑（阶段 18）后，Agent 类已经从早期“状态+循环+调度揉在一个类”的上帝类，全面演进为对标 Tau / Pi 的轻量有状态外壳（AgentHarness）：

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

┌─────────────────────────────────────────────────────────────────┐
│                      class Agent (AgentHarness)                 │
│              "轻量宿主外壳：持有状态 + 暴露原生事件流"             │
│                                                                  │
│  ┌───────────── __init__ 装配 ─────────────┐                     │
│  │  llm            ← 注入（模型边界）       │                     │
│  │  session        ← 会话（只追加存储驱动） │                     │
│  │  registry       ← 工具注册表            │                     │
│  │  messages       ← 状态/记忆（自愈清洗）  │                     │
│  │  message_queue  ← 干预队列 (Steer/Followup)                    │
│  │  hooks          ← HookRegistry（拦截管线）│                    │
│  └─────────────────────────────────────────┘                     │
│                                                                  │
│  ┌────────── 一等公民核心驱动：prompt_stream() ──────────┐        │
│  │ 异步生成器：run_agent_loop(llm, tools, messages...)    │        │
│  │       │                                               │        │
│  │       ▼ 逐一 yield 生命周期事件流                      │        │
│  │ AgentStart ➔ TurnStart ➔ MessageUpdate ➔ TurnEnd...   │        │
│  └───────────────────────────────────────────────────────┘        │
│                                                                  │
│  ┌────────── 经典便利门面：run() ──────────┐                     │
│  │  async def run(user_input):            │                     │
│  │      async for event in prompt_stream: │                     │
│  │          return final_text             │                     │
│  └────────────────────────────────────────┘                     │
│                                                                  │
│  ┌────────── 观察者监听：subscribe() ──────┐                     │
│  │  subscribe(listener) -> unsubscribe    │                     │
│  └────────────────────────────────────────┘                     │
│                                                                  │
│  ┌────────── 外部纯函数微内核 (loop.py) ───┐                     │
│  │  run_agent_loop: 彻底接管两阶段调度与双层循环                  │
│  │  _provider_context: 前置剥离空异常 + 自愈断头                 │
│  └─────────────────────────────────────────┘                     │
└─────────────────────────────────────────────────────────────────┘

对应代码：

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 Agent:
    def __init__(
        self,
        *,
        llm: LLM,
        tools: list[Tool],
        session: Session,
        system_prompt: str | None = None,
        max_iterations: int | None = None,
        context_budget: int | None = None,
        hooks: list[tuple[type[Event], Callable]] | None = None,
        ...
    ): ...

    # 1. 现代化原生事件流生成器（面向 TUI / CLI 打字机 / 前端）
    async def prompt_stream(self, user_input: str) -> AsyncIterator[Event]:
        """将用户输入送入微内核，实时产生事件流，同时驱动 session 落盘。"""
        # 前置决策点 UserInput / AgentStart ...
        async for event in run_agent_loop(
            llm=self.llm,
            model=self.model,
            system=system_prompt,
            messages=self.messages,
            tools=self.registry,
            context_manager=self._ctx,
            get_steering_messages=self._get_steering_messages,
            get_follow_up_messages=self._get_follow_up_messages,
            before_model_call=self.hooks.emit,
            before_tool_call=self.hooks.emit,
            after_tool_call=self.hooks.emit,
        ):
            # 同步落盘与外部观察者广播
            yield event

    # 2. 经典便利门面（100% 向后兼容老脚本与测试）
    async def run(self, user_input: str) -> str | None:
        """运行并返回最终文本（内部消费 prompt_stream）。"""
        final_text = None
        async for event in self.prompt_stream(user_input):
            if isinstance(event, AgentEnd):
                final_text = event.final_text
        return final_text

    # 3. 观察者订阅 API
    def subscribe(self, listener: Callable[[Event], None]) -> Callable[[], None]:
        """订阅事件流（返回注销句柄）。"""
        self._subscribers.append(listener)
        return lambda: self._subscribers.remove(listener)

    def abort(self) -> None: ...  # 协作式取消当前任务并自愈落盘
    def steer(self, message: str) -> None: ...  # 即时转向
    def follow_up(self, message: str) -> None: ...  # 排队追加

三、调度总装枢纽：prompt_stream 全流程剖析与 run() 职责解构

在现代 Agent 架构中，对外暴露的驱动接口形成了清晰的分工：

• async def prompt_stream(...)：系统最高阶的一等公民事件流生成器（Producer & Orchestrator）；
• async def run(...)：仅 10 行代码的阻塞式便利消费门面（Pure Consumer）。

1. prompt_stream 的六阶段全生命周期流水线

当外部调用 async for event in agent.prompt_stream("帮我重构代码"): 时，它严格按以下 6 大阶段推进：

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

用户输入 user_input
       │
       ▼
【阶段 1: 状态复位与扩展懒加载】
  └─ 复位 _aborted = False，生成新的 CancellationToken 取消令牌
  └─ 若插件/扩展尚未加载，触发 self.extension_manager.load()
       │
       ▼
【阶段 2: 用户输入安全门禁 (Hook 1: UserInputHook)】
  └─ 审查输入是否违规或需脱敏：
     ├─ block=True ──> 发射 AgentEnd(stop_reason="blocked") 并直接 return 退出！
     └─ 改写输入 ────> 更新 user_input 为脱敏后的新文本
       │
       ▼
【阶段 3: 会话树同步与转录本拓扑自愈】
  └─ 读取当前 Session 路径历史：restored = system + get_full_history_messages()
  └─ 运行 repair_tool_history(restored) 瞬间缝合所有断头调用，免疫 API 400
       │
       ▼
【阶段 4: 启动提示词门禁 (Hook 2: AgentStartHook)】
  └─ 审查/改写 System Prompt (允许插件动态植入全局规则)
       │
       ▼
【阶段 5: 微内核点火启动 (run_agent_loop)】
  └─ 将准备好的参数喂给 run_agent_loop，获得异步生成器 loop_gen
       │
       ▼
【阶段 6: 三向分发大循环 (async for event in loop_gen)】
  ├─ 1. 磁盘持久化分支 ──> 监听到 MessageEnd ➔ self.session.add_message 落盘
  ├─ 2. 旁路广播分支   ──> 遍历 self._subscribers 广播给第三方监听器
  └─ 3. 主干流式分支   ──> yield event 实时推给外部调用方

逐阶段技术细节解密：

1. 阶段 1（状态复位与扩展懒加载）：
   为本次调用创建专属的 CancellationToken。采用懒加载策略：在第一次真正发起问答时才执行 await self.extension_manager.load()，保证 Agent 初始化零异步卡顿；
2. 阶段 2（用户输入门禁 UserInputHook）：
   防守前置，零成本拦截。在用户输入触碰模型或写入历史前先行安检。若插件判定违规（block=True），直接就地发射 AgentEnd(stop_reason="blocked") 退出，不花一分钱模型费用；若需要脱敏（updated_input），则无缝篡改后放行；
3. 阶段 3（会话树指针同步与转录本自愈）：
   • 解决 /rewind（会话回滚）后的指针同步问题，严格以磁盘 Session 树当前指针为基准重新拉齐内存历史；
   • 运行 repair_tool_history，即使上次任务中途被强制杀死或崩溃，残留的断头工具调用也能被毫秒级修复，绝对杜绝大模型 API 报 400 Bad Request；
4. 阶段 4（系统提示词门禁 AgentStartHook）：
   允许策略插件根据当前用户身份或环境，动态向 system_prompt 中追加安全规约；
5. 阶段 5（微内核点火启动）：
   将组装完毕的消息历史、工具注册表、取消信号和队列回调打包，调用 run_agent_loop 启动纯函数引擎；
6. 阶段 6（三向分发大循环 Tee Splitter）：
   微内核吐出的每一个事件，在此处被“一鱼三吃”：
   • 向磁盘写：一旦捕获到 MessageEnd（说明消息已完整终态定型），立刻调用 self.session.add_message 通过原子重命名安全落盘到 .jsonl；
   • 向广播发：遍历 self._subscribers 列表，安全通知所有旁路监听器（日志、监控、状态栏）；
   • 向调用方推：通过 yield event 原生推向终端屏幕或 WebSocket。

2. prompt_stream 与 run() 的职责解构

很多传统框架的代码之所以臃肿，是因为把“流式输出”和“最终结果返回”混在了同一个方法里。而在现代微内核架构中，两者被进行了极其清晰的“生产者 vs 消费者”职责解构：

极简消费门面：run() 的优雅退化

正是因为 prompt_stream 把所有的脏活累活（安全拦截、自愈、驱动微内核、写文件）全部在底层包揽了，上层的便利方法 run() 才能做到前所未有的纯粹与轻薄——它退化成了仅仅 10 行代码的普通事件消费者：

1
2
3
4
5
6
7
8
9
10
11
12
13

async def run(self, user_input: str) -> str | None:
    """追加 user 消息 → 内部消费 prompt_stream 事件流 → 返回最终文本。"""
    final_text = None
    async for event in self.prompt_stream(user_input):
        if isinstance(event, AgentEnd):
            if event.stop_reason == "cancelled":
                return "(cancelled)"
            if event.stop_reason == "blocked":
                return event.final_text or "(blocked)"
            if event.stop_reason == "error":
                raise RuntimeError(event.final_text or "Error during model stream")
            final_text = event.final_text
    return final_text

这种职责解构的收益极为巨大：

1. 单一事实来源（Single Source of Truth）：整个框架只有 prompt_stream 一条真正的执行流水线，任何安全补丁、持久化优化或时序更新只需改动一处，run() 自动享受到所有红利；
2. 兼得双重体验：既能满足现代交互界面对毫秒级流式响应的严苛要求，又能保留单测和简单脚本中 ans = await agent.run(...) 一行调用的极致清爽！

3. 只读事件订阅管道：subscribe 与 _notify 广播器

在生产级 Agent 中，UI 状态更新、日志埋点、监控探针等外部系统需要实时感知 Agent 状态，但绝不能侵入或打断核心推理流程。Agent 类实现了对标 Pi（agent-session.js）与 Tau（harness.py）的只读轻量观察者管道：

1
2
3
4
5

# 订阅生命周期事件流（返回注销句柄）
unsubscribe = agent.subscribe(my_ui_listener)

# 业务注销（消除内存泄漏隐患）
unsubscribe()

_notify 广播器的六大工程保证：

1
2
3
4
5
6
7

async def _notify(self, event: Event) -> None:
    """将生命周期事件安全广播给所有旁路订阅者（对标 Tau AgentHarness._notify）。"""
    for sub in list(self._subscribers):
        with contextlib.suppress(Exception):
            res = sub(event)
            if inspect.isawaitable(res):
                await res

1. 快照遍历（Snapshot Iteration）：list(self._subscribers) 保证在迭代过程中即使外部调用了 unsubscribe() 也绝不会引发并发修改异常；
2. 异常完全隔离（Never-Throw Guarantee）：外部监听器抛出的任何异常被 contextlib.suppress(Exception) 彻底吞默，第三方的 bug 绝对无法击垮 Agent 主循环；
3. 同异步无缝自适应：自动检查 inspect.isawaitable(res)，无论监听器写成普通同步函数还是 async def 异步协程，框架均能自适应等待；
4. 闭包注销句柄（Teardown Token）：解耦索引，注销时按对象引用精准剔除；
5. 全量关键触点覆盖：在用户输入拦截阻断、系统提示词阻断、微内核主事件迭代循环、以及上下文压缩回写等所有生命周期转折点统一触发广播。

核心架构演进：为什么要把 run_agent_loop 抽离为独立纯函数？

在阶段 18 对标 Tau 架构的重塑中，我们做出了一个决定性的重构动作：将原先内联在 Agent 类中长达 260 行的双层循环彻底剥离，提炼为 src/my_agent_core/loop.py 中的独立纯函数生成器 run_agent_loop。

这一解耦设计带来了两大根本性的架构质变：

1. 纯函数无状态微内核的威力（五大工程质变）

传统框架喜欢把循环写死在类方法内部（class Agent: def run(self): while True:），导致循环与类状态深度咬合。在长对话或复杂场景下，这种设计弊端尽显。而抽离为独立的 async def run_agent_loop(...) 后，系统获得了五大不可替代的红利：

1. 彻底消除隐式状态突变（零 self，绝对确定性）：
   • 在类方法中，循环随时可能修改 self.xxx，多轮对话后状态满天飞；
   • 纯函数微内核完全没有 self，零隐式状态！所有输入项（大模型门面 llm、消息列表 messages、工具注册表 tools、取消令牌 signal）全部通过参数显式传入。输入确定则输出事件流绝对确定，排查调试的确定性拉满。
2. 通用计算发动机：多场景极致复用（Write Once, Run Everywhere）：
   • run_agent_loop 就像一台通用的汽车发动机，不绑定任何特定底盘，能够随处无缝挂载：
     • 交互式 CLI：配上 Agent 外壳，驱动 Session 树持久化与终端彩色打字机；
     • Web / API 流式服务：直接挂载到 FastAPI 的 StreamingResponse 或 WebSocket，中间事件流无需中间人直接推送到前端；
     • 大规模无头评测（Headless Benchmarks）：跑 SWE-bench 评测 1000 道题时，无需任何 Session 落盘与复杂插件装配，直接拿微内核裸跑，内存开销降低 80%，极速并发；
     • Subagent 子代理：轻量级临时执行手脚任务，无需背负庞大的 Agent 实例。
3. 单元测试速度提升 100 倍（零装配、纯内存秒跑）：
   • 以前测试循环逻辑，必须先装配一整台 Agent 大机器（Session 树、MemoryStore、Skills、Plugins），单测准备代码冗长且慢；
   • 现在在 tests/test_agent_loop_pure.py 中，直接传入 FakeLLM 和列表，0.001 秒测完并发工具与动态转向（11 项复杂微内核单测在 0.05 秒内全通）。
4. 动静分离，职责正交（SRP 原则）：
   • agent.py（类）只负责静态的结构与持久化（持有 Session 树、长期记忆、扩展装配）；
   • loop.py（函数）只负责动态的过程推理与调度（怎么推理、怎么并发调工具、怎么转向、怎么中断）。
5. 双层循环拓扑与动态干预毫秒级响应：
   • 内层微观循环在工具执行间隙动态收割 get_steering_messages() 实现即时转向；
   • 外层宏观循环在轮次自然收尾处收割 get_follow_up_messages() 自动衔接追问任务。

2. 将中间“事件流”提升为系统级的一等公民（First-Class Stream）

大模型的运行本质上是一个基于时间轴的渐进流式过程：Token 在一个字一个字蹦出、思考块在逐步形成、工具在按需触发并耗时运行。

传统 Callback / Hook 的三大灾难（打洞插管）：

在传统黑盒设计中，函数只能最后 return final_text。外部若想知道中间状态，必须在外面定义一堆回调函数（on_token、on_tool_start）硬塞给 Agent。

• 控制权倒置：调用方不能主动拉取数据，只能被框架被动调用；
• 异常黑洞：回调函数一旦抛出异常，会直接击垮主循环，或者被框架静默吞掉；
• 无法优雅暂停与取消：外部用户在前端点了停止，回调依然在疯狂发射，根本停不下来。

生成器（AsyncIterator[Event]）的一等公民之美：

在现代 run_agent_loop 架构下，事件流就是函数的正常返回值本身！没有任何侧路管子，没有控制权倒置！

1. 调用方掌握 100% 绝对主动权（传送带模型）：
   外部调用者只需写出自然优雅的 async for 循环，站在传送带旁边接住事件：

   1 2 3 4 5 6 7 8 9 10 11 12

   # 外部使用代码极度自然、主动： async for event in agent.prompt_stream("帮我计算 37 * 19"): if isinstance(event, MessageUpdate): print(event.chunk.delta, end="", flush=True) # 原生打字机！ elif isinstance(event, ToolExecutionStart): print(f"\n[正在调用工具: {event.tool_name}]") elif isinstance(event, ToolExecutionEnd): print(f"\n[工具执行完毕: {event.result}]") # 想停就停：直接 break 即可触发生成器安全清理与底层中断自愈！ if user_clicked_stop: break

2. 让上层门面实现前所未有的轻薄：
   因为事件流成了一等公民，外层的 agent.run() 彻底被掏空，退化成了仅 10 行代码的纯事件消费者：

   1 2 3 4 5 6 7

   async def run(self, user_input: str) -> str | None: """自己只做普通消费者，消费事件流并返回最终文本。""" final_text = None async for event in self.prompt_stream(user_input): if isinstance(event, AgentEnd): final_text = event.final_text return final_text

   一套微内核底座，同时完美满足了“前端极致细粒度的响应式流渲染”与“后端一行代码调用的极简体验”！

3. 架构深思：Pi 的 ExtensionAPI 如何将“只读事件”与“拦截决策点”相辅相成？

在深入对比 Pi（@earendil-works/pi-coding-agent）与 Tau 的设计后，我们终于看清了困扰很多开发者的核心谜团：为什么在很多初学者的 Agent 代码里，事件和 Hook 会被写成一团乱麻？而 Pi 却能写得如此规整有序？

答案就在于：Pi 深刻认清了“只读事件（传信息）”与“决策拦截点（改控制流）”是两套完全正交的体系，并通过 ExtensionAPI 的三层架构实现了二者的完美相辅相成！

① 概念正交化：纯只读事件 vs 决策拦截点

② 为什么初学者（包括我们重构初期）会陷入杂乱？

• 概念混淆的陷阱：我们把所有的只读事件和拦截点硬塞进了同一个 HookRegistry，导致微内核里充斥着几十处毫无意义的 if hook_registry: await hook_registry.emit(ev)；
• 双通道视觉噪音：每个事件既要 yield 给外部，又要 await emit 给内部。这就好比广播电台播报“天亮了”（纯通知），却非要在广播室门口设个安检站一本正经地问安检员“天亮了需要拦截吗？”。

③ Pi 的三层解耦美学：ExtensionAPI 与微内核的精密咬合

Pi 极为高明地通过三层架构化解了这一矛盾：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

┌─────────────────────────────────────────────────────────────┐
│ 【外部插件层 (Extension Developer)】                        │
│  面向开发者提供极简统一入口：pi.on(event, handler)           │
│  • 监听只读事件：pi.on("message_update", handler)            │
│  • 拦截安全门禁：pi.on("tool_call", handler)                 │
└──────────────────────────────┬──────────────────────────────┘
                               │ 智能分流
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 【中间调度层 (ExtensionManager)】                           │
│  • 只读事件 ➔ 接入单向广播观察者队列 (Observer Queue)         │
│  • 决策拦截点 ➔ 串联为洋葱模型中间件流水线 (Middleware)      │
└──────────────────────────────┬──────────────────────────────┘
                               │ 注入纯函数插槽
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 【底层微内核 (runAgentLoop)】                               │
│  • 微内核对 ExtensionAPI 零感知！不持有全局 Hook 注册表     │
│  • 纯只读事件：微内核只管一行代码 yield event (纯单向推流)   │
│  • 五大决策点：微内核仅通过入参暴露 5 个具体干净的回调插槽   │
│    (beforeToolCall, afterToolCall, transformContext...)     │
└─────────────────────────────────────────────────────────────┘

这种相辅相成的设计妙在哪里？

1. 对插件作者（API 面）：极致的直觉与统一。开发者不需要记两套完全不同的 API，只要在同一个 pi.on 后面写函数，TypeScript 类型系统会自动约束当前事件“能不能拦截、可以返回什么”；
2. 对中间层（调度面）：职责清晰的智能分流。只读事件走广播，安全门禁走洋葱流水线（支持多插件按序改写参数与一票短路阻断）；
3. 对微内核（底层引擎）：极致的轻盈与规整。微内核根本不需要感知插件的存在，也不需要充斥 30 处 Hook 检查样板代码；它只在 5 个固定的命脉节点上留出回调插槽，其余时间顺风顺水地跑自己的纯状态机！

这正是顶级工业级 Agent 框架的架构魅力所在：上层 API 极其人性统一，底层微内核极其纯粹高效，中间通过清晰的分流管道相辅相成！

4. 旁路广播总线：subscribe() 订阅机制的底层本质与实战全景

很多开发者在接触到 agent.prompt_stream(...) 后会产生一个经典困惑：

“既然我已经能通过 async for event in agent.prompt_stream(...) 直接拿到所有事件了，为什么还需要 agent.subscribe(...) 订阅机制？这不是多此一举吗？”

答案就在于：prompt_stream 是单通道的“独食吸管”，而 subscribe 是多方共享的“公共大喇叭”。

① 为什么必须要有 subscribe？（两大刚需场景）

1. 配合最常用的极简门面 await agent.run()：
   大部分日常业务或测试中，我们只想要最终答案：ans = await agent.run("计算 37*19")。
   然而，run() 内部已经把 prompt_stream 给消费吞掉了，外部只能拿到最终字符串。如果你此时既想用清爽的 agent.run()，又想在控制台打印工具调用日志，或者在状态栏转圈，唯一的办法就是提前挂载 agent.subscribe(my_listener)！
2. 克服 Python 生成器的“单消费者限制”：
   Python 的异步生成器不能被多个 async for 同时读取。如果你的系统需要同时做三件事：
   • 主流程：在终端打印大模型的话；
   • 日志系统：把所有工具调用记录进 agent.log；
   • 监控系统：把消耗的 Token 上报云端；
     若没有 subscribe，你必须把日志和监控代码全揉进主流程的 async for 里，代码变成大泥球。有了 subscribe，日志和监控模块只需各自独立 agent.subscribe(...)，旁路静默监听，与主业务彻底解耦。

② agent.subscribe(…) 的入参到底是什么？

入参是一个“函数”（由你编写的回调函数 Callable）。
该函数必须接收一个参数（框架会把发生的 event 实体传给它）。支持三种形式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 形式 1: 普通函数（最直观常用）
def my_listener(event: Event):
    print("收到事件:", event)

agent.subscribe(my_listener)  # 传入函数名，切勿带括号！

# 形式 2: 异步函数（适合写数据库或发网络请求）
async def my_async_listener(event: Event):
    await websocket.send_json({"event": type(event).__name__})

agent.subscribe(my_async_listener)

# 形式 3: 匿名函数（适合一行调试）
agent.subscribe(lambda ev: logger.info("Event: %s", ev))

③ 订阅只能看到事件类型吗？（绝对不是！）

每个推送到订阅函数的 event，都是一个满载真实业务数据的强类型实体对象，你能从中拿到所有的具体数据：

• 工具启动（ToolExecutionStart）：
  • event.tool_name：工具名（如 "bash"、"read"、"write"）；
  • event.args：大模型传入的真实入参字典（如 {"path": "src/app.py"}）；
  • event.tool_call_id：工具调用唯一 ID。
• 工具结束（ToolExecutionEnd）：
  • event.result：工具执行的完整输出内容（如文件全文、命令执行结果）；
  • event.is_error：执行是否出错（True / False）；
  • event.terminate：是否触发了阶段 7 提前熔断标记。
• 大模型流式吐字（MessageUpdate）：
  • event.delta：刚刚吐出的那一个字/词（驱动终端流式打字机）。
• 任务终结（AgentEnd）：
  • event.final_text：最终交付给用户的完整回答文本；
  • event.iterations：总共跑了多少轮推理；
  • event.stop_reason：为何结束（"end_turn"、"max_iterations" 等）。
• 上下文压缩（ContextCompacted）：
  • event.tokens_before 与 event.tokens_after：压缩前后的真实 Token 水位。

实战中，你可以轻松写出一个赏心悦目的终端监听器：

1
2
3
4
5
6
7
8
9
10
11

def pretty_printer(event: Event):
    if isinstance(event, ToolExecutionStart):
        print(f"🔧 [工具启动] {event.tool_name}，参数: {event.args}")
    elif isinstance(event, ToolExecutionEnd):
        status = "❌ 失败" if event.is_error else "✅ 成功"
        print(f"{status} [工具完成] 结果: {event.result}")
    elif isinstance(event, MessageUpdate):
        print(event.delta, end="", flush=True)

agent.subscribe(pretty_printer)
await agent.run("计算 37 * 19")

④ 底层实现原理（三部曲机制）

subscribe 在底层代码中仅有约 15 行，设计极其紧凑：

1
2
3
4
5
6
7
8

1. 登记通讯录 (subscribe)
   agent.subscribe(listener) ──> self._subscribers.append(listener)

2. 逐个打电话广播 (prompt_stream 内部)
   微内核产生新事件 ──> for sub in list(self._subscribers): sub(event)

3. 随时退订除名 (unsubscribe 闭包)
   调用返回的 unsub() ──> self._subscribers.remove(listener)

1. 闭包式退订令牌（Teardown Token）：
   subscribe 返回一个私有闭包 unsubscribe()。调用方无需持有监听器列表或索引，直接调用 unsub() 即可瞬间安全退订，从机制上彻底杜绝内存泄漏；
2. list(self._subscribers) 浅拷贝快照：
   在广播循环中使用快照遍历，防止监听器在执行过程中自行动态注销导致 RuntimeError: list changed size during iteration 崩溃；
3. contextlib.suppress(Exception) 异常隔离：
   第三方订阅函数内部若抛出任何未捕获异常，全部被静默隔离，绝对不影响 Agent 主调度循环的稳定性；
4. 同步/异步自适应（isawaitable）：
   底层通过 inspect.isawaitable(res) 自动检测返回值，若订阅函数是 async def，主线程自动进行 await 调度。

⑤ 关键疑惑澄清：订阅之后，回调函数到底什么时候会被调用？（实时时序拆解）

很多刚接触观察者模式的同学常常会问：“我调用 agent.subscribe(my_callback) 之后，这个函数到底是立刻执行？还是等任务结束一起批量执行？”

核心真相：

1. 订阅登记的那一刻：一次都不会调用！
   它仅仅是把你传入的函数对象加入到 self._subscribers 通讯录列表中，然后进入静默待命状态。
2. 真正的触发时刻：调用 await agent.run("提问") 开始后，事件产生的瞬间【立刻当场执行】！
   看 agent.py 内部的消费时序（第 415 行）：

   1 2 3 4

   async for event in loop_gen: # 微内核刚 yield 吐出一个事件的瞬间，下一行立刻打电话通知你： await self._notify(event) yield event

   微内核每前进一步，在 1 微秒之内，你的回调函数就会在当前协程/线程里立刻被当场执行。只有等你的回调函数执行完毕，微内核才会接着往下走！
3. 时序绝对保序（In-band / Real-time）：
   从提问到结束，一个单轮任务中你的回调函数会被实时触发十几次：
   • 启动瞬间 ➔ 调你 1 次：AgentStart
   • 轮次开始 ➔ 调你 1 次：TurnStart
   • 模型吐字 ➔ 高频连续调你：每产生 1 个增量 Token 就立刻调你 1 次 MessageUpdate（驱动打字机实时上屏！）
   • 模型写完 ➔ 调你 1 次：MessageEnd
   • 工具开始 ➔ 调你 1 次：ToolExecutionStart（驱动卡片弹出开始转圈）
   • 工具跑完 ➔ 调你 1 次：ToolExecutionEnd（驱动卡片变绿打勾）
   • 轮次闭环 ➔ 调你 1 次：TurnEnd
   • 任务终结 ➔ 调你 1 次：AgentEnd

这种“即时串联广播”确保了事件先发生必先通知、后发生必后通知，毫秒级无延迟同步！

⑥ 项目实战：利用这个订阅机制，我们真实实现了哪六大杀手级功能？

在 my-pi-agent 中，这个看似只有十几行代码的 subscribe 广播总线，实际上是撑起整个项目产品化运转的“地基”：

1. 功能 1：100% 像素级 Pi-TUI 终端交互体验（my-pi-tui）：
   • 订阅 agent_start ➔ 输入框上沿实时挖槽启动 Braille 10 帧高频旋转指示器 ── ⠸ Working ──；
   • 订阅 message_update ➔ 逐字驱动流式打字机与思维链（Thinking）折叠；
   • 订阅 tool_execution_start/update/end ➔ 驱动圆角工具卡片挂载、100ms 增量日志刷新与打勾 ✓ / 打叉 ✗ 动画；
   • 订阅 auto_retry_start ➔ 遇到 429 报错时在顶栏呈现倒计时 重试中 (1/3) 2.0s...；
   • 订阅 agent_end ➔ 停止所有旋转，边框平滑自愈复原为直线，安全归还光标。

     架构价值：Python 内核保持 100% 纯无头，没有一行写死的 UI 代码！前端通过 RPC 订阅这一套事件流，即可像素级复刻 Pi 原厂终端。

2. 功能 2：分会话双轨制全链路 Debug 调试与审计系统（tracer.py）：
   • 启动 --debug 时，调试器通过 agent.subscribe(self.tracer) 零侵入旁路接入；
   • 订阅 TurnStart/MessageEnd ➔ 毫秒级测算大模型推理耗时与 Token 增量，写入 <id>.debug.log；
   • 订阅 ToolExecutionStart/End ➔ 测算工具物理执行耗时；
   • 订阅所有里程碑事件 ➔ 自动脱敏密码与 API Key（替换为 ***REDACTED***），写入标准事件流 <id>.events.jsonl；
   • 切换会话时自动 rebind，实现会话级物理隔离。
3. 功能 3：核心微内核“防偷懒与遗漏”智能监工（TaskGuardHook）：
   • 在 agent.py 内部订阅 TurnEnd：

     1	self.subscribe(lambda ev: guard.on_turn_end(ev) if isinstance(ev, TurnEnd) else None)

   • 每当一轮对话结束时，监工看一眼 tasks.json 工单看板：如果大模型还有 pending 状态的待办卡片却试图偷懒结束，监工立刻自动给大模型发一条即时转向（Steering）提醒词：
     “系统提示：你还有未完成的待办任务，请不要提前结束，继续执行下一步！” 彻底解决大模型做了一半就交卷的顽疾！
4. 功能 4：Terminal-Bench 2.1 自动化评测的数据采集器（my-pi-eval）：
   • 评测调度器通过 inner_agent.subscribe(tracer) 旁路接入；
   • 在零侵入沙箱的前提下，全自动将每道题的最终得分、执行耗时、真实 Token 消耗以及 8200 万缓存命中精准记录下来，输出标准的 metrics.json，为打出 83.53% 胜率战报提供了铁证数据链。
5. 功能 5：第三方 Extension 插件生态扩展（ExtensionAPI.on）：
   • 允许第三方开发者通过 @api.on(TurnEnd) 编写扩展，实现“任务跑完后发一条微信/飞书 Webhook 提醒”，完全不需要改动任何框架源码。
6. 功能 6：全量 814 个离线测试的高精度黑盒断言：
   • 单测代码只需执行 agent.subscribe(events.append) 把事件收集到内存列表，通过断言事件序列（如 assert any(isinstance(e, ToolExecutionStart) for e in events)）完成高保真白盒/黑盒回归，毫秒级跑完 814 个测试。

5. 微内核调度流水线独立专文导航

至此，Agent 类已经蜕变成一个高度轻量、专注会话状态管理与全系统依赖注入的 Harness 宿主外壳。

关于微内核内部的双层状态机循环（宏观 Follow-up 追问 + 微观 ReAct 转向）、单轮 8 步时序流水线、大模型推理车间（_assistant_turn）以及工业级工具执行车间（_execute_tools_turn）的深度解析，已独立拆分并收录于专属博客中：

👉 《my-pi-agent—loop微内核.md》

前言

为了更深入理解agent的工程实现，本文会逐步从底层搭建一个agent（不借助任何框架如langchain等），agent是在循环中调用工具的模型，直到给定任务完成（ReAct架构），如下图所示。

核心组件如下

工具系统

先纠正一个常见误解

模型是怎么知道”该调工具了”还是”该直接输出答案”的？这个判断是 LangChain/LangGraph 实现的吗？

“判断”根本不是 LangChain/LangGraph 实现的，是模型本身的能力。

OpenAI 等厂商对模型做过 function calling（工具调用）专项训练：模型学会了一件事——当请求里带有工具描述、且对话内容需要工具时，输出一个结构化的 tool_calls；不需要时，输出普通文本。这个决策发生在 OpenAI 服务器上的模型推理过程中，LangChain 源码里没有、也不可能有一行”决定何时调工具”的逻辑。

工具调用流程

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

=== 1. 上行翻译：Python 函数 -> 模型看得懂的 JSON schema ===
{
  "type": "function",
  "function": {
    "name": "multiply",                        ← 来自函数名
    "description": "Multiply two integers.",   ← 来自 docstring
    "parameters": {                            ← 来自类型标注 a: int, b: int
      "properties": {
        "a": { "type": "integer" },
        "b": { "type": "integer" }
      },
      "required": ["a", "b"],
      "type": "object"
    }
  }
}

=== 2. 实际发给 OpenAI 的完整请求 payload ===
{
  "model": "gpt-4.1-mini",
  "stream": false,
  "tools": [ ...上面那段 schema... ],           ← tools 是请求的顶层字段
  "messages": [
    { "content": "Use the multiply tool to calculate 37 times 19.", "role": "user" }
  ]
}

=== 3. 下行翻译：OpenAI 原始响应 -> AIMessage.tool_calls ===
message 类型: AIMessage
content: ''
tool_calls: [{'name': 'multiply', 'args': {'a': 37, 'b': 19}, 'id': 'call_abc123', 'type': 'tool_call'}]
路由判断 bool(msg.tool_calls) = True -> 去 tools 节点

=== 4. 模型决定直接回答（不调工具）时 ===
tool_calls: []
bool(msg2.tool_calls) = False -> 去 __end__

工具参数

人类选择工具前需了解工具的功能、使用场景和输入参数。大模型同理——模型依据这些信息选择合适的工具。按以下JSON格式提供工具信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

{
    "type": "function",
    "function": {
        "name": "get_current_weather",
        "description": "当你想查询指定城市的天气时非常有用。",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "城市或县区，比如北京市、杭州市、余杭区等。"
                }
            },
            "required": ["location"]
        }
    }
}

• type字段固定为"function"；
• function字段为 Object 类型；
  • name字段为自定义的工具函数名称，建议使用与函数相同的名称，如get_current_weather或get_current_time；
  • description字段是对工具函数功能的描述，大模型会参考该字段来选择是否使用该工具函数。
  • parameters字段是对工具函数入参的描述，类型是 Object ，大模型会参考该字段来进行入参的提取。如果工具函数不需要输入参数，则无需指定parameters参数。
    • type字段固定为"object"；
    • properties字段描述了入参的名称、数据类型与描述，为 Object 类型，Key 值为入参的名称，Value 值为入参的数据类型与描述；
    • required字段指定哪些参数为必填项，为 Array 类型。

发起 Function Calling 前，在代码中定义工具信息数组（tools），包含每个工具的函数名、描述和参数定义。该数组在后续请求时作为参数传入。

工具注册机制

把「一个函数变成可调用的工具」其实分两步：

1. @tool（打包）：把一个函数 → 一个 Tool（说明书 + 函数本体）。产出的是一个对象。
2. ToolRegistry.register（注册）：把一堆 Tool 收进注册表（内部 name → Tool 字典）。产出的才是一张能按名查人的花名册，registry.execute 就靠它，凭模型回传的名字字符串反查到真函数。

函数注册为工具

1
2
3
4

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b

这段和下面完全等价：

1
2
3
4
5

def multiply(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b

multiply = tool(multiply)   # ← @tool 就是帮你写了这一行

所以 tool 就是个普通函数：吃进去一个函数，吐出来一个 Tool。@ 只是语法糖。

这里藏着一个你最好亲自验证一下的事实——执行完 multiply = tool(multiply) 之后，multiply 这个名字已经不是函数了，而是一个 Tool 对象。原来的函数被塞进了 Tool.func 里存着。

1
2
3
4
5
6

class Tool:
    """一个可被模型调用的工具：函数本体 + 发给模型的 JSON schema（类化重构后）。"""
    func: Callable[..., Any]            # 函数本体
    name: str                           # 工具名（默认函数名）
    description: str                    # 描述（默认 docstring）
    params_model: type[BaseModel]       # pydantic 参数模型（create_model 动态建模）

Callable[..., Any] 是什么

Callable 来自 from typing import Callable（第 12 行），它是一个泛型类型，写法是 Callable[[参数类型...], 返回类型]。比如：

1 2 3

Callable[[int, str], bool] # └──┬───┘ └┬┘ # 接收 int和str 返回 bool

完整调用链

第 1 步：装饰时把函数存进 Tool

1
2
3
4
5
6

return Tool(
    name=func.__name__,          # 比如 "get_weather"
    ...
    func=func,                   # ← 原函数本体存进来了
)

第 2 步：调用时按名字找回这个包裹（registry.execute 内部）

1
2

name = tc["function"]["name"]           # 模型说："我要调 get_weather"
target = registry.get(name)             # 查注册表，拿到对应的 Tool 对象（没有则 None）

第 3 步：解析模型给的参数（registry.execute 内部）

1
2
3

args = json.loads(tool_call.function.arguments)
# 模型传来的是 JSON 字符串，比如 '{"city": "北京"}'
# 解析后变成 Python dict：{"city": "北京"}

第 4 步：用 func 真正调用（Tool.execute 内部）

1
2

result = target.execute(args)           # 校验 + 执行（pydantic 参数校验，永不抛）
# 等价于：get_weather(city="北京")

**args 是字典解包，把 {“city”: “北京”} 展开成关键字参数 city=”北京” 传给 func。这就是「利用 Tool 对象的 func 调用函数」的确切时刻。

create_model 动态建模工具参数

为什么必须动态？

框架是「库」，不知道你会写什么工具

my_agent_core 是被 main.py 使用的库。库的代码在写的时候，根本不知道使用者会注册哪些工具：

1
2
3
4
5
6
7
8

@tool
def get_weather(city: str) -> str: ...        # 用户可能写 1 个字段

@tool
def multiply(a: int, b: int) -> int: ...      # 可能写 2 个字段

@tool
def search_docs(query: str, tags: list[str], limit: int = 5) -> str: ...  # 3 个字段，类型各异

每个工具的参数形状都不一样。如果模型类是静态的，框架就得在源码里把「所有可能的工具签名」都写成类——那是不可能的。唯一的出路是：模型类在运行时、根据实际收到的函数来造。

实现

1
2
3
4
5
6

# tools.py:76
model = create_model(
    f"{func.__name__}_Args",          # 类名，如 "get_weather_Args"
    __config__=ConfigDict(extra="forbid"),
    **fields,                         # ← 关键
)

**fields 把 {“city”: (str, …)} 展开成关键字参数，等价于直接写：

create_model("get_weather_Args", __config__=..., city=(str, ...))

schema 在完整闭环里的角色

1
2
3
4
5
6
7
8

① 框架 → 模型：发送 schema（"我有这些工具，参数格式如下"）
                    │
② 模型 → 框架：返回 tool_call
       name: "multiply"
       arguments: '{"a": 37, "b": 19}'   ← 模型按契约生成的合规参数
                    │
③ 框架执行：json.loads(arguments) → {"a": 37, "b": 19}
            func(**args) → 703            ← registry.execute 内部

第 ② 步值得多看一眼：’{“a”: 37, “b”: 19}’ 这个 JSON 字符串是模型自己生成的——它读了 schema，知道该给 a 和 b 各传一个整数，于是按格式”填表”。schema 写得好不好（尤其 description），直接决定模型用得对不对。这也是为什么文件开头把这一层叫「上行翻译层」：把 Python 函数翻译成模型能读懂的 JSON 说明书。

生成的 Tool.parameters 就是一份标准 JSON Schema：

1 2 3 4 5 6 7 8

{ "type": "object", "properties": { "a": {"type": "integer"}, "b": {"type": "integer"}, }, "required": ["a", "b"], }

读作：「参数是一个对象，含 a、b 两个字段，都是整数，都必填」。注意第 56~57 行的循环就是在逐参数填这张「表」。

然后 Tool.to_openai_schema（或 registry.get_schemas 批量）再包一层 OpenAI API 要求的外壳：

1 2 3 4 5 6 7 8

[{ "type": "function", "function": { "name": "multiply", "description": "Multiply two integers.", "parameters": { ...上面那份 schema... }, }, }]

最终实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

agent.py（循环）      → 只做：调用 registry.execute + 配对写回
   │
registry.py（注册表） → 收完整 tool_call：json.loads + 查表 + 错误转 ToolResult
   │
tools.py（工具本体）  → Tool 类：动态建模 + to_openai_schema + execute + __call__
   │
ToolResult（结果）    → 永不抛，ok/data/error，serialize 转字符串
---------------------
@tool def get_weather(city: str) -> str: ...   # 一行定义
       ↓
Tool 类（动态建模 + to_openai_schema + execute + __call__）
       ↓
ToolRegistry.register(tool) / execute(tool_call)   # 注册表分发
       ↓
agent.py: registry.execute(tc).serialize()   # 循环只做调用 + 写回

src/my_agent_core/tools/core.py

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

ToolResult（dataclass）

@dataclass
class ToolResult:
    ok: bool                   # 是否成功
    data: Any = None           # 成功时的结果数据
    error: str | None = None   # 失败时的错误消息
    terminate: bool = False    # 熔断退出标记（用于阶段 7 提前退出 ReAct 循环）
    meta: dict[str, Any] = field(default_factory=dict) # 结构化诊断元数据

    def serialize(self) -> str:
        # 转成写入 messages 的字符串；失败返回错误文本

Tool（类）

class Tool:
    def __init__(
        self,
        func,
        *,
        name: str | None = None,
        description: str | None = None,
        params_model: type[BaseModel] | None = None,
        is_parallel_safe: bool = False,       # 关键：声明并发安全性（默认 False 保护因果顺序）
        timeout: float | None = None,          # 执行超时时间（结合 asyncio.wait_for）
        raw_schema: dict[str, Any] | None = None, # 外部/MCP 预编译 Schema 透传
    ): ...

    def execute(self, args: dict) -> ToolResult:
        # 校验 + 异步执行 + 超时保护，永不抛：pydantic 校验失败或工具异常 → ToolResult(ok=False, error=...)

tool（模块级函数，装饰器工厂）

def tool(
    func=None,
    *,
    name: str | None = None,
    description: str | None = None,
    params_model: type[BaseModel] | None = None,
    is_parallel_safe: bool = False,
    timeout: float | None = None,
):
    # @tool 装饰器工厂：支持 @tool 和 @tool(name=..., is_parallel_safe=True)
    # 内部构造 Tool 对象并返回

my_agent_core/registry.py

1
2
3
4
5
6
7
8
9
10
11
12

ToolRegistry（类）

class ToolRegistry:
    def __init__(self): ...

    def register(self, tool: Tool) -> None: ...   # 注册工具（同名静默覆盖）
    def unregister(self, name: str) -> None: ... # 注销工具
    def get(self, name: str) -> Tool | None: ... # 查表
    def get_schemas(self) -> list[dict]: ...     # 导出 OpenAI 兼容 tools 列表

    async def execute_batch(self, tool_calls: list[dict]) -> list[ToolResult]:
        """批量执行工具调用：一票否决制因果时序保护调度。"""

核心调度机制：一票否决因果保序并发（Unanimous Parallel）

当大模型在单轮 ReAct 交互中同时发出多个工具调用（例如同时下发 5 个 tool_calls）：

1. 全员通过才放行并发：
   系统检查该批次目标工具对象的并发标记：all(tool.is_parallel_safe for tool in batch_tools)；
   只有当该批调用的全部工具均为 is_parallel_safe=True（例如全为只读的 read, grep, find, list），系统才调用 asyncio.gather 并发极速推进，总耗时由 $O(N)$ 骤降至 $O(1)$！
2. 一票否决回退保序串行：
   一旦批次中包含了哪怕一个 is_parallel_safe=False（例如包含写操作 edit, write, todo 或外部未知工具），整批工具调用立刻自动降级为严格按大模型输出的原序串行执行！
   架构价值：从调度器层面从根本上杜绝了“本应先写后读的操作，因并发导致先读到了脏数据”的因果时序倒置（Causal Inversion）风险！

架构演进展望：对标 Tau 的 AgentToolResult 协议升级

在阶段 18 对标 Tau 架构的深度重塑中，工具返回协议将进一步向前迈进：

1. 多模态内容与诊断隔离（AgentToolResult）：
   • content: list[TextContent | ImageContent]：原生向大模型喂回文本块与图片块；
   • details: JSONValue：专门向前端 TUI / CLI 传递诊断元数据（如 exit_code, duration, truncated_bytes），该字段完全不消耗大模型 Token，彻底解耦“给模型看的文本”与“给前端展示的数据”；
2. 动态工具扩展（added_tool_names）：
   工具（如插件加载器或 MCP 动态发现）执行后，可通过此字段向会话声明动态新开放的工具集，触发上下文窗口的动态 Token 补算；
3. 提前交接终止信号（terminate）：
   允许工具（如人机交互问卷或不可逆故障）主动向调度循环发出终止信号，无需等待模型多跑一轮。

内部工具

框架层除了让用户自己 @tool，还内置了一批工具——放在 my_agent_core.tools.builtin 包，用工厂函数构造。其中四个文件工具对标 pi 的基本四件套：

两个共同的关键点

① 工厂函数收 root——路径逃逸防护的边界

四个工具都长这样（工厂函数吃 root，吐 Tool）：

1
2
3
4

def make_read_tool(root: str | Path) -> Tool:
    def read(path: str, limit: int | None = None) -> str:
        ...
    return Tool(func=read, name="read")

root 是「工作区根目录」，工具内部每条路径都先过一遍 _safe_path：

1
2
3
4
5

def _safe_path(root: Path, p: str) -> Path:
    path = (root / p).resolve()          # 消掉 .. 和符号链接
    if not path.is_relative_to(root):    # 逃出根目录了？
        raise ValueError(f"Path escapes workspace: {p}")
    return path

resolve() 把 ../etc/passwd 折算成绝对路径，is_relative_to(root) 检查它还在不在根里，逃逸就报错——这是文件工具的第一道安全门。

② 错误不抛，且提供精细化纠错提示（Prompt-Quality Errors）

四个工具内部都是 try/except，把异常转成极具指导意义的错误字符串返回，给模型提供明确的自我纠错线索：

• read：越界时明确返回文件实际总行数（如 Offset 200 is beyond end of file ('app.py' has only 80 lines total)）；
• edit：未找到时提示检查缩进与换行，多处匹配时提示提供更多上下文；
• bash：超时时自动捕获并保留超时前打印的已输出日志，方便模型判断是否卡在交互输入（如 -y）。

③ FileMutationQueue 细粒度文件锁（并发安全与性能兼得）

write 和 edit 工具内部通过 FileMutationQueue 按文件绝对路径获取 asyncio.Lock。修改不同文件时全员并发执行，修改同一文件时自动排队串行，兼备极致性能与写安全性。

bash 的额外两道防护

1
2
3
4
5
6
7
8

_DANGEROUS = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]

def bash(command: str) -> str:
    if any(d in command for d in _DANGEROUS):
        return "Error: Dangerous command blocked"
    r = subprocess.run(command, shell=True, cwd=root,
                       capture_output=True, text=True, timeout=120)
    return (r.stdout + r.stderr).strip() or "(no output)"

• 危险命令黑名单：rm -rf /、sudo 这类直接拦下。是「尽力而为」的字符串匹配，不是真沙箱（真隔离得靠容器/权限层，那是 coding agent 层的事）。
• 120 秒超时：sleep 1000 这种挂死命令不会无限等，超时返回 "Error: Timeout (120s)"。

为什么是「工厂函数」而不是「直接一个工具」

因为 root 是显式的、每个应用不同——框架层不知道你的工作区在哪。所以工具是 make_read_tool(root) 现造的，调用方（将来 coding agent 层）传自己的 workspace 根进去，再 Agent(tools=[make_read_tool(root), ...]) 装配。这和「@tool 装饰器直接定义」的区别在于：内置工具需要一个运行时才知道的参数（root）注入。

永不抛出：工具出错也是一条消息

一、 观念转变：异常 vs 消息（谁才是接收者？）

理解这一节的关键，在于搞清楚错误是给谁看的：

1
2
3
4
5
6
7
8

❌ 传统思路（把错误当成异常）：
工具报错 (FileNotFoundError) ──> 穿透调用栈 ──> 打断 Agent Loop ──> 终端崩溃 ──> 只能由人类程序员重新启动。
【接收者是 Python 解释器 / 调用栈】

✅ Agent 哲学（把错误当成一条消息）：
工具报错 (FileNotFoundError) ──> 框架捕获 ──> 包装成普通消息: ToolResultMessage(isError=True)
                              ──> 追加进上下文 ──> 喂给大模型 ──> 模型看懂了：“路径错了，我先 ls 看看目录” ──> 自动修复！
【接收者是大模型】

二、 统一出口：6 种错误，1 种产物

在工具调用的完整生命周期中，可能会有 6 个不同阶段抛错，但无论哪一步挂掉，Pi 都保证绝不向外抛异常，全部归一化为一条标准的
ToolResultMessage：

1
2
3
4
5
6
7
8

LLM 输出 ToolCall
  │
  ├── 1. 工具不存在 (如模型幻觉造了一个未注册的工具) ────> ToolResultMessage { isError: true, content: "Tool xxx not found" }
  ├── 2. prepareArguments 参数预处理抛错 ───────────────> ToolResultMessage { isError: true, content: 预处理异常 }
  ├── 3. Schema 参数强校验失败 (如传错类型) ───────────> ToolResultMessage { isError: true, content: Pydantic 校验错误 }
  ├── 4. beforeToolCall 权限拦截 (如危险命令被阻断) ─────> ToolResultMessage { isError: true, content: 拦截原因 }
  ├── 5. tool.execute 运行时崩溃 (如 500/超时/文件不存在) ─> ToolResultMessage { isError: true, content: 运行时异常 }
  └── 6. afterToolCall 后置处理抛错 ────────────────────> ToolResultMessage { isError: true, content: 后置异常 }

对 Agent Loop 而言：它看到的结果永远是一个干净的 ToolResultMessage，因此主循环的 while 可以安全地继续转动，把结果喂给下一轮大模型。

经典对比（看看 Pi 是怎么写第一层的）：

1. Read 工具（read.ts）

• ❌ 差的报错：raise Exception(“Read error”) → 模型两眼一抹黑。

• ✅ Pi 的报错：Offset 200 is beyond end of file (100 lines total) → 模型立即明白：“文件只有 100 行，那我下次传 offset=50”。

  2. Bash 工具（bash.ts）

• ❌ 差的报错：raise Exception(“Command failed”)。

• ✅ Pi 的做法（教科书级）：
  把“执行到一半已经被捕获的 stdout 输出” + “退出码 / 超时状态” 打包在一起返回：

  1 2 3 4 5

  Command failed with exit code 1. --- Output before error --- npm ERR! code ENOENT npm ERR! syscall open npm ERR! path /package.json

  模型看到这段具体的错误输出，就能像人一样分析根因。

异常 → 消息：编码前后对比

1
2
3
4
5
6
7
8
9
10
11
12
13
14

工具抛出的原始异常（except 之前）：       编码后的 ToolResultMessage（except 之后）：
FileNotFoundError: [Errno 2] No such     {
  → 一路穿透管道                            role: "toolResult",
  → 打断 Agent Loop                         toolCallId: "call_abc",
  → 事件序列不完整，UI 卡死                  toolName: "read",
                                            content: [{
                                              type: "text",
                                              text: "[Errno 2] No such file or directory"
                                            }],
                                            isError: True   ← 唯一标记
                                          }
                                          → 追加到对话历史
                                          → 下一轮发给模型
                                          → 模型看到后自己决定怎么办

三、 异常流动的全景拓扑图：从底层报错到双轨流向与前端表现

在传统的单体或普通 Web 开发中，一段代码抛出 Exception 会顺着调用栈向上穿透，如果没有 try...catch 兜底，整个进程直接崩溃退出。
但在 ReAct 智能体架构中，异常绝不是程序的灾难，异常是大模型获取反馈、进行自我修正的最宝贵数据！

因此，当一个工具调用发生意外时，异常会经历完整的 7 层流转、蜕变与分流：

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

┌────────────────────────────────────────────────────────────────────────┐
│ 1. 工具执行层 (Tools: bash, read, edit...)                             │
│    抛出 Python 异常 (FileNotFoundError) 或 非零退出码 (exit_code != 0)    │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │ 原始异常 / 失败对象
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│ 2. 装饰器门禁层 (@tool / ToolRegistry)                                 │
│    【Never-Throw 不变式拦截】: 捕获一切 Exception ➔ 包装为结构化对象      │
│    ToolResult(ok=False, error="...", data=None)                        │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │ 标准 ToolResult(ok=False)
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│ 3. ReAct 微内核流水线 (loop.py: _execute_tools_turn)                    │
│    - res.serialize() ➔ 提取错误字符串                                   │
│    - is_err = not res.ok ➔ 标记 is_error=True                          │
│    - 广播事件 1: yield ToolExecutionEnd(is_error=True, result=obs)     │
│    - 广播事件 2: 生成 Message(role="tool", is_error=True) ➔ 写入内存    │
└───────────────────┬───────────────────────────────┬────────────────────┘
                    │                               │
       强类型事件流  │                               │ Tool 消息追加
                    ▼                               ▼
┌──────────────────────────────────────┐  ┌──────────────────────────────┐
│ 4. 表现层流向 (RPC ➔ TUI)            │  │ 5. 大脑层流向 (LLM 自愈闭环)  │
│  - RpcServer 序列化为 JSON-RPC 发送   │  │  - Session 原子写盘 (.jsonl) │
│  - tracer 记录 [TOOL_CALL_END] ERROR │  │  - 次轮 LLM 看到报错信息     │
│  - TUI 变红底、打红叉 ✗、显示 (失败)  │  │  - 模型反思并自我修正重试    │
└──────────────────────────────────────┘  └──────────────────────────────┘

七层详细流动过程拆解

1. 第 1 层：工具层暴雷
   • 参数畸形：大模型生成了非法的入参（如 read 漏传 path），Pydantic 入口抛出 ValidationError；
   • 系统级硬异常：目标文件不存在抛出 FileNotFoundError，或权限拒绝抛出 PermissionError；
   • 外部进程失败：bash 执行命令由于命令不存在或环境不合规，返回非零退出码（如 exit code 1 或 255）。
2. 第 2 层：装饰器拦截（Never-Throw 保证：把“灾难”变成“数据”）
   • Tool.execute() 将整个工具执行包裹在安全沙箱中，绝不上抛任何未捕获异常；
   • 任何硬异常均被转化为统一数据载体：ToolResult(ok=False, error="...")。
3. 第 3 层：微内核加工（状态判定与事件解耦）
   • loop.py 接过 ToolResult：
     • 计算 is_err = not res.ok（因为 ok=False，所以 is_err=True）；
     • 提取 obs = res.serialize()（提取出详细报错文本）；
     • 派发 ToolExecutionEnd(is_error=True, result=obs) 事件通知宿主与终端；
     • 构建 Message(role="tool", content=obs, metadata={"tool_call_id": call.id, "is_error": True})。
4. 第 4 层：会话持久化（原子落盘）
   • 外层 Agent.prompt_stream 捕捉到 MessageEnd，将带有 is_error: True 的 tool 消息原子写盘至 sessions/.../<id>.jsonl；
   • 保证会话转录本严格遵守双向配对不变式（每一个 tool_call 都有对应的 tool 回复，哪怕它是报错）。
5. 第 5 层：可观察性与日志记录（时序存档）
   • tracer.py 在 .debug.log 中记录 [TOOL_CALL_END] tool=bash status=ERROR duration=...ms；
   • 在 .events.jsonl 中如实写入 {"type": "tool_execution_end", "isError": true, "result": "..."}。
6. 第 6 层：TUI 终端渲染（红叉视觉反馈）
   • 前端接收到 tool_execution_end 事件：
     • 终止正在旋转的 Braille 转圈动效；
     • 依据权威字段 if (event.isError) 立即渲染亮红色的 ✗ 与 (失败)；
     • 卡片背景切换为警告底色 toolErrorBg，向人类开发者如实警示。
7. 第 7 层：大模型自我修正闭环（ReAct 的灵魂）
   • 内核将失败的 tool_msg 喂入次轮 ReAct 循环；
   • 大模型在上下文里看到了上一步的详细报错，触发自我反思（Reasoning），自动修改参数重试或改变解决策略。

四、 避坑陷阱：为什么工具“言行不一”会导致前端画绿勾？（真实事故复盘）

在实际工程落地中，我们曾踩过一个极具教育意义的深坑：工具明明执行失败了，终端上却赫然画出了一个成功的绿色对勾 ✓！

1
2
3

事故现场：
✓ bash (command=find src -name "*.py" | head -100) (0.0s)
Command failed with exit code 255: ...

1. 致命的隐式假成功

当时 bash.py 内部写下了这样一段代码：

1
2
3

if exit_code != 0:
    # 开发者直觉：命令挂了，我返回一段说明失败的文本
    return f"Command failed with exit code {exit_code}:\n{output}"

但框架层（@tool）的判定规则是极度严谨的：

1
2
3
4
5
6
7
8

try:
    result = await func(...)
except Exception as exc:
    return ToolResult(ok=False, error=str(exc))

# 只要你的 Python 函数平平安安 return 了一个普通字符串，没有抛出 Exception：
# 框架一律判定为：执行成功！
return ToolResult(ok=True, data=result)

1
2

is_err = not res.ok  # 因为 ok=True，所以 is_err = False！
yield ToolExecutionEnd(is_error=is_err, ...)

2. 前端是“无辜且严谨”的

前端 TUI 组件的渲染规则遵循严格的协议解耦：

1
2
3
4
5

if (this.isError) {
    icon = theme.fg("error", "✗"); // 红叉
} else {
    icon = theme.fg("success", "✓"); // 绿勾
}

3. 根治法则

在工具实现中，任何业务失败或非零退出码，绝不能只返回普通错误字符串，必须显式返回 ToolResult(ok=False, error=msg)，保持类型系统判定与业务内容的高度统一，让绿勾准确蜕变为红叉！

4. 核心深思：前端靠事件，模型靠上下文（双轨正交解耦）

思考题：前端是依靠 ToolExecutionEnd 这个事件，而自适应纠错则是把报错信息存入 messages 里面对吗？

回答：完全正确！这是 Agent Harness 架构中至关重要的关注点分离（Separation of Concerns）：

• ToolExecutionEnd 是“给人类看的”：负责 TUI 终端界面的视觉反馈。前端是一个纯粹事件驱动的状态机，它不关心也不应该去解析几十万 Token 的庞大历史树，只认 ToolExecutionEnd 里的 isError 布尔值，以毫秒级响应决定画绿勾 ✓ 还是红叉 ✗，避免遍历消息历史带来的卡顿与闪烁；
• messages 里的 Tool 消息是“给大模型看的”：负责 ReAct 循环的推理闭环。大模型看不到屏幕，它反思与自我纠偏的唯一依据，就是下一轮发给它的 messages 列表中包含的 role: "tool", content: "报错详情...", is_error: True。

在微内核 loop.py 中，工具执行完毕后会同时向左（发射事件给人类）和向右（追加消息给模型）一分为二：

1
2
3
4
5
6

# 1. 向左走：通过事件流发射给人类与 TUI (纯表现层)
yield ToolExecutionEnd(tool_call_id=call.id, tool_name=call.name, result=obs, is_error=is_err)

# 2. 向右走：封装成消息追加进 messages 供给下一轮大模型 (纯推理层)
yield MessageStart(tool_msg)
yield MessageEnd(tool_msg)

这使得人类交互体验（UX）与大模型认知流（Cognitive Loop）彻底正交解耦、互不干扰。

五、 写自定义工具时的最佳实践

借鉴 Bash 工具的写法，自定义工具的 execute 应该长这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

async def execute_custom_tool(params: MyArgs) -> ToolResult:
    try:
        # 1. 正常业务逻辑
        result_data = await do_some_io(params)
        return ToolResult(ok=True, data=result_data)

    except KnownBusinessErrorA as err:
        # 2. 主动识别错误 A：给出具体的线索与修复建议
        return ToolResult(
            ok=False,
            error=f"Resource '{params.id}' not found. Available IDs are: {get_available_ids()}"
        )

    except KnownBusinessErrorB as err:
        # 3. 主动识别错误 B：给出范围限制
        return ToolResult(
            ok=False,
            error=f"Timeout after {params.timeout}s. Try increasing timeout or splitting query."
        )

    except Exception as err:
        # 4. 未知异常：尽量带上原始异常细节，交给框架
        return ToolResult(ok=False, error=f"Unexpected failure: {type(err).__name__}: {err}")

终极防线：工具尚未执行就被中途中断？——对标 Tau 的 tool_history 三阶段自愈状态机

在前一章中，我们详细分析了“永不抛出（Never-Throw）”原则——它保证了当代码执行进工具函数内部时，即便遇到异常也会被包装为 ToolResult(ok=False, error=...)，避免 Agent 进程崩溃。

但是在真实大模型 Agent 系统中，还潜伏着一个更隐蔽、破坏力更强的系统级死穴：如果工具压根“没来得及执行完”，或者在执行前夕就被外部打断了呢？

1. 致命的“断头工具调用（Dangling Tool Call）”

各大主流大模型（OpenAI、Anthropic、DeepSeek 等）在设计 Function Calling 协议时，有着严苛的图灵机契约：

🚨 模型 API 契约约束：
如果一条 assistant 消息声明了 tool_calls: [{"id": "call_123", ...}]，那么在对话历史中，其紧邻的下一条消息必须是 role: "tool" 且带上匹配的 tool_call_id: "call_123"。

如果出现以下任何一种真实突发场景：

1. 用户主动取消：模型发起了一个耗时 30 秒的编译命令工具，用户等不及直接按了 Ctrl+C 或触发了 await agent.abort()；
2. 网络异常截断：模型吐出了工具调用描述后，网络突然抖动中断；
3. 并发工具部分失败：模型一次性发起了 3 个并发工具调用，第 1 个执行成功，第 2 个发生致命系统错误直接退出了当轮 ReAct 迭代。

这会导致会话历史里留下了一条“只有 ToolCall，没有对应 ToolResult”的消息——即断头调用。

现实案例剖析：一个“断头”如何让会话彻底脑死亡？

我们来看一个真实发生过的典型崩溃场景：

1. 模型决定调用工具：一次性生成了两个工单 call_001（读 a.txt）和 call_002（一个耗时 30 秒的巨型编译命令）。
2. 正常执行前半截：call_001 顺利执行完成，结果回填。
3. 中途意外打断：正在执行 call_002 时，用户等不及了按下了 Ctrl+C，或者代码里触发了 await agent.abort()。
4. 致命空档产生：此时 Python 进程被打断，call_002 根本没有产生任何返回值。
5. 坏死数据固化：会话历史直接停留在这一秒，并持久化到了硬盘的 .jsonl 文件中。

此时磁盘里的消息记录变成了这种残缺状态（断头了！）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

[
  {"role": "user", "content": "帮我处理一下代码"},
  {
    "role": "assistant",
    "tool_calls": [
      {"id": "call_001", "name": "read"},
      {"id": "call_002", "name": "build"}  // ← 发起了 call_002
    ]
  },
  {
    "role": "tool",
    "tool_call_id": "call_001",
    "content": "a.txt 的内容是 Hello"
  }
  // 🚨 致命缺失：由于中途中断，历史里根本没有 call_002 的结果！
]

接下来会发生什么悲剧？

当下一次用户再发一句新的话（比如：“算了，不要编译了，给我讲个笑话吧”）：
Agent 会把上面这段包含断头的历史加上新的提问，一起打包发给 OpenAI / DeepSeek / Claude。

大模型服务端的网关校验器一扫历史，发现 assistant 发起了 call_002，后面居然没有对应的 tool 结果，网关直接无情抛出：

1
2
3

HTTP 400 Bad Request:
An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'.
Missing tool response for: call_002.

更致命的是：因为这条残缺的历史记录已经保存在你的硬盘会话文件里了，以后只要你加载这个会话，无论发什么新指令，大模型永远报 400！这个会话文件就彻底“脑死亡”报废了！

tool_history.py 是如何化解这场悲剧的？

tool_history.py 就像一个“消息转录本的智能外科医生兼安检门”。在会话从磁盘恢复后、以及每次送给大模型之前，它都会对整个消息链进行拓扑自愈。

当它扫描到上述残缺历史时，发现 call_002 悬空无下文，它会在内存中自动就地合成一条合法的工具结果插进去：

1
2
3
4
5
6

{
  "role": "tool",
  "tool_call_id": "call_002",
  "content": "Tool call interrupted by user",  // 明确告诉大模型：这个工具被用户打断了
  "metadata": {"is_error": true}
}

这样一来，两全其美：

1. 大模型的 API 契约瞬间被满足了：每个 call 都有对应的 tool 结果紧随其后，API 绝对不会再报 400 拒绝服务！
2. 大模型的认知逻辑也顺畅了：模型看到 Tool call interrupted by user，在上下文中就自然理解“哦，原来刚才那个任务被用户取消了”，下一轮它就能基于这个事实正常回答！

2. 为什么简单的遍历补齐搞不定？

最朴素的想法是：“遍历消息列表，只要发现某条 Assistant 消息有 tool_calls，后面没跟 Tool 消息就硬塞一条假消息进去。”

在严苛的工程实践中，这种写法会踩进大坑：

1. 跨轮同名 ID 复用：某些开源或商用模型在多轮长对话中，可能会重复生成相同的 tool_call_id（如多次调用都叫 "call_0"）。朴素遍历极易把第 5 轮的真实合法结果“偷”给第 1 轮的同名调用，导致第 5 轮反而变成了断头！
2. 位置错位与孤儿结果（Orphan Results）：偶尔因并发乱序或重试，存在没有被任何 Assistant 声明引用的游离 ToolResult。如果在上下文中放行，同样会触发 API 400 报错。

3. 对标 Tau 的三阶段确定性拓扑自愈算法

为了彻底扫清断头死锁，我们在阶段 17 深度对齐了 Tau (tau_agent.tool_history) 的三阶段确定性自愈状态机，并在核心层实现了 my_agent_core/tool_history.py。

算法核心流转拓扑如下：

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

原始可能有断头/孤儿的历史消息列表
               │
               ▼
┌─────────────────────────────────────────────────────────────┐
│ 【Phase 1: 预留就近配对 (Reserve Adjacent Pairs)】           │
│  先扫描当前位置与期望位置完全匹配的合法结果，建立强绑定排他    │
│  锁定，占住位点，绝对防止被跨轮同名 ID 或贪心扫描错误抢夺！   │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 【Phase 2: 贪心匹配与断头补齐 (Greedy Match or Synthesize)】 │
│  对所有未被锁定的 ToolCall 进行向后贪心扫描：                │
│  • 优先在调用位置之后寻找合法的真实结果；                    │
│  • 若彻底找不到，自动合成标准中断结果：                      │
│    Message(role="tool", content="Tool call interrupted by   │
│            user", metadata={"tool_call_id": id,             │
│                             "is_error": True})              │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 【Phase 2.5: 真实结果反超 (Real Result Priority)】           │
│  若后续扫描发现由于遍历顺序遗漏了真实合规的结果，立即撤销    │
│  已合成的中断占位，让真实结果反超生效！                      │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 【Phase 3: 转录本重塑与孤儿清理 (Reconstruction & Pruning)】 │
│  • 按照 Assistant 声明的调用顺序，严格紧随插入匹配的工具结果 │
│  • 凡是没有被任何 ToolCall 认领的游离结果（孤儿），全部剔除！│
└─────────────────────────────────────────────────────────────┘
                               │
                               ▼
                      100% 结构合法的自愈转录本

4. 核心源码落地与诊断模型 (tool_history.py)

自愈函数返回一个结构化不可变诊断数据类 ToolHistoryRepair：

1
2
3
4
5
6
7
8
9
10
11
12
13

# src/my_agent_core/tool_history.py

_INTERRUPTED_TOOL_RESULT = "Tool call interrupted by user"

@dataclass(frozen=True, slots=True)
class ToolHistoryRepair:
    """修复后的合法转录本以及结构化诊断计数。"""
    messages: tuple[Message, ...]       # 自愈后的合法消息序列
    changed: bool = False               # 是否发生了纠偏与修复
    synthesized_results: int = 0        # 补齐的中断结果条数
    dropped_orphan_results: int = 0     # 丢弃的游离孤儿结果条数
    dropped_duplicate_results: int = 0  # 丢弃的重复结果条数
    reordered_results: int = 0          # 纠正乱序重排的条数

核心修复算法 repair_tool_history 的精简实现：

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

def repair_tool_history(messages: Sequence[Message]) -> ToolHistoryRepair:
    """对会话历史进行确定性拓扑自愈，确保所有工具调用均合法闭合。"""
    # 提取所有 (msg_idx, call_offset) 的工具调用事件
    call_occurrences = ...
    # 按 tool_call_id 索引所有真实存在的 tool 消息位置
    results_by_id = ...

    selected_results: dict[tuple[int, int], tuple[int | None, Message]] = {}
    used_result_positions: set[int] = set()
    synthesized_results = 0

    # ── Phase 1: 预留已就近配对的调用 ──
    for occurrence, call, expected_pos in call_occurrences:
        if expected_pos < len(messages) and _get_tool_call_id(messages[expected_pos]) == call.get("id"):
            selected_results[occurrence] = (expected_pos, messages[expected_pos])
            used_result_positions.add(expected_pos)

    # ── Phase 2: 剩余调用贪心匹配或补齐中断结果 ──
    for occurrence, call, _ in call_occurrences:
        if occurrence in selected_results:
            continue
        call_id = str(call.get("id", ""))
        candidates = results_by_id.get(call_id, [])
        matched_pos, matched_msg = _find_best_candidate(candidates, used_result_positions, occurrence[0])

        if matched_msg is not None:
            selected_results[occurrence] = (matched_pos, matched_msg)
            used_result_positions.add(matched_pos)
        else:
            # 补齐标准中断工具结果
            synthetic = Message(
                role="tool",
                content=_INTERRUPTED_TOOL_RESULT,
                metadata={"tool_call_id": call_id, "is_error": True},
            )
            selected_results[occurrence] = (None, synthetic)
            synthesized_results += 1

    # ── Phase 2.5: 真实结果反超合成中断 ──
    # ... 确保真实结果绝不被虚假中断误吞 ...

    # ── Phase 3: 重建转录本、孤儿丢弃与保序重排 ──
    repaired: list[Message] = []
    for msg_idx, message in enumerate(messages):
        if message.role == "tool":
            # 游离孤儿丢弃；已配对工具在 assistant 之后紧邻插入，此处直接跳过
            continue

        repaired.append(message)
        if message.role == "assistant":
            for offset in range(1, len(_get_tool_calls(message)) + 1):
                _, tool_res = selected_results[(msg_idx, offset)]
                repaired.append(tool_res)  # 严格保序紧随其后

    return ToolHistoryRepair(messages=tuple(repaired), changed=..., synthesized_results=synthesized_results, ...)

5. 架构级双重防护网：内层 Never-Throw 护函数，外层 tool_history 护转录本

通过这次 Tau 对齐演进，整个工具调用系统形成了固若金汤的双层立体防御架构：

至此，哪怕用户在工具并发执行的任意毫秒暴力掐断任务，或者强行杀死进程后重启系统，会话也能在毫秒级内自动恢复闭合，彻底攻克了大模型应用中最棘手的会话持久化死锁顽疾！

6. 工业级七阶段工具执行流水线：并发批处理与实时进度流

在完成了工具参数校验和拓扑自愈后，当大模型在一轮中发起了多个工具调用时，我们进入了最核心的 工具执行车间 (_execute_tools_turn)。

严格对齐 Pi 官方架构契约与 Tau 微内核设计，工具执行被划分为七个确定性阶段：

1. 阶段 1：输出截断防御检查 (_fail_tool_calls_from_truncated_message)：
   当检测到模型输出触达 Token 上限被截断（stop_reason == "length"）时，断然拒绝执行任何工具，防止流式 salvage 拼出残缺参数导致代码写崩或命令腰斩。自动合成警告错误并回传模型引导重新完整发起调用。
2. 阶段 2：Preflight 广播 (ToolExecutionStart)：
   在审批与执行前，率先按 source order 广播 ToolExecutionStart，使 UI 能够毫秒级渲染工具准备运行状态。
3. 阶段 3：串并行决策网关 (ToolRegistry.execute_batch)：
   悲观读写分流：全只读安全工具（is_parallel_safe=True）启用 asyncio.gather 全并发加速；只要包含任一写入/串行工具，整批退化为保序串行执行，防止因果时序倒置。
4. 阶段 4：前置审查审批与改参 (before_tool_call)：
   通过 _coerce_tool_call 归一化入参，调用 before_tool_call 审批，支持安全阻断（block）与参数就地热修改（updated_args）。
5. 阶段 5：并发批执行与流式进度回传 (ToolExecutionUpdate)：
   采用 asyncio.Queue 与 loop.call_soon_threadsafe 跨线程安全桥接，支持长耗时工具（如 Bash 编译或子代理）在运行态向外广播累积快照（Cumulative Snapshot）。生命周期锁存（accepting_updates）确保工具返回后丢弃迟到回调。
6. 阶段 6：后置改写与单工具终态广播 (after_tool_call & ToolExecutionEnd)：
   调用 after_tool_call 支持结果脱敏与改写（updated_result），广播包含 terminate 状态的 ToolExecutionEnd。
7. 阶段 7：转录本保序归档与批量优雅熔断 (MessageStart/End & should_terminate)：
   无论并发执行完成顺序如何，回传大模型的 role="tool" 消息严格按 Assistant 原始 Source Order 恢复排布。若批次中任一工具（any() 语义）或 Hook 返回 terminate=True，立即终结 ReAct 循环，保全 final_text 并正常交付结果。

阶段 5 核心攻坚：生产者-消费者管道与三大技术死结破解

在实现第 5 阶段（并发执行与实时流式进度回传）时，架构面临了三个看似不可调和的技术冲突：

1
2
3
4
5
6
7
8
9
10
11
12
13

【冲突 1】异步生成器 vs 并发批量执行
  • 需求 A：多个工具必须并发跑（比如同时读 3 个文件），底层用的是 asyncio.gather(...)；
  • 需求 B：_execute_tools_turn 是一个生成器，必须一有进度就立刻 yield 出去。
  • 痛点：在 asyncio.gather 的深处是不能直接向外层生成器 yield 的！

【冲突 2】主事件循环 vs 工作线程池
  • 需求 A：很多工具是同步的（如普通的 Python 函数、调用操作系统的 bash），必须扔进线程池
           asyncio.to_thread 跑，不能卡死主线程；
  • 痛点：Python 的异步队列 asyncio.Queue 是严格单线程的，工作线程直接碰队列就会崩溃闪退！

【冲突 3】实时流式读取 vs 任务结束防死锁
  • 需求 A：前台必须一直监听队列，只要有日志就拿出来；
  • 痛点：前台怎么知道后台什么时候“全部跑完了”？如果盲目等，后台跑完后前台就会永久卡死（死锁）。

破局方案：前后台解耦的“单向传送带”管道模型

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

┌──【后台 Producer: runner 异步任务】──┐
│                                       │
│  并发执行 tool_1, tool_2, tool_3 ...   │
│                   │                   │
│                   ▼ 产生进度          │
│      on_update("50% complete")        │
│                   │                   │
│                   ▼                   │
│   safe_put_update (线程安全路由)       │
│     ├─ 主线程: queue.put_nowait()     │
│     └─ 工作线程: call_soon_threadsafe │
│                   │                   │
│                   ▼ 写入              │
└─────────────> ┌────────┐ <────────────┘
                │ 异步队列│
                │ queue  │
                └────────┘
                    │
                    ▼ 读取
┌──【前台 Consumer: 生成器主循环】────┐
│                                       │
│  while True:                          │
│      item = await queue.get()         │
│      if item is _SENTINEL: break      │
│      yield item  (实时推给前端 UI!)    │
│                                       │
└───────────────────────────────────────┘

核心代码落地：5 个严丝合缝的实现步骤

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

# ── 阶段 B: 并发批执行与实时进度流式广播 ──
if prepared_calls:
    # 步骤 1: 建立线程安全事件队列与独一无二的哨兵
    queue: asyncio.Queue[Event | object] = asyncio.Queue()
    _SENTINEL = object()
    loop = asyncio.get_running_loop()
    loop_thread_id = threading.get_ident()

    # 步骤 2: 跨线程安全网关路由
    def safe_put_update(ev: Event) -> None:
        if threading.get_ident() == loop_thread_id:
            queue.put_nowait(ev)
        else:
            with contextlib.suppress(RuntimeError):
                loop.call_soon_threadsafe(queue.put_nowait, ev)

    # 步骤 3: 柯里化闭包工厂，给每个工具定制专属 on_update
    def make_on_update(call_id: str, tool_name: str, args: dict[str, Any]) -> Callable[[Any], None]:
        def on_update(partial: Any) -> None:
            safe_put_update(
                ToolExecutionUpdate(
                    tool_call_id=call_id,
                    tool_name=tool_name,
                    args=args,
                    partial_result=partial,
                )
            )
        return on_update

    calls_to_run = [
        (call.name, current_args, make_on_update(call.id, call.name, current_args), call.id)
        for _, call, current_args in prepared_calls
    ]

    # 步骤 4: 后台并发批处理任务（无论成败，finally 必送哨兵防死锁）
    async def _run_batch() -> list[ToolResult]:
        try:
            return await registry.execute_batch(calls_to_run, signal=signal)
        finally:
            if threading.get_ident() == loop_thread_id:
                queue.put_nowait(_SENTINEL)
            else:
                with contextlib.suppress(RuntimeError):
                    loop.call_soon_threadsafe(queue.put_nowait, _SENTINEL)

    runner = asyncio.create_task(_run_batch())
    try:
        # 前台消费循环：见事件就 yield，见哨兵就 break
        while True:
            item = await queue.get()
            if item is _SENTINEL:
                break
            if isinstance(item, Event):
                yield item
        batch_out = await runner
        for (idx, _, _), res in zip(prepared_calls, batch_out, strict=False):
            direct_results[idx] = res
    except Exception as exc:
        for idx, _, _ in prepared_calls:
            if idx not in direct_results:
                direct_results[idx] = ToolResult(ok=False, error=f"Tool execution failed: {exc}")
    finally:
        # 步骤 5: 外层中断时的防暴清场
        if not runner.done():
            runner.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await runner

这套架构将“不可流式的多任务并发”优雅转化为“可流式的单通道事件流”，用 call_soon_threadsafe 抹平了同步线程与异步主循环的鸿沟，用 _SENTINEL 彻底封死了死锁漏洞，达到了真正的工业级水准！

深度辨析：宏观串并行 vs 微观同异步（及异步队列的本质）

很多开发者在理解工具并发与流式回传时，容易把“宏观调度的串并行”和“底层函数的同异步”混淆。实际上，这是两个完全正交的独立维度：

1. 核心澄清：异步队列里流动的到底是什么？

请务必注意：
异步队列 queue 里面装的不是工具执行的最终返回值（ToolResult）！
最终结果是在所有工具都跑完后，由 batch_out = await runner 一次性拿到的。
队列里流动的，纯粹是工具在运行中途发射出来的“中间流式进度事件”（ToolExecutionUpdate）。

2. 如果工具是【并行】，底层具体怎么运转？

假设模型单轮发起了 3 个工具调用：tool_1（async def 协程）、tool_2（普通 def 计算）、tool_3（普通 def 命令），且均声明为 is_parallel_safe=True：

• 宏观调度：registry.execute_batch 使用 asyncio.gather 将它们同时打出去并发运行；
• 微观分流：tool_1 在主事件循环中跑协程；tool_2 和 tool_3 被 asyncio.to_thread 分配给系统线程池中的两个独立子线程并行跑；
• 进度汇入：
  • tool_1 在主线程调用 on_update $\to$ safe_put_update 识别为主线程 $\to$ queue.put_nowait 直接丢入；
  • tool_2 在子线程调用 on_update $\to$ safe_put_update 识别为工作线程 $\to$ 通过 loop.call_soon_threadsafe 安全跨线程预约投递；
• 前台感知：前台 while True: item = await queue.get() 无论谁的进度先到，就立刻把谁先 yield 出来给外部界面，呈现出多个任务交织滚动的极致流式体验。

3. 如果工具是【串行】，又是如何处理的？

假设模型发起了一个写操作 edit（is_parallel_safe=False）和一个读操作 read：

• 宏观调度：触发“一票否决”，registry.execute_batch 退化为顺序遍历：for tc in tool_calls: await self.execute_tool(tc, ...)；
• 微观分流与队列复用：
  • 先执行 edit：中途产生的进度事件依次塞入 queue $\to$ 前台立刻实时打印，直到 edit 彻底结束；
  • 接着执行 read：中途产生的进度事件塞入同一个 queue $\to$ 前台接着打印，直到 read 结束；
  • 整批串行任务结束：触发 finally 塞入 _SENTINEL 哨兵，前台收工退出；
  • 整套传送带与哨兵机制 100% 无缝复用，零多余逻辑。

实战指南：什么样的工具写成 async def？什么样的工具写成普通 def？

在 Agent 系统的工程实践中，判断工具应该写成异步（async def）还是同步（普通 def），核心依据只有一个：

这个工具在执行时，主要是在“等外部事件（网络/其他服务/子代理）”，还是在“让本地 CPU/操作系统猛跑”？

1. 必须 / 适合写成【异步工具】（async def）

核心特征：需要长时间等待外部世界响应，等待期间可以主动让出 CPU 控制权（await），让主线程去处理其他任务。

• 子代理委派（Subagent / Task）：例如 task(prompt="审查代码", agent="reviewer")。子 Agent 还要经历自己的推理与工具循环，耗时可达数秒甚至数分钟，必须通过 await subagent.run() 异步挂起。
• 网络检索与外部 API（Web Search / Crawl / MCP 协议）：底层依赖 httpx.AsyncClient 或 aiohttp，等待远程服务器网络握手与数据传输。
• 浏览器自动化（Playwright / Browser）：等待页面加载（domcontentloaded）、等待元素渲染、网络空闲等，天然全是非阻塞异步调用。
• 定时器与轮询工具：内部使用 await asyncio.sleep(delay) 优雅挂起，绝不卡死主线程。

2. 通常写成【同步工具】（普通 def）

核心特征：调用的几乎全是本地现成计算资源、操作系统 API 或传统三方阻塞库，代码自上而下顺次执行。

• 本地快速小文件读写与编辑（read / write / edit）：在现代 SSD 上读写几十 KB 文本只需 0.1~0.5 毫秒，直接使用 open() 或 pathlib.Path.read_text() 简单直观，无需额外创建异步事件调度开销。
• CPU 密集型计算与正则扫描（grep / calculator / ast-grep）：底层是纯 CPU 满负荷运算或 C/Rust 原生扩展，中途根本没有空闲等待时间，写成 async def 毫无意义（无处可 await）。
• 简单封装传统阻塞库的系统工具（bash）：许多开发者习惯直接调用 subprocess.run(cmd, shell=True)，这种阻塞操作适合作为普通 def。

3. 框架的“零心智负担”自适应无感桥接

为了让工具开发者彻底摆脱心智负担，my-pi-agent 在 Tool.execute 内部建立了自适应分流机制：

1
2
3
4
5
6

if self.is_async:
    # async def: 直接在主事件循环中高性能协程 await
    result = await async_func_call()
else:
    # 普通 def: 框架自动封装进 asyncio.to_thread，发配给后台线程池执行，绝对不阻塞主事件循环！
    result = await asyncio.to_thread(func_call)

工程收益：开发者想怎么写就怎么写。写同步无需担心卡死 Agent，写异步能够极致榨干事件循环吞吐量。

7. 深入底层：并发调度与同异步执行全景（宏观批处理 vs 单工具执行）

在阅读工具执行源码时，很多人容易把“宏观调度的串并行”与“底层函数的同异步”混为一谈，甚至产生误解：“是不是并行工具就是用子线程跑，串行工具就是直接跑？”

实际上，在 my-pi-agent 的工具流水线中，这是两个完全解耦的独立正交维度：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

┌─────────────────────────────────────────────────────────────────────────────┐
│                    两个完全正交的架构维度                                   │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  【维度 1：宏观批处理 (Macro Batch Scheduling)】                             │
│    • 关心的核心问题：这批工具是一起开工，还是一个接一个排队？               │
│    • 决定者与位置：ToolRegistry.execute_batch (基于 is_parallel_safe 属性)   │
│    • 实现手段：                                                             │
│      - 全并发：整批全为 is_parallel_safe=True ➔ asyncio.gather(...) 并发    │
│      - 保序串行：含任一写入工具 ➔ 悲观回退，for 循环依次 await               │
│                                                                             │
│  【维度 2：单工具内部执行 (Micro Single-Tool Execution)】                   │
│    • 关心的核心问题：这个工具自身的 Python 代码会卡死事件循环主线程吗？     │
│    • 决定者与位置：Tool.execute (基于 inspect.iscoroutinefunction 静态自省) │
│    • 实现手段：                                                             │
│      - 原生异步：async def ➔ 直接 await tool.func(...) (主线程协程调度)     │
│      - 同步阻塞：普通 def ➔ asyncio.to_thread(_sync_run) 扔进线程池        │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

1. 维度 1：宏观批处理调度（并发还是串行？）

由 ToolRegistry.execute_batch 全权掌舵。其采用“一票否决制”的悲观读写分流策略：

1. 全并发加速（asyncio.gather）：
   • 触发条件：大模型在一轮中呼叫的所有工具，其 tool.is_parallel_safe 均为 True（例如同时读取 3 个只读文件、查询 2 个外部只读 API）。
   • 调度方式：使用 asyncio.gather(*[self.execute_tool(...) for ...]) 一起并发调度，整体耗时从累加缩减为“取决于最慢的那一个”。
2. 保序串行（悲观回退）：
   • 触发条件：这一批调用中哪怕包含任一一个 is_parallel_safe=False 的工具（如写入文件 write、编辑文件 edit、运行终端命令 bash）。
   • 调度方式：整批工具立刻放弃并发，严格退化为按大模型在提示词里的原始声明顺序（Source Order），使用普通的 for 循环逐个串行执行。
   • 架构不变式：严防因果倒置（例如大模型本意是“先编辑代码，再执行测试”，如果盲目并发，可能导致测试在代码还没写完前就抢跑报错）。

2. 维度 2：单工具内部执行（异步协程还是线程池？）

无论宏观上是 asyncio.gather 并发还是 for 循环串行，每个具体工具在执行时，依然由其自身的实现形态决定在哪个线程运行：

1
2
3
4
5
6
7

# Tool.execute 的核心分流逻辑
if inspect.iscoroutinefunction(self.func):
    # 原生异步工具：零线程开销，直接在当前主事件循环上跑
    result = await self.func(...)
else:
    # 同步阻塞工具：坚决不能卡死主线程，封装进线程池独立子线程跑
    result = await asyncio.to_thread(_sync_run)

怎么判断工具该写成【异步工具（async def）】还是【同步工具（普通 def）】？

核心依据只有一个：这个工具在执行时，主要是在“等外部世界响应”，还是在“占用本地 CPU 或操作系统的同步进程”？

3. 两套机制交织碰撞：异步队列与跨线程安全

现在把两张拼图合在一起，整个底层流式通信的全景就清晰无比了：

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

┌─────────────────────────────────────────────────────────────────────────────┐
│ 宏观调度 (execute_batch): 决定是 asyncio.gather 还是串行 for 循环            │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │ 遍历调起单个工具 (execute_tool)
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 单工具执行 (Tool.execute):                                                   │
│   ├─ Case A: 若工具是 async def (如 task) ────────► 在主线程跑               │
│   │                                                 │                       │
│   │                                                 ▼ 调 on_update           │
│   │                                         直接 queue.put_nowait()          │
│   │                                                                         │
│   └─ Case B: 若工具是同步 def (如 bash)   ────────► 扔进工作子线程跑        │
│                                                     │                       │
│                                                     ▼ 调 on_update           │
│                                      跨线程必须用 call_soon_threadsafe 桥接!  │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │ 全部安全汇入
                                       ▼
                     ┌──────────────────────────────────┐
                     │ 主线程专属 asyncio.Queue (传送带) │
                     └─────────────────┬────────────────┘
                                       │ await queue.get()
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ 外部异步生成器 (_execute_tools_turn): 实时 yield ToolExecutionUpdate 给 UI  │
└─────────────────────────────────────────────────────────────────────────────┘

• 无论这批工具在宏观上是并发还是串行：后台都运行在独立的 runner = asyncio.create_task(...) 任务中，前台始终是 while True: item = await queue.get() 的流畅消费者；
• 无论具体工具是运行在主线程的协程、还是运行在子线程里的阻塞代码：safe_put_update 都能通过 threading.get_ident() == loop_thread_id 自动识别，丝滑抹平线程边界，保证事件安全、保序地呈现在用户眼前！

深度进阶：工具调用中断机制与 Steering 转向时序哲学

在掌握了七阶段执行流水线与转录本自愈后，我们必须面对真实交互环境中最考验架构内功的两大“深水区”场景：

1. 中途中断（Interrupt）：如果一个工具跑到一半被用户强行喊停（按 Esc），剩余没跑的工具怎么办？恢复对话时大模型如何自愈接续？
2. 即时转向（Steering）：如果用户在工具执行中途打字插话，插入时机到底应该在“单项工具结束后”还是“整轮 Turn 结束后”？

一、 工具中途中断全景流动与恢复处理（结合实战日志）

1. 中断信号的全景流动链路：从人类手指到操作系统进程树强杀

当你按下 Escape（或 Ctrl+C，或通过 API 调用 abort()）的瞬间，信号会在微秒级内沿着以下链路向下传递并瞬时斩断：

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

1. 人类操作: 终端按下 Escape
     │
     ▼
2. 前端 TUI (interactive-mode.ts):
   - 发起 JSON-RPC: {"method": "abort"}
   - 立即清理流式助手组件 (chatContainer.removeChild)
   - 撤回待发消息回输入框
     │
     ▼
3. 后端网关 (rpc_server.py) ➔ Agent 门面 (agent.py):
   - 设置 self._aborted = True
   - 触发取消令牌: self._current_signal.cancel()
     │
     ▼ 广播给所有注册了回调的子系统
4. 正在执行的工具进程 (bash.py / tools):
   - signal.add_callback 触发!
   - 1ms 内调用操作系统原生 taskkill /F /T 杀死完整子进程树!
     │
     ▼ 正在等待的 await proc.wait() 瞬间解冻并退出
5. 微内核工具车间 (_execute_tools_turn):
   - 捕获到 signal.is_cancelled()
   - 【核心操作】: 将当前中断的和剩下没跑的工具，全部填入 "Tool call interrupted by user"
     │
     ▼
6. 状态闭环:
   - 补齐发射所有配对的 Tool 消息并原子落盘至 .jsonl
   - 发射 AgentEnd(stop_reason="cancelled") 优雅停机

2. 核心解密：正在跑的工具 vs 还没跑的工具，分别如何处理？

假设大模型在单轮中并发发起了 3 个工具工单：

• call_01: read("file1.py")（耗时短，已完成）
• call_02: bash("npm test")（正在后台执行，耗时长）
• call_03: read("file2.py")（排在队尾，尚未开始执行）

此时用户按下 Escape。如果系统处理不当直接粗暴掐死，磁盘里就会留下残缺的 call_02 和 call_03，导致下一次发消息时大模型 API 报 400 死锁。

微内核（src/my_agent_core/loop.py）的处理策略如下：

关键原则：为什么未执行的 call_03 也必须写一条结果？
因为大模型的 API 规则是绝对铁律：只要 Assistant 的 tool_calls 里出现了 call_03，后面就必须有一条 role="tool" 带上匹配的 call_03。哪怕写的是“被用户中断”，也绝不能缺席！

3. 真实日志现场复盘（会话 01a0d14f）

在真实日志 01a0d14f 中，大模型在第 1 阶段第 3 轮同时发起了两个读文件工具：

• call_00: read(path="PROGRESS.md")
• call_01: read(path="src/my_agent_core/loop.py")

用户在此刻按下 Escape 中断。查看当时的事件流 01a0d14f.events.jsonl：

1
2
3
4
5
6
7
8
9
10
11
12
13

35: type=turn_start
36: type=message_start role=assistant
37: type=message_end   role=assistant (发起了 call_00 和 call_01)

═══════════════════════【中断收割与虚构补齐】═══════════════════════
38: type=message_start role=tool  (call_00)
39: type=message_end   role=tool  content="Tool call interrupted by user"
40: type=message_start role=tool  (call_01)
41: type=message_end   role=tool  content="Tool call interrupted by user"
══════════════════════════════════════════════════════════════════

42: type=turn_end
43: type=agent_end     stop_reason="cancelled"

再看磁盘持久化文件 01a0d14f.jsonl：

1
2
3

{"role": "assistant", "tool_calls": [{"id": "call_00", "name": "read"}, {"id": "call_01", "name": "read"}]}
{"role": "tool", "content": "Tool call interrupted by user", "metadata": {"tool_call_id": "call_00", "is_error": true}}
{"role": "tool", "content": "Tool call interrupted by user", "metadata": {"tool_call_id": "call_01", "is_error": true}}

4. 会话恢复时：剩下的工具是如何处理与智能接续的？

任务被打断后终端回到等待输入状态。过了几分钟，用户在终端输入新指令（例如说 "继续刚才的读取" 或单纯说 "你好"）。

系统在恢复时，经历了 “转录本自愈防护 ➔ 大模型心智理解 ➔ 智能接续” 三个阶段：

1. 第一道安全网：转录本自愈状态机 (tool_history.py)
   若发生电脑断电或进程强杀等极端意外，导致上面的补齐逻辑没跑完，磁盘里留下断头 ToolCall；下一次会话启动时，repair_tool_history 状态机会先在内存中自动补齐 Tool call interrupted by user 替身，确保发往大模型网关的请求绝不报 400！
2. 第二道自愈：大模型的心智认知
   在恢复后的新一轮中，大模型在上下文里看到了完整的因果链条：

   1 2 3 4 5 6

   [上一轮 Assistant]: 我决定调用 call_01 (读 file1), call_02 (跑测试), call_03 (读 file2) [上一轮 Tool 01]: file1 的代码内容 (成功) [上一轮 Tool 02]: Tool call interrupted by user (被用户中断) [上一轮 Tool 03]: Tool call interrupted by user (被用户中断) ───────────────────────────────────────────────────────────────── [用户最新输入]: "继续" (或者提了别的新问题)

   大模型具有上下文推理能力：
   • 如果用户说“继续”：模型看懂了“call_01 已经拿到数据，但 call_02 和 call_03 上次被人类喊停”，于是自动重新派发 call_02 和 call_03，不会傻傻去重跑已经成功的 call_01；
   • 如果用户问了新问题（例如“你好”）：模型理解“上一轮已被取消，现在开启新话题”，顺畅承接对话，绝不精神分裂。

二、 Steering（即时转向）的时序边界本质：为什么必须在整轮 Turn 结束后插入？

1. 致命场景：如果“某个单项工具一结束就插入”会发生什么？

现代大模型支持并发工具调用（Parallel Tool Calling）。假设模型单轮发起了 3 个工具：

• call_01: read("file1.py")（耗时 5ms）
• call_02: read("file2.py")（耗时 10ms）
• call_03: bash("npm test")（耗时 200ms）

如果按照直觉“某个工具一结束就插入用户转向消息”：

• call_01 跑完（5ms），系统发现用户打入了转向指令 "不要跑测试了，换个写法"，立刻插进对话历史；
• 此时 call_02 和 call_03 还在后头陆续完成返回。

发往大模型 API 的历史记录变成了这种畸形序列：

1
2
3
4
5

1. [Assistant]: 发起了 3 个工具调用 [call_01, call_02, call_03]
2. [Tool 消息]: call_01 的结果
3. [User 消息]: "不要跑测试了，换个写法"  <─── 💥 灾难发生了！插在了工具结果中间！
4. [Tool 消息]: call_02 的结果
5. [Tool 消息]: call_03 的结果

后果是毁灭性的：各大模型网关（OpenAI、Anthropic、DeepSeek）会直接抛出 HTTP 400 校验死锁：

❌ API 报错原文：
“Invalid message sequence: Messages with role ‘tool’ must be a response to a preceding message with ‘tool_calls’. Found ‘user’ message in between.”

会话当场报废。

2. 为什么必须在“整轮 Turn 结束（Turn Boundary）”插入？

微内核 src/my_agent_core/loop.py 严格遵循了 Turn 边界契约：

1. 模型推理：大模型吐出本轮要调用的所有 tool_calls；
2. 工具车间批处理：_execute_tools_turn 并发把这批工具全部执行完毕；
3. 配对落盘：把每一个工具结果按顺序封装为 role: "tool" 消息追加进上下文；
4. 发射 turn_end：宣告本轮所有工具事实已闭环形成稳态！

只有在此时，才是唯一的安全插入窗口（Safe Point）！

1
2
3
4
5
6
7
8
9
10
11

# ── 1. 执行整轮工具批处理（必须等它全部跑完）
async for ev in _execute_tools_turn(tool_calls=tool_calls, ...):
    yield ev

# ── 2. 本轮工具全部结束，发射 Turn 结束事件
yield TurnEnd()

# ── 3. 此时才安全检查 Steering 队列并提取！
if message_queue.has_steering():
    pending_messages = message_queue.get_steering_messages()
    # 带着这批转向指令，安全进入下一轮 Turn 的大模型推理！

此时大模型在下一轮看到的历史记录是完美合法的：

1
2
3
4
5
6

1. [Assistant]: 发起了 3 个工具调用 [call_01, call_02, call_03]
2. [Tool 消息]: call_01 的结果
3. [Tool 消息]: call_02 的结果
4. [Tool 消息]: call_03 的结果
────────────────────────────【一整轮 Turn 结束的合法边界】
5. [User 消息]: "不要跑测试了，换个写法"  <─── 稳稳当当插在最后！

3. 真实日志对比（会话 01a0d0f5）

在会话日志 01a0d0f5.events.jsonl 中，真实记录了【有 Steering】与【无 Steering】的事件差异：

• 【有 Steering 时】：

  1 2 3 4 5 6

  14: type=turn_end <--- 第 1 轮结束 15: type=message_start role=user content="暂停一下" <--- 紧随 turn_end 插入！ 16: type=message_end role=user content="暂停一下" 17: type=turn_start iteration=2 <--- 第 2 轮带转向启动 18: type=message_start role=assistant 19: type=message_end role=assistant content="好的，已暂停 ⏸ ..."

• 【无 Steering 时】：

  1 2 3

  36: type=turn_end <--- 第 1 轮结束 37: type=turn_start iteration=2 <--- 无 user 消息，直接开第 2 轮！ 38: type=message_start role=assistant (模型自主决定继续发起新的工具调用)

总结对比

pi架构详解

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

pi 启动
  │
  ├─► project_trust（仅用户/全局和 CLI 扩展生效，在加载项目资源之前触发）
  ├─► session_start { 原因: "startup"（启动） }
  └─► resources_discover { 原因: "startup"（启动） }
      │
      ▼
用户发送 prompt ───────────────────────────────────────────┐
  │                                                        │
  ├─►（优先检查扩展命令，若匹配则直接旁路/绕过）            │
  ├─► input（输入事件：可拦截、转换或直接处理）            │
  ├─►（若未被处理，进行 Skill/Template 技能与模板展开）    │
  ├─► before_agent_start（可注入消息、修改系统提示词）     │
  ├─► agent_start                                          │
  ├─► message_start / message_update / message_end         │
  │                                                        │
  │   ┌─── turn 轮次（在 LLM 调用工具期间重复执行） ───┐   │
  │   │                                            │       │
  │   ├─► turn_start                               │       │
  │   ├─► context（可修改消息列表/上下文）         │       │
  │   ├─► before_provider_headers（可修改请求头）  │       │
  │   ├─► before_provider_request（可检查或替换请求载荷 Payload）
  │   ├─► after_provider_response（响应状态码 + 响应头，在消费流之前）
  │   │                                            │       │
  │   │   LLM 给出响应，可能会调用工具:            │       │
  │   │     ├─► tool_execution_start               │       │
  │   │     ├─► tool_call（可进行阻断/拦截）       │       │
  │   │     ├─► tool_execution_update              │       │
  │   │     ├─► tool_result（可修改工具返回结果）  │       │
  │   │     └─► tool_execution_end                 │       │
  │   │                                            │       │
  │   └─► turn_end                                 │       │
  │                                                        │
  ├─► agent_end                                            │
  └─► agent_settled（无剩余重试/压缩/排队追加消息，进入沉淀等待状态）
                                                           │
用户发送下一条 prompt ◄────────────────────────────────────┘

/new（新建会话）或 /resume（切换会话）
  ├─► session_before_switch（可取消切换）
  ├─► session_shutdown（关闭旧会话）
  ├─► session_start { 原因: "new" | "resume", 上一次的会话文件? }
  └─► resources_discover { 原因: "startup" }

/fork 或 /clone（分叉或克隆会话）
  ├─► session_before_fork（可取消分叉）
  ├─► session_shutdown
  ├─► session_start { 原因: "fork", 上一次的会话文件 }
  └─► resources_discover { 原因: "startup" }

/name 或 pi.setSessionName()（会话重命名）
  └─► session_info_changed（会话信息变更）

/compact 或自动压缩（上下文压缩）
  ├─► session_before_compact（可取消或自定义压缩逻辑）
  └─► session_compact

/tree 历史树导航
  ├─► session_before_tree（可取消或自定义导航逻辑）
  └─► session_tree

/model 或 Ctrl+P（选择/循环切换模型）
  ├─► thinking_level_select（若切换模型导致思考深度发生变化或被截断）
  └─► model_select

思考深度变更（通过设置、快捷键绑定或 pi.setThinkingLevel() 触发）
  └─► thinking_level_select

退出（Ctrl+C、Ctrl+D、SIGHUP、SIGTERM）
  └─► session_shutdown（会话关闭清理）

pi的这个设计和langchain的Middleware十分类似

如果把这些控制逻辑全写在 Agent 的主循环里，代码会变得极其臃肿。因此，Pi 的 Extension 系统与 LangChain Agent Middleware 都选择将控制权（Control Layer）与执行层（Execution Layer）解耦，在 Agent 生命周期的关键切面上暴露钩子。