TypeScript 多 Agent 系统实战：Orchestrator 与 Pipeline 模式解析

单个 Agent 在生产环境中会触及真实的极限。长任务会超出上下文窗口。复杂目标在不同阶段需要不同的工具。当子任务相互独立时，顺序推理效率低下。

多 Agent 系统通过将工作分解到专业化的 Agent 中来解决这些问题，这些 Agent 可以并行运行。本文深入讲解两种模式——Orchestrator/Subagent 和 Pipeline——完全使用 TypeScript 实现。

多 Agent 系统的优势并非抽象概念。考虑一个研究任务："分析我们竞争对手的定价页面并总结关键差异。"

单个 Agent 顺序工作必须：获取第 1 页、处理、获取第 2 页、处理、获取第 3 页、处理，然后撰写分析。每一步都在消耗上下文窗口。总时间是所有步骤之和。

基于编排器的方法：并行生成三个 Agent，每个竞争对手一个。每个 Agent 独立获取并处理其页面。总时间大约是最慢 Agent 的时间，而非所有时间之和。

Single agent (serial):
  Task → [fetch A → fetch B → fetch C → analyze] → Result
  Time: T(A) + T(B) + T(C) + T(analyze)

Multi-agent (parallel):
                    ┌→ [Agent A: fetch + process] ─┐
  Task → [Orchestrator] → [Agent B: fetch + process] → [Synthesize] → Result
                    └→ [Agent C: fetch + process] ─┘
  Time: max(T(A), T(B), T(C)) + T(synthesize)

The Orchestrator/Subagent Pattern

编排器模式分为三个阶段：将目标分解为子任务、执行子任务（在可能的情况下并行）、然后综合结果。

首先，定义在编排器和子 Agent 之间流动的数据结构：

// src/lib/agent/multi/types.ts

export interface SubTask {
  id: string;
  title: string;
  description: string;
  toolSet: 'browser' | 'file' | 'code' | 'rag' | 'general';
  dependsOn?: string[];   // IDs of tasks that must complete first
  priority: 'high' | 'medium' | 'low';
  timeoutMs?: number;
}

export interface SubTaskResult {
  taskId: string;
  status: 'success' | 'failed' | 'timeout' | 'skipped';
  output: string;
  durationMs: number;
}

export interface OrchestratorPlan {
  goal: string;
  tasks: SubTask[];
  estimatedParallelGroups: string[][];  // Which tasks can run concurrently
}

dependsOn 是让依赖图工作的关键——需要另一个任务输出的任务在此列出其 ID。estimatedParallelGroups 是 LLM 建议哪些任务可以同时运行。

Phase 1: Decomposition

编排器调用一次 LLM，将自然语言目标转换为结构化计划：

async decompose(goal: string): Promise<OrchestratorPlan> {
  const { text } = await callLLM([{
    role: 'user',
    content: `Decompose this goal into subtasks. Output JSON only.

Goal: ${goal}

Format:
{
  "goal": "...",
  "tasks": [
    {
      "id": "task_1",
      "title": "...",
      "description": "detailed enough for another AI to complete independently",
      "toolSet": "browser|file|code|rag|general",
      "dependsOn": [],
      "priority": "high|medium|low",
      "timeoutMs": 60000
    }
  ],
  "estimatedParallelGroups": [["task_1", "task_2"], ["task_3"]]
}`,
  }], { system: ORCHESTRATOR_SYSTEM, temperature: 0 });

  return JSON.parse(text.replace(/```
{% endraw %}
json\n?|\n?
{% raw %}
```/g, '').trim());
}

规划阶段使用 temperature: 0——你需要确定性的、结构化的输出，而非创造性。

Phase 2: Parallel Execution with Concurrency Control

执行阶段同时遵守依赖图和并发限制：

async executeParallel(
  plan: OrchestratorPlan,
  maxConcurrency = 3,
): Promise<Map<string, SubTaskResult>> {
  const results = new Map<string, SubTaskResult>();
  const completed = new Set<string>();
  const failed = new Set<string>();

  for (const group of plan.estimatedParallelGroups) {
    // Only run tasks whose dependencies have completed successfully
    const executable = group.filter(taskId => {
      const task = plan.tasks.find(t => t.id === taskId);
      if (!task) return false;
      return (task.dependsOn ?? []).every(
        dep => completed.has(dep) && !failed.has(dep)
      );
    });

    if (executable.length === 0) continue;

    const batchResults = await this.executeBatch(
      executable.map(id => plan.tasks.find(t => t.id === id)!),
      results,
      maxConcurrency,
    );

    for (const [id, result] of batchResults) {
      results.set(id, result);
      result.status === 'success' ? completed.add(id) : failed.add(id);
    }
  }

  return results;
}

