AI Agent 能力进化路线：从单任务执行到多 Agent 协作的三个阶段与关键技术栈

一、引言

AI Agent 的能力边界在过去一年中经历了快速扩展。从最初的单轮对话工具，演进到能够执行多步骤任务的自主体，再到当前的多 Agent 协作系统。这一演进过程不是简单的功能叠加，而是对 Agent 架构设计、任务规划能力和环境交互模式的重新思考。

单任务 Agent 的核心挑战在于工具调用的准确性。多步骤 Agent 需要解决任务分解和中间状态管理问题。多 Agent 协作系统则引入了 Agent 间的通信协议、任务分配策略和结果聚合机制。每个阶段的架构复杂度呈指数级增长，但能力的提升也带来了新的应用场景。

本文基于 AI Agent 能力的三个阶段划分，梳理从单任务执行到多 Agent 协作的技术演进路径。重点分析各阶段的核心技术栈、架构设计决策和工程实践中的关键陷阱。

二、三阶段能力模型与架构演进

AI Agent 的能力进化可以清晰地划分为三个阶段，每个阶段对应不同的架构复杂度和应用场景。

阶段一：单任务 Agent（工具调用阶段）

架构核心是一个大语言模型（LLM）加上工具调用（Function Calling）能力。Agent 接收用户指令，选择合适的工具执行，解析工具返回结果并生成回复。整个流程是同步的、单轮的。

关键技术栈：LLM（如 GPT-4、Claude 3.5）、函数调用框架（如 OpenAI Functions、Anthropic Tool Use）、工具集成层（如 LangChain Tools、FunctionGraph）。

核心挑战：工具选择的准确性、参数提取的完整性、工具返回结果的解析质量。工程中需要大量提示词工程和错误处理机制。

阶段二：多步骤 Agent（任务规划阶段）

架构引入了任务规划器（Planner）和状态管理器。Agent 能够将复杂任务分解为多个子步骤，按顺序或并行执行，并在执行过程中根据中间结果动态调整后续步骤。

关键技术栈：任务规划框架（如 LangGraph、AutoGen）、状态管理（如 Redux、Zustand 或专用的 Agent State 库）、记忆系统（短期记忆用于当前任务，长期记忆用于跨任务知识积累）。

核心挑战：任务分解的质量、步骤间的状态传递、错误恢复机制。如果某一步骤失败，Agent 需要能够回退或寻求替代方案，而非直接终止任务。

阶段三：协作 Agent 系统（多 Agent 阶段）

架构引入了多个专业化的 Agent，每个 Agent 负责特定领域或特定类型的任务。系统需要一个协调者（Orchestrator）或基于消息的协作协议，使 Agent 之间能够传递任务、共享上下文、聚合结果。

关键技术栈：多 Agent 框架（如 AutoGen、CrewAI、LangGraph Multi-Agent）、Agent 通信协议（如 Agent Protocol、自定义消息格式）、任务分配策略（如基于能力的路由、负载均衡）。

核心挑战：Agent 间的通信效率、任务分配的优化、结果一致性的保证。当多个 Agent 给出矛盾的结果时，系统需要能够识别并解决冲突。

三、关键技术实现

以下代码展示了阶段二的多步骤 Agent 实现，采用 LangGraph 框架，重点展示任务规划、状态管理和条件分支的实现。

// src/agents/MultiStepAgent.ts
// 多步骤AI Agent实现 - 阶段二架构

import { ChatOpenAI } from '@langchain/openai';
import { ChatAnthropic } from '@langchain/anthropic';
import {
  StateGraph,
  END,
  START,
  MemorySaver,
} from '@langchain/langgraph';
import { HumanMessage, AIMessage } from '@langchain/core/messages';

/// @notice Agent状态接口定义
/// 设计决策：使用强类型定义状态，支持类型安全的状态访问
export interface AgentState {
  messages: (HumanMessage | AIMessage)[];  // 对话历史
  currentTask: string;                       // 当前任务描述
  taskPlan: TaskStep[];                     // 任务计划（分解后的步骤）
  currentStepIndex: number;                 // 当前执行步骤索引
  stepResults: Record<string, StepResult>;  // 各步骤的执行结果
  errors: ExecutionError[];                 // 执行过程中的错误记录
  isComplete: boolean;                      // 任务是否完成
  needsHumanInput: boolean;                 // 是否需要人工介入
}

