Agent Composition Pattern 详解 - Agent 组合模式

什么是 Agent Composition？

生活类比

你一个人能干所有事吗？搜索、写作、代码、设计、翻译...

不行。所以你组了个团队：搜索员负责找资料，写手负责写内容，程序员负责写代码，设计师负责出图。

每个人都有专长，组合在一起就是一支"超级团队"。

这就是 Agent Composition——把多个专门的 Agent 像乐高一样拼装起来，组成一个强大的系统。

就像复仇者联盟——钢铁侠+美国队长+雷神+黑寡妇，各自单打不过，合体就是宇宙最强。

正式定义：Agent Composition Pattern 是将多个 AI Agent 按照特定结构组合在一起的架构模式，通过定义 Agent 之间的协作方式、数据流和决策权，实现复杂任务的分解与协同完成。

核心组合模式

模式 1：Sequential（序列组合）

Agent A Agent B Agent C ┌────────┐ ┌────────┐ ┌────────┐ │ 搜索 │ ─→│ 分析 │ ─→│ 写作 │ ─→ 最终输出 │ Agent │ │ Agent │ │ Agent │ └────────┘ └────────┘ └────────┘ 像流水线：前一个的输出是后一个的输入

// 序列组合 - 流水线模式
async function sequentialPipeline(task) {
  // 步骤1：搜索 Agent
  const searchResult = await runAgent('search-agent', {
    query: task.query
  });
  
  // 步骤2：分析 Agent（使用搜索结果）
  const analysis = await runAgent('analysis-agent', {
    data: searchResult,
    focus: task.focus
  });
  
  // 步骤3：写作 Agent（使用分析结果）
  const article = await runAgent('writer-agent', {
    analysis: analysis,
    style: task.style
  });
  
  return article;
}

模式 2：Parallel（并行组合）

┌─ Agent A ──┐ 任务 ──→ │ 搜索英文 │ ─┐ └────────────┘ │ ┌─ Agent B ──┐ ├──→ 合并输出 │ 搜索中文 │ ─┤ └────────────┘ │ ┌─ Agent C ──┐ │ │ 搜索日文 │ ─┘ └────────────┘ 像多线程：多个 Agent 同时干不同的事

// 并行组合 - 同时搜索多个来源
async function parallelSearch(task) {
  // 同时搜索多个来源
  const [webResults, paperResults, newsResults] = await Promise.all([
    runAgent('web-search-agent', { query: task.query }),
    runAgent('paper-search-agent', { query: task.query }),
    runAgent('news-search-agent', { query: task.query })
  ]);
  
  // 合并结果
  const merged = mergeResults(webResults, paperResults, newsResults);
  return merged;
}

// 并行的好处：3个搜索各2秒 = 总共2秒（而不是6秒）

模式 3：Hierarchical（层级组合）

┌─────────────────────────┐ │ Manager Agent (管理者) │ │ 接收任务 → 分配 → 合并 │ └────┬───────┬───────┬─────┘ │ │ │ ┌────────▼─┐ ┌──▼──────▼──┐ ┌▼──────────┐ │ Worker A │ │ Worker B │ │ Worker C │ │ 搜索 │ │ 写作 │ │ 审核 │ └──────────┘ └─────────────┘ └───────────┘ 像公司架构：Manager 分配任务，Worker 执行

// 层级组合 - Manager/Worker 模式
async function hierarchicalTask(task) {
  // Manager 分析任务并拆分
  const subtasks = await runAgent('manager-agent', {
    task: task,
    action: 'decompose'
  });
  
  // 分配给不同的 Worker
  const results = await Promise.all(
    subtasks.map(subtask => 
      runAgent(subtask.assignedAgent, subtask)
    )
  );
  
  // Manager 合并结果
  const finalResult = await runAgent('manager-agent', {
    action: 'merge',
    results: results
  });
  
  return finalResult;
}

模式 4：Dynamic（动态组合）

┌────────────────────────────────────────┐ │ Orchestrator (编排器) │ │ │ │ 分析任务 → 选择最佳 Agent 组合 → 执行 │ │ ↕ ↕ │ │ 评估结果 → 需要更多 Agent? → 动态添加 │ └────────────────────────────────────────┘ 像 Uber：根据需求动态调度

