拆解 10 万星 AI Agent 项目：值得借鉴的软件工程实践

作者：袖梨 2026-09-20

AI Agent 项目往往容易被模型能力和功能演示抢走注意力，但决定工具能否持续演进的，仍是基础的软件工程质量。开源项目 pi 提供了一个很有代表性的样本：从正交分层、事件建模到原子任务边界、测试隔离与供应链防护，每项选择都在回应真实的维护成本和运行风险。

pi（earendil-works/pi）是一个开源的 AI Agent 工具箱，包含统一的多模型 LLM 接口、Agent 运行时、终端 UI 库，以及基于它们打造的编程助手 CLI。这个项目本身已经小有名气，GitHub 上有 10 万+ star，社区里甚至出现了 Rust、Go、Python 的独立移植版本。

抛开「AI」这个话题热度，这个项目真正值得学的是一系列扎实的工程决策，而且大多数跟 AI 没有直接关系。下面每个主题都配一段示意代码（不是原项目源码，是根据它的设计思路写的简化示例），帮助把抽象的原则落到具体写法上。

一、把系统拆成「薄」而「正交」的层

不要把所有逻辑塞进一个大工具里，而是按职责拆包：协议层（和 LLM 说话）、运行时层（工具调用、状态管理）、界面层（终端渲染）、产品层（把前三层组装成 CLI）。

拆分之后，协议层要足够「薄」，薄到可以只靠一个字段做路由，而不是为每个厂商单独写一套代码：

// pi-ai: 用一个 api 字段分发到具体 provider,
// 而不是为每个「品牌」单独写适配器
async function streamSimple(model: Model, context: Context, options: StreamOptions) {
  switch (model.api) {
    case "anthropic-messages":
      return new AnthropicProvider().stream(model, context, options);
    case "openai-completions":
      // OpenRouter / Together / Groq / DeepSeek / xAI 等几十个「品牌」
      // 全部走这一条分支,只是 model.baseUrl 不同
      return new OpenAiProvider().stream(model, context, options);
    case "google-generative-ai":
      return new GoogleProvider().stream(model, context, options);
  }
}

关键点：先去分辨真正不同的协议形态有几种（这里只有 3 种），而不是按表面上的品牌数量去写适配器。少写几十倍的重复代码。

拆分之后每一层都可以单独被复用、单独测试、单独维护版本号。事实也证明了这一点——pi-ai 这个「薄」的协议层，已经被完整移植成了 Rust 版本，复用同一套消息/工具类型定义。一套接口设计能不能脱离原来的编程语言存活，是检验它是不是「真的解耦」的一个很实用的标准。

二、把「运行过程」建模成一棵可订阅的事件树

Agent 执行一次任务的过程，被建模成固定的三段式事件：start → update*（可重复）→ end，层层嵌套。

type AgentEvent =
  | { type: "run_start"; runId: string }
  | { type: "turn_start" }
  | { type: "message_start"; role: "assistant" }
  | { type: "message_update"; delta: string }      // 可以触发很多次
  | { type: "message_end" }
  | { type: "tool_start"; toolName: string; args: unknown }
  | { type: "tool_update"; progress: unknown }      // 可以触发很多次
  | { type: "tool_end"; result: unknown }
  | { type: "turn_end" }
  | { type: "run_end" };

// 任何一方——终端界面、Web 界面、日志系统——只需要订阅事件流,
// 就能重建完整状态,不需要理解 Agent 内部实现
function onEvent(event: AgentEvent) {
  switch (event.type) {
    case "message_update":
      ui.appendToCurrentBubble(event.delta);
      break;
    case "tool_start":
      ui.showSpinner(event.toolName);
      break;
    case "tool_end":
      ui.hideSpinner();
      ui.showToolResult(event.result);
      break;
    // ...
  }
}

这就是「事件溯源」思路在 UI 状态同步上的应用，能用在任何需要多端同步、断线重连恢复状态的系统上——不只是 AI。

三、抢占/排队：先定「不可分割的操作单元」，再设计规则

用户在 Agent 正在执行工具调用时发来新消息，该怎么处理？规则很克制：绝不打断正在执行的工具调用，只在「一轮模型调用结束后」才检查排队消息。

class AgentLoop {
  private steeringQueue: UserMessage[] = [];

