15 KiB

Raw Blame History

Pi Agent 架构总览

本文档是对两个参考项目的架构解读，作为后续实现 focus（纯 Rust std + workspaces 版）的参考蓝图。

pi-main (TypeScript, 原版): https://github.com/earendil-works/pi — 设计的源头

pi_agent_rust-main (Rust, 移植): https://github.com/Dicklesworthstone/pi_agent_rust

阅读建议：配合 01-types-and-protocol.md（类型逐行精读）与 02-agent-loop.md（循环逐行精读）一起看。

一、分层架构

Pi 是一个严格的分层架构。agent 层是纯逻辑，不依赖任何 I/O —— 这是整个设计的核心洞察。

┌─────────────────────────────────────────────────────────┐
│  coding-agent 层  (CLI / TUI / 文件工具)                  │  ← 产品层
│    pi-main: packages/coding-agent                         │
│    rust:    src/main.rs + src/interactive.rs + src/cli.rs │
├─────────────────────────────────────────────────────────┤
│  harness 层  (会话持久化 / 上下文压缩 / 系统提示 / 工具注册)  │  ← 基础设施
│    pi-main: packages/agent/src/harness/                   │
│    rust:    src/session.rs + src/compaction.rs            │
├─────────────────────────────────────────────────────────┤
│  agent 层    (消息类型 / agent 循环 / 工具执行 / 状态)       │  ← 纯逻辑（灵魂）
│    pi-main: packages/agent/src/agent-loop.ts + types.ts   │
│    rust:    src/agent.rs（核心循环）+ src/model.rs         │
├─────────────────────────────────────────────────────────┤
│  ai 层       (协议类型 / 流式接口 / provider HTTP 适配)      │  ← 网络
│    pi-main: packages/ai/                                  │
│    rust:    src/provider.rs + src/providers/ + src/sse.rs │
└─────────────────────────────────────────────────────────┘

层	职责	纯 std 可行性
ai 层	多 provider 统一流式 API、SSE 解析、HTTP/TLS	❌ TLS 是硬墙；可抽 `Transport` trait
agent 层	消息类型、agent 循环、工具 trait、状态管理、事件	⚠️ 大部分可行，需手写 JSON
harness 层	会话树持久化(JSONL)、上下文压缩、系统提示模板	✅ 完全可行
coding-agent 层	CLI、文件工具(read/write/edit/bash/grep)	✅ 完全可行（std::fs + std::process）

二、核心数据流（一次完整 run）

用户调用 harness.prompt("做 X")
        │
        ▼
┌─ runAgentLoop ────────────────────────────────────────────┐
│  emit(agent_start)                                        │
│  把 user message 加入 context.messages                    │
│                                                            │
│  ┌─ runLoop（双层循环）──────────────────────────────────┐  │
│  │  外层 while: follow-up 循环                            │  │
│  │    内层 while: steering + 工具循环                     │  │
│  │      │                                                 │  │
│  │      │  (1) streamAssistantResponse ─── 边界转换 ──┐   │  │
│  │      │      transformContext (AgentMsg→AgentMsg)   │   │  │
│  │      │      convertToLlm    (AgentMsg→Message[])   │   │  │
│  │      │      streamFn(model, llmContext) ───────────┼──▶│  │ provider
│  │      │      for await event: 累积流式 assistant     │   │  │ (SSE→Event)
│  │      │      return finalMessage ◀──────────────────┘   │  │
│  │      │                                                 │  │
│  │      │  (2) 提取 toolCalls                            │  │
│  │      │      validateToolArguments                     │  │
│  │      │      beforeToolCall hook (可 block)            │  │
│  │      │      execute (sequential | parallel)           │  │
│  │      │      afterToolCall hook (可覆盖结果)            │  │
│  │      │      → ToolResultMessage[]                     │  │
│  │      │                                                 │  │
│  │      │  (3) prepareNextTurn (harness 每轮重建上下文)   │  │
│  │      │      shouldStopAfterTurn?                       │  │
│  │      │      getSteeringMessages() 再 poll             │  │
│  │      └─────────────────────────────────────────────┘  │  │
│  │    getFollowUpMessages()                              │  │
│  └───────────────────────────────────────────────────────┘  │
│  emit(agent_end)                                          │
└────────────────────────────────────────────────────────────┘

四个关键设计点（详见 02-agent-loop.md）：

边界处转换：AgentMessage[] 永远不直接发给 LLM，convertToLlm 是唯一翻译点。
流式消息是"活"的：流式过程中 assistant 消息就躺在 context.messages 末尾，随 delta 原地修改。
prepareNextTurn 重建：harness 每 turn 后从 session 重载消息、重算系统提示、重建工具列表。
Steering vs Follow-up：steering 在 agent 工作时插入（内层 poll），follow-up 在 agent 要停下时检查（外层 poll）。

