12 AI优秀实现五件套（开发者可直接复刻版）

这一篇是“压缩包总结”：把最值得抄的 5 套实现，直接变成你的落地模板。

五件套总览

1. 执行链路分层状态机
2. 上下文卫生与压缩恢复
3. 工具策略与审批闸门
4. 模型回退与错误归一化
5. 子智能体编排与技能快照

---

套件1 执行链路分层状态机

对应源码：

• src/agents/pi-embedded-runner/run.ts
• src/agents/pi-embedded-runner/run/attempt.ts
• src/agents/pi-embedded-subscribe.ts
• src/agents/pi-embedded-runner/runs.ts

可抄要点：

1. run 只做调度，不做细节执行
2. attempt 负责单次事务（prompt + wait compaction + cleanup）
3. subscribe 只做事件聚合
4. runs 维护 active handle（steer/abort）

状态机：

queued -> attempting -> streaming -> compacting(optional) -> completed/failed -> cleanup

---

套件2 上下文卫生与压缩恢复

对应源码：

• src/agents/context-window-guard.ts
• src/agents/pi-embedded-runner/history.ts
• src/agents/session-transcript-repair.ts
• src/agents/pi-embedded-runner/run/compaction-timeout.ts

可抄要点：

1. 运行前窗口预检：resolveContextWindowInfo + evaluateContextWindowGuard
2. 进入 prompt 前固定顺序：

• sanitizeSessionHistory
• limitHistoryTurns
• sanitizeToolUseResultPairing

3. 压缩等待：waitForCompactionRetry()
4. 压缩超时快照：selectCompactionTimeoutSnapshot(...)

---

套件3 工具策略与审批闸门

对应源码：

• src/agents/pi-tools.ts
• src/agents/tool-policy-pipeline.ts
• src/agents/pi-tools.before-tool-call.ts
• src/gateway/exec-approval-manager.ts
• src/gateway/server-methods/exec-approval.ts

可抄要点：

1. 工具先“全集构建”，后“分层过滤”
2. before_tool_call 支持“改参/阻断”
3. after_tool_call 在成功和失败路径都触发审计
4. 审批状态机必须有 request -> wait -> resolve -> timeout

---

套件4 模型回退与错误归一化

对应源码：

• src/agents/model-fallback.ts
• src/agents/failover-error.ts
• src/agents/pi-embedded-helpers/errors.ts

可抄要点：

1. resolveFallbackCandidates(...) 统一候选生成
2. coerceToFailoverError(...) 统一错误语义
3. shouldRethrowAbort(...) 保留用户中断语义
4. 输出结构化 attempts，便于观测和告警

---

套件5 子智能体编排与技能快照

对应源码：

• src/agents/tools/sessions-spawn-tool.ts
• src/agents/subagent-registry.ts
• src/agents/subagent-announce.ts
• src/agents/skills/workspace.ts
• src/agents/skills/refresh.ts
• src/agents/skills/env-overrides.ts

可抄要点：

1. spawn 必须独立会话 + 防递归
2. 子 run 必须注册、可恢复、可清理
3. 子结果回传主会话，不直接硬发给用户
4. 技能采用 snapshot + version，不要每轮全盘扫描

---

可直接复制的工程模板（最小骨架）

async function runAgent(req: Req) {
  return enqueue(req.sessionLane, () =>
    enqueue(req.globalLane, async () => {
      const handle = activeRuns.register(req.sessionId);
      try {
        const tools = buildAllowedTools(req);
        const session = createSession(req, tools);
        const sub = subscribe(session);
        await session.prompt(req.prompt);
        await sub.waitForCompactionRetry();
        return collectResult(sub);
      } catch (err) {
        return await runWithModelFallback({
          candidates: resolveFallbackCandidates(req),
          run: (provider, model) => rerunAttempt(req, provider, model),
        });
      } finally {
        activeRuns.clear(req.sessionId, handle);
      }
    }),
  );
}

交付前自检（五件套对应五项）

1. 运行链路：同 session 串行、abort 可中断、finally 能清理。
2. 上下文：无孤儿 tool_result，压缩重试可等待。
3. 工具安全：策略顺序固定，审批支持 timeout。
4. 回退：attempts 带 reason/status/code。
5. 编排：子任务可恢复，技能改动能自动升版。