Agent 架构

职责概述

解决的问题：单个 AI 对话很难处理"重构整个微服务架构"这种大任务。需要一种机制把大任务拆成小任务，分配给多个 AI Agent 并行执行，再汇总结果。Agent 架构就是 CC 的"多 AI 协作"方案。

应用场景：① 用户说"把这个项目从 JS 迁移到 TS" → 主 Agent 拆分出多个子 Agent 分别处理不同模块 ② 后台 Agent 异步执行耗时任务（如全量测试），不阻塞主对话 ③ 代码审查 Agent 在完成后自动触发 review ④ 多个 Agent 组成团队，通过消息传递协调工作。

一句话理解：就像一个项目经理带着几个开发——经理拆任务、分配给组员、收集结果，组员各自独立干活，有事儿互相喊。

架构设计

任务类型层次

5 种 Task 类型对比

核心数据流

AgentTool.call() 路由流程

Agent 层次结构

SendMessage 通信模式

关键类型与接口

TaskType 与 TaskStatus（Task.ts:6-29）

export type TaskType =
  | 'local_bash'          // Shell 命令任务
  | 'local_agent'         // 本地 agent 任务
  | 'remote_agent'        // 远程 agent 任务
  | 'in_process_teammate' // 进程内 teammate
  | 'local_workflow'      // 工作流任务
  | 'monitor_mcp'         // MCP 监控任务
  | 'dream'               // 梦境代理

export type TaskStatus =
  | 'pending' | 'running' | 'completed' | 'failed' | 'killed'

export function isTerminalTaskStatus(status: TaskStatus): boolean {
  return status === 'completed' || status === 'failed' || status === 'killed'
}

Task 接口（Task.ts:72-76）

export type Task = {
  name: string
  type: TaskType
  kill(taskId: string, setAppState: SetAppState): Promise<void>
}

TaskStateBase 共享状态（Task.ts:45-57）

export type TaskStateBase = {
  id: string
  type: TaskType
  status: TaskStatus
  description: string
  toolUseId?: string
  startTime: number
  endTime?: number
  totalPausedMs?: number
  outputFile: string     // 输出文件路径（磁盘持久化）
  outputOffset: number   // 输出读取偏移量
  notified: boolean      // 是否已通知用户
}

Task ID 生成（Task.ts:98-106）

const TASK_ID_PREFIXES: Record<string, string> = {
  local_bash: 'b', local_agent: 'a', remote_agent: 'r',
  in_process_teammate: 't', local_workflow: 'w',
  monitor_mcp: 'm', dream: 'd',
}

export function generateTaskId(type: TaskType): string {
  const prefix = getTaskIdPrefix(type)
  const bytes = randomBytes(8)  // 36^8 ≈ 2.8 万亿组合
  let id = prefix
  for (let i = 0; i < 8; i++) {
    id += TASK_ID_ALPHABET[bytes[i]! % TASK_ID_ALPHABET.length]
  }
  return id  // 例如 "a3f7k2j9n"
}

InProcessTeammateTaskState（tasks/InProcessTeammateTask/types.ts:22-76）

export type InProcessTeammateTaskState = TaskStateBase & {
  type: 'in_process_teammate'
  identity: TeammateIdentity       // agentId, agentName, teamName, color
  prompt: string
  model?: string                   // 可选模型覆盖
  selectedAgent?: AgentDefinition  // 可选 agent 定义
  abortController?: AbortController
  currentWorkAbortController?: AbortController
  awaitingPlanApproval: boolean    // 等待计划审批
  permissionMode: PermissionMode   // 独立权限模式
  result?: AgentToolResult
  progress?: AgentProgress
  messages?: Message[]             // 对话历史（UI 镜像，最多 50 条）
  pendingUserMessages: string[]    // 待消费的用户消息队列
  isIdle: boolean                  // 是否空闲（等待消息）
  shutdownRequested: boolean       // 关闭请求标志
  onIdleCallbacks?: Array<() => void>  // idle 通知回调
}

AgentTool 输出类型（AgentTool.tsx:140-191）

// 同步完成
{ status: 'completed', prompt: string, ...agentToolResult }

// 异步启动
{ status: 'async_launched', agentId, description, prompt, outputFile, canReadOutputFile }

// Teammate 启动（内部类型）
{ status: 'teammate_spawned', prompt, teammate_id, agent_id, name, color,
  tmux_session_name, tmux_window_name, tmux_pane_id, team_name,
  is_splitpane, plan_mode_required }

// 远程启动
{ status: 'remote_launched', taskId, sessionUrl, description, prompt, outputFile }

SendMessage 路由类型（SendMessageTool.ts:92-131）