private async executeBatch(
  tasks: SubTask[],
  previousResults: Map<string, SubTaskResult>,
  maxConcurrency: number,
): Promise<Map<string, SubTaskResult>> {
  const results = new Map<string, SubTaskResult>();
  const active: Promise<void>[] = [];

  for (const task of tasks) {
    const promise: Promise<void> = this.executeTask(task, previousResults)
      .then(result => { results.set(task.id, result); })
      .catch(err => {
        results.set(task.id, {
          taskId: task.id,
          status: 'failed',
          output: err instanceof Error ? err.message : String(err),
          durationMs: 0,
        });
      })
      .finally(() => { active.splice(active.indexOf(promise), 1); });

    active.push(promise);

    if (active.length >= maxConcurrency) {
      await Promise.race(active);  // Wait for one slot to free up
    }
  }

  await Promise.allSettled(active);
  return results;
}

并发控制模式值得理解：Promise.race(active) 等待第一个活动任务完成后，从池中移除它并允许下一个任务启动。并发运行的任务数量永远不会超过 maxConcurrency。

Executing Individual Subtasks

每个子任务在自己的 ReAct Agent 实例中运行，配有适合其 toolSet 的工具。上游结果作为上下文注入：

private async executeTask(
  task: SubTask,
  previousResults: Map<string, SubTaskResult>,
): Promise<SubTaskResult> {
  const start = Date.now();

  // Inject upstream results as context
  const context = (task.dependsOn ?? [])
    .map(depId => {
      const dep = previousResults.get(depId);
      return dep ? `[${depId}] ${dep.output.slice(0, 1000)}` : '';
    })
    .filter(Boolean)
    .join('\n\n');

  const prompt = context
    ? `Background (from upstream tasks):\n${context}\n\nCurrent task: ${task.description}`
    : task.description;

  const agent = new ReActAgent({
    tools: this.getToolsForTaskType(task.toolSet),
    maxSteps: 8,
  });

  try {
    const timeout = task.timeoutMs ?? 120_000;
    const result = await Promise.race([
      agent.run(prompt),
      new Promise<never>((_, reject) =>
        setTimeout(() => reject(new Error('TIMEOUT')), timeout)
      ),
    ]);

    return {
      taskId: task.id,
      status: result.stopped === 'error' ? 'failed' : 'success',
      output: result.answer,
      durationMs: Date.now() - start,
    };
  } catch (error) {
    const isTimeout = error instanceof Error && error.message === 'TIMEOUT';
    return {
      taskId: task.id,
      status: isTimeout ? 'timeout' : 'failed',
      output: isTimeout ? `Timed out after ${task.timeoutMs ?? 120_000}ms` : String(error),
      durationMs: Date.now() - start,
    };
  }
}

注意：超时是通过 Promise.race 实现的。这会停止等待 Agent，但不会终止底层工作——Agent 的 API 调用可能仍在运行。在生产环境中，你需要通过 AbortController 实现真正的取消。

所有子任务完成后，最后一次 LLM 调用生成答案：

async synthesize(
  goal: string,
  plan: OrchestratorPlan,
  results: Map<string, SubTaskResult>,
): Promise<string> {
  const summaries = plan.tasks
    .map(task => {
      const result = results.get(task.id);
      return `[${task.title}]\nStatus: ${result?.status ?? 'skipped'}\n${result?.output ?? 'No output'}`;
    })
    .join('\n\n---\n\n');

  const { text } = await callLLM([{
    role: 'user',
    content: `Based on all subtask results, answer the original goal.

Original goal: ${goal}

Subtask results:
${summaries}

Provide a complete, well-structured final answer. If some subtasks failed, explain the impact.`,
  }], { temperature: 0.5, maxTokens: 3000 });

  return text;
}

并非所有多 Agent 工作都适合编排器模型。有些任务本质上是顺序的——每个阶段转换前一阶段的输出。对于这些，Pipeline 更清晰。

// src/lib/agent/multi/pipeline.ts

export interface PipelineStage {
  name: string;
  description: "string;"
  tools: Tool[];
  maxSteps?: number;
  buildPrompt: (previousOutput: string, originalInput: string) => string;
  validate?: (output: string) => { valid: boolean; reason?: string };
}

