hermes agent的记忆系统

双层记忆架构全景

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

┌────────────────────────────────────────────────────────────────────────┐
│                          Hermes Agent 记忆体系                          │
├────────────────────────────────────────────────────────────────────────┤
│                                                                        │
│  【第一层：内置精选文件记忆 (Builtin Curated Memory)】                  │
│    • 定位：持久化、高信号、强控制、极低 Token 消耗                      │
│    • 载体：MEMORY.md (Agent 笔记) + USER.md (用户画像)                   │
│    • 机制：启动时冻结快照注入 System Prompt + 运行时受控 memory 工具维护 │
│  ────────────────────────────────────────────────────────────────────  │
│                                                                        │
│  【第二层：可插拔外部提供商 (Pluggable Memory Providers)】              │
│    • 定位：开放生态、大规模语义检索、跨设备云同步 (Honcho / Mem0 等)    │
│    • 机制：MemoryManager 单例控制 (一次只开一个)，提供 per-turn prefetch │
│                                                                        │
└────────────────────────────────────────────────────────────────────────┘

对于单机本地 Agent 而言，第一层（内置精选文件记忆） 是 Hermes 最核心的灵魂所在。

内置文件记忆的工程设计

1. 核心不变式：Frozen Snapshot（冻结快照）与 Prefix Cache 保护

大模型 API（如 Claude Prompt Caching、OpenAI/DeepSeek Prefix Caching）能够极大降低推理延迟与成本，但前提是 System
Prompt 和开头的历史消息必须严格保持不变（哈希一致）。

传统的做法是：Agent 一旦学到新知识，就立刻修改 System Prompt。这会导致整条上下文的前缀缓存全部失效，下一轮必须重新
Prefill。

Hermes 的解法：双状态并行（Parallel States）

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

                启动时: 读磁盘 ──► 捕获 Frozen Snapshot (冻结快照)
                                         │
                                         ▼ 注入 System Prompt (本会话全程静止!)
                                ┌─────────────────┐
                                │  System Prompt  │ (KV Cache 命中率 100%)
                                └─────────────────┘
                                         ▲
                                         │ 绝不修改快照!
LLM 调用 memory(action="add") ───────────┴──► 立即落盘 MEMORY.md (持久化)
                                                │
                                                ▼
                                     下次启动/新会话时才加载最新快照

• System Prompt 快照冻结：会话进行中，无论调用多少次 memory 工具写入，当前会话的 System Prompt 绝对静止，Prefix Cache
  绝不失效；

• 磁盘实时持久化：新写入的记忆立即原子落盘，下一次开启新会话时自动生效。

  2. 双 Store 与字符预算硬限制（Character Budget Limits）

  Hermes 将长期记忆严格拆分为两个 Markdown 文件，并且采用字符数（Characters）而非 Token 数作为预算硬约束（因为字符计算是
  模型无关且 100% 确定的）：

1
2
3
4
5
6
7

┌───────────┬────────────┬────────────────────────────────────────────────────────────┐
│ 记忆载体  │ 预算上限   │ 存储内容与语义                                             │
├───────────┼────────────┼────────────────────────────────────────────────────────────┤
│ MEMORY.md │ 2,200 字符 │ Agent 自己的笔记（环境事实、项目规范、工具踩坑、业务结论） │
├───────────┼────────────┼────────────────────────────────────────────────────────────┤
│ USER.md   │ 1,375 字符 │ 用户画像（偏好风格、语言习惯、角色设定、禁忌事项）         │
└───────────┴────────────┴────────────────────────────────────────────────────────────┘

为什么设置严格的字符上限？

防止长期使用后记忆无界膨胀冲垮大模型上下文。当总字符即将超过上限时，系统会拒绝追加写入，并把当前所有条目打印给模型，迫
使模型在当前轮次内调用 replace 合并精简旧记忆（Consolidation）或 remove 删除过时记忆。

3. 单工具多 Action (memory) 与唯原子串定位

Hermes 没有提供零散的 remember / recall / search_memory 工具，而是收敛为一个高度紧凑的单工具：

1
2
3
4
5
6
7
8
9
10

{
  "name": "memory",
  "parameters": {
    "target": "memory | user",
    "action": "add | replace | remove",
    "content": "新条目内容 (用于 add)",
    "old_text": "唯原子串定位 (用于 replace / remove)",
    "new_content": "替换后的内容 (用于 replace)"
  }
}

① 特殊条目分隔符：\n§\n

Hermes 选用 §（Section Sign）作为条目分隔符，条目内部允许包含多行 Markdown，避免普通换行造成切分歧义。

② 唯原子串定位（Unique Substring Match）

在 replace 和 remove 操作中，模型不需要记住数字 ID（如 id=12），也不需要复制整段冗长的原始文本，只需传入一段具有唯一性
的文本子串（old_text）：

• 歧义拦截（Ambiguity Guard）：如果 old_text 同时匹配了多条记录，系统会直接拦截并报错，把命中的多条记录全部列出，要求
  模型给出更精确的关键词，彻底杜绝误删误改；

• 精准替换：只命中唯一样本时才执行替换/删除并原子写盘。

  一、特殊条目分隔符：\n§\n

  1. 它要解决什么痛点？

  如果我们把长期记忆存成一个 Markdown 文本文件（MEMORY.md），程序该如何区分“条目 1”和“条目 2”？

• 如果按普通换行（\n）切分：
  如果某一条记忆是多行的（比如包含代码片段、列表或换行说明），程序就会错误地把它切成好几条碎片！

• 如果存成 JSON：
  人类打开 MEMORY.md 直接阅读或手工编辑时很不友好，而且大模型写 JSON 容易出现转义符号错误。

  2. Hermes 的解法：用 § 做切分线

  §（Section Sign，章节符）是一个在日常代码和自然语言中极少出现的特殊字符。Hermes 规定两条记忆之间必须用 \n§\n 隔开。

实际效果示例：

假设你的 MEMORY.md 里面存了 2 条记忆，文件内容长这样：

1
2
3
4
5
6
7

项目构建规范：
1. 必须使用 uv 管理依赖
2. 运行测试命令为 `uv run pytest`
§
数据库配置备忘：
- 本地端口为 5432
- 用户名 postgres，密码在 .env 中

程序怎么读写？

• 读取解析：entries = file_text.split(“\n§\n”) ➔ 稳稳当当地拿到 2 个完整的字符串列表，哪怕每个条目内部换了 10 行，也不
  会被切碎！

• 保存写盘：”\n§\n”.join(entries) ➔ 重新拼装成漂亮的 Markdown 文件。

  二、唯一子串定位（Unique Substring Match）

  │ （注：前面文档里的“唯原子串”是 唯一子串（Unique Substring） 的意思）

  1. 它要解决什么痛点？

  当大模型想要修改（replace）或删除（remove）某条旧记忆时，它该如何告诉程序“我要改哪一条”？

• 方案 A（按数字编号 ID=1, 2, 3 定位）：

  • ❌ 缺点：大模型特别容易记错 ID；而且一旦删除了第 1 条，后面的编号全变了，很容易误删其它记忆。

• 方案 B（必须 100% 完整复制原句）：

  • ❌ 缺点：极费 Token，而且大模型只要漏复制一个标点或空格，程序就报错“找不到该内容”。

  2. Hermes 的解法：只要提供一段“唯一的特征片段”

  大模型只需要传入旧记忆里独一无二的一小段关键词（old_text），程序在后台进行子串匹配：

  1 2	# 找到所有包含 old_text 的记忆条目 matches = [i for i, entry in enumerate(entries) if old_text in entry]

  3. 运行时的三种情况：

✅ 情况 1：精准唯一命中（执行修改）

• 现有记忆条目：
  • 条目 A：”用户偏好使用 Python 3.11 编写代码”
  • 条目 B：”项目部署在阿里云 K8s 集群”
• 模型调用工具：
  • memory(action=”replace”, old_text=”Python 3.11”, new_content=”用户偏好使用 Python 3.12”)
• 系统判定：只有条目 A 包含 “Python 3.11”（命中数量 = 1）➔ 精准定位，替换成功！

⚠️ 情况 2：歧义多处命中（主动拦截报警，防止误删！）

• 现有记忆条目：
  • 条目 A：”用户在前端偏好使用 Vue3”
  • 条目 B：”用户在后端偏好使用 FastAPI”
• 模型调用工具：
  • memory(action=”remove”, old_text=”偏好使用”)
• 系统判定：old_text=”偏好使用” 同时出现在了条目 A 和条目 B 里（命中数量 = 2）！
• 系统的防误删保护触发：
  系统立刻拒绝删除，并给模型报错返回：
  │ “Ambiguous match: found 2 entries matching ‘偏好使用’. Please provide a more specific old_text.”
  │ （匹配到多条记录，存在歧义！请提供更精准的特征词，比如 ‘前端偏好’ 或 ‘后端偏好’）

❌ 情况 3：未命中

• 模型传入了一个不存在的词 ➔ 系统返回 Text ‘xxx’ not found in memory，提示模型检查现有清单。

什么该保存，什么该跳过

主动保存这些内容

Agent 会自动保存——无需你主动要求。当它学到以下内容时会保存：

• 用户偏好： “我更喜欢 TypeScript 而非 JavaScript” → 保存到 user
• 环境事实： “此服务器运行 Debian 12，安装了 PostgreSQL 16” → 保存到 memory
• 纠正信息： “Docker 命令不要用 sudo，用户已在 docker 组中” → 保存到 memory
• 约定： “项目使用 tab 缩进、120 字符行宽、Google 风格 docstring” → 保存到 memory
• 已完成的工作： “2026-01-15 将数据库从 MySQL 迁移到 PostgreSQL” → 保存到 memory
• 明确请求： “记住我的 API 密钥每月轮换一次” → 保存到 memory

跳过这些内容

• 琐碎/显而易见的信息： “用户询问了 Python”——太模糊，没有实用价值
• 容易重新发现的事实： “Python 3.12 支持 f-string 嵌套”——可以网络搜索
• 原始数据转储： 大型代码块、日志文件、数据表——对记忆来说太大
• 会话特定的临时内容： 临时文件路径、一次性调试上下文
• 已在上下文文件中的信息： SOUL.md 和 AGENTS.md 的内容

提示词与 Schema 中是如何定义的？

在 Hermes 和 my-agent-core 中，大模型在开始思考前，能同时看到两处关键指引：

1. 工具描述里的执行契约（memory Tool Schema）

在每个工具注册到大模型时，memory 工具的 description 字段里写满了具体的触发条件：

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

【WHEN 触发时机】
- 当用户表达偏好、纠正你的错误或提供个人信息时，主动保存；
- 当你了解到关于用户开发环境、项目规范或工作流的稳定事实时，主动保存；
- 优先级排序：用户偏好与纠错 > 环境与项目事实 > 流程经验。
- 黄金准则：最优秀的记忆，是让用户永远不需要重复说第二遍。

【TARGETS 存储目标】
- 'user'：用户画像（姓名、角色、代码偏好、沟通风格、忌讳事项）；
- 'memory'：Agent 自己的笔记（环境事实、项目规范、依赖怪癖、踩坑教训）。

【SKIP 明确禁止记录的内容】
- 严禁记录临时任务进度、已完成工作日志、临时 TODO 待办（用 session 或 todo 管理）；
- 严禁记录容易随时重新发现的事实或大段原始数据；
- 严禁记录多步骤可复用的操作流程（操作流程应写进 Skill，而非 Memory）。

2. System Prompt 里的持久化上下文块（）

在 System Prompt 首条消息中，会动态注入当前已有的记忆，并给模型明确的角色定位：

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

<MEMORY_CONTEXT>
以下是你跨会话保留的长期记忆。
当你从对话中学习到新的事实、用户偏好或被用户纠错时，请主动调用 `memory` 工具进行更新维护。

## MEMORY.md (Agent 笔记)
- 项目使用 uv 管理依赖，测试命令为 uv run pytest
- Python 版本为 3.11+，禁止使用 3.10 以下废弃语法

## USER.md (用户画像)
- 用户偏好简洁、强类型的代码风格
- 拒绝过度设计，单函数尽量保持在 50 行以内
</MEMORY_CONTEXT>

如何发现并触发“更新记忆”

通道一：大模型主动识别与实时更新（主通道）

