AI Agent 的软件架构：计划、工具、记忆和权限

// ❌ 坏做法：计划一旦生成不可修改
interface RigidPlan {
  steps: PlanStep[] // 生成后不再变化
  currentIndex: number
}
 
function executeRigidPlan(plan: RigidPlan) {
  for (const step of plan.steps) {
    // 即使上一步结果异常，也硬着头皮继续
    const result = await executeStep(step)
    plan.currentIndex++
  }
}
 
// ✅ 好做法：计划可根据执行结果修正
interface AdaptivePlan {
  steps: PlanStep[]
  currentIndex: number
  maxReplans: number // 防止无限重规划
  replanCount: number
}
 
async function executeAdaptivePlan(plan: AdaptivePlan, context: ExecutionContext) {
  while (plan.currentIndex < plan.steps.length) {
    const step = plan.steps[plan.currentIndex]
    const result = await executeStep(step)
 
    if (result.anomaly && plan.replanCount < plan.maxReplans) {
      // 执行结果与预期不符，重新规划后续步骤
      const newSteps = await replanner.replan({
        completedSteps: plan.steps.slice(0, plan.currentIndex + 1),
        remainingGoal: step.originalGoal,
        anomaly: result.anomaly,
      })
      plan.steps = [...plan.steps.slice(0, plan.currentIndex + 1), ...newSteps]
      plan.replanCount++
    }
 
    plan.currentIndex++
  }
}

// 第一层：工具注册——按角色过滤可用工具
interface ToolRegistry {
  // ❌ 坏做法：返回所有工具
  // getTools(): Tool[]
 
  // ✅ 好做法：按角色和场景返回工具子集
  getTools(context: { role: string; scenario: string }): Tool[] {
    return allTools.filter(tool =>
      tool.allowedRoles.includes(context.role) &&
      tool.allowedScenarios.includes(context.scenario)
    )
  }
}
 
// 第二层：调用前拦截——独立权限检查
interface ToolInterceptor {
  async intercept(call: ToolCall, user: UserInfo): Promise<InterceptResult> {
    const tool = registry.get(call.name)
 
    // 高风险工具需要二次确认
    if (tool.riskLevel === 'high' && !user.hasConfirmed) {
      return { allowed: false, reason: '需要用户确认', requireConfirm: true }
    }
 
    // 参数校验：不信任模型生成的参数
    const validation = tool.schema.safeParse(call.arguments)
    if (!validation.success) {
      return { allowed: false, reason: `参数校验失败: ${validation.error}` }
    }
 
    return { allowed: true }
  }
}
 
// 第三层：调用后验证——不能只看返回码
interface ToolResultVerifier {
  async verify(call: ToolCall, result: ToolResult): Promise<VerifiedResult> {
    // ❌ 坏做法：直接返回原始结果
    // return result
 
    // ✅ 好做法：检查结果是否合理
    if (result.status === 'error') {
      return { status: 'failed', error: result.error, retryable: result.retryable }
    }
 
    // 检查返回内容是否与调用意图一致
    if (tool.expectedOutputPattern && !tool.expectedOutputPattern.test(result.output)) {
      return { status: 'suspicious', warning: '返回结果格式异常', raw: result.output }
    }
 
    return { status: 'success', data: result.output }
  }
}

// ❌ 坏做法：无限堆积上下文，直到触碰窗口上限
class NaiveMemory {
  private messages: Message[] = []
 
  add(message: Message) {
    this.messages.push(message) // 只增不减
  }
 
  getContext(): Message[] {
    return this.messages // 可能已经超出窗口
  }
}
 
// ✅ 好做法：分层记忆 + 主动压缩 + 外部存储
interface MemoryLayer {
  working: Message[]      // 工作记忆：最近 N 步
  shortTerm: Message[]    // 短期记忆：当前会话摘要
  longTerm: string[]      // 长期记忆：外部存储的检索结果
}
 
class StructuredMemory {
  private memory: MemoryLayer = { working: [], shortTerm: [], longTerm: [] }
  private workingLimit = 20 // 工作记忆保留最近 20 条
 
  add(message: Message) {
    this.memory.working.push(message)
 
    // 工作记忆超出上限时，压缩旧消息到短期记忆
    if (this.memory.working.length > this.workingLimit) {
      const toCompress = this.memory.working.splice(0, 10)
      const summary = this.compressor.summarize(toCompress)
      this.memory.shortTerm.push({ role: 'system', content: summary })
    }
  }
 
  getContext(): Message[] {
    return [
      ...this.memory.shortTerm,  // 会话摘要在前
      ...this.memory.working,    // 最近步骤在后（注意力最强区域）
    ]
  }
 
  // 按需从长期记忆检索
  async recall(query: string): Promise<string[]> {
    return this.vectorStore.search(query, { topK: 5 })
  }
}

// ✅ 在状态机中加入退出条件和升级机制
interface StateMachineConfig {
  maxRetriesPerStep: 3       // 单步最大重试次数
  maxTotalSteps: 20          // 总步骤上限
  escalationCondition: (ctx: ExecutionContext) => boolean
}
 