export type MessageRouting = {
  recipient: string       // 目标 teammate 名称或 'all'
  content: string         // 消息内容
  summary?: string        // 消息摘要
  action?: 'message' | 'broadcast' | 'shutdown_request' | 'shutdown_approval'
          | 'shutdown_rejection' | 'plan_approval' | 'plan_rejection'
}

设计模式与亮点

更多亮点

• 动态后台化：LocalAgentTask 和 LocalShellTask 都支持从同步模式动态转为后台模式。backgroundAgentTask() 将任务从 foreground 队列移到后台，释放主循环。用户可以在运行中按 Escape 触发后台化。
• 消息数量上限：TEAMMATE_MESSAGES_UI_CAP = 50 限制 teammate 的 UI 镜像消息数量，防止长时间运行的 teammate 消耗过多内存。完整对话保存在磁盘。
• 分布式关闭协议：SendMessageTool 实现了 request → approval/rejection 关闭协商。Teammate 不能被强制终止（除非 kill），必须通过协商优雅关闭。支持 reason 反馈。
• 计划审批流程：plan_mode_required 的 teammate 在执行前必须提交计划等待 leader 审批。Leader 通过 handlePlanApproval/handlePlanRejection 决定是否允许执行，支持 feedback。
• Worktree 隔离：AgentTool 支持 isolation: 'worktree' 参数，为子 agent 创建独立的 git worktree。Fork 子 agent 自动检测 worktree 并注入通知。完成后自动清理。
• Task ID 前缀：每种任务类型有唯一前缀（a/b/r/t/w/m/d），便于在日志和 UI 中快速识别任务类型。8 位随机字符提供 2.8 万亿组合，抵抗暴力攻击。
• Agent 自定义 MCP：initializeAgentMcpServers()（runAgent.ts:95）为每个 agent 初始化独立的 MCP 服务器实例，避免跨 agent 的 MCP 状态污染。
• 进度追踪：ProgressTracker（LocalAgentTask.tsx:41-49）实时追踪 agent 的 token 使用量、工具调用次数和当前活动。支持 SDK 消费者通过 tool_progress 事件获取实时更新。

◈ 可视化处理拓扑图

AgentTool.call() 是所有子 agent 请求的统一入口。它根据参数组合进行多级路由判断，最终将任务分发到 5 种执行路径之一。以下拓扑图展示完整的路由决策、执行引擎和结果回收流程。

Phase 2 — 系统提示注入与执行引擎

Phase 3 — 同步 Agent 消息循环

⇉ 核心处理流程详解

Agent 系统的核心是一个从用户请求到最终输出的完整处理链路。入口是 AgentTool.call()（AgentTool.tsx:239），它负责路由决策、资源准备、生命周期管理，最终将子 agent 的结果返回给主对话。以下是每一步的详细处理过程：

★ 设计精华

1. Fork 路径的 Cache-Identical 前缀优化

Fork 子 agent 复用父进程的完整 system prompt 和工具定义数组（useExactTools: true），确保 API 请求的字节前缀完全一致，从而命中 prompt cache。这是性能优化的极致体现——不是"相似"，而是"字节级一致"。代价是 Fork 子 agent 继承了父进程的 thinking 配置和交互模式，但这正是设计意图：用最少的 cache_creation tokens 获得最大的上下文共享。

availableTools: isForkPath ? toolUseContext.options.tools : workerTools,
// useExactTools 让 runAgent 跳过 resolveAgentTools()，直接使用传入的工具
// 同时继承父进程的 thinkingConfig 和 isNonInteractiveSession
...(isForkPath && { useExactTools: true }),

2. 递归 Fork 双重防护

Fork 子 agent 保留了 Agent 工具（用于 cache 前缀一致），但必须防止递归 Fork。系统使用两级防护：主防护是 context.options.querySource（设为 'agent:builtin:fork'），这个值存在于 options 对象上而非消息中，因此能抵抗 autocompact 的消息重写。备用防护是 isInForkChild()（forkSubagent.ts:78），扫描消息中是否存在 Fork 标记——当 querySource 因某种原因未传递时兜底。

if (toolUseContext.options.querySource === `agent:builtin:${FORK_AGENT.agentType}`
    || isInForkChild(toolUseContext.messages)) {
  throw new Error('Fork is not available inside a forked worker.')
}

3. Worktree 隔离的智能清理

Agent 可以在独立 git worktree 中运行（isolation: 'worktree'），实现文件系统级别的沙箱。清理逻辑尤为精妙：cleanupWorktreeIfNeeded()（AgentTool.tsx:644）比较 worktree 创建时的 HEAD commit 和当前状态——有变更则保留 worktree，无变更则自动删除。Hook-based worktree 永远保留（因为无法检测 VCS 变更）。

