my-pi-agent--架构设计

发表于 更新于 分类于 ， 阅读次数：

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 生命周期的关键切面上暴露钩子。

image-20260801154819449

功能维度 LangChain Middleware 机制 Pi Extension 生命钩子 (Hooks) 共同解决的场景
Agent 启动入口 before_agent input, before_agent_start 拦截用户输入、预处理 Prompt
上下文修改 before_model, modify_model_request context, before_provider_request 上下文裁剪（Compaction）、RAG 动态注入、Payload 替换
网络/Header 控制 Client Transport Interceptor before_provider_headers, after_provider_response 动态切换 API Key、注入自定义 Header、监听响应头
工具调用拦截 after_model / HumanInTheLoopMiddleware tool_call (可直接返回 block) 人工审批（HITL）、安全阻断拦截
工具结果处理 ContextEditingMiddleware / Tool Interceptor tool_result 返回值脱敏（PII Redaction）、长输出截断
自动摘要与压缩 SummarizationMiddleware session_before_compact, session_compact 对话历史自动压缩

概念解析：Trace与Turn

Trace（一次完整运行）

一个 Trace 是从用户按下回车、到 Agent 彻底停下来、发出 agent_end 事件的整个过程。一个 Trace 包含多个 Turn。

1
2
3
4
5
6
7
一个 Trace（一次 agent_start 到 agent_end）
│
├── Turn 1：调模型 → 模型返回 toolUse（要读文件）→ 执行 read 工具
│
├── Turn 2：带着工具结果再调模型 → 模型返回 toolUse（还要改文件）→ 执行 edit 工具
│
└── Turn 3：带着工具结果再调模型 → 模型返回 stop（改好了，没有工具调用）→ agent_end

Turn（一个轮次）

一个 Turn 的定义非常精确：一次模型调用 + 这次调用触发的所有工具执行。

每个 Turn 由一对 turn_start 和 turn_end 事件包裹。关键点：一个 Turn 只有一次模型调用。 模型返回了 toolUse → 执行那批工具 → 发送 turn_end → 这个 Turn 就结束了。把工具结果喂回去再调模型，那是下一个 Turn。

所以 Trace 和 Turn 的关系就是

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Trace（一次完整运行）
│  agent_start
│
├── Turn 1
│   │  turn_start
│   ├── 调模型 → toolUse → 执行工具（read + grep）
│   │  turn_end
│   │
├── Turn 2
│   │  turn_start
│   ├── 调模型 → toolUse → 执行工具（edit）
│   │  turn_end
│   │
├── Turn 3
│   │  turn_start
│   ├── 调模型 → stop → 没有工具
│   │  turn_end
│   │
│  agent_end

注意：首轮 Turn 的 turn_start 是在 runAgentLoop() 入口就发出的，然后 runLoop() 内用 firstTurn 标志跳过首圈的 turn_start，避免重复。

流程全景

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
你按下回车："帮我读一下 src/main.ts"
│
│  ① 你的输入变成一条消息
│
UserMessage { role: "user", content: "帮我读一下 src/main.ts" }
│
│  ② 进入循环（agentLoop 入口）—— agent_start（一个 Trace 开始了）
│
└── runLoop()
    │
    │  ③ 消息转换（AgentMessage → LLM 认识的 Message）
    │
    │  ┌── Turn 1 ──────────────────────────────────────────┐
    │  │  turn_start                                         │
    │  │  ④ 调用 Model（每 Turn 仅一次模型调用）               │
    │  │  streamSimple(model, { systemPrompt, messages })    │
    │  │       ↓ 逐 token 流式返回                            │
    │  │  AssistantMessage {                                  │
    │  │      content: [ ..., ToolCall { name: "read", ... } ],│
    │  │      stopReason: "toolUse"  ← 有工具调用，继续转      │
    │  │  }                                                  │
    │  │  ⑤ 执行 Tool（工具的五步管道，详见第5章）              │
    │  │  ToolResultMessage { content: [{ text: "文件内容" }] }│
    │  │  turn_end                                            │
    │  └─────────────────────────────────────────────────────┘
    │
    │  循环判断：stopReason 是 toolUse → hasMoreToolCalls = true → 继续
    │
    │  ┌── Turn 2 ──────────────────────────────────────────┐
    │  │  turn_start                                         │
    │  │  ⑥ 第二次调用 Model（工具结果已追加到消息列表）         │
    │  │  streamSimple(model, { messages: [..., toolResult] })│
    │  │       ↓ 模型看到文件内容，开始解释                      │
    │  │  AssistantMessage {                                  │
    │  │      content: [ TextContent { text: "这个文件..." } ],│
    │  │      stopReason: "stop"  ← 没有工具调用，准备停        │
    │  │  }                                                  │
    │  │  turn_end                                            │
    │  └─────────────────────────────────────────────────────┘
    │
    │  循环判断：hasMoreToolCalls = false，pendingMessages 为空
    │  → 内层循环退出
    │  → 外层循环检查 followUp → 空 → 外层循环退出
    │
    └── agent_end（一个 Trace 结束，共 2 个 Turn）