在正常对话交互中，大模型是记忆更新的第一责任人。Hermes 通过在 工具 Schema Description 与 System Prompt 中注入极度明确
的 “触发信号指引（WHEN / PRIORITY / SKIP）”，让大模型在 ReAct 循环中主动调用 memory 工具。

1. 触发场景与优先级规则（WHEN & PRIORITY）

Hermes 在 MEMORY_SCHEMA 描述中规定了清晰的记忆更新等级：

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

┌──────────────────────────────────────────────────────────────┐
│  最高优先级：用户纠错与偏好声明 (User Preferences & Corrections) │
│  - 用户纠正你："记住，以后不要用 xxx" / "改成用 pnpm"           │
│  - 用户分享个人画像：技术栈、时区、命名习惯、禁忌事项           │
│  => 立即调用: memory(target="user", action="add/replace")     │
├──────────────────────────────────────────────────────────────┤
│  次高优先级：环境事实与工具踩坑 (Environment Facts & Quirks)     │
│  - 发现项目特性："这个库在 Python 3.12 下有兼容性 bug"          │
│  - 部署/构建约定："发布前必须先编译静态资源"                     │
│  => 立即调用: memory(target="memory", action="add/replace")   │
├──────────────────────────────────────────────────────────────┤
│  过滤垃圾信息 (SKIP - 绝不记录)                               │
│  - 任务进度、已完成工作日志、临时 TODO (交给 session / task)   │
│  - 容易随时重新获取的事实、大段原始数据 dump                   │
│  - 可复用的多步骤操作流程 (应写入 Skill，而非 Memory)           │
└──────────────────────────────────────────────────────────────┘

2. 真实工作流示例

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

用户: "我本地没有装 Docker，所有的测试都用本地 sqlite 跑，不要去启动容器。"
  │
  ▼ LLM 思考 (ReAct 内部)
  │ [检测到高优先级用户环境与偏好声明]
  │ "这属于长期有效事实，未来所有会话都适用，必须立即持久化。"
  ▼
发起工具调用:
  memory(target="user", action="add", content="本地无 Docker，测试均使用本地 sqlite 运行")
  │
  ▼ 工具执行成功落盘
  │
继续回答用户:
  "已记住您的环境配置，后续测试将直接使用本地 sqlite 运行..."

────────────────────────────────────────────────────────────────────────────────

通道二：生命周期钩子提取与兜底（Hermes 外部理论参考与对比）

在调研开源项目 Hermes Agent 时，Hermes 设计了生命周期的后台被动兜底机制（在会话快接近上限即将压缩前，调用外部模型自动扫描历史提取事实）：

my-pi-agent 架构取舍注记：在我们的框架设计中，刻意没有引入被动的后台提取通道，而是纯粹采用通道一（大模型主动调用受控工具）。
原因：被动后台提取会在运行时不可控地修改记忆文件，破坏我们严格守护的 Frozen Snapshot（冻结快照） 机制，进而导致大模型的 Prompt Prefix Cache 频繁失效。主动维护不仅保持了前缀缓存 100% 稳定，而且将记忆的选择权完全交给了当前上下文的大模型。
当对话轮次过多触发 Context 压缩裁切时，如果不提前提取，早期的重要对话细节就会被压缩丢失。Hermes 会在裁切前触发
on_pre_compress，自动提炼有价值的偏好并持久化到文件；

1. on_session_end(messages)（会话结束提炼）：
   在会话正常退出或保存时，对整场会话进行复盘总结。

模块整体布局

memory.py 主要由两个实体组成：

1. MemoryStore：底层的条目化存储仓储，管理 Markdown 文件持久化、字符预算与 Frozen Snapshot（冻结快照）；

2. make_memory_tool：上层的工具工厂函数，将 MemoryStore 包装为符合 OpenAI Function Calling 标准的 memory 工具（供大模
   型调用）。

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

   ┌────────────────────────────────────────────────────────────────────────┐ │ memory.py │ ├────────────────────────────────────────────────────────────────────────┤ │ │ │ 【核心常量与契约】 │ │ ENTRY_DELIMITER = "\n§\n" │ │ MEMORY_CHAR_LIMIT = 2200 (Agent 笔记上限) │ │ USER_CHAR_LIMIT = 1375 (用户画像上限) │ │ │ │ 【底层仓储：MemoryStore】 │ │ - load_from_disk(): 读 MEMORY.md/USER.md + 保序去重 + 捕获冻结快照 │ │ - _atomic_save(): mkstemp + fsync + os.replace 原子落盘 │ │ - add() / replace() / remove(): 增删改 + 唯一子串匹配 + 预算超限拦截 │ │ - format_all_for_system_prompt(): 渲染 <MEMORY_CONTEXT> 提示词块 │ │ │ │ 【工具工厂：make_memory_tool(store)】 │ │ - @tool 包装为标准 memory 工具 (target, action, content, ...) │ │ - 内部错误捕获与 Never-Throw 保证 (转换为 ToolResult 引导 LLM 自愈) │ │ │ └────────────────────────────────────────────────────────────────────────┘

MemoryStore

内部核心状态

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

class MemoryStore:
    def __init__(
        self,
        mem_dir: Path | str,
        memory_char_limit: int = MEMORY_CHAR_LIMIT,  # 默认 2200 字符
        user_char_limit: int = USER_CHAR_LIMIT,      # 默认 1375 字符
    ) -> None:
        self.mem_dir = Path(mem_dir)
        self.limits: dict[str, int] = {
            "memory": memory_char_limit,
            "user": user_char_limit,
        }
        self.files: dict[str, str] = {
            "memory": "MEMORY.md",
            "user": "USER.md",
        }
        # ① Live 实时状态：工具增删改时立即变化
        self._entries: dict[str, list[str]] = {"memory": [], "user": []}
        # ② 冻结快照：启动或 reset 时捕获一次，本会话内绝对静止
        self._snapshot: dict[str, str] = {"memory": "", "user": ""}

核心亮点：双状态并行（Parallel States）

MemoryStore 内部最精妙的设计是维护了 _entries 和 _snapshot 两个互不干扰的字典：

• _entries（实时运行态）：保存当前内存中最新的记忆条目列表。当模型调用 add/replace/remove 时，它会实时增删；
• _snapshot（系统提示词冻结态）：在启动时把文件内容拼成纯文本固化下来。当前会话无论怎么写，_snapshot 永远不变！这就是
  为什么大模型的 System Prompt 不会频繁变动，Prefix Cache（前缀缓存）命中率能达到 100% 的根本原因。

核心方法

1
2
3
4
5
6
7
8
9

                                   MemoryStore 核心方法全景
                                             │
           ┌─────────────────────────────────┼─────────────────────────────────┐
           ▼                                 ▼                                 ▼
   【启动与重置阶段】                【运行修改阶段 (写落盘)】        【提示词组装阶段 (读快照)】
    load_from_disk()             - add(target, content)  -format_for_system_prompt()
(读盘 -> 保序去重 -> 捕获快照)  - replace(target, old, new)    -                                                                                  format_all_for_system_prompt()
                               - remove(target, old)       (生成<MEMORY_CONTEXT> XML 块)
                                  - _atomic_save(target)

方法调用时序全景图

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

══════════════════════════════════════════════════════════════════════════════════════
阶段 1：Agent 实例化冷启动（应用启动阶段）
══════════════════════════════════════════════════════════════════════════════════════

 用户代码: agent = Agent(..., memory_dir="./memory")
   │
   ├─► 【memory_dir 三态语义】:
   │     • None: 自动探测 <cwd>/.my-pi-agent/memory，存在才开启
   │     • False: 显式禁用 (子代理必须强制设为 False，实现全局记忆绝对隔离)
   │     • str | Path: 显式指定存储目录并装配
   │
   ▼
   ├─► ① 构造 MemoryStore(mem_dir)
   │
   ├─► ② 调用: store.load_from_disk()
   │     └─ 读磁盘 MEMORY.md / USER.md ➔ 去重 ➔ 捕获 _snapshot 冻结快照
   │
   ├─► ③ 调用: store.format_all_for_system_prompt()
   │     └─ 提取快照，拼装为 <MEMORY_CONTEXT>... 注入首条 System Message
   │
   └─► ④ 调用: make_memory_tool(store) ➔ 注册 memory 工具到 Agent 工具表


══════════════════════════════════════════════════════════════════════════════════════
阶段 2：会话运行中（Agent.run 内联 ReAct 循环）
══════════════════════════════════════════════════════════════════════════════════════

 用户: "记住，这个项目必须用 pytest 跑测试"
   │
   ▼ LLM 思考后发起工具调用: memory(target="memory", action="add", content="...")
   │
   ├─► ⑤ 调用: store.add("memory", "项目使用 pytest 跑测试")
   │     ├─ 检查是否超过 2200 字符限制
   │     ├─ 触发私有: store._atomic_save("memory") ➔ 写入磁盘
   │     └─ 【注意】绝不修改阶段 1 的 _snapshot（System Prompt 保持静止，保缓存）
   │
 (若用户说: "测试框架改成 vitest 吧")
   │
   ▼ LLM 发起工具调用: memory(target="memory", action="replace", old_text="pytest", ...)
   │
   ├─► ⑥ 调用: store.replace("memory", old_text="pytest", new_content="...")
   │     ├─ 定位含 "pytest" 的唯一条目（多处命中则报错防误改）
   │     └─ 触发私有: store._atomic_save("memory") ➔ 更新磁盘
   │
 (若用户说: "把刚才关于测试的备注删掉")
   │
   ▼ LLM 发起工具调用: memory(target="memory", action="remove", old_text="vitest")
   │
   └─► ⑦ 调用: store.remove("memory", old_text="vitest")
         └─ 触发私有: store._atomic_save("memory") ➔ 磁盘删除


══════════════════════════════════════════════════════════════════════════════════════
阶段 3：跨会话召回（第二天开启全新的 Session B）
══════════════════════════════════════════════════════════════════════════════════════

 用户代码: new_agent = Agent(..., memory_dir="./memory")  # 新实例
   │
   ├─► 重新走阶段 1 的 ① ② ③ 步 (load_from_disk 读到昨天落盘的新内容)
   │
   ▼ 用户提问: "这个项目用什么跑测试？"
   │ (此时 System Prompt 的 <MEMORY_CONTEXT> 里已经有昨天的记忆了)
   │
   └─► ⑧ LLM 0 工具调用，直接回答: "根据记忆，该项目使用 vitest 运行测试。"


══════════════════════════════════════════════════════════════════════════════════════
阶段 4：会话重置（调用 agent.reset() 时）
══════════════════════════════════════════════════════════════════════════════════════

 用户代码: agent.reset()
   │
   ▼
   ├─► ⑨ 重新调用: store.load_from_disk() ➔ 从磁盘重新载入最新记忆，刷新快照
   └─► ⑩ 重新调用: store.format_all_for_system_prompt() ➔ 重新拼装 System Prompt

add 方法：追加新记忆

1

def add(self, target: str, content: str) -> str

1. 核心作用

当 Agent 发现新的长期有效事实、用户习惯或纠错信息时，将这条新记忆追加到对应的记忆库（MEMORY.md 或 USER.md）末尾并原子
写盘。

2. 内部执行步骤与安全防线

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

传入: target ("memory"|"user"), content ("内容...")
  │
  ├─► [校验 1：目标与非空检查] ── target 必须合法，content.strip() 不能为空
  │
  ├─► [校验 2：精确重复拦截] ── 检查 content 是否已经在已有条目中
  │     └─ 若已存在: 提示 "Entry already exists"，不重复写盘，省 I/O
  │
  ├─► [校验 3：预算硬上限检查 (Budget Check)]
  │     └─ 试算: (现有全部内容 + 新条目 + 分隔符) 的总字符数
  │     └─ 若 > 上限: 【超限拦截】拒绝写入！
  │           - 打印错误信息
  │           - 把库里当前所有条目列出来打印给模型
  │           - 引导模型先调用 replace/remove 精简或删除旧记忆 (Consolidate)
  │
  └─► [执行：写入与落盘]
        - 追加到 live 条目列表: self._entries[target].append(content)
        - 触发原子写盘: self._atomic_save(target)
        - 返回成功信息: "Added to memory (120/2200 chars used)."

replace 方法：修改 / 替换旧记忆

1

def replace(self, target: str, old_text: str, new_content: str) -> str

1. 核心作用