4. 消息过滤保证 API 一致性

filterIncompleteToolCalls()（runAgent.ts:866）在多个关键路径上确保消息序列符合 API 规范：每个 tool_use 必须有对应的 tool_result。当 Fork 子 agent 的父消息包含多个 tool_use 但只需处理其中一部分时，该函数为未配对的 tool_use 补充 placeholder result，防止 API 拒绝请求。同样在 agent 终止时清理未完成的工具调用。

export function filterIncompleteToolCalls(messages: Message[]): Message[] {
  // 收集所有 tool_use ID 和已匹配的 tool_result ID
  // 对于未匹配的 tool_use，在其后的 assistant 消息前
  // 插入带空结果的 user 消息
}

5. 模块化 Task 类型系统

Task 类型（Task.ts:6-76）定义了 6 种 TaskType，每种对应不同的执行模型：local-agent（本地同步/异步）、local-shell（后台命令）、in-process-teammate（进程内持久化队友）、remote-agent（远程执行）等。getTaskByType() 工厂函数根据类型分发不同的 Task handler，每种 handler 只实现自己关心的生命周期方法（kill、spawn、render），实现了开放-封闭原则。

◈ Agent 实践借鉴 — 客服 Agent 多 Agent 协作设计

场景映射：客服 agent 在协作上遇到的真实问题

在客服场景中，多 agent 协作是刚需而非锦上添花。典型痛点包括：

• 技术问题升级：客户来电，一线 agent 处理不了的技术问题需要转给技术支持 agent，转交时上下文容易丢失，客户被迫重复描述问题。
• 退单审批流转：退单超过 5000 元需要主管 agent 审批，一线 agent 必须带上操作建议（如"建议同意退款，原因是物流破损"）提交审批，主管审批后一线 agent 才能执行。
• 大促期间并行查询：一个 agent 查订单状态，另一个 agent 同时查物流轨迹，两个结果合并后才能给客户完整答复。串行查询太慢，客户等不及。
• VIP 专属权限：VIP 客户有专属 agent，能直接操作普通坐席没权限的功能（如免审批全额退款、优先发货），需要权限隔离。

借鉴 CC + 客服改造

1. 客服 agent 的任务类型定义（CC 5 种 → 客服 3 种）

CC 定义了 7 种 TaskType（local_agent/local_bash/in_process_teammate/remote_agent/local_workflow/monitor_mcp/dream），但客服场景只需要 3 种核心模式：

// CC 的 5 种运行模式 → 客服场景精简为 3 种
type ServiceTaskType =
  | 'sync_query'      // 同步处理：查订单、查物流、回答FAQ
  | 'async_background' // 异步后台：发短信通知、同步CRM记录
  | 'escalation'      // 升级转交：技术支持、主管审批

type ServiceTaskStatus =
  | 'pending' | 'handling' | 'waiting_approval'
  | 'completed' | 'failed' | 'timeout'

// 统一任务基类——借鉴 CC 的 TaskStateBase
interface ServiceTask {
  taskId: string
  type: ServiceTaskType
  status: ServiceTaskStatus
  customerPhone: string
  startTime: number
  endTime?: number
  // 输出持久化到磁盘，避免内存溢出
  outputFilePath: string
  kill(): Promise<void>
}

// 优先级路由：借鉴 CC AgentTool.call() 的级联路由
function dispatchServiceTask(params: ServiceTaskParams): ServiceTaskResult {
  // 优先级 1: 升级转交（金额超限/技术问题/VIP专属）
  if (needsEscalation(params)) {
    return escalateToSpecialist(params)
  }
  // 优先级 2: 异步后台（通知/日志/数据同步）
  if (isBackgroundTask(params)) {
    return launchBackgroundTask(params)  // fire-and-forget
  }
  // 默认: 同步处理（一线 agent 直接处理）
  return handleSyncQuery(params)
}

2. agent 间上下文传递协议（借鉴 CC SendMessageTool）

CC 的 SendMessageTool 使用 pendingUserMessages 队列 + onIdleCallbacks 实现邮箱式通信。客服场景改造为"带摘要的上下文传递"——不是传递完整对话（太长），而是传递关键信息摘要：

// 借鉴 CC 的 MessageRouting → 客服场景的上下文传递协议
interface HandoffContext {
  fromAgent: string          // 来源 agent（如 "一线坐席-A03"）
  toAgent: string            // 目标 agent（如 "技术支持-B07"）
  customerProfile: {
    phone: string
    vipLevel: 'normal' | 'silver' | 'gold' | 'platinum'
    tags: string[]           // 如 ["高频投诉", "退款偏好"]
  }
  issueSummary: string       // 问题摘要（不是完整对话！）
  actionsTaken: string[]     // 已执行的操作（如 ["已查订单 ORD-20240101", "已查物流 显示签收"]）
  recommendation?: string    // 操作建议（如 "建议同意退款，物流显示破损"）
  urgency: 'low' | 'medium' | 'high' | 'critical'
}