项目结构设计

当前 my-pi-agent 严格遵循清晰解耦的三层 Python 架构设计（Tau 式单包拓扑与 hatchling 导出）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
src/my_coding_agent               ← 【第 3 层：产品与应用层】（开箱即用代码助手）
        │                           • CodingAgent 门面组装
        │                           • 7 大工作区安全编码工具 (read/write/edit/bash/grep/find/ls)
        │                           • FileMutationQueue 单文件细粒度并发互斥锁
        │                           • 原生异步 MCP 客户端扩展 (AsyncExitStack)
        │                           • PermissionGate 业务权限审查门禁 (Accept-on-Diff)
        ▼
src/my_agent_core                 ← 【第 2 层：框架核心层】（通用 Agent 运行时微内核）
        │                           • 纯函数无状态微内核 run_agent_loop (loop.py)
        │                           • 轻量 Harness 宿主外壳 Agent (支持 prompt_stream 原生事件流)
        │                           • 对话转录本自愈与断头保护引擎 (tool_history.py)
        │                           • 模块化会话存储子系统 session/ (9种多态实体、纯追加持久化)
        │                           • 14 个生命周期事件与五大决策拦截点 (events & hooks)
        │                           • 统一 Todo 看板与 BackgroundRunner 进程树强杀
        │                           • 离散 Epoch 块级压缩管线 (80% 水位线门控 + L3 磁盘溢出 + L4 结构化摘要)
        │                           • 声明式 Skills、Subagents 多智能体与 Plugin 系统
        ▼
src/my_agent_llm                  ← 【第 1 层：模型边界层】（底层网络与协议转换地基）
                                    • 统一 LLM 门面 (chat / stream / achat / achat_stream)
                                    • 多 Provider 直连 (OpenAI / DeepSeek / Anthropic / Antigravity)
                                    • 流式 Tool Calls 增量聚合与 Token Usage 锚定

项目架构

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
┌─────────────────────────────────────────────────────────────────────────┐
│ 【1. 宿主外壳 AgentHarness】 (agent.py 瘦身 260+ 行)                    │
│   • 纯粹的状态与队列容器，对外暴露极窄接口                              │
│   • 一等公民事件流：async def prompt_stream() -> AsyncIterator[Event]   │
│   • 观察者订阅 API：subscribe(listener) -> unsubscribe 句柄             │
│   • 经典门面：run() 退化为仅 10 行消费 prompt_stream 的便利包装        │
└────────────────────────────┬────────────────────────────────────────────┘
                             │ 驱动
                             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│ 【2. 纯函数无状态微内核】 (loop.py: run_agent_loop)                     │
│   • 纯异步生成器，与类状态彻底解耦                                      │
│   • 内层负责 ReAct 微观步骤 + steering 动态转向                         │
│   • 外层负责宏观任务流转 + follow-up 自动收割                           │
│   • CancellationToken 协作式取消，成对发射标准事件流                    │
└──────────────┬──────────────────────────────────────────┬───────────────┘
               │ 前置清洗                                 │ 数据持久化
               ▼                                          ▼