模式对比

模式	通信方式	延迟	复杂度	适用场景
Sequential	线性传递	高（累加）	低	流水线任务
Parallel	并发执行	低（取最长）	中	独立子任务
Hierarchical	Manager 协调	中	高	复杂分解
Dynamic	按需连接	不确定	最高	未知任务

OpenClaw 实战应用

1. SubAgent 序列模式

// OpenClaw SubAgent 组合
// 使用 sessions_spawn 创建子 Agent
const searchAgent = await sessions_spawn({
  task: "搜索最近AI行业热点新闻",
  label: "search-worker",
  runtime: "subagent",
  streamTo: "parent"
});

// 等待搜索完成
const searchResults = await searchAgent.result;

// 传给写作 Agent
const writerAgent = await sessions_spawn({
  task: `基于以下信息写AI新闻日报：${searchResults}`,
  label: "writer-worker",
  runtime: "subagent"
});

2. SubAgent 并行模式

// OpenClaw 并行 SubAgent
const [hackerNews, githubTrending, arxiv] = await Promise.all([
  sessions_spawn({
    task: "搜索 Hacker News 上的 AI 热帖",
    label: "hn-worker",
    runtime: "subagent"
  }),
  sessions_spawn({
    task: "获取 GitHub Trending 的 AI 项目",
    label: "gh-worker",
    runtime: "subagent"
  }),
  sessions_spawn({
    task: "获取 arXiv 上的最新 AI 论文",
    label: "arxiv-worker",
    runtime: "subagent"
  })
]);

// 合并所有结果
const allResults = await Promise.all([
  hackerNews.result,
  githubTrending.result,
  arxiv.result
]);

3. 层级模式：Manager + Workers

// OpenClaw SubAgent 层级模式
// 主 Agent (Manager) 分配任务给子 Agent (Workers)

// 1. Manager 拆分任务
const subtasks = [
  { task: "搜索5个OpenClaw教程主题", worker: "researcher" },
  { task: "搜索5个AI Agent术语", worker: "researcher" },
  { task: "搜索5个竞品动态", worker: "researcher" }
];

// 2. 并行分配给 Worker
const workers = subtasks.map(st =>
  sessions_spawn({
    task: st.task,
    agentId: st.worker,
    runtime: "subagent",
    label: st.worker
  })
);

// 3. 收集结果
const results = await Promise.all(workers.map(w => w.result));

// 4. Manager 合并（在主 Agent 中完成）

4. 持久 Session 模式

// OpenClaw Thread-bound Session
// 适合需要多轮交互的场景
await sessions_spawn({
  task: "你是妙趣AI的写作助手",
  mode: "session",     // 持久会话
  thread: true,         // 绑定到线程
  runtime: "subagent",
  label: "persistent-writer"
});

// 后续可以发送消息继续交互
await sessions_send({
  sessionKey: "persistent-writer",
  message: "帮我写一篇关于 Agent Composition 的文章"
});

常见坑点

踩坑实录

坑1：Agent 太多，成本爆炸

每个 Agent 都要调 LLM，10 个 Agent 就是 10 次 LLM 调用。需要合并 Agent、减少不必要的层级。

坑2：上下文丢失

A Agent 的输出传给 B Agent，中间丢失了关键信息。需要设计好中间数据格式，确保信息完整传递。

坑3：循环依赖

A 等 B，B 等 C，C 等 A...死锁了。组合前画好依赖图，确保没有循环。

坑4：错误传播

C Agent 出错，整个流水线失败。需要错误隔离和降级机制——C 挂了，跳过它，用默认值继续。

坑5：调试困难

10 个 Agent 组合出错，到底是谁的锅？需要完善的可观测性（参考 Observability Stack）。

最佳实践

简单优先：能用一个 Agent 解决的，不要用多个
明确接口：每个 Agent 的输入输出格式要标准化
并行化：独立子任务尽量并行执行
错误隔离：一个 Agent 失败不应影响整个系统
上下文压缩：Agent 之间传递的信息要精简
可观测性：记录每个 Agent 的执行情况
成本控制：监控总 LLM 调用次数和 Token 消耗
逐步构建：先串行再并行再层级，逐步复杂化

Agent Composition Pattern