// 上下文传递——借鉴 CC 的 handleMessage + pendingUserMessages
function handoffToSpecialist(context: HandoffContext): void {
  const targetAgent = findAvailableAgent(context.toAgent)
  // 摘要注入 target agent 的消息队列
  targetAgent.pendingMessages.push(formatHandoffPrompt(context))
  // 借鉴 CC 的 onIdleCallbacks：如果目标 agent 空闲，立即唤醒
  if (targetAgent.isIdle) {
    targetAgent.onIdleCallbacks.forEach(cb => cb())
  }
}

// 格式化转交提示——关键信息在前，不是原始对话堆砌
function formatHandoffPrompt(ctx: HandoffContext): string {
  return `[客户转交] 来自 ${ctx.fromAgent}
客户: ${ctx.customerProfile.phone} (${ctx.customerProfile.vipLevel})
问题: ${ctx.issueSummary}
已处理: ${ctx.actionsTaken.join('; ')}
建议: ${ctx.recommendation || '无'}
紧急度: ${ctx.urgency}`
}

3. 并行查询 + 结果合并（借鉴 CC Fork cache-identical）

CC 的 Fork 模式通过共享 system prompt 实现 cache-identical 前缀。客服场景借鉴其"共享上下文、独立执行"的思想，做并行查询：

// 借鉴 CC Fork 的"共享上下文、独立执行" → 客服并行查询
async function parallelQuery(customerId: string, issues: string[]) {
  // 共享上下文：客户基本信息（避免每个 agent 重复查）
  const sharedContext = {
    customerId,
    customerProfile: await getCustomerProfile(customerId),
  }

  // 并行派生多个查询任务——类似 CC 的 Fork 子 agent
  const tasks = issues.map(issue => ({
    agent: selectAgentForIssue(issue),
    context: { ...sharedContext, query: issue },
  }))

  // Promise.allSettled——不因单个查询失败而放弃全部
  const results = await Promise.allSettled(
    tasks.map(t => t.agent.query(t.context))
  )

  // 合并结果——每个查询只返回摘要，不是原始数据
  return results.map((r, i) => ({
    issue: issues[i],
    status: r.status,
    summary: r.status === 'fulfilled' ? r.value.summary : '查询失败',
  }))
}

// 使用示例：大促期间同时查订单 + 物流
const queryResults = await parallelQuery('CUS-12345', [
  '查询订单 ORD-20240101 的状态',
  '查询订单 ORD-20240101 的物流轨迹',
])
// → [{ issue: '订单状态', summary: '已发货，预计明天到达' },
//    { issue: '物流轨迹', summary: '最新节点: 广州中转站，今天14:00' }]

4. 升级审批流程（借鉴 CC 计划审批机制）

CC 的 plan_mode_required teammate 需要提交计划等待 leader 审批。客服场景改造为"一线 agent 提交操作建议 → 主管审批"：

// 借鉴 CC 的 plan_approval/plan_rejection → 客服升级审批
interface EscalationRequest {
  requestId: string
  fromAgent: string
  customerPhone: string
  issueType: 'refund' | 'compensation' | 'technical' | 'complaint'
  amount?: number            // 退单金额
  recommendation: string     // 一线 agent 的建议
  evidence: string[]         // 支撑材料（订单截图、物流证据等）
}

// 一线 agent 提交审批
async function submitApproval(req: EscalationRequest): Promise<void> {
  const supervisor = await findAvailableSupervisor(req.issueType)
  // 借鉴 CC 的 pendingUserMessages 队列
  supervisor.pendingApprovals.push(req)
  // 通知主管（类似 CC 的 onIdleCallbacks）
  notifySupervisor(supervisor.id, `新的审批请求: ${req.issueType}`)
}

// 主管审批——借鉴 CC 的 handlePlanApproval / handlePlanRejection
async function handleApproval(
  requestId: string,
  decision: 'approved' | 'rejected',
  feedback?: string
): Promise<void> {
  const req = findEscalationRequest(requestId)
  if (decision === 'approved') {
    // 自动执行一线 agent 的建议（如发起退款）
    await executeRecommendation(req.recommendation)
  }
  // 回传审批结果给一线 agent
  notifyAgent(req.fromAgent, {
    requestId,
    decision,
    feedback,  // 如 "同意退款，但下次先核实物流再提交"
  })
}

落地清单

代码索引