┌──────────────────────────────┐ ┌────────────────────────────────────────┐
│ 【3. 转录本自愈与防400引擎】 │ │ 【4. 模块化会话存储子系统】 (session/) │
│   • _provider_context 清洗   │ │  • entries.py : 9 种多态 Pydantic v2   │
│     剥离空中断失败轮次       │ │  • tree.py    : 纯内存 DAG 算法 (防环) │
│   • tool_history.py 自愈     │ │  • memory.py  : SessionState 纯函数折叠│
│     三阶段状态机合成中断     │ │  • storage.py : 只追加纯异步协议+内存驱动│
│     彻底消灭断头 API 400 死锁│ │  • jsonl.py   : 追加存储、文件锁与自愈 │
└──────────────────────────────┘ └────────────────────────────────────────┘

为什么使用ts而不是py

1. 行业现状：为什么标杆项目（Pi、Claude Code）优先选择 TS？

在当前的 AI Agent 工业界，像 Pi（earendil-works/pi）和 Claude Code 官方首先选择 TypeScript / Node.js，核心原因在于：

  1. 天然的单线程异步事件循环：V8 引擎和 Node.js 原生基于事件循环，没有 Python GIL（全局解释器锁）的历史包袱，做异步 I/O、流式打字机和并发工具调度非常顺手；
  2. Web 与终端生态丰富：TypeScript 拥有成熟的前端组件生态，方便直接在同一个语言生态下构建跨平台终端（Ink/TUI）或 Web 界面；
  3. 强类型元数据（TypeScript 类型系统）：其结构化类型系统在编写复杂的泛型管道和事件分发时极为灵活。

2. 我们的选择：为什么要在 Python 中从零手搓出相同水准的 Agent？

尽管 TypeScript 很流行，但在真实的 AI 研发、数据分析与算法工程界，Python 依然是绝对无可撼动的“第一公民”语言。

市面上大多数 Python Agent 框架（如早期的 LangChain 等）充斥着重型抽象与黑盒嵌套，一旦遇到并发、流式截断或会话分叉就束手无策。我们发起 my-pi-agent 的使命，正是为了证明：

利用现代 Python 3.11+ 的原生异步生态（asyncio + Pydantic v2 + 结构化多态联合体），完全可以从零构建出一套与 Pi / Tau 具有同等甚至更高性能、架构优雅、零过度设计的纯粹 Agent 运行时！

  • 纯原生异步（Native Asyncio）：消除传统 Python 线程锁（GIL）死锁风险，通过单线程协程与操作系统内核 I/O 多路复用（IOCP / epoll），轻松调度上百个并发后台命令；
  • 微内核纯函数化：将 ReAct 循环抽离为无状态异步生成器 run_agent_loop，事件流作为一等公民，外部消费流畅如丝；
  • 工业级自愈与防僵尸：自研 tool_history.py 三阶段状态机消除断头 400 校验死锁，全平台子进程树递归强杀（taskkill /F /T 与 killpg）杜绝孤儿进程；
  • 全量 100% 离线测试驱动：不依赖任何笨重第三方框架，全套 814 项离线单元测试并发秒级跑完，每个模块接缝分明、立即可用！

架构思考：为什么 Pi 坚持 “No Background Bash”？（对比 Claude Code 的后台陷阱）

在 Coding Agent 的执行环境与工具链设计中，长期运行的命令（如前端 Dev Server、长耗时测试套件、持续构建监控或交互式 REPL 调试器）是开发者的家常便饭。

主流商业 Agent（典型如 Anthropic 官方的 Claude Code，简称 CC）选择在框架内部自建一套后台任务管理系统（内置 background bash）。然而，Pi 的作者 Mario Zechner 却在设计中旗帜鲜明地坚持 “No background bash”，并强调 “Bash is all you need”。

这一取舍背后蕴含着深刻的 Agent 框架设计哲学。

1. Claude Code 内部后台机制的 4 大致命缺陷

在 Agent 内部强行用语言层代码（如 Node.js / Python 子进程池）造一个“微型操作系统进程管理器”，带来了严重的工程反噬：

① 上下文压缩后的“失忆症”与僵尸进程（The Amnesia & Zombie Problem）

