swarm

Swarm

并行处理许多相互独立的项目。create 会创建表句柄； run 会将工作分派到各行，并将结果合并回表中。一行 = 一个工作单元 — swarm 会自动处理批次。

流程

1. 创建。 根据来源创建表 — 文件、glob 模式或 预解析的记录。每个项目占一行。返回一个句柄。
2. 运行。 跨各行分派 instruction 模板。结果会合并 回表中。返回 { completed, failed, skipped, failures }。
3. 汇总。 使用 rows() 和普通 JS 进行计数、筛选或汇总。 不要为汇总另外启动子 Agent。
4. 重试。 使用 filter: { column: "<col>", exists: false } 重新运行， 仅重新处理失败的行。

选择来源

glob / filePaths — 一个文件 = 一行。适用于每个文件都是独立 工作单元的情况。每行会获得 { id, file }；子 Agent 会通过 {file} 占位符自行读取文件。

tasks — 直接传入预先构建的记录。适用于数据存储在文件中（JSONL、CSV、JSON 数组）的情况。先在 eval 中读取并解析文件，然后传入记录。一条记录 = 一行 — 不要将 多个项目合并到同一行。

对于较小的文件（约 500 行以内），在一个代码块中解析并创建：

const { create } = await import("@/skills/swarm");
const raw = await tools.readFile({ file_path: "/data.jsonl" });
const records = raw.trim().split("\n").map(l => JSON.parse(l));
const table = await create({ tasks: records });
console.log(table);

对于较大的文件，请以每次 500 行的方式分块读取，以避免内容被截断：

const { create } = await import("@/skills/swarm");
let records = [];
let offset = 0;
while (true) {
  const chunk = await tools.readFile({ file_path: "/data.txt", offset, limit: 500 });
  const lines = chunk.split("\n").filter(l => l.trim());
  for (const l of lines) { records.push({ id: `r${records.length}`, text: l }); }
  if (lines.length < 500) break;
  offset += 500;
}
const table = await create({ tasks: records });
console.log(table);

当文件过大，无法在一次 eval 调用中完成解析和分派时，请拆分为两个代码块。只有调用 swarm 函数的代码块需要导入：

// eval 1: parse only — no swarm import needed
const raw = await tools.readFile({ file_path: "/data.jsonl" });
globalThis.records = raw.trim().split("\n").map(l => JSON.parse(l));
console.log(`Parsed ${globalThis.records.length} records`);

// eval 2: create and dispatch
const { create, run } = await import("@/skills/swarm");
const table = await create({ tasks: globalThis.records });
const result = await run(table.id, {
  instruction: "Classify {text}",
  responseSchema: {
    type: "object",
    properties: { label: { type: "string" } },
    required: ["label"],
  },
});
console.log(result);

传入 filePaths: ["/data.jsonl"] 会创建一个仅有一行的表， 该行指向这个文件 — 而不是文件中的每条记录各占一行。

何时使用 subagentType

对于分类、提取、标注，以及单次模型调用并使用结构化输出即可完成的任务，请省略 subagentType。这是 默认选项，成本和耗时都显著更低 — 每次分派都是直接的模型调用，不使用工具，也不进行迭代。

当任务需要工具、文件访问或多步骤推理时，设置 subagentType。每次分派都会由指定的子 Agent 执行完整的 Agent 式循环。

// Direct model call — classification, no tools needed
await run(table.id, {
  instruction: "Classify {text}",
  responseSchema: { type: "object", properties: { label: { type: "string" } }, required: ["label"] },
});

// Subagent — needs to read files and reason over multiple steps
await run(table.id, {
  subagentType: "reviewer",
  instruction: "Review {file} for security issues.",
  responseSchema: { type: "object", properties: { finding: { type: "string" } }, required: ["finding"] },
});

指令与上下文

instruction 是一个用于逐项处理的模板，其中包含 {column} 占位符。 占位符由框架解析 — 你的列名会在提示词中作为对旁边所列值的引用出现，而不会作为原始模板语法出现。子 Agent 负责执行工作 — 不要自行用 JS 处理项目并将结果写入行中。

context 是会添加到每个子 Agent 提示词前的自由格式文本。可用于提供共享背景信息：领域术语、分类规则、示例等。

const { create, run } = await import("@/skills/swarm");

const table = await create({ glob: "src/**/*.ts" });
const r = await run(table.id, {
  subagentType: "reviewer",
  instruction: "Review {file} for security issues. List findings or write 'no issues'.",
  context: "TypeScript Express backend using Prisma ORM. Focus on injection, auth bypass, path traversal.",
  responseSchema: {
    type: "object",
    properties: { review: { type: "string" } },
    required: ["review"],
  },
});
console.log(r);
// → { completed: 45, failed: 2, skipped: 0, failures: [...] }

结构化输出

必须提供 responseSchema。Schema 属性会成为每行的顶层列，并限制子 Agent 可以返回的内容。

const { run } = await import("@/skills/swarm");
await run(table.id, {
  instruction: "Classify: {text}",
  responseSchema: {
    type: "object",
    properties: {
      sentiment: { type: "string", enum: ["positive", "negative", "neutral"] },
    },
    required: ["sentiment"],
  },
});
// Row after: { id: "r1", text: "...", sentiment: "positive" }

批处理

默认情况下，swarm 会自动批处理，使分派总数少于 10。对于较小的表（≤10 行），每行都会单独调用子 Agent。对于较大的表， 系统会自动将多行分组。

设置 batchSize 以控制分组：