async function runWithBounds(config: StateMachineConfig, plan: AdaptivePlan) {
  let totalSteps = 0
 
  while (plan.currentIndex < plan.steps.length) {
    if (totalSteps >= config.maxTotalSteps) {
      // 超出总步骤上限，停下来而不是继续转圈
      return { status: 'escalated', reason: '总步骤超限，需要人工介入' }
    }
 
    const result = await executeStep(plan.steps[plan.currentIndex])
 
    if (result.failed) {
      const stepRetries = getRetryCount(plan.currentIndex)
      if (stepRetries >= config.maxRetriesPerStep) {
        // 同一步骤重试 3 次仍失败，说明方向可能有根本问题
        return { status: 'escalated', reason: `步骤 ${plan.currentIndex} 重试耗尽` }
      }
    }
 
    totalSteps++
    plan.currentIndex++
  }
 
  return { status: 'completed' }
}

// ✅ 工具按风险分级，高风险操作需要用户确认
interface ToolPermission {
  riskLevel: 'low' | 'medium' | 'high' | 'critical'
  autoApprove: boolean
  requireConfirmation?: boolean
}
 
const toolPermissions: Record<string, ToolPermission> = {
  read_file:       { riskLevel: 'low',      autoApprove: true },
  write_file:      { riskLevel: 'medium',   autoApprove: true },
  create_pr:       { riskLevel: 'medium',   autoApprove: false, requireConfirmation: true },
  run_command:     { riskLevel: 'high',     autoApprove: false, requireConfirmation: true },
  delete_file:     { riskLevel: 'critical', autoApprove: false, requireConfirmation: true },
}
 
// 在工具调用前拦截
async function guardedToolCall(call: ToolCall, user: UserInfo) {
  const perm = toolPermissions[call.name]
 
  if (!perm.autoApprove && perm.requireConfirmation) {
    const confirmed = await requestUserConfirmation({
      tool: call.name,
      arguments: call.arguments,
      riskLevel: perm.riskLevel,
      message: `Agent 想要调用 ${call.name}，是否允许？`,
    })
    if (!confirmed) {
      return { status: 'denied', reason: '用户拒绝' }
    }
  }
 
  return await executeTool(call)
}

// ✅ 阶段性总结：用摘要替换过程详情
async function compactPhaseMemory(
  phaseName: string,
  messages: Message[],
): Promise<Message[]> {
  // 保留阶段结论，丢弃中间过程
  const summary = await llm.generate({
    system: '将以下对话阶段压缩为结构化摘要，保留关键发现和结论，丢弃中间搜索和试错过程。',
    messages,
  })
 
  return [{
    role: 'system',
    content: `[阶段总结 - ${phaseName}]\n${summary}`,
  }]
}
 
// 在主循环中阶段性调用
async function runWithCompaction(plan: AdaptivePlan) {
  let phaseMessages: Message[] = []
 
  for (const step of plan.steps) {
    const result = await executeStep(step)
    phaseMessages.push(step.message, result.message)
 
    // 每个阶段结束时压缩
    if (step.isPhaseEnd) {
      const compacted = await compactPhaseMemory(step.phaseName, phaseMessages)
      memory.replace(phaseMessages, compacted) // 用摘要替换原始消息
      phaseMessages = compacted
    }
  }
}

// ❌ 坏做法：用 console.log 散点记录
console.log('Calling tool:', toolName)
const result = await callTool(toolName, args)
console.log('Tool result:', result)
 
// ✅ 好做法：结构化 trace，每个事件有类型、时间和关联
interface AgentTraceEvent {
  traceId: string          // 同一次运行的所有事件共享
  spanId: string           // 当前事件唯一标识
  parentSpanId?: string    // 父事件（支持嵌套）
  type: 'tool_call' | 'reasoning' | 'state_transition' | 'memory_op'
  timestamp: number
  duration?: number
  input?: unknown
  output?: unknown
  error?: { type: string; message: string }
  metadata?: Record<string, unknown>
}
 
class AgentTracer {
  private events: AgentTraceEvent[] = []
 
  async traceToolCall(traceId: string, parentSpanId: string, call: ToolCall) {
    const spanId = crypto.randomUUID()
    const start = Date.now()
 
    try {
      const result = await executeTool(call)
      this.events.push({
        traceId, spanId, parentSpanId,
        type: 'tool_call',
        timestamp: start,
        duration: Date.now() - start,
        input: { tool: call.name, arguments: call.arguments },
        output: result,
      })
      return result
    } catch (error) {
      this.events.push({
        traceId, spanId, parentSpanId,
        type: 'tool_call',
        timestamp: start,
        duration: Date.now() - start,
        input: { tool: call.name, arguments: call.arguments },
        error: { type: error.name, message: error.message },
      })
      throw error
    }
  }
 
  // 导出完整 trace 用于回放
  exportTrace(traceId: string): AgentTraceEvent[] {
    return this.events
      .filter(e => e.traceId === traceId)
      .sort((a, b) => a.timestamp - b.timestamp)
  }
}