三、两个类型宇宙

Pi 区分 LLM Message（provider 说的语言）和 AgentMessage（可扩展的 transcript）：

// LLM Message（pi-ai/types.ts）—— 三种角色，role 做顶层分发
Message = UserMessage | AssistantMessage | ToolResultMessage

// AgentMessage（agent/types.ts）—— 在 Message 之上扩展自定义角色
type AgentMessage = Message | CustomAgentMessages[keyof CustomAgentMessages]
//                                       ^^^^^^^^^^^^^^^^^^^^
//                        coding-agent 通过 declaration merging 加:
//                        bashExecution / custom / branchSummary / compactionSummary

自定义消息不直接发给 LLM，由 convertToLlm 在边界转成 Message[]。这让 transcript 能承载 UI 专属信息而不污染 LLM 上下文。

内容块（ContentBlock）的 type 做内部分发：

块类型	出现在	说明
`TextContent`	user / assistant / toolResult	普通文本
`ThinkingContent`	assistant only	推理过程（带签名防篡改）
`ImageContent`	user / toolResult（非 assistant）	base64 图片
`ToolCall`	assistant only	工具调用请求

详细类型定义见 01-types-and-protocol.md。

四、流式模型

Pi 没有用 EventEmitter 或回调，而是手搓了 EventStream<T, R>：

class EventStream<T, R> implements AsyncIterable<T> {
    push(event: T)         // 生产者 API
    end(result?: R)
    result(): Promise<R>   // 独立的"最终结果" promise
    [Symbol.asyncIterator]()  // for await 消费
}

同一个对象既是可迭代流（边收 delta），又能 await response.result()（拿完整消息）。

流式事件协议 AssistantMessageEvent —— 每个事件都带完整 partial 快照，消费者无需自己累积：

| { type: "start"; partial: AssistantMessage }
| { type: "text_delta";  contentIndex; delta; partial }
| { type: "toolcall_delta"; contentIndex; delta; partial }
| { type: "done";  reason: "stop"|"length"|"toolUse"; message }
| { type: "error"; reason: "aborted"|"error"; error }

五、工具抽象

interface AgentTool<TParameters, TDetails> extends Tool {
    label: string
    prepareArguments?: (args: unknown) => Static<TParameters>   // 预验证 shim
    execute: (
        toolCallId: string,
        params: Static<TParameters>,
        signal?: AbortSignal,
        onUpdate?: (partialResult: AgentToolResult<TDetails>) => void,  // 流式更新
    ) => Promise<AgentToolResult<TDetails>>
    executionMode?: "sequential" | "parallel"   // per-tool 覆盖
}

interface AgentToolResult<T> {
    content: (TextContent | ImageContent)[]   // 模型看到的
    details: T                                 // UI/日志用的（不发给模型）
    addedToolNames?: string[]                  // 延迟加载工具
    terminate?: boolean                        // batch 级停止提示
}

关键设计：content（模型看到）和 details（UI 看）分离。terminate 是 batch 级的——只有当批内所有结果都设 terminate: true 才真正停止（一个"继续"结果可否决停止）。

两种执行模式（agent-loop.ts:413-556）：

sequential：prepare→execute→finalize 一个一个来
parallel：所有 prepare 顺序完成，execute 并发（Promise.all），tool_execution_end 按完成顺序发射，但 ToolResultMessage 按 assistant 源顺序发射

Rust 版多了 TS 没有的 ToolEffects 位标志调度（tools.rs:37-155）：每个工具声明 READ|WRITE|NETWORK|PROCESS，plan_tool_effect_batches 把兼容工具分批并发。

六、Provider 实现（以 Anthropic 为例）

streamSimple → stream。stream 职责：构造 HTTP 请求 → 解析 SSE → 翻译成 AssistantMessageEvent。

Anthropic SSE → pi-ai 事件映射（anthropic-messages.ts:559-732）：

Anthropic SSE	pi-ai 动作
`message_start`	种入 `responseId` + 初始 `usage`
`content_block_start` (text/thinking/tool_use)	push 对应空块；emit `*_start`
`content_block_delta` `text_delta`/`thinking_delta`	累加到块；emit `*_delta`
`content_block_delta` `input_json_delta`	`partialJson += ...`; `arguments = parseStreamingJson(partialJson)`
`content_block_stop`	toolCall: 解析最终 arguments，删除 scratch 字段
`message_delta`	`mapStopReason`（end_turn→stop, max_tokens→length, tool_use→toolUse）

七、Harness 层（可选基础设施）