• 数字 — 所有行使用统一的批次大小。batchSize: 1 强制逐行分派；batchSize: 20 每 20 行分为一组。
• 函数 — (row, rowCount) => number。为每一行返回所需的批次大小。相同批次大小的行会先分到一起，然后再分块。可混合分派，让部分行单独处理，其他行进行批处理。

const { create, run } = await import("@/skills/swarm");
const table = await create({ tasks: items });

// Complex items get individual attention; simple ones batch together
await run(table.id, {
  instruction: "Analyze {text}",
  responseSchema: {
    type: "object",
    properties: { analysis: { type: "string" } },
    required: ["analysis"],
  },
  batchSize: (row) => (row.token_count > 1000 ? 1 : 10),
});

计算后，批次大小会限制在 [1, 50] 范围内。

汇总

run() 之后，使用 rows() 和普通 JS — 无需额外启动子 Agent。

const { rows } = await import("@/skills/swarm");
const data = await rows(table.id, { columns: ["sentiment"] });
const counts = {};
data.forEach(r => { counts[r.sentiment] = (counts[r.sentiment] || 0) + 1 });
console.log(counts);
// → { positive: 120, negative: 45, neutral: 35 }

串联多轮处理

run 会原地更新表 — 串联调用即可累积列。

const { create, run } = await import("@/skills/swarm");
const table = await create({ tasks: interviews });
await run(table.id, {
  instruction: "Classify sentiment of {text}",
  responseSchema: {
    type: "object",
    properties: { sentiment: { type: "string", enum: ["positive", "negative", "neutral"] } },
    required: ["sentiment"],
  },
});
await run(table.id, {
  filter: { column: "sentiment", equals: "negative" },
  instruction: "Summarize why {text} had negative sentiment.",
  responseSchema: {
    type: "object",
    properties: { summary: { type: "string" } },
    required: ["summary"],
  },
});

仅执行操作的任务

当子 Agent 执行操作（写入文件、应用修复）而不是返回数据时，请使用包含状态或标记字段的简单 schema。重试仍可使用 exists: false 筛选条件。

const { create, run } = await import("@/skills/swarm");
const fixedSchema = {
  type: "object",
  properties: { fixed: { type: "string" } },
  required: ["fixed"],
};
const table = await create({ glob: "src/**/*.ts" });
await run(table.id, {
  subagentType: "fixer",
  instruction: "Add missing JSDoc to all exported functions in {file}.",
  responseSchema: fixedSchema,
});
// retry any that failed
await run(table.id, {
  subagentType: "fixer",
  instruction: "Add missing JSDoc to all exported functions in {file}.",
  responseSchema: fixedSchema,
  filter: { column: "fixed", exists: false },
});

筛选

{ column: "status", equals: "done" }
{ column: "status", notEquals: "done" }
{ column: "category", in: ["A", "B"] }
{ column: "result", exists: false }      // not yet processed
{ and: [filter1, filter2] }
{ or: [filter1, filter2] }

技术说明

• 仅在调用 swarm 函数的代码块中导入 @/skills/swarm。 准备数据（读取文件、解析数据、将数据存入 globalThis）不需要导入。 只解构你会用到的内容：{ create }、{ run }、{ create, run } 等。
• 控制台输出上限约为 5 KB。 切勿记录原始文件内容—— 只记录数量和简短样例。
• 在 eval 中调用 readFile 会返回原始内容，不带行号前缀。 每次调用最多读取 500 行。对于超过 500 行的文件，请使用递增的 offset 循环读取。
• 根据文件构建表格时，请在 eval 中读取文件。 在沙盒中读取的数据会留在沙盒内，不会进入 Agent 的上下文窗口。
• 切勿直接写入 .swarm/。 始终使用 create()。
• 子 Agent 所需的所有内容都必须放在 instruction + context 中。 子 Agent 无法查看 Agent 的上下文。
• 行 ID 必须唯一。 如果来源产生重复 ID，create() 会拒绝。 对于 tasks，这由调用方负责；对于 glob / filePaths，ID 会根据父目录自动消歧。
• 未知列会立即导致失败。 如果 instruction 引用了 {foo}，但没有匹配的行提供 foo，run() 会在分派任何子 Agent 之前抛出错误。

API 参考

create(source)

创建表格。返回句柄 { id, count, columns }。

来源	描述
{ glob: "src/**/*.ts" } 或 { glob: ["src/**/*.ts", "lib/**/*.ts"] }	按一个或多个模式匹配文件。列：id、file
{ filePaths: ["a.ts", "b.ts"] }	显式文件列表。列：id、file
{ tasks: [{ id: "t1", text: "..." }] }	自定义行。每行都必须有 id

run(tableId, options)

为各行分派工作。返回 { completed, failed, skipped, failures }。

选项	默认值	描述
instruction	（必填）	含有 {column} 占位符的模板
responseSchema	（必填）	JSON Schema（type: "object"）——属性会成为行列
context	—	添加到每个子 Agent 提示词前面的说明文字
filter	—	仅分派匹配的行
subagentType	—	要分派给的子 Agent 名称。设置后会运行完整的 Agent 循环；省略时则直接调用模型
batchSize	自动	数字或 (row, rowCount) => number。自动模式最多分派 10 项；1 = 每行一项；函数 = 按行设置批次大小
concurrency	10	同时分派的子 Agent 数量上限（限制在 1–10 之间）

rows(tableId, options?)

检索行。用于检查和基于 JS 的聚合。

选项	描述
filter	仅返回匹配的行
columns	投影到指定列
limit	返回的行数上限