  async runTurn() {
    const assistantMsg = await this.callModel();

    if (assistantMsg.toolCalls.length > 0) {
      // 工具调用批次是原子单元,执行期间绝不插队
      for (const call of assistantMsg.toolCalls) {
        await this.executeTool(call);       // 不会被打断
      }
    }

    this.emit("turn_end");

    // 只有到了这里——模型调用之间的安全边界——才把排队消息
    // 当作新的 user message 追加进去
    const queued = this.steeringQueue.splice(0);
    if (queued.length > 0) {
      this.context.messages.push(...queued.map(toUserMessage));
    }
  }

  // 用户发消息时,不会直接打断当前执行,而是先入队
  onUserMessage(msg: UserMessage) {
    this.steeringQueue.push(msg);
  }
}

原因很直接：assistant 发起 tool_use → tool_result 必须配对完整，中途硬插东西会产生「孤儿」工具结果，让模型困惑甚至报错。先想清楚系统里「不可分割的最小操作单元」是什么，再围绕它设计抢占/排队规则，而不是反过来。这条原则用在后台任务系统、审批流程、协作编辑器上同样成立。

四、测试：把「请求对不对」和「对方服务器给不给对结果」分开测

// ✅ 好的做法:用本地 mock server 验证「我发的请求格式对不对」
// 完全不连真实的模型服务,零成本、零网络抖动
test("anthropic provider builds correct request payload", async () => {
  const mockServer = createMockHttpServer((req) => {
    expect(req.body.model).toBe("claude-sonnet-5");
    expect(req.body.messages).toEqual([{ role: "user", content: "hi" }]);
    return mockSSEResponse(["Hello", " there"]);
  });

  const events = await collectEvents(
    new AnthropicProvider(mockServer.url).stream(model, context, options)
  );
  expect(events).toContainEqual({ type: "message_update", delta: "Hello" });
});

// ? 真正连线上服务的 E2E 测试单独隔离、默认不跑
test.skipIf(!process.env.LIVE_API_KEY)(
  "real anthropic API smoke test",
  async () => { /* ... */ }
);

前者本地秒跑、可以跑几百个用例覆盖各种边界情况；后者留给少量、昂贵、按需触发的验证。很多团队图省事把两者混在一起写集成测试，结果 CI 又慢又不稳定又烧钱。

「我发出去的东西对不对」和「对方服务愿不愿意正常响应」是两类完全不同的风险，应该用两套完全不同代价的测试去覆盖——这适用于任何依赖外部 API（尤其是收费、限流、不稳定的 API）的项目。

五、供应链安全：依赖变更当代码变更去审

不只是态度，是几条配置层面的具体做法：

# .npmrc
save-exact=true       # 直接依赖锁死精确版本,不用 ^/~ 范围
min-release-age=2     # 包发布不满 2 天不允许安装,避免拉到刚发布、
                       # 还没被社区验证过的可疑版本

# CI 里
npm ci --ignore-scripts          # 安装时不执行任何生命周期脚本
npm audit --omit=dev             # 定期扫描已知漏洞
npm audit signatures --omit=dev  # 校验包签名

# pre-commit hook 思路:lockfile 被当成唯一真相
# 未经允许的 lockfile 变动直接拦截
if git diff --cached --name-only | grep -q package-lock.json; then
  if [ -z "$PI_ALLOW_LOCKFILE_CHANGE" ]; then
    echo "lockfile 变动需要显式确认,设置 PI_ALLOW_LOCKFILE_CHANGE=1"
    exit 1
  fi
fi

一个能执行 shell 命令、改文件的工具，依赖链一旦被污染，攻击面就是用户整台机器。「依赖升级当成正常代码改动一样走审查」，是任何被广泛安装的工具/库都应该有的态度。

六、诚实地承认没做安全沙箱

// 核心里不存在这样的东西:
// function checkPermission(action: "read_file" | "exec" | "network"): boolean

// 官方文档直接说明:核心默认以启动它的用户权限运行,
// 需要更强边界,请自己用容器/沙箱包裹整个进程:
//
//   docker run --rm -v $(pwd):/workspace --network=none pi-image pi
//
// 而不是在核心里塞一套自己都测不完整的伪权限系统

很多项目为了「看起来安全」，会在核心里塞一套自己都测不完整的权限系统，结果给用户一种虚假的安全感。「没做的事情诚实地说没做，并指明正确的做法」，比「看起来做了但其实没做全」要专业得多。

七、成本（Token）是运行时的一等公民