export class AgentPipeline {
  constructor(
    private stages: PipelineStage[],
    private maxRetries = 2,
  ) {}

  async run(initialInput: string): Promise<{ finalOutput: string; success: boolean }> {
    let currentOutput = initialInput;

    for (const stage of this.stages) {
      let retries = 0;
      let success = false;

      while (retries <= this.maxRetries && !success) {
        const agent = new ReActAgent({
          tools: stage.tools,
          maxSteps: stage.maxSteps ?? 6,
          systemPrompt: stage.description,
        });

        const result = await agent.run(
          stage.buildPrompt(currentOutput, initialInput)
        );

        if (stage.validate) {
          const validation = stage.validate(result.answer);
          if (!validation.valid) {
            retries++;
            continue;
          }
        }

        currentOutput = result.answer;
        success = true;
      }

      if (!success) {
        throw new Error(`Pipeline stage "${stage.name}" failed after ${this.maxRetries} retries`);
      }
    }

    return { finalOutput: currentOutput, success: true };
  }
}

每个阶段的 validate 函数是这里关键的设计决策。与其希望每个 Agent 产生可用的输出，不如定义"有效"的含义，并在无效时自动重试。这使得 Pipeline 比单阶段 Agent 可靠得多。

Example: Research Report Pipeline

三个阶段：收集来源、提取洞察、撰写报告。

const researchPipeline = new AgentPipeline([
  {
    name: 'Information Gathering',
    description: "'You are a researcher. Collect information from the web.',"
    tools: [fetchWebpageTool],
    maxSteps: 6,
    buildPrompt: (_, originalInput) =>
      `Gather key information on this topic, including recent developments and data:\n\n${originalInput}`,
    validate: output => ({
      valid: output.length > 200,
      reason: 'Insufficient information gathered',
    }),
  },
  {
    name: 'Analysis',
    description: "'You are an analyst. Extract insights from raw research.',"
    tools: [],
    buildPrompt: (previousOutput, originalInput) => `
Original topic: ${originalInput}

Raw research:
${previousOutput}

Extract: key findings, main trends, supporting data points, open questions.`,
    validate: output => ({
      valid: output.includes('finding') || output.includes('trend'),
      reason: 'Analysis must include findings or trends',
    }),
  },
  {
    name: 'Report Writing',
    description: "'You are a technical writer. Produce a structured report.',"
    tools: [writeFileTool],
    buildPrompt: (previousOutput, originalInput) => `
Topic: ${originalInput}

Analysis:
${previousOutput}

Write a structured report with: Executive Summary, Key Findings, Analysis, Conclusion.`,
  },
]);

const result = await researchPipeline.run(
  'Current state of TypeScript adoption in backend development'
);

多 Agent 系统的故障方式往往不透明。Pipeline 中间的 Agent 产生了糟糕的输出；下一个 Agent 静默地基于它工作；最终结果是错误的，却没有明显的信号说明原因。

为编排器添加进度回调，为每个阶段添加结构化日志：

const orchestrator = new Orchestrator({
  onProgress: (event) => {
    console.error(`[${event.type}] ${event.message}`);
    // In production: emit SSE event to frontend, write to tracing system
  },
});

onProgress 回调在每个关键里程碑触发：计划完成、组开始、任务开始、任务完成（带状态）、综合开始、全部完成。通过这些事件，你可以在 UI 中显示实时进度，并且在故障时拥有完整的审计跟踪。

What to Watch Out For

Token 成本会累积。每个子 Agent 运行自己的 ReAct 循环。一个 6 任务的编排工作，每个 Agent 5 步，在综合之前就是 30 次 LLM 调用。合理预算，并积极设置 maxSteps。

计划是建议，不是保证。LLM 可能将实际上有隐式依赖的任务放入 estimatedParallelGroups。在执行前始终验证依赖图的一致性。

故障模式不同。在单个 Agent 中，故障很明显——没有输出。在多 Agent 系统中，一个任务可能静默失败而其他任务成功，综合步骤可能用听起来合理的输出掩盖缺口。每个阶段的明确状态跟踪和输出验证不是可选项。

Promise.race 处理超时不会取消。超时触发后，Agent 底层的 API 调用仍在运行。在生产环境中，将 AbortController 信号传递到每个 LLM 调用。

本文改编自《AI Engineering with TypeScript》第十六章——一本关于在 Leanpub 或 Amazon 上构建 AI Agent 的综合指南