当旧记忆发生变更（如用户换了技术栈、配置更新）、或者需要将多条冗长的旧记忆合并精简为一句时，通过关键词定位旧条目并进行
内容更新。

2. 内部执行步骤与安全防线

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

传入: target, old_text ("旧条目特征词"), new_content ("新内容...")
  │
  ├─► [校验 1：参数非空检查] ── old_text 和 new_content 均不能为空
  │
  ├─► [步骤 2：唯一子串定位 (Unique Substring Match)]
  │     └─ 扫描库中所有包含 old_text 的条目:
  │        matches = [i for i, entry in enumerate(entries) if old_text in entry]
  │
  ├─► [分支判断 A：命中 0 条]
  │     └─ 报错返回: "Text 'xxx' not found"，提醒模型检查关键词
  │
  ├─► [分支判断 B：命中 2 条及以上 (歧义命中)]
  │     └─ 🚨【歧义拦截安全网触发！】
  │     └─ 拒绝执行修改！
  │     └─ 把命中的多条不同记录全部列出返回给模型
  │     └─ 强制要求模型提供更具区分度的特征词（彻底防止误改相邻记忆）
  │
  ├─► [分支判断 C：精准命中唯一样本]
  │     ├─ 试算替换后的总字符数是否超过预算限制
  │     ├─ 更新 live 条目列表: self._entries[target][idx] = new_content
  │     └─ 触发原子写盘: self._atomic_save(target)
  │     └─ 返回成功信息: "Replaced in memory (150/2200 chars used)."

remove 方法：删除过时记忆

1

def remove(self, target: str, old_text: str) -> str

1. 核心作用

当某些记忆已不再适用、或者库满了需要清理低价值的过期记录时，通过关键词定位并彻底从磁盘中抹除该条目。

2. 内部执行步骤与安全防线

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

传入: target, old_text ("要删除的特征词")
  │
  ├─► [校验 1：参数非空检查] ── old_text 不能为空
  │
  ├─► [步骤 2：唯一子串定位]
  │     └─ matches = [i for i, entry in enumerate(entries) if old_text in entry]
  │
  ├─► [分支 A：未找到] ── 返回 "Text 'xxx' not found"
  │
  ├─► [分支 B：歧义多处命中]
  │     └─ 🚨【防误删保护触发！】
  │     └─ 拒绝删除！打印所有冲突条目，要求模型提供更精确的子串
  │
  └─► [分支 C：精准命中 1 条]
        - 从列表中剔除该索引条目
        - 触发原子写盘: self._atomic_save(target)
        - 返回成功信息: "Removed from memory (90/2200 chars used)."

make_memory_tool工具

为什么它是一个工厂函数 make_memory_tool(store)？

你可能会问：“为什么不直接写一个普通的 @tool def memory(…)，而要写一个 make_memory_tool(store) 工厂函数？”

这体现了专业框架的 依赖注入（Dependency Injection） 与 闭包数据隔离（Closure Isolation） 思想：

1
2
3
4
5
6
7
8
9
10

     Agent A 实例                       Agent B 实例
          │                                  │
          ▼                                  ▼
MemoryStore(dir="./mem_a")         MemoryStore(dir="./mem_b")
          │                                  │
          ▼                                  ▼
make_memory_tool(store_a)          make_memory_tool(store_b)
          │                                  │
          ▼                                  ▼
 Tool 实例 A (绑定 store_a)         Tool 实例 B (绑定 store_b)

1. 多实例隔离：如果一个进程里同时跑了两个不同的 Agent（比如 Agent A 操作目录 A，Agent B 操作目录 B），工厂函数通过
   Python 闭包，让每个 memory 工具实例精准绑定自己专属的 MemoryStore，互不干扰；
2. 生命周期绑定：Agent 在初始化时，只用调一次 make_memory_tool(self.memory_store)，就能把工具挂载进注册表。

Action 分发与人性化参数容错

大模型在生成 JSON 参数时，偶尔会有小偏差（例如把 replace 的新内容放进 content 而不是 new_content）。make_memory_tool
在分发时做了精细的校验与容错：

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

if action == "add":
    # 1. add 必须有 content
    if not content:
        raise ValueError("`content` is required when action is 'add'.")
    return store.add(target, content)

elif action == "replace":
    # 2. replace 必须有定位关键词 old_text
    if not old_text:
        raise ValueError("`old_text` is required when action is 'replace'.")

    # 【人性化容错】：模型可能传了 new_content，也可能把新内容写在 content 里
    effective_new = new_content or content
    if not effective_new:
        raise ValueError("`new_content` is required when action is 'replace'.")
    return store.replace(target, old_text, effective_new)

elif action == "remove":
    # 3. remove 必须有定位关键词 old_text
    if not old_text:
        raise ValueError("`old_text` is required when action is 'remove'.")
    return store.remove(target, old_text)

else:
    raise ValueError(f"Unknown action '{action}'. Must be 'add', 'replace', or 'remove'.")

Never-Throw 架构保证（永不崩溃与自愈）

如果模型调用时漏传了字段（例如发起了 action=”replace” 却没传 old_text），代码里抛出了 raise ValueError(…)，这会导致
Python 进程报错退出吗？

绝对不会！

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

LLM 发起错误调用: memory(action="replace") ──(漏传 old_text)──┐
                                                               │
                                                               ▼
                make_memory_tool 内部抛出: ValueError("`old_text` is required...")
                                                               │
                                                               ▼
           ┌────────────────────────────────────────────────────────┐
           │  底层的 Tool.execute() 拦截层 (Never-Throw Guard)       │
           │  捕获 ValueError 异常，包装为标准 ToolResult 结构:      │
           │  ToolResult(ok=False, error="`old_text` is required...")│
           └───────────────────────────┬────────────────────────────┘
                                       │
                                       ▼
      把错误字符串塞进 Tool 消息写回对话历史: role="tool", content="Error: `old_text` is required..."
                                       │
                                       ▼
                 下一轮 LLM 看到错误提示，立即自我修正并重新调用！

这就是我们框架的核心设计原则：工具内部抛出的所有业务校验异常，都会被底层的 Tool.execute() 拦截转化为给大模型的结构化反
馈，Agent 永远不会因为参数错误而崩溃，而是引导大模型自我纠错。

pi 的 extension 机制

extension 就是一个 TypeScript 模块，默认导出一个工厂函数，拿到一个 ExtensionAPI 对象，往里面注册东西。

定义在 core/extensions/types.ts:1193 的 ExtensionAPI，分六大类：

机制原理

pi 的扩展系统采用了高度解耦的 “静态注册面（ExtensionAPI） + 动态运行上下文（ExtensionContext）” 设计：

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

               ┌────────────────────────────────────────────────────────┐
               │              Extension Factory Entry                   │
               │        export default function (pi: ExtensionAPI)      │
               └───────────────────────────┬────────────────────────────┘
                                           │
             ┌─────────────────────────────┴─────────────────────────────┐
             ▼ (启动装配阶段)                                              ▼ (运行时触发阶段)
┌────────────────────────────────────────┐                  ┌────────────────────────────────────────┐
│        ExtensionAPI (pi 对象)          │                  │       ExtensionContext (ctx 对象)      │
│  【静态注册面】                         │                  │  【动态运行上下文】                     │
│  - pi.registerTool(...)                │                  │  - ctx.ui (弹窗/选择/通知/底部状态栏)  │
│  - pi.on(event, handler)               │ ──触发事件注入──► │  - ctx.sessionManager (会话树只读访问) │
│  - pi.registerCommand(...)             │                  │  - ctx.modelRegistry (模型与认证)      │
│  - pi.registerProvider(...)            │                  │  - ctx.signal (中止信号 AbortSignal)   │
│  - pi.sendMessage() / sendUserMessage()│                  │  - ctx.cwd / ctx.mode / ctx.hasUI      │
└────────────────────────────────────────┘                  └────────────────────────────────────────┘

1. ExtensionAPI（pi 实体）：
   • 传给扩展入口函数的参数；
   • 负责声明“这个扩展有什么能力”（注册了哪些工具、订阅了哪些事件、扩展了哪些斜杠命令）。
2. ExtensionContext（ctx 实体）：
   • 当事件触发、命令执行或工具被调用时，作为运行时参数动态传入 Handler；
   • 负责提供“当前环境的即时上下文与交互能力”（当前会话树、TUI 交互接口、Abort 取消信号等）。

生命周期拦截

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

用户输入 (User Prompt)
  │
  ├─► [1. input 事件] ──────────► 扩展可直接 handled (不调LLM) 或 transform (改写Prompt)
  ├─► [2. before_agent_start] ──► 扩展可动态注入上下文消息、改写系统提示词 SystemPrompt
  │
  ▼
进入 Agent ReAct 循环
  │
  ├──► [3. context 事件] ───────► 发送给模型前，扩展可非破坏性过滤/压缩历史 messages
  ├──► [4. before_provider_request] ──► 拦截并修改发往 OpenAI/Anthropic 的原始 HTTP 请求 Payload
  │
  │    LLM 返回 tool_call:
  │    ├──► [5. tool_call 事件] ──► 扩展可原地修改 args 参数，或返回 { block: true } 拦截！
  │    ├──► (执行工具真实逻辑)
  │    └──► [6. tool_result 事件] ─► 扩展可后置篡改返回给模型的 content / details
  │
  └──► [7. turn_end / agent_end]

• input：截获用户原始输入。返回 { action: "handled" } 可直接由扩展自行处理而不触发 LLM；返回 { action: "transform", text: "..." } 可对 Prompt 进行预处理改写。
• before_agent_start：在 Agent 循环启动前触发，支持动态追加上下文消息，或修改本轮调用的 systemPrompt。
• context：在每次调用 LLM 前触发，提供对发往模型的 messages 进行过滤与裁剪（非破坏性视图）。
• tool_call（核心拦截点）：在工具实际执行前触发。支持原地修改 event.input（参数修补），或返回 { block: true, reason: "..." } 阻断高危命令执行。
• tool_result：在工具执行后触发，支持链式修改返回给模型的 content 或供前端消费的 details。

自定义工具体系：pi.registerTool

扩展是如何把一个普通函数变成大模型能调用的工具的？

1. 自动契约合成：
   @api.tool 内部复用框架底层的 Pydantic 动态建模，自动从 Python 函数签名、类型标注和 docstring 提取生成标准的 OpenAI
   / Anthropic Function Calling Schema。
2. 后加载覆盖机制（Overriding Power）：
   在 Agent.init 的装配时序中：
   注册内置工具与用户工具 → 加载 Extension 扩展工具
   因为扩展是在最后阶段被调用的，所以如果扩展中注册了一个同名工具（如 read、bash），它会静默覆盖掉默认的内置实现。
   • 架构价值：开发者无需修改核心代码，就能通过扩展将原生的本地文件读取工具替换为“带权限审计的只读沙箱工具”。

本地命令调度与反射路由（Bypass-LLM Dispatching）

为了向用户提供 0 Token 消耗的本地交互通道，ExtensionAPI 提供了命令路由系统：

1
2
3
4
5
6
7
8
9
10
11

用户输入: "/echo hello"
  │
  ▼ CLI 前置拦截 (不调大模型)
ExtensionManager.handle_command("echo", "hello")
  │
  ├─ 1. 查表获取 handler
  ├─ 2. 反射探测形参: inspect.signature(handler).parameters
  │     ├─ 若 len == 0: 执行 handler()
  │     └─ 若 len > 0: 执行 handler("hello")
  ▼
直接向控制台返回结果 (0 Token 消耗，不污染会话历史)

通过 Python 的 inspect 反射能力，实现了对无参命令（如 /stats、/compact）和带参命令（如 /echo foo）的参数自动适配。

案例

在实际开发中，一个扩展往往会同时组合使用这三项能力，形成功能闭环：

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

# .agents/extensions/todo_extension.py
"""一个完整的待办事项（Todo）扩展：
- 注册 todo_add 工具供大模型调用
- 注册 /todos 命令供用户在终端随时查看
- 监听 AgentEnd 事件在任务结束时统计待办
"""
from my_agent_core.events import AgentEnd

# 模块级共享状态
TODOS: list[str] = []

