多基因 Agent 流水线组合

单个基因已经很强大，但真正的魔法在于组合。基因代数 — Seq、Par、Cond、Try 和 Transform — 让你把简单基因连接成复杂的 Agent 流水线，同时保持类型安全、可验证和自动优化。

本教程将带你构建一个实际的流水线：搜索网络 → 摘要结果 → 格式化输出 — 然后扩展它，加入并行执行、条件分支和错误恢复。

前置条件

• 已安装 Rotifer CLI（npm i -g @rotifer/playground）
• 已初始化项目（rotifer init my-pipeline && cd my-pipeline）
• 了解基本基因概念（参见 5 分钟创建你的第一个基因）

第一步：了解构建块

项目已包含 genesis 基因。查看现有基因：

rotifer arena list

┌──────┬─────────────────────┬────────────┬────────┬──────────┐
│ #    │ Name                │ Domain     │ F(g)   │ Fidelity │
├──────┼─────────────────────┼────────────┼────────┼──────────┤
│ 1    │ genesis-web-search  │ search.web │ 0.9200 │ Native   │
│ 2    │ genesis-search-lite │ search.web │ 0.7100 │ Native   │
│ 3    │ genesis-code-format │ code.format│ 0.8800 │ Native   │
└──────┴─────────────────────┴────────────┴────────┴──────────┘

我们将用 genesis-web-search 作为第一阶段。接下来需要一个摘要基因。

第二步：创建摘要基因

mkdir -p genes/summarizer

编写 genes/summarizer/index.ts：

export async function express(input: {
  text: string;
  maxLength?: number;
}) {
  const maxLen = input.maxLength || 200;
  const sentences = input.text.split(/[.!?]+/).filter(Boolean);

  let summary = "";
  for (const sentence of sentences) {
    if ((summary + sentence).length > maxLen) break;
    summary += sentence.trim() + ". ";
  }

  return {
    summary: summary.trim(),
    wordCount: summary.split(/\s+/).length,
  };
}

包装并提交：

rotifer wrap summarizer --domain text.summarize
rotifer compile summarizer
rotifer arena submit summarizer

第三步：第一个组合 — Seq（顺序执行）

Seq 按顺序执行基因，将每个基因的输出作为下一个基因的输入：

Seq(A, B, C) = A → B → C

但是 genesis-web-search 的输出是 { results: [...] }，而 summarizer 期望 { text: string }。我们需要 Transform 来桥接 schema：

import { Seq, Transform } from "@rotifer/algebra";

const searchAndSummarize = Seq(
  "genesis-web-search",
  Transform((searchResult) => ({
    text: searchResult.results.map(r => r.snippet).join(" "),
    maxLength: 200,
  })),
  "summarizer"
);

创建 Agent：

rotifer agent create researcher --genes genesis-web-search summarizer

运行：

rotifer agent run researcher --input '{"query": "quantum computing 2026"}' --verbose

--verbose 标志显示每个阶段的中间输入和输出。

第四步：并行执行 — Par

如果你想同时搜索多个数据源？Par 并发执行基因：

Par(A, B, C) = A ‖ B ‖ C → [resultA, resultB, resultC]

import { Par, Seq, Transform } from "@rotifer/algebra";

const multiSourceSearch = Seq(
  Par(
    "genesis-web-search",
    "genesis-web-search-lite"
  ),
  Transform((results) => ({
    text: results.flat().map(r => r.results?.map(x => x.snippet)).flat().join(" "),
    maxLength: 300,
  })),
  "summarizer"
);

两个搜索通过线程池并行运行。结果收集到数组中，然后由 Transform 合并后传给摘要基因。

第五步：条件分支 — Cond

Cond 根据运行时谓词路由执行：

import { Cond, Seq, Transform } from "@rotifer/algebra";

const adaptivePipeline = Seq(
  "genesis-web-search",
  Cond(
    (result) => result.results.length > 10,
    Seq(
      Transform((r) => ({
        text: r.results.map(x => x.snippet).join(" "),
        maxLength: 500,
      })),
      "summarizer"
    ),
    Transform((r) => ({
      summary: r.results[0]?.snippet || "未找到结果。",
      wordCount: 0,
    }))
  )
);

如果搜索返回超过 10 条结果，进行摘要处理。否则，直接返回第一条摘要。

第六步：错误恢复 — Try

Try 尝试主基因，如果失败则回退到备用基因：

import { Try, Seq, Transform } from "@rotifer/algebra";

const resilientPipeline = Seq(
  Try(
    "genesis-web-search",
    "genesis-web-search-lite"
  ),
  Transform((r) => ({
    text: r.results.map(x => x.snippet).join(" "),
    maxLength: 200,
  })),
  "summarizer"
);

如果主搜索失败（网络错误、速率限制等），执行自动回退到 lite 版本。无需手动错误处理。

第七步：完整流水线

将所有算子组合成一个生产级流水线：

import { Seq, Par, Cond, Try, Transform } from "@rotifer/algebra";

const productionPipeline = Seq(
  // 阶段 1：弹性多源搜索
  Try(
    Par("genesis-web-search", "genesis-web-search-lite"),
    Par("genesis-web-search-lite")  // 回退：单一数据源
  ),

  // 阶段 2：合并并行结果
  Transform((results) => ({
    text: results.flat()
      .map(r => r.results?.map(x => x.snippet))
      .flat()
      .filter(Boolean)
      .join(" "),
    resultCount: results.flat().reduce((n, r) => n + (r.results?.length || 0), 0),
  })),

  // 阶段 3：自适应摘要
  Cond(
    (data) => data.resultCount > 5,
    Seq(
      Transform((d) => ({ text: d.text, maxLength: 400 })),
      "summarizer"
    ),
    Transform((d) => ({ summary: d.text.slice(0, 200), wordCount: 0 }))
  ),

  // 阶段 4：格式化输出
  "genesis-code-format"
);

创建并运行 Agent：

rotifer agent create research-bot \
  --genes genesis-web-search genesis-web-search-lite summarizer genesis-code-format

rotifer agent run research-bot \
  --input '{"query": "rotifer protocol gene evolution"}' \
  --verbose

类型安全

组合代数在组合时强制执行 schema 兼容性。如果你连接了两个不兼容的基因，会得到清晰的错误：

Error[E0032]: Type mismatch in Seq composition
  → gene 'web-search' output: { results: SearchResult[] }
  → gene 'summarizer' input:  { text: string, maxLength?: number }

  help: Add a Transform between the genes to reshape the data

这在运行时之前就能捕获数据流 bug。

组合基因的适应度

组合有自己的适应度评分：

算子	适应度公式
Seq	min(F(组件)) × 延迟惩罚
Par	avg(F(组件)) × 并行奖励
Try	F(主) × 成功率 + F(备用) × (1 - 成功率)

Arena 将组合作为整体评估，因此存在选择压力推动高效的组合结构。

你学到了什么

• Seq 将基因链接成顺序流水线
• Par 并发运行基因，提升速度和冗余
• Cond 根据运行时条件路由执行
• Try 提供自动错误恢复与回退
• Transform 桥接基因之间的 schema 不匹配
• 组合是类型安全的，并且有复合适应度评分

---

深入阅读： 参见完整的组合模式指南了解所有算子、类型约束和适应度公式。Agent CLI 命令参见 Agent 参考。