Claude Code 工程模式
Claude Code 是 Anthropic 官方 CLI 产品，其源码展现了业界顶级的 AI Agent 工程实践。本 skill 提炼其核心设计模式。
核心模式概览
| 模式 | 解决的问题 | 适用场景 |
|---|
| 启动并行化 | 启动延迟高 | 需要 I/O 预取的 agent 系统 |
| 状态机 Query Loop | 递归栈溢出、状态追踪困难 | 多轮对话、复杂任务编排 |
| 流式工具并发执行 | 工具执行阻塞模型输出 | 多工具调用场景 |
| 多层上下文压缩 | 上下文溢出 | 长对话、大项目 |
| 错误恢复 withhold | 中间错误暴露给用户 | 任何可能失败的 API 调用 |
| 熔断器 | 失败操作无限重试 | 自动重试逻辑 |
| 工具并发安全声明 | 工具执行冲突 | 多工具并发系统 |
1. 启动并行化 + Profiling
问题：Agent 启动需要加载配置、连接服务、预取数据，串行执行导致启动慢。
解决方案：所有阻塞操作并行化，用 checkpoint 测量瓶颈。
// 入口文件最顶部 — 模块加载前就开始计时
profileCheckpoint('main_entry');
// 并行启动所有 I/O
const [mdmResult, keychainResult, commandsResult] = await Promise.all([
startMdmRawRead(), // 并行读取 MDM 配置
startKeychainPrefetch(), // 并行预取 keychain
getCommands(), // 并行加载命令
]);
profileCheckpoint('all_parallel_done');
关键技巧：
- 纯文件读取可以在信任对话框显示前开始（无执行风险）
--bare 模式跳过所有非必要预取，专为脚本调用优化- 用
profileCheckpoint() 定位瓶颈，不要猜测
应用建议：
// 而不是
await loadConfig();
await connectMCP();
await prefetchMemory();
// 应该
await Promise.all([
loadConfig(),
connectMCP(),
prefetchMemory(),
]);
2. 状态机 Query Loop
问题：多轮对话用递归实现会导致栈溢出，且难以追踪状态变化原因。
解决方案：用 while(true) + 不可变 State 对象，每次 continue 记录原因。
type State = {
messages: Message[]
toolUseContext: ToolUseContext
transition: Continue | undefined // 记录上次 continue 的原因
maxOutputTokensRecoveryCount: number
hasAttemptedReactiveCompact: boolean
// ...
}
type Continue =
| { reason: 'next_turn' }
| { reason: 'max_output_tokens_recovery'; attempt: number }
| { reason: 'reactive_compact_retry' }
| { reason: 'stop_hook_blocking' }
// ...
async function* queryLoop(params: QueryParams) {
let state: State = initialState(params)
while (true) {
const { messages, toolUseContext, transition } = state
// 执行查询逻辑...
if (needsFollowUp) {
state = {
messages: [...messages, ...toolResults],
toolUseContext: updatedContext,
transition: { reason: 'next_turn' },
}
continue // 不用递归，用 continue
}
return { reason: 'completed' }
}
}
为什么不用递归：
- 递归栈会增长，长对话可能溢出
transition 字段让每次 continue 的原因可追踪- 测试可以断言
transition.reason 而不是检查消息内容
Continue 的各种路径：
| 原因 | 触发条件 | 处理方式 |
|---|
next_turn | 正常工具调用后 | 合并消息继续 |
max_output_tokens_recovery | 输出被截断 | 注入恢复提示 |
reactive_compact_retry | 上下文太长 | 压缩后重试 |
stop_hook_blocking | hook 要求继续 | 注入 hook 消息 |
3. 流式工具并发执行（StreamingToolExecutor）
问题：传统模式是等模型输出完毕再执行工具，浪费大量时间。
解决方案：工具在流式输出时就开始执行，结果缓冲后按序 yield。
class StreamingToolExecutor {
private tools: TrackedTool[] = []
// 添加工具到队列，立即开始执行（如果并发条件允许）
addTool(block: ToolUseBlock, assistantMessage: AssistantMessage) {
const isConcurrencySafe = this.checkConcurrencySafe(block)
this.tools.push({ block, status: 'queued', isConcurrencySafe })
void this.processQueue() // 立即处理，不等待
}
// 获取已完成的结果（非阻塞）
*getCompletedResults() {
for (const tool of this.tools) {
if (tool.status === 'completed' && tool.results) {
tool.status = 'yielded'
yield* tool.results
}
}
}
}
// 使用方式
for await (const message of callModel()) {
if (message.type === 'assistant') {
for (const toolBlock of message.toolUseBlocks) {
executor.addTool(toolBlock, message) // 立即开始执行
}
}
// 同时 yield 已完成的结果
for (const result of executor.getCompletedResults()) {
yield result
}
}
并发安全分区：
// 工具声明自己是否并发安全
class ReadTool {
isConcurrencySafe(input: Input): boolean {
return true // 只读，可并发
}
}
class BashTool {
isConcurrencySafe(input: Input): boolean {
return false // 可能写文件，串行
}
}
// 调度器根据声明分区
function partitionToolCalls(tools: ToolUseBlock[]): Batch[] {
// 连续的并发安全工具合并成一个 batch
// 非并发安全工具单独一个 batch
}
Bash 错误级联取消：
// Bash 出错时取消所有并行工具
if (tool.block.name === 'BASH' && isErrorResult) {
this.hasErrored = true
this.siblingAbortController.abort('sibling_error')
}
4. 多层上下文压缩
问题：单一压缩策略无法平衡性能和信息保留。
解决方案：5 层防御，从轻到重。
Layer 1: History Snip → 删除旧消息（最轻量，~0 cost）
Layer 2: Microcompact → 压缩单个工具结果（缓存友好）
Layer 3: Context Collapse → 折叠历史段落（保留粒度）
Layer 4: Auto Compact → 全量摘要（最重量）
Layer 5: Reactive Compact → 被动响应 API 413 错误
各层触发条件：
| 层 | 触发条件 | 特点 |
|---|
| Snip | 消息数 > 阈值 | 删除最旧的非关键消息 |
| Microcompact | 单个工具结果 > 阈值 | 只压缩该结果，不动其他 |
| Collapse | 上下文 > 90% | 折叠旧段落为摘要 |
| Auto Compact | 上下文 > 93% | 全量摘要 |
| Reactive Compact | API 返回 413 | 最后防线 |
熔断器设计：
const MAX_CONSECUTIVE_FAILURES = 3
if (tracking?.consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
// 停止重试，避免浪费 API 调用
return { wasCompacted: false }
}
5. 错误恢复 Withhold 模式
问题：流式输出中遇到可恢复错误，如果直接 yield 给调用方，调用方会终止。
解决方案：先 withhold（扣留）错误，尝试恢复，成功则调用方无感知。
// 流式输出中
if (isRecoverableError(message)) {
withheld = true
// 不 yield，先尝试恢复
}
// 尝试恢复
const recovered = await tryRecovery()
if (recovered) {
// 继续新的查询循环，调用方完全不知道出过错
state = { messages: recovered.messages, ... }
continue
}
// 恢复失败才 yield 错误
yield withheldError
适用场景：
prompt_too_long → 尝试压缩后重试max_output_tokens → 升级 token 限制或注入恢复提示media_size_error → 压缩图片后重试
6. Task 系统的类型设计
问题：任务 ID 难以识别类型，日志可读性差。
解决方案：Task ID 带类型前缀。
const TASK_ID_PREFIXES = {
local_bash: 'b',
local_agent: 'a',
in_process_teammate: 't',
remote_agent: 'r',
workflow: 'w',
monitor: 'm',
dream: 'd',
}
// 生成: "t3f8a2b1c9" 一眼知道是 teammate
function generateTaskId(type: TaskType): string {
const prefix = TASK_ID_PREFIXES[type]
const random = crypto.randomBytes(8).toString('hex').slice(0, 8)
return prefix + random
}
7. 其他实用模式
内容哈希替代随机 UUID
// 避免每次 API 调用都破坏 prompt cache
const contentHash = hashContent(settings)
const tempFile = `${cacheDir}/settings_${contentHash}.json`
using 关键字管理资源
// 自动 dispose，即使发生异常
using pendingMemoryPrefetch = startRelevantMemoryPrefetch(messages)
// ... 代码 ...
// 离开作用域时自动调用 pendingMemoryPrefetch[Symbol.dispose]()
Tombstone 消息
// fallback 时用 tombstone 标记需要移除的消息，而不是直接删除
yield { type: 'tombstone', message: orphanedMessage }
Speculation（预测执行）
// 用户还在打字时，开始预测并执行
type SpeculationState = {
status: 'idle'
} | {
status: 'active'
id: string
messagesRef: { current: Message[] }
timeSavedMs: number // 累计节省的时间
}
参考文档
快速应用清单
设计 AI Agent 系统时，检查以下问题：