/// @notice 任务步骤定义
export interface TaskStep {
  id: string;                    // 步骤唯一标识
  description: string;           // 步骤描述
  toolName: string;             // 使用的工具名称
  toolArgs: Record<string, any>; // 工具参数（可引用之前步骤的结果）
  dependsOn: string[];          // 依赖的前置步骤
  status: 'pending' | 'running' | 'completed' | 'failed' | 'skipped';
  maxRetries: number;           // 最大重试次数
  currentRetry: number;         // 当前重试次数
}

/// @notice 步骤执行结果
export interface StepResult {
  stepId: string;
  success: boolean;
  output: any;                   // 步骤输出（可被后续步骤引用）
  error?: string;               // 错误信息（如果失败）
  executionTime: number;         // 执行耗时（毫秒）
  timestamp: number;            // 执行时间戳
}

/// @notice 执行错误记录
export interface ExecutionError {
  stepId: string;
  error: string;
  timestamp: number;
  recoverable: boolean;          // 是否可恢复（决定是否重试）
}

/// @notice 多步骤Agent核心类
export class MultiStepAgent {
  private model: ChatOpenAI | ChatAnthropic;
  private tools: Record<string, Function>;  // 可用工具集合
  private graph: StateGraph<typeof AgentState>;
  private memory: MemorySaver;
  
  constructor(
    modelConfig: { provider: 'openai' | 'anthropic'; modelName: string },
    tools: Record<string, Function>
  ) {
    // 设计决策：支持多个LLM提供商，提升灵活性
    if (modelConfig.provider === 'openai') {
      this.model = new ChatOpenAI({ modelName: modelConfig.modelName });
    } else {
      this.model = new ChatAnthropic({ modelName: modelConfig.modelName });
    }
    
    this.tools = tools;
    this.memory = new MemorySaver();
    
    // 构建状态图
    this.graph = this.buildGraph();
  }
  
  /// @notice 构建Agent状态图
  /// 设计决策：使用LangGraph的StateGraph，支持循环和条件分支
  private buildGraph(): StateGraph<typeof AgentState> {
    const graph = new StateGraph<typeof AgentState>({
      channels: {
        messages: { value: (x, y) => x.concat(y), default: () => [] },
        currentTask: { value: (x, _y) => x, default: () => '' },
        taskPlan: { value: (x, _y) => x, default: () => [] },
        currentStepIndex: { value: (x, _y) => x, default: () => 0 },
        stepResults: { value: (x, y) => ({ ...x, ...y }), default: () => ({}) },
        errors: { value: (x, y) => x.concat(y), default: () => [] },
        isComplete: { value: (x, _y) => x, default: () => false },
        needsHumanInput: { value: (x, _y) => x, default: () => false },
      },
    });
    
    // 添加节点
    graph.addNode('planner', this.plannerNode.bind(this));
    graph.addNode('executor', this.executorNode.bind(this));
    graph.addNode('evaluator', this.evaluatorNode.bind(this));
    graph.addNode('human_input', this.humanInputNode.bind(this));
    
    // 添加边（定义流程）
    graph.addEdge(START, 'planner');
    
    // 设计决策：条件边，根据任务计划决定下一步
    graph.addConditionalEdges('planner', this.routeFromPlanner.bind(this), {
      executor: 'executor',
      human_input: 'human_input',
      [END]: END,
    });
    
    graph.addConditionalEdges('executor', this.routeFromExecutor.bind(this), {
      evaluator: 'evaluator',
      planner: 'planner',  // 动态重新规划
      human_input: 'human_input',
      [END]: END,
    });
    
    graph.addConditionalEdges('evaluator', this.routeFromEvaluator.bind(this), {
      executor: 'executor',
      planner: 'planner',
      [END]: END,
    });
    
    graph.addEdge('human_input', 'planner');
    
    return graph;
  }
  