def extension(api):
   # 1. 【工具注册】：让大模型可以在思考中记录待办
   @api.tool(description="添加一条待办事项")
   def todo_add(item: str) -> str:
       TODOS.append(item)
       return f"已成功添加待办: {item}"

   # 2. 【命令调度】：用户在终端输入 /todos 时直接打印，不花大模型 Token
   @api.command("todos", description="查看所有待办列表")
   def cmd_todos():
       if not TODOS:
           return "当前没有待办事项。"
       return "\n".join(f"{i+1}. {t}" for i, t in enumerate(TODOS))

   # 3. 【事件订阅】：每次对话轮次彻底结束时，若有待办则在终端输出提示
   @api.on(AgentEnd)
   def notify_on_end(event: AgentEnd, api):
       if TODOS:
           print(f"\n[Todo 扩展提示] 当前还有 {len(TODOS)} 项待办未完成，输入 /todos 查看。")

异步消息导流与插队机制（Message Steering）

当 Agent 正在忙碌地运行一个多轮 ReAct 循环时，扩展如何向模型传达新指令？

ExtensionAPI 定义了两种核心的异步分发模式：

1
2
3
4
5

1. 实时插队纠偏 (deliverAs: "steer"):
   LLM 生成 -> 执行工具 1 -> [扩展插队注入 user 消息] -> 下一轮 LLM 生成 (立即生效纠偏!)

2. 静默排队跟进 (deliverAs: "followUp"):
   LLM 生成 -> 执行工具 1 -> 执行工具 2 -> 任务彻底完成 -> [触发下一轮新任务]

• steer（方向盘）：当前轮次的工具一跑完，下一次调模型前强行把消息插入队列，实现对模型的“实时急刹车与路线矫正”；
• followUp（待办队列）：等整个 Agent 任务全部跑完空闲后，才开启下一轮新对话。

如何利用 extension 实现subagent

subagent 实现 = 注册一个名叫 subagent 的”执行器/转接器”工具，具体 agent 是它的运行时配置数据，不是各自的工具。

工具表里永远只有一个 subagent，它的参数是 { agent: "某个名字", task: "..." }（index.ts:448 的 schema）。模型填的是字符串，然后 execute 去 agents 目录里查对应的 md 文件。这也解释了为什么要 --append-system-prompt 写临时文件——agent 的 system prompt 是运行时读到的字符串，不是编译进工具的代码。

pi 的 subagent 不是框架里的一等公民，而是用一个通用扩展点 pi.registerTool 实现出来的。 全部代码就一个文件 index.ts，主 agent 的模型把”委派任务”当成一次普通工具调用来完成。

入口：只注册了一个工具

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

// index.ts:460
export default function (pi: ExtensionAPI) {
  pi.registerTool({
    name: "subagent",
    label: "Subagent",
    description: "Delegate tasks to specialized subagents with isolated context...",
    parameters: SubagentParams,   // TypeBox schema
    async execute(_toolCallId, params, signal, onUpdate, ctx) {
      // 1. 查找指定子代理定义 (agents/<name>.md)
      // 2. 创建独立的子 Session 实例 (Fresh Context)
      // 3. 运行子 Agent 独立 ReAct 循环
      // 4. 返回子 Agent 的最终总结文本
    },
    renderCall(...)  { /* 怎么画调用 */ },
    renderResult(...) { /* 怎么画结果 */ },
  });
}

如何利用 extension 实现mcp

MCP（Model Context Protocol）连接器同样不需要作为框架核心的硬编码逻辑，而是通过 Extension 机制以“外部工具动态桥接”的方式实现：

1
2
3
4
5
6

MCP 扩展启动 (extension(api))
  ├── 1. 读取配置文件 (.mcp.json)，启动 MCP Server 子进程 (Stdio / SSE 连接)
  ├── 2. 与 MCP Server 进行握手与协议初始化 (initialize)
  ├── 3. 调用 tools/list 获取 MCP Server 暴露的所有外部工具定义
  ├── 4. 将 MCP 工具批量翻译为本地 Tool 对象
  └── 5. 循环调用 api.register_tool(translated_tool) 注册进 Agent!

1. 协议转换：Extension 在 initialize 握手后，通过 tools/list 拿到 MCP Server 声明的所有工具与 Schema；
2. 工具注册：遍历 MCP 工具，将每个外部工具动态包装为本地的 Tool，调用 api.register_tool 注入 Agent；
3. 调用转发：当模型调用该工具时，包装函数的 execute 拦截调用，将参数打包为 tools/call 请求通过 Stdio/SSE 转发给 MCP Server 进程，并把执行结果原样喂回给大模型。

extensions.py

在我们的 Python 框架中，提炼了 pi 的核心设计，通过 extensions.py 实现了极简的框架级扩展三件套：

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

┌──────────────────────────────────────────────────────────────┐
│                    ExtensionManager                          │
│  - 目录发现 (discover: 递归扫描 **/*.py，跳过 _ 开头文件)     │
│  - 动态加载 (load_extension: importlib 加载模块)             │
│  - 坏扩展隔离 (load: try...except 捕获异常，仅告警不崩进程)   │
│  - 命令调度 (handle_command: 参数个数自适应)                 │
└──────────────┬───────────────────────────────────────────────┘
               │ 构造并持有
               ▼
┌──────────────────────────────────────────────────────────────┐
│                      ExtensionAPI                            │
│  1. 事件订阅: api.on(EventCls, handler) -> (event, api) 双参 │
│  2. 工具注册: @api.tool(...) / api.register_tool             │
│  3. 命令注册: @api.command("name") / api.register_command     │
└──────────────────────────────────────────────────────────────┘

ExtensionAPI

在 my-pi-agent 框架中，ExtensionAPI 是整个扩展机制的 “开发者契约面（Developer Surface）”。

当外部写一个扩展（如 mcp.py 或 .agents/extensions/my_tool.py）时，入口函数接收到的唯一参数就是 api: ExtensionAPI：

1
2
3

def extension(api: ExtensionAPI):
    # 扩展开发者只跟 api 对象打交道
    ...

ExtensionAPI 的定位是 “将 Agent 内部复杂的事件系统（HookRegistry）、工具注册表（ToolRegistry）和命令系统，收敛为最简单
直观的三套对外接口”。

核心能力 1：事件订阅系统 —— api.on

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

class ExtensionAPI:
    @overload
    def on(
        self, event_cls: type[Event], handler: None = None
    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]: ...

    @overload
    def on(self, event_cls: type[Event], handler: Callable[..., Any]) -> None: ...

    def on(
        self, event_cls: type[Event], handler: Callable[..., Any] | None = None
    ) -> Any:
        """注册事件 handler（可作装饰器）。handler 签名 (event, api)：
        返回 None=观察，返回 HookResult=干预。"""

        def _register(h: Callable[..., Any]) -> Callable[..., Any]:
            def wrapped(event: Event):
                # 关键点：将底层的单参 (event) 包装并注入 (event, self) 双参
                return h(event, self)

            self.agent.hooks.register(event_cls, wrapped)
            return h

        if handler is not None:
            _register(handler)
            return None
        return _register

统一干预数据结构：HookResult 与五大决策点

所有生命周期拦截点共享统一强类型的 HookResult 数据结构：

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

@dataclass(frozen=True)
class HookResult:
    """Hook 回调的干预结果。返回 None = 纯观察，返回 HookResult = 干预。"""

    block: bool = False
    reason: str | None = None

    # 决策点 1 (input): 改写用户输入文本
    updated_input: str | None = None

    # 决策点 2 (before_agent_start): 动态改写 System Prompt
    updated_system_prompt: str | None = None

    # 决策点 3 (context): 临时改写发给大模型的 messages 视图（不污染 Session 树）
    updated_messages: list[Message] | None = None

    # 决策点 4 (tool_call): 改写工具入参
    updated_args: dict | None = None

    # 决策点 5 (tool_result): 改写工具出参
    updated_result: str | None = None

框架在 Agent.run() 中完整对齐了 Pi 的 五大生命周期决策拦截点：

1. UserInput（input）：在用户输入进入 Session 之前触发，支持 block=True 阻断输入或 updated_input 改写输入；
2. AgentStart（before_agent_start）：在准备好系统消息后触发，支持 updated_system_prompt 动态更新首条 system 消息；
3. BeforeModelCall（context）：在 _ctx.prepare() 产出 view 后、调用大模型前触发，支持 updated_messages 临时改写送给大模型的视图；
   • 关键架构不变式：updated_messages 仅临时修改当前这次发给 LLM 的 view 变量，真实 self.messages 与 Session 磁盘 JSONL 保持绝对纯净，实现 零历史污染；
4. ToolExecutionStart（tool_call）：工具执行前触发，支持 block=True 拦截危险调用或 updated_args 改写入参；
5. ToolExecutionEnd（tool_result）：工具执行后触发，支持 updated_result 改写返回给模型的出参。

流式 Token 级实时熔断（MessageUpdate）：除上述 5 大决策点外，MessageUpdate 同样继承了 Interceptable。在大模型逐 Token 流式生成的过程中，扩展如果检测到危险输出片段，返回 HookResult(block=True) 即可瞬间掐断生成，并且框架会直接丢弃未完成的半截文本（不写入 Session 树），彻底避免模型在下一轮对话中产生“续写断句”的严重幻觉。

核心能力 2：工具注册系统 —— api.tool 与 api.register_tool

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

def register_tool(self, tool: Tool) -> None:
    """注册工具 → agent.registry（撞名静默覆盖，registry 语义）。"""
    self.agent.registry.register(tool)

def tool(self, **kwargs: Any):
    """@api.tool(description=...) 装饰器：@tool 包装 + register_tool。"""

    def decorator(func) -> Tool:
        # 复用底层 pydantic 动态建模装饰器
        t = _tool(**kwargs)(func)
        self.register_tool(t)
        return t

    return decorator

• @api.tool(…)：面向普通的 Python 业务函数。扩展开发者写一个原生 Python 函数，加上 @api.tool，内部自动通过 Pydantic
  提取函数签名、类型标注和 docstring，生成 OpenAI/Anthropic 兼容的 JSON Schema，并注册进 Agent。
• api.register_tool(tool: Tool)：面向复杂的动态工具（如 MCP 远程工具、Subagent 委派工具）。直接传入已经构造好的 Tool
  实体对象（例如带有 raw_schema 的 MCP 工具）。

核心能力 3：命令注册系统 —— api.command 与 api.register_command

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

def register_command(
    self, name: str, handler: CommandHandler, description: str = ""
) -> None:
    """注册命令（name 不含 /）。"""
    self._commands[name] = handler

def command(self, name: str, description: str = ""):
    """@api.command("now") 装饰器。"""

    def decorator(func: CommandHandler) -> CommandHandler:
        self.register_command(name, func, description)
        return func

    return decorator

def get_commands(self) -> dict[str, CommandHandler]:
    """已注册命令的拷贝（name → handler）。"""
    return self._commands.copy()

• 将用户输入的命令名（如 “mcp” 或 “now”）与对应的 Python 处理函数映射在 self._commands 字典中；
• get_commands() 返回字典的浅拷贝（.copy()），保护内部字典不被外部意外修改。

ExtensionManager

ExtensionManager 就是面向 Agent 框架内部的“扩展总管与调度仓储（Repository）”。

它对标了框架内的 SkillManager 与 SubagentManager，专门负责：

1. 解析扫描目录（三态语义：默认探测 / 显式禁用 / 外部指定）；
2. 递归发现磁盘上的 .py 扩展文件（自动跳过 _ 开头的私有辅助文件）；
3. 通过 importlib 动态加载模块并执行 extension(api) 握手（原生支持 async def 协程与同步 def 入口，支持异步长连接初始化）；
4. 单点故障隔离（坏插件隔离保护：try...except 捕获异常，单个扩展崩溃仅告警、不影响主 Agent 启动）；

5. 用户斜杠命令的反射分发与调度（自动适配 0 参 / 1 参 Handler）。

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

   1. 启动装配期 (Agent.__init__) ┌──────────────────────────────────────────────────────────┐ │ Agent.__init__() │ │ ├── self._register_tools(tools) │ │ ├── self.extension_manager = ExtensionManager(self, ..)│ │ └── self.extension_manager.load() │ │ │ │ │ ▼ (遍历发现 .py 文件) │ │ ExtensionManager.load_extension("mcp.py") │ │ │ │ │ ▼ (执行插件入口) │ │ mcp.extension(self.api) │ │ ├── api.register_tool(t) ──► 注入 Agent.registry │ │ ├── api.on(Event) ──► 注入 Agent.hooks │ │ └── api.command("mcp") ──► 注入 api._commands │ └──────────────────────────────────────────────────────────┘ 2. 运行交互期 ├── [模型调用工具] ──► 查表分发至 Extension 注入的工具 (ReAct 循环) ├── [生命周期事件] ──► 触发 Extension 注册的 Hook (洋葱拦截) └── [用户输入 /cmd] ──► extension_manager.handle_command("cmd", args)

生命周期与方法全景图

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

  【启动装配阶段】

       1. __init__(agent, extension_dirs)
          │  ├─ 绑定 agent
          │  ├─ 创建 ExtensionAPI(agent)
          │  └─ 解析目录路径 (三态: 默认探测 / 显式禁用 / 指定目录)
          ▼
       2. load() ─── 批量加载与单点故障隔离保护
          │
          ├─► 3. discover(directory) ─── 递归扫描文件
          │      │  扫描目录及子目录下的 **/*.py
          │      └─ 自动跳过 _ 开头的私有文件 (_helper.py 等)
          │
          │      ┌─ 针对发现的每个 .py 文件逐个调用 ──────────┐
          │      │  (用 try...except 隔离: 坏扩展只告警不崩溃) │
          ▼      ▼                                             │
       4. load_extension(path) ─── 动态加载单个扩展            │
          │  ├─ importlib 动态载入模块                         │
          │  ├─ 智能识别入口函数: extension(api)               │
          │  └─ 执行 extension_func(self.api)                  │
          │        ├── 注册工具 ──► 注入 Agent.registry        │
          │        ├── 订阅事件 ──► 注入 Agent.hooks           │
          │        └── 注册命令 ──► 存入 self.api._commands    │
          └────────────────────────────────────────────────────┘

────────────────────────────────────────────────────────────────────────────

  【运行交互阶段 (0 Token 消耗)】

  用户在终端输入: "/mcp status"
          │
          ▼
       5. handle_command("mcp", "status") ─── 斜杠命令反射调度
          │  ├─ 查表: 从 self.api._commands 找到对应 handler
          │  ├─ 参数自适应:
          │  │     ├─ 若为无参函数 cmd()  ──► 执行 handler()
          │  │     └─ 若为带参函数 cmd(args) ──► 执行 handler("status")
          │  └─ 未知命令抛出清晰错误提示
          ▼
  向终端控制台直接输出结果

mcp

mcp.py 是框架的 MCP (Model Context Protocol) 客户端内置扩展。

分层说明：在最新的架构重构中，MCP 客户端以 Extension 插件的形式置于产品层（src/my_coding_agent/mcp.py），使框架层 my_agent_core 保持极简纯净，同时产品层可随时按需插拔加载。

它不仅是一个完整的 MCP 客户端实现，同时也是一个遵循框架标准规范的 Extension 插件（包含工具注册与命令注册）。

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

            .mcp.json (工作区配置)
                 │
                 ▼
        MCPClientManager (多 Server 管理)
                 │
     ┌───────────┴───────────┐
     ▼                       ▼
MCPConnection (Server A)  MCPConnection (Server B)
[AsyncExitStack 管理]     [AsyncExitStack 管理]
├── stdio_client 子进程   ├── stdio_client 子进程
└── ClientSession (MCP)   └── ClientSession (MCP)
     │                       │
     └───────────┬───────────┘
                 ▼
         包装为内部 Tool 对象
  (raw_schema=远程Schema, is_parallel_safe=True)
                 │
                 ▼
     ExtensionAPI.register_tool() ──► 注入 Agent.registry
     ExtensionAPI.command("mcp")  ──► 注册 /mcp 状态命令

配置数据类：MCPServerConfig

1
2
3
4
5
6

@dataclass
class MCPServerConfig:
    name: str
    command: str
    args: list[str]
    env: dict[str, str] | None = None

对应 .mcp.json 中配置的单个 Server 条目，如：

1
2
3
4
5
6
7
8

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "D:/data"]
    }
  }
}