interface CompactionResult {
  summary: string;
  tokensBefore: number;
  tokensAfterEstimate: number;  // 压缩效果直接算进结果里,
                                  // 不是事后靠日志拼凑
}

interface Usage {
  inputTokens: number;
  outputTokens: number;
  cachedTokens: number;   // 缓存命中的部分单独统计
  cacheHitCostSaved: number;
}

// 调用时用 session ID 做缓存键,复用命中的 prompt cache
const response = await provider.stream(model, context, {
  ...options,
  promptCacheKey: session.id,
});

成本核算和实际状态变化绑在一起产生，而不是靠外部埋点去凑，数据的准确性和及时性会好很多。任何按量计费或对成本敏感的系统（不止 AI，云资源调度、批处理任务都类似），把成本核算前置进状态模型本身，收益会很明显。

八、边界情况：该报错就报错，别悄悄生成空壳结果

async function compactSession(messages: Message[]): Promise<CompactionResult> {
  const eligible = messages.filter(isEligibleForCompaction);

  // ❌ 早期版本的隐患写法:
  // if (eligible.length === 0) return { summary: "", tokensAfterEstimate: 0 };
  // 悄悄吐出一个空摘要,表面上「能跑」,实际污染了下游上下文

  // ✅ 修复后:没有可处理的消息就直接拒绝
  if (eligible.length === 0) {
    throw new Error("No eligible messages to compact — refusing to summarize.");
  }

  return await summarize(eligible);
}

任何「生成式」的处理流程——摘要、聚合、汇总——遇到空输入，默认应该 fail loud，而不是 fail silent 地吐出一个看似正常的结果。后者短期内「看起来能跑」，但会悄悄污染下游数据，而且因为不报错，事后极难排查。

九、上下文组装用分层文件，而不是代码里拼字符串

项目根目录/
├── AGENTS.md            # 基础规则(项目通用,人和 Agent 都读)
├── SYSTEM.md             # 整体替换默认系统提示词(可选)
├── APPEND_SYSTEM.md      # 在默认系统提示词基础上追加(可选)
└── .pi/skills/           # 可插拔的技能模块

// 最终的系统提示词由几层显式叠加而成,而不是这样写死在代码里:
// const systemPrompt = `你是一个助手...` + userConfig + moreStuff + "..."

function buildSystemPrompt(project: ProjectConfig): string {
  const base = project.systemMdExists
    ? readFile("SYSTEM.md")          // 完全替换默认值
    : DEFAULT_SYSTEM_PROMPT;

  const appended = project.appendSystemMdExists
    ? readFile("APPEND_SYSTEM.md")   // 在基础上追加
    : "";

  return [base, appended, ...loadSkills(project)].join("nn");
}

使用者不用碰代码，只改配置文件就能精确控制最终效果，还能被 git diff、被非工程师维护。这个思路能推广到任何「内容组装」场景，不止是 prompt：配置生成、模板渲染、多环境部署配置……把「内容由哪几层、按什么顺序叠加而成」显式建模成文件系统里的层级结构，比在代码里维护一堆字符串拼接和模板变量更易读。

十、一套运行时，三种入口

// 核心运行时只有一份,三种「外壳」复用同一个 AgentHarness

// 1. 交互式终端
const tui = new TerminalUI(new AgentHarness(config));
tui.start();

// 2. 无头模式,走 JSONL RPC,方便脚本/CI 调用
const rpc = new JsonlRpcServer(new AgentHarness(config));
rpc.listen(process.stdin, process.stdout);

// 3. 直接作为 SDK 嵌入你自己的应用
const harness = new AgentHarness(config);
harness.on("message_update", (delta) => myApp.streamToUser(delta));
await harness.run("帮我重构这个函数");

先把「核心运行时」和「某一种具体的交互界面」彻底解耦，界面永远只是运行时的一种呈现方式——反过来做的项目，事后剥离出来的 SDK 往往千疮百孔。

配合这一点，pi 甚至把「给 Agent 用的组件间通信、状态同步、插件加载」这类基础设施，单独抽成了一个完全独立、不带任何「AI」业务语义的通用包（chord），高层的远程会话协议是构建在它之上的。当你发现自己在为业务系统写「组件通信/状态复制/插件系统」这类基础设施代码时，值得停下来想一想：这段代码里有多少是跟具体业务强相关的？大概率很少——那就值得单独拆出来，当一个没有业务语义的基础库去设计。