  /// @notice 规划器节点：分解任务为步骤
  private async plannerNode(state: AgentState): Promise<Partial<AgentState>> {
    // 设计决策：使用LLM生成任务计划，而非硬编码规则
    // 这提升了Agent处理未知任务类型的能力
    const planningPrompt = `
你是一个任务规划器。基于以下用户任务，生成执行步骤计划。

用户任务: ${state.currentTask}

已有工具: ${Object.keys(this.tools).join(', ')}

要求:
1. 每个步骤明确指定使用的工具和参数
2. 步骤之间如果有依赖关系，明确标注
3. 参数可以使用之前步骤的输出（使用 {{stepId.output}} 引用）

输出JSON格式的任务计划。
`;
    
    const response = await this.model.invoke([
      new HumanMessage(planningPrompt),
    ]);
    
    // 解析LLM返回的任务计划
    const taskPlan = this.parseTaskPlan(response.content.toString());
    
    return {
      taskPlan,
      currentStepIndex: 0,
    };
  }
  
  /// @notice 执行器节点：执行当前步骤
  private async executorNode(state: AgentState): Promise<Partial<AgentState>> {
    const currentStep = state.taskPlan[state.currentStepIndex];
    
    if (!currentStep) {
      return { isComplete: true };
    }
    
    // 设计决策：检查依赖步骤是否已完成
    const dependenciesMet = currentStep.dependsOn.every(
      stepId => state.stepResults[stepId]?.success === true
    );
    
    if (!dependenciesMet) {
      // 依赖未满足，跳过此步骤
      return {
        taskPlan: state.taskPlan.map(step =>
          step.id === currentStep.id ? { ...step, status: 'skipped' as const } : step
        ),
        currentStepIndex: state.currentStepIndex + 1,
      };
    }
    
    // 解析工具参数（支持引用之前步骤的输出）
    const resolvedArgs = this.resolveToolArgs(
      currentStep.toolArgs,
      state.stepResults
    );
    
    try {
      // 执行工具
      const startTime = Date.now();
      const toolFunc = this.tools[currentStep.toolName];
      if (!toolFunc) {
        throw new Error(`工具 ${currentStep.toolName} 未找到`);
      }
      
      const output = await toolFunc(resolvedArgs);
      const executionTime = Date.now() - startTime;
      
      // 记录成功结果
      const stepResult: StepResult = {
        stepId: currentStep.id,
        success: true,
        output,
        executionTime,
        timestamp: Date.now(),
      };
      
      return {
        stepResults: { [currentStep.id]: stepResult },
        taskPlan: state.taskPlan.map(step =>
          step.id === currentStep.id 
            ? { ...step, status: 'completed' as const } 
            : step
        ),
        currentStepIndex: state.currentStepIndex + 1,
      };
    } catch (error) {
      // 设计决策：区分可恢复错误和不可恢复错误
      const recoverable = currentStep.currentRetry < currentStep.maxRetries;
      
      const executionError: ExecutionError = {
        stepId: currentStep.id,
        error: error instanceof Error ? error.message : String(error),
        timestamp: Date.now(),
        recoverable,
      };
      
      if (recoverable) {
        // 可恢复：增加重试计数，保持当前步骤
        return {
          errors: [executionError],
          taskPlan: state.taskPlan.map(step =>
            step.id === currentStep.id 
              ? { ...step, currentRetry: step.currentRetry + 1, status: 'pending' as const } 
              : step
          ),
        };
      } else {
        // 不可恢复：标记步骤失败，请求重新规划或人工介入
        return {
          errors: [executionError],
          needsHumanInput: true,
          taskPlan: state.taskPlan.map(step =>
            step.id === currentStep.id 
              ? { ...step, status: 'failed' as const } 
              : step
          ),
        };
      }
    }
  }
  
  /// @notice 评估器节点：评估当前执行状态
  private async evaluatorNode(state: AgentState): Promise<Partial<AgentState>> {
    // 设计决策：使用LLM评估任务完成情况，而非简单的步骤计数
    const allCompleted = state.taskPlan.every(
      step => step.status === 'completed' || step.status === 'skipped'
    );
    
    if (allCompleted) {
      // 所有步骤完成，评估最终结果质量
      const evaluationPrompt = `
评估以下任务执行结果是否满足了用户需求：

用户任务: ${state.currentTask}

步骤结果:
${state.taskPlan.map(step => {
  const result = state.stepResults[step.id];
  return `- ${step.description}: ${result?.success ? '成功' : '失败'}`;
}).join('\n')}

给出评估结论：completed 或 needs_iteration
`;
      
      const response = await this.model.invoke([
        new HumanMessage(evaluationPrompt),
      ]);
      
      const isComplete = response.content.toString().includes('completed');
      
      return { isComplete };
    }
    
    // 还有未完成的步骤，继续执行
    return {};
  }
  