单连接管理器：MCPConnection

这是与单个 MCP 子进程打交道的连接实体。

1
2
3
4
5

class MCPConnection:
    def __init__(self, config: MCPServerConfig):
        self.config = config
        self._session: ClientSession | None = None
        self._exit_stack = contextlib.AsyncExitStack()

• 核心亮点：AsyncExitStack 上下文管理
  MCP 官方 SDK 的 stdio_client 和 ClientSession 都是异步上下文管理器（async with）。在不退出当前方法的情况下长期保持连接，使用 AsyncExitStack 可以将多个异步上下文压栈保存，在需要关闭时
  调用 aclose() 一键安全释放。

• 建立连接与握手 (start)：

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

  async def start(self) -> None: params = StdioServerParameters( command=self.config.command, args=self.config.args, env=server_env, ) read_stream, write_stream = await self._exit_stack.enter_async_context( stdio_client(params) ) session = await self._exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) self._session = session await session.initialize() # 完成 MCP 初始化握手协议

• 调用工具 (call_tool)：

  1	async def call_tool(self, name: str, arguments: dict[str, Any]) -> ToolResult:

  • Never-Throw 保障：捕获所有异常，绝不向上抛出导致 Agent 崩溃，而是包装为 ToolResult(ok=False, error=…) 让大模型自我修正。
  • 内容格式化：将 MCP 返回的多段 content 抽取为纯文本，并识别 MCP 协议中的 isError 标记。

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

┌──────────────────────────────────────────────────────────┐
│                      MCP 架构分层                         │
├──────────────────────────────────────────────────────────┤
│                                                          │
│  【第二扇门】ClientSession (协议层)                       │
│    • list_tools()  • call_tool()  • JSON-RPC 消息解析     │
│  ─────────────────────────┬────────────────────────────  │
│                           │ 依赖底层流传输                │
│                           ▼                              │
│  【第一扇门】stdio_client (传输层)                        │
│    • 启动 Node.js/Python 进程  • 管道管理 (stdin/stdout)  │
│                                                          │
└──────────────────────────────────────────────────────────┘

1. 先有第一扇门，才有第二扇门：
   如果没有 stdio_client 启动进程并提供 read/write_stream，ClientSession 就根本不知道该向哪里读写协议数据。

2. 关闭时必须倒序退出：

   • 先关 ClientSession（告诉对方我们要结束会话了，把未完成的 RPC 请求取消）；
   • 再关 stdio_client（彻底杀掉子进程，释放操作系统进程句柄与内存）。

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

   ┌───────────────────────────────────────────────────────────────┐ │ 第一步：stdio_client(params) 【物理与传输层：打通管道】 │ │ 操作系统的进程与字节流 (Process & Pipes) │ │ 输入: command="npx", args=[...] │ │ 产出: 裸管道 (read_stream, write_stream) │ └───────────────────────────────┬───────────────────────────────┘ │ 传递裸管道 ▼ ┌───────────────────────────────────────────────────────────────┐ │ 第二步：ClientSession(read, write) 【协议与业务层：听懂语言】 │ │ JSON-RPC 2.0 与 MCP 业务对象 │ │ 输入: 裸管道 (read_stream, write_stream) │ │ 产出: 高级会话对象 session (能调 list_tools/call_tool) │ └───────────────────────────────────────────────────────────────┘

mcp的底层通信协议：JSON-RPC 2.0

MCP 的所有消息交互全部建立在 JSON-RPC 2.0 规范之上。客户端和服务端通过互相发送结构化的 JSON 文本进行交流。

常见的几种通信帧：

1. 工具列表发现（tools/list）

• Client 发出请求：

  1 2 3 4 5 6

  { "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} }

• Server 返回结果（带 JSON Schema 参数契约）：

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

  { "jsonrpc": "2.0", "id": 1, "result": { "tools": [ { "name": "calculate_tax", "description": "计算个人所得税", "inputSchema": { "type": "object", "properties": { "income": { "type": "number", "description": "税前收入" } }, "required": ["income"] } } ] } }

  2. 工具调用执行（tools/call）

• Client 发出执行指令：

  1 2 3 4 5 6 7 8 9

  { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "calculate_tax", "arguments": { "income": 20000 } } }

• Server 返回执行结果：

  1 2 3 4 5 6 7 8 9 10

  { "jsonrpc": "2.0", "id": 2, "result": { "content": [ { "type": "text", "text": "应纳个税: 1590 元" } ], "isError": false } }

传输层模式：Stdio vs SSE/HTTP

MCP 规范定义了两种主要的物理传输通道：

1
2
3
4
5
6
7
8
9
10

1. Stdio（标准输入输出，本地子进程模式 —— 我们框架采用的模式）：
   Client (Python Agent)
     │  stdin (向子进程写 JSON)
     ├───► [MCP Server 子进程 (如 Node.js / Python)]
     ◄───┘ stdout (从子进程读 JSON)
   优点：零网络端口暴露，启动即用，极高安全性，适合本地工具。

2. SSE / HTTP（Server-Sent Events 远程流式模式）：
   Client ─── HTTP POST / SSE ───► 远程云端 MCP Server (企业内网服务)
   优点：适合连接跨机器的大型企业数据库或云端 API。

多服务协调器：MCPClientManager

如果把单个 MCPConnection 比作 “专线接线员”，那么 MCPClientManager 就是 “接线总调度中心”。

在实际项目中，用户通常会在 .mcp.json 中配置多个 MCP 服务（比如一个 GitHub 查 PR、一个 Postgres 查数据库、一个 Filesystem 读文件）。
MCPClientManager 负责管理所有这些服务的生命周期编排与工具适配。

一、连接池状态：init

1
2
3

class MCPClientManager:
    def __init__(self):
        self.connections: dict[str, MCPConnection] = {}

• 维护一个全局字典 self.connections，以服务名称为 key（如 “github”、”filesystem”），存储对应的 MCPConnection 实例；

• 方便后续按服务名查找连接、状态检查（/mcp 命令）和统一关闭。

  二、配置解析：load_config(path)

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

  def load_config(self, path: Path | str) -> list[MCPServerConfig]: p = Path(path) if not p.exists(): return [] try: data = json.loads(p.read_text(encoding="utf-8")) except Exception as exc: raise ValueError(f"Invalid JSON in {p}: {exc}") from exc servers = data.get("mcpServers", {}) configs = [] for name, srv in servers.items(): configs.append( MCPServerConfig( name=name, command=srv.get("command", ""), args=srv.get("args", []), env=srv.get("env"), ) ) return configs

• 契约对齐：完全对齐 Claude Desktop / Cursor / Pi 的标准 .mcp.json 格式：

  1 2 3 4 5 6 7 8 9

  { "mcpServers": { "sqlite": { "command": "uvx", "args": ["mcp-server-sqlite", "--db-path", "test.db"], "env": {"DEBUG": "1"} } } }

• 将 JSON 转换为强类型的 MCPServerConfig 数据类列表。

  三、协议适配与工具转换：connect_server(config) —— 最核心方法

  这是整个类最精妙的部分，它完成了从 “MCP 远程服务” 到 “Agent 内部标准 Tool 对象” 的桥接转换：

  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

  async def connect_server(self, config: MCPServerConfig) -> list[Tool]: # 1. 建立长连接并完成握手 conn = MCPConnection(config) await conn.start() self.connections[config.name] = conn # 2. 动态获取远程支持的所有工具 mcp_tools = await conn.list_tools() wrapped_tools = [] # 3. 逐个包装成 Agent 框架的标准 Tool 对象 for t in mcp_tools: tool_name = t.name schema = getattr(t, "input_schema", getattr(t, "inputSchema", {})) # 闭包生成执行函数 def _make_handler(target_conn: MCPConnection, target_name: str): async def _handler(args: dict[str, Any]) -> ToolResult: return await target_conn.call_tool(target_name, args) return _handler wrapped = Tool( func=_make_handler(conn, tool_name), name=tool_name, description=t.description or "", raw_schema=schema, # 关键点 1：直接透传远程 Schema timeout=120.0, is_parallel_safe=True, # 关键点 2：标记为只读并发安全 ) wrapped_tools.append(wrapped) return wrapped_tools

  这里解决了三个核心工程难点：

1. Python 循环中的闭包陷阱（Closure Late-Binding Trap）：
   如果在 for 循环里直接写 async def _handler(args): return await conn.call_tool(t.name, args)，由于 Python 延迟绑定的特性，所有工具最终都
   会调成最后一个工具！
   通过 _make_handler(conn, tool_name) 独立工厂函数，确保每个工具在内存中精确绑定属于自己的 conn 和 tool_name。

