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 Monorepo
架构设计：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| packages/my-coding-agent ← 【第 3 层：产品与应用层】（开箱即用代码助手）
│ • CodingAgent 门面组装
│ • 4 大工作区安全文件工具 (read / write / edit / bash)
│ • FileMutationQueue 单文件细粒度并发互斥锁
│ • 原生异步 MCP 客户端扩展 (AsyncExitStack)
▼
packages/my-agent-core ← 【第 2 层：框架核心层】（通用 Agent 运行时微内核）
│ • 纯函数无状态微内核 run_agent_loop (loop.py)
│ • 轻量 Harness 宿主外壳 Agent (支持 prompt_stream 原生事件流)
│ • 对话转录本自愈与断头保护引擎 (tool_history.py)
│ • 模块化会话存储子系统 session/ (9种多态实体、纯追加持久化)
│ • 12 个生命周期事件与五大决策拦截点 (events & hooks)
│ • 统一 Todo 看板与 BackgroundRunner 进程树强杀
│ • 4 层 Cheap-first 上下文压缩管线 (L3➔L1➔L2➔L4)
│ • 声明式 Skills、Subagents 多智能体与 Plugin 系统
▼
packages/my-agent-llm ← 【第 1 层：模型边界层】（底层网络与协议转换地基）
• 统一 LLM 门面 (chat / stream / achat / achat_stream)
• 三大 Provider (OpenAI / DeepSeek / Anthropic)
• 流式 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%
离线测试驱动：不依赖任何笨重第三方框架，全套 378 项离线单元测试
20 秒跑完，每个模块接缝分明、立即可用！