大模型在长程任务中必然会触发上下文压缩（Context Compaction）。

  • 一旦上下文被截断或压缩，模型会彻底忘记自己之前启动过哪些后台进程；
  • 框架没有为 Agent 提供独立的外置观察工具，导致 Agent 既不知道有进程在跑，也不知道它们的 PID。
  • 后果：本地服务的 3000、8080 等端口被持续霸占，CPU 持续被占用，成为无人认领的“孤儿僵尸进程”。直到下一次执行报错 Port already in use，用户只能手动去终端 lsof -i :3000 并执行 kill -9。

② 黑盒陷阱与极差的可观测性（Poor Observability）

在 Claude Code 的内置后台执行中，进程输出被吞入框架底层的缓冲区：

  • 终端界面只有一个转动的 Spinner，人类无法像在普通终端里那样自由滚动查看实时流式日志；
  • 只有进程彻底崩溃或异常退出时，它才会吐出最后几行截断日志；如果服务在中间假死或输出变慢，外部根本无从得知现场状态。

③ 缺乏实时交互通道，极易静默死锁（Interactive Stdin Deadlock）

许多命令在执行途中会临时弹出终端交互式确认（例如 Overwrite file? [y/N] 或 Run database migration? (yes/no)）。

  • Agent 内部后台子进程对实时标准输入（stdin）的管道交互支持极度生硬；
  • 只要遇到需要用户按键的提示，后台任务便会陷入静默死锁（Hang），用户和 Agent 都在干等，直到超时崩溃。

④ 彻底切断了“人机协同调试（Co-debugging）”的可能

  • 进程完全被闭门锁在 Agent 的宿主框架里，人类开发者无法切入现场。
  • 如果 Agent 启动了一个崩溃的后端服务或 C/C++ 调试器（LLDB/GDB），人类无法把键盘介入该进程去单步排查断点，彻底失去了掌控力与协同体验。

2. Pi 与 Unix 哲学的解法：拥抱 tmux

Mario 强调：不要在 Agent 框架里重新发明轮子，Unix 早就有经过数十年工业检验的终极方案——tmux。

1
2
3
4
5
【Claude Code 模式：框架内造轮子】
  Agent 内部进程池 ──► Background Bash ──► 黑盒缓冲区 ──► 上下文压缩后失忆、死锁

【Pi 模式：回归 Unix 哲学组合】
  Agent ──► 调用普通同步 Bash ──► 驱动系统级 tmux ──► 极致透明、人机协同、永不失联

借助 tmux，所有难题迎刃而解：

对比维度 Claude Code 内置后台模式 Pi 倡导的 Unix / tmux 方案
可观测性 弱，内部缓冲，难见现场 极强：人类随时敲 tmux attach 进入全屏查看实时全彩输出
状态持久化 易随上下文压缩（Compaction）丢失 极稳：Agent 随时可用 tmux list-sessions 查明当前活跃会话
人机协同 无法切入调试现场 完美协同：人类随时切入会话与 AI 共享终端、共同单步调试（Co-debugging）
交互能力 遇到交互确认容易死锁 原生支持：通过 tmux send-keys 传递输入，完全支持 REPL 与交互式确认
框架代码包袱 极重（需实现缓冲、心跳、强杀） 零包袱：Agent 框架仅需一个极简的同步 bash 工具

3. 对 my-pi-agent 架构设计的指导启示

在构建我们自己的 Python 版 my-pi-agent 时，必须汲取这一核心设计教训：

  1. 核心工具保持纯粹与同步确定性：
    Agent 内置的 bash 工具应当是确定、无状态、同步退出的。命令跑完即返回状态码与结果，杜绝在 Agent 内核中塞入半吊子的后台轮询管道。
  2. 拒绝安全剧场与过度抽象：
    当需要长驻服务、后台构建或 REPL 调试时，通过标准工具链（tmux 或后台脚本进程树管理）来组合解决，而不是在 Agent 框架内堆砌脆弱的胶水代码。
  3. 可观测性是一等公民：
    任何不能让人类一眼看到完整执行现场的设计，最终都会沦为难以调试的技术负债。将控制权与可见性留给标准终端环境，才是构建高可靠 Agent 框架的康庄大道。

参考博客

我所学到的是如何构建一个具有自己观点且代码简洁的智能体。 —- What I learned building an opinionated and minimal coding agent