2. raw_schema 机制解耦 Python 函数签名：
   普通工具需要 Python 函数类型注解（如 def add(a: int) -> int）来推导 JSON Schema；但 MCP 工具的代码在远程，框架通过 raw_schema=schema 直
   接透传远程给的 JSON Schema 字典，大模型能立刻看懂入参结构。

3. is_parallel_safe=True 赋予并发加速能力：
   包装出的工具自动具备前文讲过的只读并发能力（asyncio.gather 批执行），当模型同时调用多个 MCP 工具时可以并行执行。

   四、安全清理：close_all()

   1 2 3 4 5 6

   async def close_all(self) -> None: """异步关闭所有连接。""" for conn in self.connections.values(): with contextlib.suppress(Exception): await conn.close() self.connections.clear()

• 资源防泄漏：遍历所有连接逐个调用 close()（进而触发 AsyncExitStack.aclose() 回收 Stdio 子进程）；
• 容错关闭：使用 contextlib.suppress(Exception)，即使某一个子进程已经意外退出了，也不会中断其他正常子进程的回收；

• 重置连接池：清空字典。

  总结：MCPClientManager 的定位

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

  .mcp.json │ ▼ (load_config) MCPServerConfig 列表 │ ▼ (connect_server) ┌────────────────────────────────────────┐ │ MCPClientManager │ │ │ │ • conn1 (github) ──► 产出 Tool 列表│──► 统一注入 Agent 注册表 │ • conn2 (filesystem) ──► 产出 Tool 列表│ └──────────────────┬─────────────────────┘ │ ▼ (close_all) 一键安全回收全部子进程

  它是一个非常干净的装配器（Assembler）+ 门面（Facade），屏蔽了多进程管理的复杂性，对外只提供配置加载、批量连接和统一关闭能力。

功能设计

在初始设计中，委派执行逻辑（make_task_tool）曾放在 subagents.py 中。但在 2026-08-16 的设计迭代
（subagent-task-manager-design.md）中，架构进行了清晰分层：

1. subagents.py（定义与仓储）：收窄职责，只负责 Subagent 声明与 SubagentManager 发现管理。
2. subagent_tasks.py（任务生命周期，对标 OpenHands）：
   • 引入 SubagentTask（拥有 id、status、result、error）与 TaskStatus（RUNNING、COMPLETED、ERROR）；
   • SubagentTaskManager 负责任务生命周期的推进（创建任务 → 过滤工具并剔除 task → 装配子 System Prompt → 独立 Session 落盘到 subagents/ → 驱动子 Agent 运行 → 标记状态）。
3. tools/builtin/task.py（工具桥）：
   • make_task_tool 移入内置工具包，仅作为连接大模型工具调用与 SubagentTaskManager 的轻量转换层。

子agent调用机制

调用机制：一个内建工具

1
2
3
4
5
6
7
8
9
10
11

主 agent（allowedTools 里含 "Agent"）
  │ 模型根据 description 自动决定，或 prompt 里显式点名
  ▼
调 "Agent" 工具：{ subagent_type: "code-reviewer", prompt: "..." }
  │ -------------------------------
  ▼  运行时（harness）接管，模型看不到这层