  /// @notice 解析工具参数中的步骤输出引用
  /// 设计决策：支持 {{stepId.output.field}} 格式的引用解析
  private resolveToolArgs(
    args: Record<string, any>,
    stepResults: Record<string, StepResult>
  ): Record<string, any> {
    const resolved: Record<string, any> = {};
    
    for (const [key, value] of Object.entries(args)) {
      if (typeof value === 'string' && value.match(/\{\{(\w+)\.output(\.\w+)?\}\}/)) {
        // 解析引用
        const match = value.match(/\{\{(\w+)\.output(\.\w+)?\}\}/);
        if (match) {
          const stepId = match[1];
          const field = match[2]?.slice(1);  // 去掉前面的点号
          const stepResult = stepResults[stepId];
          
          if (stepResult?.success) {
            resolved[key] = field 
              ? stepResult.output[field] 
              : stepResult.output;
          } else {
            throw new Error(`依赖步骤 ${stepId} 未成功完成`);
          }
        }
      } else {
        resolved[key] = value;
      }
    }
    
    return resolved;
  }
  
  /// @notice 路由函数：从规划器出发
  private routeFromPlanner(state: AgentState): string {
    if (state.taskPlan.length === 0) {
      return 'human_input';  // 无法生成计划，请求人工输入
    }
    return 'executor';
  }
  
  /// @notice 路由函数：从执行器出发
  private routeFromExecutor(state: AgentState): string {
    if (state.needsHumanInput) {
      return 'human_input';
    }
    if (state.isComplete) {
      return END;
    }
    return 'evaluator';
  }
  
  /// @notice 路由函数：从评估器出发
  private routeFromEvaluator(state: AgentState): string {
    if (state.isComplete) {
      return END;
    }
    return 'executor';  // 继续执行下一步骤
  }
  
  private humanInputNode(state: AgentState): Partial<AgentState> {
    // 设计决策：人工介入节点，暂停执行等待用户输入
    // 实际实现中需要与前端交互系统对接
    console.log('需要人工介入:', state.errors);
    return { needsHumanInput: false };  // 重置标志
  }
  
  private parseTaskPlan(llmOutput: string): TaskStep[] {
    // 简化实现：解析LLM返回的JSON
    try {
      const parsed = JSON.parse(llmOutput);
      return parsed.steps || [];
    } catch {
      return [];
    }
  }
}

四、边界条件与能力陷阱

在 AI Agent 能力进化的过程中，以下边界条件容易被忽视。

任务规划的过度自信

LLM 生成的任务计划可能存在逻辑漏洞或遗漏关键步骤。阶段二的 Agent 需要有计划验证机制，在执行前对计划进行合理性检查。单纯的 LLM 生成缺乏执行反馈闭环，容易导致"看起来合理但执行失败"的计划。

多步骤状态管理的复杂性

随着步骤数量增加，步骤间的状态依赖关系呈组合级增长。如果缺乏清晰的状态管理策略，Agent 可能在长流程执行中丢失上下文或产生状态不一致。工程实践中需要引入状态快照和回滚机制。

多 Agent 协作的通信开销

阶段三引入多 Agent 协作后，Agent 间的通信延迟可能成为系统瓶颈。如果任务本身可以被单 Agent 高效完成，引入多 Agent 反而会降低系统性能。需要在架构设计阶段进行任务特性分析，明确多 Agent 协作的必要性和预期收益。

工具可靠性的隐藏依赖

Agent 的能力上限受限于其可用工具的可靠性。如果关键工具存在高失败率或返回质量不稳定，Agent 的整体表现将受到严重影响。需要在工具集成层引入健康检查和降级策略。

结论

AI Agent 的能力进化不是线性的功能堆叠，而是架构思维的转变。从单任务到多步骤，核心是引入规划能力；从多步骤到多 Agent，核心是引入协作协议。

对于工程团队，Agent 能力升级的优先级应该是：先确保单任务执行的可靠性，再引入多步骤规划能力，最后在确有必要时引入多 Agent 协作。过早跳过某个阶段会导致系统复杂度失控，影响可维护性和调试效率。

Agent 能力的真正价值不在于完成更多种类的任务，而在于在特定领域内完成任务的质量和可靠性。聚焦垂直场景的深度优化，比追求通用能力的广度更有工程意义。