功能	TS 位置	核心设计
系统提示	`harness/system-prompt.ts`	每轮重建（可注入 cwd/git 等动态上下文）
上下文压缩	`harness/compaction/compaction.ts`	`shouldCompact` → `findCutPoint` → `generateSummary`（滚动总结，不从头重算）
会话持久化	`harness/session/`	树结构（非列表）+ JSONL 追加写
工具注册	`harness/`	ToolRegistry + before/afterToolCall hooks

会话存储的关键设计：

是树不是列表：每个 entry 有 id + parentId，getBranch(leafId) 返回根到叶路径——会话分叉/版本管理的基础
JSONL 追加写：每条一行 JSON，永不修改，appendEntry 就是 appendFile(line + "\n")
目录隔离：encodeCwd("/home/foo") → --home-foo--

八、TS → Rust 移植对照

设计点	TS (pi-main)	Rust (pi_agent_rust)
消息类型	interface + union	enum + serde tag（字段名/标签完全对应）
流式消息	直接替换	`Arc<AssistantMessage>` + `Arc::make_mut` COW
流式事件	单一 `AssistantMessageEvent`	拆成内部 `StreamEvent`（不序列化）+ 外部 `AssistantMessageEvent`（序列化）
工具执行	per-tool `executionMode`	额外加 `ToolEffects` 位标志 + 批次调度
异步运行时	Node 事件循环	自研 `asupersync`（能力化 `Cx` + 预算 + TLS + HTTP/1.1），刻意避开 tokio/reqwest
取消机制	`AbortSignal` (DOM)	`asupersync::sync::Notify` + `AtomicBool` + `Cx::checkpoint()`

Rust 版最值得注意的是：用自研 asupersync 完全替代 tokio + reqwest + rustls。作者追求极简依赖，但即便如此，serde/serde_json/futures/async-trait 仍结构性无法省掉——这印证了"agent 层是纯逻辑但需要 JSON 序列化"的判断。

九、"纯 std" 可行性总结（针对 focus 实现）

组件	纯 std?	说明
agent 循环逻辑	✅	纯状态机调度，`for` 循环 + channel
消息类型	⚠️	需手写 JSON 编解码（~300 行极简版），或认栽用 serde
工具 trait	✅	Rust 1.75+ 原生 async trait，无需 async-trait
流式事件	✅	用 `mpsc::Receiver` 替代 `Stream`
会话持久化	✅	JSONL + std::fs
文件工具	✅	std::fs + std::process::Command
上下文压缩	✅	纯算法
provider 网络	❌	TLS 是硬墙 → 抽 `Transport` trait，实现用 curl 子进程

结论：focus 的 agent 层 + harness 层 + 工具层可以做到几乎纯 std；网络层必须留 trait 边界。建议 workspace 拆分：

focus/
├── Cargo.toml              # [workspace]
└── crates/
    ├── focus-core/         # agent 层：类型、循环、工具 trait（纯 std + 手写 JSON）
    ├── focus-transport/    # Transport trait 定义（纯 std）
    ├── focus-providers/    # provider 实现（依赖 transport，用 curl 子进程）
    ├── focus-harness/      # 会话/压缩/系统提示（纯 std）
    ├── focus-tools/        # 文件工具（纯 std）
    └── focus-cli/          # 入口（组合上述）

十、学习路线

第一阶段：吃透核心（按顺序）

01-types-and-protocol.md — types.ts 逐行精读（建立词汇表）
02-agent-loop.md — agent-loop.ts 核心循环逐行精读（Pi 的心脏）
pi-main/packages/ai/src/utils/event-stream.ts — EventStream 实现

第二阶段：真实 provider

pi-main/packages/ai/src/api/anthropic-messages.ts — stream 函数的 SSE 翻译

第三阶段：harness（可选）

pi-main/packages/agent/src/harness/agent-harness.ts — prepareNextTurn + hook 系统
pi-main/packages/agent/src/harness/compaction/compaction.ts — 上下文压缩
pi-main/packages/agent/src/harness/session/ — 会话树

Rust 版对照阅读（按需 grep，不通读）

src/model.rs — 对照 TS 类型
src/agent.rs:1453 run_loop — 对照双层循环
src/agent.rs:1960 stream_assistant_response — 对照边界转换
src/tools.rs:158-191 Tool trait — 对照 AgentTool
src/sse.rs — 可学习的纯 SSE 解析（除 SseStream 的 futures 部分）
src/agent_cx.rs — 能力化上下文（180 行，可通读）

避开（产品化代码，非核心）：extensions*.rs(2MB+ WASM/JS)、main.rs(287KB CLI)、interactive.rs(TUI)

15 KiB Raw Blame History Unescape Escape