查定义（程序式 agents dict / .claude/agents/*.md / 内置 general-purpose）
  │
spawn 子代理（新 context window，只收：自己的 prompt + Agent 工具的 prompt 字符串 + 项目 CLAUDE.md + 工具定义）
  │
父逐字收到子代理的【最终消息】作为 Agent 工具结果

关键点：对模型而言，这就是一个普通 tool_use——block.name == “Agent”，block.input.subagent_type 是代理名。模型根本不知道背后是”又起了一个 agent”，harness 负责创建子会话/执行/回收。

子agent包含的字段

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

name（默认值：文件名主干名）

description

tools（默认值：[]）：单个工具名称或工具名称列表

skills（默认值：[]）：逗号分隔的字符串或技能名称列表

model（默认值：inherit）：inherit 表示复用父级 LLM；其他值将作为 LLM 配置方案（profile）名称从 profile_store_dir 或默认 profile 存储区加载

color（可选）

max_iteration_per_run（可选，正整数）

max_budget_per_run（可选，以美元计的正数）

hooks（可选的 hook 配置）

profile_store_dir（可选的自定义 LLM profile 目录）

mcp_config（可选的 MCP 服务器映射表）；mcp_servers 为已弃用的别名

permission_mode（可选）：always_confirm、never_confirm 或 confirm_risky；省略则继承父级的确认策略

condenser（可选）：省略则使用默认的摘要压缩器（summarizing condenser）；none 或 false 禁用压缩；传入字典映射则用于配置特定压缩器

流程

1
2
3
4
5

agents/*.md ──发现──► SubagentManager ──get(name)──► Subagent 定义
                                                        │ spawn
model 调 task(prompt, agent_type) ──────────────────► 子 Agent 实例
        ▲                                              │ run(prompt)
        └──────────── 只回最终文本 ◄────────────────────┘

subagents.py

数据模型 Subagent

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

@dataclass(frozen=True)
class Subagent:
    name: str                           # 子代理唯一标识
    description: str                    # 描述：主模型根据此意图决定何时委派
    content: str                        # 正文：子代理独立的 System Prompt
    file_path: Path                     # 来源文件路径（内置则为 <builtin>）
    model: str | None = None            # 模型覆盖：None 表示继承父 Agent 的模型
    effort: str | None = None           # 推理深度（预留字段）
    max_turns: int | None = None        # 最大轮数限制（对应 maxTurns）
    tools: tuple[str, ...] | None = None       # 工具白名单：None 表示继承父级所有可用工具
    disallowed_tools: tuple[str, ...] = ()     # 工具黑名单
    skills: tuple[str, ...] | None = None      # 允许该子代理使用的 skill 名称列表

流程演示

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

[启动阶段]
Agent.__init__(subagent_dirs=...)
  ├── SubagentManager 发现并解析所有 .agents/agents/*.md
  ├── format_prompt() 将 <available_agents> 拼入 Agent 的 system_prompt 尾部
  └── Agent 自动注册内置工具 make_task_tool (即 task 工具)

[运行阶段 - ReAct 循环]
1. 用户提问："帮我审查一下最新的提交"
2. 主 Agent 决定派发：
   -> tool_call: task(prompt="审查最新 commit", subagent_type="code-reviewer")
3. TaskManager.start_task():
   ├── 从 SubagentManager 中检索 "code-reviewer" 的 Subagent 实体
   ├── 工具过滤：按 tools/disallowed_tools 过滤，强制剔除 task 工具（硬防线：禁止子代理调 task）
   ├── System Prompt 装配：仅包含 Subagent.content + 声明的 skills 清单（不继承父 Prompt）
   ├── 会话隔离：在父会话目录下创建 subagents/agent-task_00000001.jsonl 独立存储子对话
   ├── 实例化子 Agent：
   │     Agent(
   │         llm=parent.llm,
   │         tools=filtered_tools,
   │         session=child_session,
   │         system_prompt=sub_system_prompt,
   │         model=sub.model or parent.model,
   │         max_iterations=sub.max_turns or parent.max_iterations,
   │         subagent_dirs=[],   # 禁用子代理再去扫描子代理
   │         skill_dirs=[],
   │     )
   └── 执行子 Agent.run()，将子 Agent 的最终总结字符串作为 tool_result 返回给主 Agent

SubagentManager

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

[磁盘文件]
.agents/agents/
   ├── code-reviewer.md
   └── researcher.md
          │
          ▼ 扫描、过滤、解析 (utf-8-sig + parse_frontmatter)
┌──────────────────────────────────────────────────────────┐
│                   SubagentManager                        │
│  - self.subagents: {"code-reviewer": Subagent(...), ...} │
└──────────────┬────────────────────────────┬──────────────┘
               │ 1. format_prompt()          │ 2. manager.get("code-reviewer")
               ▼                             ▼
   [Agent.system_prompt 拼装]         [TaskManager._run 委派派生]
   <available_agents>                - 拿到 Subagent 定义
     <agent>...</agent>               - 取 sub.content 作为子 Agent system_prompt
   </available_agents>               - 裁剪 tools / model / maxTurns 启动子 Agent

核心方法format_prompt

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

def format_prompt(self) -> str:
    """全部 agents → XML 清单块（名字 + description）；空 → 空串（进 system）。"""
    if not self.subagents:
        return ""
    parts = ["<available_agents>"]
    for s in self.subagents.values():
        parts.append("  <agent>")
        parts.append(f"    <name>{s.name}</name>")
        parts.append(f"    <description>{s.description}</description>")
        parts.append("  </agent>")
    parts.append("</available_agents>")
    return "\n".join(parts)

• 低 Token 占用设计：
  格式化生成的 XML 仅包含 name 和 description 两个核心字段，不会将子代理的详细 System Prompt 正文塞入父 Agent。
• 自动拼接进 System Prompt：
  在 Agent.init 中，如果 SubagentManager 包含有效子代理，format_prompt() 输出的内容会被用 \n\n 自动拼接在父 Agent
  的 system_prompt 尾部：

  1 2 3 4 5 6 7 8 9 10

  <available_agents> <agent> <name>code-reviewer</name> <description>审查代码提交与潜在缺陷</description> </agent> <agent> <name>researcher</name> <description>只读模式调研技术文档和架构</description> </agent> </available_agents>

  父模型据此便能知道当前拥有哪些子代理可以委派。

subagent_tasks.py（子代理委派引擎）

架构命名解耦注记（阶段 8 演进）：为了彻底消除子代理运行时生命周期与项目工程待办看板（TaskItem / TaskStore）的命名撞车与概念歧义，本模块由早期的 tasks.py 全面重命名为 subagent_tasks.py，核心类统一命名为 SubagentTask / SubagentTaskStatus / SubagentTaskManager，界限分明。

核心数据结构：SubagentTaskStatus 与 SubagentTask

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

class SubagentTaskStatus(StrEnum):
    """委派任务三态。"""
    RUNNING = "running"
    COMPLETED = "completed"
    ERROR = "error"

@dataclass
class SubagentTask:
    """一次子代理委派任务：有 id/状态/结果，可查询、可追踪、可动态干预。"""
    id: str
    status: SubagentTaskStatus
    result: str | None = None
    error: str | None = None

    def set_result(self, result: str) -> None:
        self.result = result
        self.error = None
        self.status = SubagentTaskStatus.COMPLETED

    def set_error(self, error: str) -> None:
        self.error = error
        self.result = None
        self.status = SubagentTaskStatus.ERROR

Prompt 与权限隔离

在实例化子 Agent 之前，必须对子代理的 Prompt 边界 与 工具权限 进行精准裁剪：

1. 子代理 System Prompt 装配（_system_for）

1
2
3
4
5
6
7
8

def _system_for(sub: Subagent, parent: "Agent") -> str:
    """子代理 system = 正文 + （若有 skills）子集清单；不收父 system prompt（Claude 官方语义）。"""
    parts = [sub.content]
    if sub.skills:
        block = parent.skill_manager.format_prompt(sub.skills)
        if block:
            parts.append(block)
    return "\n\n".join(p for p in parts if p)

• 正文即人设：子代理仅使用自身的 sub.content 作为基础人设，绝不继承父 Agent 的 system_prompt（遵循 Claude 官方语义，防
  止父级复杂指令干扰子任务）。

• 按需注入 Skill 清单：如果子代理定义了 skills: [“git-diff”, “python-eval”]，则利用父级 SkillManager 过滤出对应子集的
  XML 清单并拼入 System Prompt 尾部。

  2. 工具过滤与防递归硬防线（_filter_tools）

  1 2 3 4 5 6 7 8 9

  def _filter_tools(parent: "Agent", sub: Subagent) -> list: """父工具集按白/黑名单过滤；task 永不出现（防递归）。""" tools = [t for t in parent.registry.list() if t.name != "task"] if sub.tools is not None: allowed = set(sub.tools) tools = [t for t in tools if t.name in allowed] black = set(sub.disallowed_tools) tools = [t for t in tools if t.name not in black] return tools

• 权限裁剪：先支持白名单 tools 筛选，再进行黑名单 disallowed_tools 剔除。

• 强制剔除 task 工具（硬防线）：无论配置文件的白名单写了什么，task 工具一律从子工具集中强制剥离，防止子 Agent 在运行中
  再次委派产生死循环。

TaskManager

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

class TaskManager:
   """委派任务的生命周期管理器（对标 OpenHands TaskManager）。"""

   def __init__(self, manager: SubagentManager, parent: "Agent"):
       self._manager = manager     # 查 agent 定义
       self._parent = parent       # 供 llm/工具集/skill_manager/max_iterations
       self._counter = 0

   def start_task(self, prompt: str, subagent_type: str = "default") -> Task:
       """同步阻塞：建 Task(RUNNING) → spawn → run → 更新状态 → 返回。"""
       task = self._create_task(subagent_type)
       try:
           task.set_result(self._run(prompt, subagent_type, task.id))
       except Exception as exc:
           task.set_error(str(exc))
       return task

start_task 是对外的核心方法，它通过 try…except 兜住整个派生与执行流程。无论是子代理找不到、大模型报错、还是子 Agent
抛出异常，都会被捕获并转换为 task.set_error(…)，返回包含错误描述的 Task 对象，绝不让未处理异常打崩父 ReAct 循环。

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

  ┌────────────────────────────────────────────────────────┐
  │    TaskManager._run(prompt, subagent_type, task_id)    │
  └───────────────────────────┬────────────────────────────┘
                              │
                              ▼
               ┌──────────────────────────────┐
               │ 1. 查阅子代理定义 (Lookup)   │
               │    sub = manager.get(type)   │
               └──────────────┬───────────────┘
                              │
           ┌──────────────────┴──────────────────┐
           │ 是否找到 Subagent 定义？             │
           └─────────┬───────────────────┬───────┘
        [未找到]     │                   │ [找到]
                     ▼                   │
        ┌─────────────────────────┐      │
        │ subagent_type=="default"│      │
        └────┬───────────────┬────┘      │
      [是]   │               │ [否]      │
             ▼               ▼           │
┌─────────────────┐   ┌────────────────┐ │
│使用内置默认定义 │   │抛出 ValueError │ │
│DEFAULT_SUBAGENT │   │列出可用 agents │ │
└────────────┬────┘   └───────┬────────┘ │
             │                │ (失败)   │
             └─────────┬──────┴──────────┘
                       │
                       ▼
    ┌──────────────────────────────────────┐
    │ 2. 创建子代理独立 Session            │
    │    - 路径: subagents/agent-{id}.jsonl│
    │    - 注入 Header 元数据:             │
    │      * agent_type                    │
    │      * parent_session_id             │
    │    - child_session.save()            │
    └──────────────────┬───────────────────┘
                       │
                       ▼
    ┌──────────────────────────────────────┐
    │ 3. 准备隔离环境与权限 (Sanitize)      │
    │    ├─ _system_for(sub, parent)       │
    │    │   正文 + (若有) 声明 skills 清单│
    │    └─ _filter_tools(parent, sub)     │
    │        按白/黑名单过滤，强制剔除 task│
    └──────────────────┬───────────────────┘
                       │
                       ▼
    ┌──────────────────────────────────────┐
    │ 4. 实例化子 Agent (Spawn)            │
    │    Agent(                            │
    │      llm = parent.llm,               │
    │      tools = 过滤后工具集,           │
    │      session = child_session,        │
    │      system_prompt = 子专属 Prompt,  │
    │      model = sub.model 覆盖/继承,    │
    │      max_iterations = 轮数覆盖/继承, │
    │      skill_dirs = [],                │
    │      subagent_dirs = []  <-- 防递归! │
    │    )                                 │
    └──────────────────┬───────────────────┘
                       │
                       ▼
    ┌──────────────────────────────────────┐
    │ 5. 驱动子 Agent 独立运行             │
    │    result = child.run(prompt)        │
    └──────────────────┬───────────────────┘
                       │
           ┌───────────┴───────────┐
           │ 执行是否发生未捕获异常?│
           └─────┬───────────┬─────┘
        [成功]   │           │ [抛出异常]
                 ▼           ▼
  ┌───────────────────────┐ ┌────────────────────────────┐
  │ 结果判空兜底:         │ │ 捕获并包装为:              │
  │ result or "(no summary│ │ RuntimeError("Subagent...  │
  └──────────────┬────────┘ │   failed: ...")            │
                 │          └─────────────┬──────────────┘
                 │                        │
                 ▼                        ▼
  ┌───────────────────────────────────────────────────────┐
  │   返回最终文本摘要字符串 / 异常冒泡给 start_task 捕获  │
  └───────────────────────────────────────────────────────┘

关键技术细节亮点：

1. 循环导入解解（Circular Import Resolution）：
   Agent 在初始化时需要 import make_task_tool（依赖 TaskManager），而 TaskManager._run 需要实例化 Agent。
   这里采用了 TYPE_CHECKING + _run 内部延迟 import Agent 的标准方案，优雅消解了循环依赖。
2. 子代理独立持久化（对齐 Claude Code 目录规范）：
   子代理拥有专属的 Session 文件，落盘在父 Session 所在目录下的 subagents/agent-task_00000001.jsonl 中，且在 Header 的
   metadata 中记录了：
   • agent_type：子代理类型；
   • parent_session_id：关联的父会话 ID（通过父子会话拓扑指针关联，保证可溯源且不引入死状态）。
     这使得子代理的交互全流程都可以溯源复盘，同时完全不污染父会话的消息历史。
3. Fresh Context（全新上下文）：
   子 Agent 拥有全新的消息队列，仅接收当前的 prompt 作为单条 user 消息输入，不携带父 Agent 之前的长历史，大幅节省
   Token 并提升模型注意力。

task工具

在 tools/builtin/task.py 中，make_task_tool 充当了 Agent 与 TaskManager 之间的适配层：

1
2
3
4
5
6
7
8

def make_task_tool(manager: SubagentManager, parent: "Agent") -> Tool:
    task_manager = TaskManager(manager, parent)

    def task(prompt: str, agent_type: str = "default") -> str:
        t = task_manager.start_task(prompt, agent_type)
        return t.result if t.status is TaskStatus.COMPLETED else (t.error or "(no summary)")

    return Tool(func=task, name="task")

• 大模型发起工具调用 task(prompt=”…”, agent_type=”…”)；
• 工具桥将调用转交给 task_manager.start_task；
• 工具桥根据 Task.status 提取 t.result 或 t.error，作为字符串直接喂回父模型的 ReAct 观察管道（Observation）。

如何防止子代理中再次调用 task 委派导致无限递归

在框架设计中，防止子代理（Subagent）再次调用 task 委派导致无限递归死循环（Anti-Recursion Loop），采用的是 “切断继承 + 抑制自生成” 的双重硬防线设计，并辅以 认知层隔离。

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

                 子代理派生请求 (TaskManager._run)
                                │
        ┌───────────────────────┴───────────────────────┐
        │                                               │
        ▼ 【第一道防线：防继承】                          ▼ 【第二道防线：防自生】
_filter_tools(parent, sub)                     Agent(..., subagent_dirs=[])
        │                                               │
        ▼                                               ▼
强制过滤 t.name != "task"                      SubagentManager 为空 (len == 0)
        │                                               │
        ▼                                               ▼
传入子 Agent 的 tools 参数无 task               bool(subagent_manager) == False
        │                                               │
        │                                               ▼
        │                                      Agent._register_tools
        │                                      跳过 make_task_tool 注册
        │                                               │
        └───────────────────────┬───────────────────────┘
                                │
                                ▼
                   【子 Agent ToolRegistry】
                   物理上彻底不存在 task 工具
                                │
                                ▼
                   【第三道防线：Prompt 认知隔离】
                   System Prompt 无 <available_agents>

1. 第一道防线：切断引用继承 —— _filter_tools 强制剔除 8 大内置敏感工具（对标 learn-claude-code）

在 subagent_tasks.py 中，当 SubagentTaskManager 从父 Agent 提取可用工具准备传递给子 Agent 时，在工具过滤的最源头就进行了物理级硬剥离：

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

def _filter_tools(parent: "Agent", sub: Subagent) -> list[Tool]:
    """父工具集按白/黑名单过滤；内置敏感工具永不出现（防递归与脏写）。"""
    # 核心：必须同时剔除委派、长期记忆与 Todo 看板工具，防止子代理篡改全局状态或递归衍生
    builtins = (
        "task", "memory", "todo",
        "task_create", "task_update", "task_get", "task_list", "todo_write",
    )
    tools = [t for t in parent.registry.list() if t.name not in builtins]

    if sub.tools is not None:
        allowed = set(sub.tools)
        tools = [t for t in tools if t.name in allowed]

    black = set(sub.disallowed_tools)
    tools = [t for t in tools if t.name not in black]
    return tools

• 无视白名单声明：即使开发者在 .agents/agents/xxx.md 的 Frontmatter 中写了 tools: read, bash, task，由于第一步 t.name not in builtins 已经把 task、memory、todo 从列表中彻底过滤掉，后续的白名单判定也绝不可能拿到它们。
• 解决的问题：不仅防止子 Agent 拿到 task 工具发生递归委派，更防止子 Agent 误调用 memory 或 todo 污染全局项目状态。

2. 第二道防线：抑制自发生装配 —— 5 大隔离参数触发注册短路（对标 OpenHands 工厂隔离）

仅仅从父级工具列表中剔除敏感工具还不够，因为 Agent 自身在初始化时具备自动探测本地技能、插件、记忆、工单并自动装配的能力。

因此，在 SubagentTaskManager._run 中实例化子 Agent 时，显式传入了全套 5 大隔离参数（禁用态）：

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

child = Agent(
    llm=self._parent.llm,
    tools=_filter_tools(self._parent, sub),
    session=child_session,
    system_prompt=_system_for(sub, self._parent),
    model=sub.model or self._parent.model,
    max_iterations=sub.max_iterations or self._parent.max_iterations,
    skill_dirs=[],       # 1. 禁用技能自探测 (仅按需继承声明的技能清单)
    subagent_dirs=[],    # 2. 禁用子代理自探测 (防递归衍生)
    plugin_dirs=[],      # 3. 禁用插件自探测 (防插件重新注册 task 工具)
    memory_dir=False,    # 4. 显式禁用长期记忆共享
    task_store=False,    # 5. 显式禁用项目看板工单共享
)

内部连锁反应机制：

1. 空仓储生成：SubagentManager([]) 收到空列表，不会扫描任何目录，self.subagents 为空字典 {}。
2. 利用 Python 容器协议 __len__ 实现布尔截断：
   在 subagents.py 中，SubagentManager 实现了 __len__：

   1 2	def __len__(self) -> int: return len(self.subagents)

   当对象实现了 __len__ 时，bool(obj) 会依据 len(obj) > 0 来判定。因为 len == 0，所以 bool(subagent_manager) 严格为 False。
3. Agent._register_tools 自动短路：

   1 2 3 4 5 6 7 8 9

   def _register_tools(self, tools: list[Tool]) -> None: for t in tools: self.registry.register(t) # 因为 self.subagent_manager 为 False，以下整个代码块直接跳过！ if self.subagent_manager: if self.registry.get("task") is not None: raise ValueError("Tool name 'task' conflicts...") self.registry.register(make_task_tool(self.subagent_manager, self))

• 解决的问题：防止子 Agent 在初始化时“自发生成”并注册一个新的 task 工具。

3. 第三道防线：认知层隔离 —— 抑制 <available_agents> 清单注入

大模型发起工具调用的前提是“知道存在这项能力”。

由于第二道防线中 SubagentManager 为空，在拼接子代理的 System Prompt 时：

1
2
3

def format_prompt(self) -> str:
    if not self.subagents:
        return ""  # 仓储为空时直接返回空字符串

• 子 Agent 的 System Prompt 尾部不会被注入任何 <available_agents> XML 块；
• 子模型在认知层面根本不知道系统具备“子代理委派”功能，从源头消除了产生“尝试调用子代理”的幻觉可能。

如何跟踪task状态

在设计任务跟踪与生命周期管理时，我们借鉴了 OpenHands（openhands/tools/task/manager.py） 的任务状态机体系与会话拓扑设计，实现了结构化、可落盘且零死状态的状态跟踪体系。

1. Task 状态机与生命周期模型

将传统的单次函数返回值升级为具备明确生命周期的 Task 实体与三态状态机：

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

class TaskStatus(StrEnum):
    RUNNING = "running"      # 任务正在由子 Agent 独立执行
    COMPLETED = "completed"  # 任务执行成功并返回了有效的文本总结
    ERROR = "error"          # 任务因未知名 agent_type、异常或超轮数终止

@dataclass
class Task:
    id: str                  # 格式为 task_00000001 的递增唯一标识
    status: TaskStatus
    result: str | None = None
    error: str | None = None

    def set_result(self, result: str) -> None:
        self.result = result
        self.error = None
        self.status = TaskStatus.COMPLETED

    def set_error(self, error: str) -> None:
        self.error = error
        self.result = None
        self.status = TaskStatus.ERROR

• 原子互斥更新：set_result 与 set_error 确保 result 与 error 状态严格互斥；
• 同步与异步一致性：在当前的同步阻塞执行模式下，Task 从 RUNNING 启动，并在 TaskManager.start_task 返回时即确定为 COMPLETED 或 ERROR；同时为未来演进至 async / background 后台异步委派留出了统一接口。

2. 基于“会话拓扑指针”的溯源跟踪（对齐 OpenHands）

在追踪一个任务会话时，我们摒弃了容易造成冗余死状态的局部深度计数器（如 spawn_depth），改为采用与 OpenHands / Anthropic SDK 一致的 “父子拓扑指针（Parent Linkage）” 方案：

1. 子会话独立物理隔离：
   子会话保存在父会话目录下的 subagents/agent-{task_id}.jsonl 中，完全不污染父会话的对话历史。
2. JSONL Header 元数据溯源：

   1 2 3 4 5 6 7 8 9

   { "id": "20260816-220000-abcd1234", "created_at": "2026-08-16T22:00:00.000000", "cwd": "D:/code/project", "metadata": { "agent_type": "code-reviewer", "parent_session_id": "20260816-215000-efgh5678" } }

   • parent_session_id：天然标明当前会话是一个子代理任务，且精确指向派生它的父会话 ID；
   • agent_type：标明当前子代理执行时所使用的角色定义。

3. 全链路容错与结果回传

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

主模型调 task 工具
  │
  ▼
TaskManager.start_task
  ├── 创建 Task(id="task_00000001", status=RUNNING)
  ├── 尝试执行 _run(...)
  │     ├── 成功 ──► task.set_result(summary) ──► status = COMPLETED
  │     └── 失败 ──► task.set_error(err_msg)  ──► status = ERROR
  ▼
make_task_tool 工具桥
  ├── COMPLETED ──► 返回 task.result
  └── ERROR     ──► 返回 task.error
  │
  ▼
作为 Observation 喂回主模型 ReAct 循环（永不抛未捕获异常）

通过这一闭环，无论子代理成功完成还是中途遇到未知角色/运行报错，任务状态都能被完整捕获并平滑反馈给主模型，主模型据此可继续自我纠错或汇总最终答复。

4. 动态干预通道与异步生命周期追踪

在 SubagentTaskManager 内部维护了 _active_agents: dict[str, Agent] 字典，提供以下原生异步干预方法：

• async def start_task(prompt: str, agent_type: str = "default") -> SubagentTask: 异步拉起子代理任务；
• def steer_task(task_id: str, message: str) -> None: 对正在运行的子代理实例进行定向转向（Steer 即时插队）；
• def follow_up_task(task_id: str, message: str) -> None: 对子代理追加排队消息（Follow-up 任务衔接）。

子agent的异常处理流程

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

步骤 ①：子 Agent 运行崩溃
       │  child.run(prompt) 抛出底层异常 (例如: TimeoutError)
       ▼
步骤 ②：TaskManager._run 捕获并包装
       │  except Exception as exc:
       │      raise RuntimeError("Subagent 'code-reviewer' failed: TimeoutError") from exc
       ▼
步骤 ③：TaskManager.start_task 捕获并沉淀为状态（不再往上抛异常！）
       │  # tasks.py -> start_task
       │  task = self._create_task(...)
       │  try:
       │      task.set_result(self._run(...))
       │  except Exception as exc:
       │      task.set_error(str(exc))   <-- 错误字符串存入 task.error, status 切为 ERROR
       │  return task                   <-- 返回 Task 实体对象给工具桥
       ▼
步骤 ④：make_task_tool 工具桥解包为文本返回给大模型
       │  # tools/builtin/task.py -> make_task_tool
       │  t = task_manager.start_task(...)
       │  return t.result if t.status is COMPLETED else (t.error or "(no summary)")
       │  # 此处返回纯文本: "Subagent 'code-reviewer' failed: TimeoutError"
       ▼
步骤 ⑤：父 Agent 的 ReAct 循环接收为工具执行结果 (Observation)
       │  作为一条 role="tool" 的消息写进父 Agent 的 messages 历史，
       │  父模型看到这行报错，开始思考并自我修正（换工具重试或向用户报告）。

为什么需要异步，哪里需要异步

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

1. 模型层 (src/my_agent_llm) 【已就绪!】
   ├── llm.achat(...) ────────► 异步网络请求
   └── llm.achat_stream(...) ──► 异步生成器 (Async Generator)，吐出增量 StreamChunk

2. 事件层 (my_agent_core/events.py) 【已就绪!】
   ├── MessageUpdate ─────────► 携带增量 Token 文本
   └── ToolExecutionUpdate ───► 携带工具运行中的流式进度

3. 工具层 (my_agent_core/tools/) 【待接入】
   ├── 支持 async def my_tool(...) 异步工具定义
   └── 支持 ToolRegistry.execute_batch 并发执行多个工具

4. Agent 核心层 (my_agent_core/agent.py & loop.py) 【核心落地点】
   ├── 一等公民事件流生成器：Agent.prompt_stream(prompt) -> AsyncIterator[Event]
   ├── 经典便利异步门面：async def run(prompt) -> str | None
   ├── 纯函数微内核驱动：run_agent_loop(...) 无状态异步生成器
   ├── 异步调用 ContextManager.prepare()
   ├── 异步驱动 llm.achat_stream 接收 Token 并发射 MessageUpdate
   └── 异步并发调度 tool_calls 并发射 ToolExecutionUpdate

价值 1：工具并发执行（Parallel Tool Execution）

• 场景：当 Agent 在做代码重构或审查时，大模型经常会一次性输出 5 个文件读取请求（read(“a.py”), read(“b.py”),
  read(“c.py”)…）。

• 价值：同步只能逐个读；异步可以直接用 asyncio.gather 同时并发读取 5 个文件或调用 5 个外部 API，耗时直接从 O(N) 降到
  O(1)。

  价值 2：打字机流式输出与实时事件（Streaming UX）

• 场景：大模型生成一段长篇代码（可能需要 15 秒）。

• 价值：

  • 同步只能等 15 秒后一次性拿到完整文本；
  • 异步通过我们在 events.py 中预留的 MessageUpdate 与 ToolExecutionUpdate，可以做到 Token 级别的毫秒级流式回显，在
    终端或 Web 前端呈现丝滑的打字机动画。

  价值 3：即时可取消性与信号响应（Graceful Cancellation）

• 场景：大模型写出了一个死循环 bash 命令，或者正在进行一个错误的超长推理。

• 价值：

  • 在异步事件循环中，用户按一次 Esc 或 Ctrl+C，框架可以在事件循环中触发 asyncio.CancelledError，毫秒级掐断网络连接
    、杀掉运行中的子进程，并把当前已产生的上下文安全存盘，绝不破坏 Session 树。

  价值 4：多用户并发服务能力（Web / Server 部署）

• 场景：未来如果你把这个 Agent 包装成一个 FastAPI 接口，或者部署到 Web 平台供多人同时使用。

• 价值：
  • 如果是同步代码：一个用户提问（耗时 10 秒），这个 Python 进程线程就被占死，其他所有用户全都在排队等待；
  • 如果是异步代码：单个 Python 进程可以轻松并发服务成百上千个用户同时对话。

工具并发

如何防止工具并发冲突与因果时序倒置

第一道机制：is_parallel_safe 声明与“一票否决”因果保护

1. 要解决的核心问题：防止“并发写冲突”与“因果时序倒置（Causal Inversion）”

假设大模型在一轮推理中，同时发起了 2 个有先后因果关系的工具调用：

• tool_0: write(“config.json”, “port=8080”)（写操作，有副作用）

• tool_1: read(“config.json”)（只读，目的是确认刚才写入的新配置）

  如果粗暴地把只读工具（tool_1）提前拿去并发执行，就会导致 read 先于 write 发生，读出未修改前的旧数据，发生严重的因果时序倒置 Bug！

  2. 机制实现原理：Pi 风格的“一票否决制”

  在工具定义时，每个工具声明自己是否是“并发安全的（只读无副作用）”：

• @tool(is_parallel_safe=True)：如 read、get_weather、db_query（只读）；

• 默认 is_parallel_safe=False：如 write、edit、bash（有写操作/副作用）。

  在 ToolRegistry.execute_batch 执行前，采用一票否决判定：

• 全员只读放行并发：当且仅当批次中的每一个工具均为 is_parallel_safe=True 时，才使用 asyncio.gather 全并发执行；

• 含写全批保序串行：只要批次中包含任何一个有写副作用的工具（或未知工具），整批工具立即放弃并发，严格按照大模型输出的原始先后顺序依次串行执行！

  1 2 3 4 5 6 7 8 9

  场景 A (全只读): [ 0: read(A), 1: read(B), 2: grep(C) ] │ ▼ (全员 is_parallel_safe=True) 【asyncio.gather 全员并发加速 ⚡】 场景 B (含写入): [ 0: write(A), 1: read(A) ] │ ▼ (检测到 write: 一票否决降级) 【严格按 0 ➔ 1 原始因果顺序串行执行 🛡️】

  ────────────────────────────────────────────────────────────────────────────────

  第二道机制：asyncio.gather 并发（极速执行）

  1. 要解决的核心问题：消灭累加的“网络与 I/O 等待耗时”

  传统的同步模式下，读 3 个远程 API 耗时 2s + 2s + 2s = 6s。

  2. 机制实现原理：

  当批次判定为全员只读安全时，使用 asyncio.gather 同时把所有协程扔进事件循环，让操作系统在后台同时发起网络/文件 I/O。耗时取决于最慢的那个，直接从 $O(N)$ 降到 $O(1)$。

  1	全只读批次 ──► asyncio.gather(task_0, task_1, task_2) ──► 多个只读工具在后台同时跑 ⚡

  ────────────────────────────────────────────────────────────────────────────────

  第三道机制：严格保序回填（协议对齐）

  1. 要解决的核心问题：大模型对 tool_call_id 的强顺序依赖

  大模型在发请求时，它心里的顺序是：[0号工具, 1号工具, 2号工具]。

• 在并发执行时，2 号工具可能 0.1 秒就跑完了，而 0 号工具花了 3 秒；

• 如果我们按“谁先跑完谁先写回”，消息列表会变成 [2号结果, 1号结果, 0号结果] → 大模型 API 校验直接报错，或者把 2 号的结果误当成 0 号的结果！

  2. 机制实现原理：“结果索引保序对齐”

  无论工具是并发执行还是串行回退执行，返回的 ToolResult 列表严格与入参 tool_calls 的索引位置完全对齐：

  1 2 3 4

  大模型 tool_calls 列表: [ 0: call_0, 1: call_1, 2: call_2 ] │ ▼ 生成的 results 结果列表: [ res_0, res_1, res_2 ] (索引 100% 完美对应!)

  然后，框架按照这个保序的数组逐条生成 role: "tool" 消息追加到 messages 与 session 中，安全喂给下一轮大模型。

  ────────────────────────────────────────────────────────────────────────────────

  总结工具批调度的决策全景图

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

  大模型返回工具批次: [ tool_0, tool_1, ... ] │ ▼ 【检查是否包含任何写操作/非安全工具？】 / \ [ 是 ] [ 否 ] / \ ▼ ▼ 【一票否决: 串行降级】 【全员并发: asyncio.gather】 严格按 0 ➔ 1 顺序执行 全员只读工具同时发起 I/O ⚡ │ │ └──────────┬──────────┘ │ ▼ 【严格保序回填入库】 与原始 tool_call_id 严格对齐 │ ▼ 安全、合规、零时序倒置地喂回大模型！