From 985fd5c6f697f4fcf74c9d19badbc3d4072c625f Mon Sep 17 00:00:00 2001 From: Kaiyi Date: Fri, 29 May 2026 15:14:23 +0800 Subject: [PATCH] feat(agent-core): add SwarmCoordinator (plan, parallel workers, synthesize) --- .../agent-core/src/agent/swarm/coordinator.ts | 91 +++++++++++++++++++ .../agent-core/src/agent/swarm/prompts.ts | 57 ++++++++++++ .../agent-core/test/swarm/coordinator.test.ts | 83 +++++++++++++++++ 3 files changed, 231 insertions(+) create mode 100644 packages/agent-core/src/agent/swarm/coordinator.ts create mode 100644 packages/agent-core/src/agent/swarm/prompts.ts create mode 100644 packages/agent-core/test/swarm/coordinator.test.ts diff --git a/packages/agent-core/src/agent/swarm/coordinator.ts b/packages/agent-core/src/agent/swarm/coordinator.ts new file mode 100644 index 000000000..a80605b35 --- /dev/null +++ b/packages/agent-core/src/agent/swarm/coordinator.ts @@ -0,0 +1,91 @@ +import { mapWithConcurrency } from './concurrency'; +import { parsePlan } from './parse'; +import { + DEFAULT_WORKER_TOOLS, + PLANNER_SYSTEM_PROMPT, + SYNTHESIZER_SYSTEM_PROMPT, + renderPlannerPrompt, + renderPlannerRetryPrompt, + renderSynthesizerPrompt, +} from './prompts'; +import type { SwarmCoordinatorDeps, SwarmPlan } from './types'; + +export class SwarmCoordinator { + constructor(private readonly deps: SwarmCoordinatorDeps) {} + + private progress(text: string): void { + this.deps.onProgress?.(text); + } + + async run(rootTask: string): Promise { + this.deps.signal.throwIfAborted(); + this.progress('Planning subtasks…'); + const plan = await this.decompose(rootTask); + this.progress(`Planned ${String(plan.subtasks.length)} subtasks`); + + await this.runWave(plan); + + this.progress('Synthesizing results…'); + const result = await this.deps.spawnSubagent({ + profileName: 'swarm-synthesizer', + systemPrompt: SYNTHESIZER_SYSTEM_PROMPT, + tools: [], + prompt: renderSynthesizerPrompt(plan), + description: 'Swarm synthesizer', + signal: this.deps.signal, + }); + return result.result; + } + + private async decompose(rootTask: string): Promise { + const first = await this.deps.spawnSubagent({ + profileName: 'swarm-planner', + systemPrompt: PLANNER_SYSTEM_PROMPT, + tools: [], + prompt: renderPlannerPrompt(rootTask), + description: 'Swarm planner', + signal: this.deps.signal, + }); + const plan = parsePlan(rootTask, first.result); + if (plan !== null) return plan; + + const retry = await this.deps.spawnSubagent({ + profileName: 'swarm-planner', + systemPrompt: PLANNER_SYSTEM_PROMPT, + tools: [], + prompt: renderPlannerRetryPrompt(rootTask, first.result), + description: 'Swarm planner (retry)', + signal: this.deps.signal, + }); + const retried = parsePlan(rootTask, retry.result); + if (retried !== null) return retried; + + throw new Error('Swarm planner failed to produce a valid plan after one retry'); + } + + private async runWave(plan: SwarmPlan): Promise { + const limit = this.deps.maxConcurrency ?? 4; + await mapWithConcurrency(plan.subtasks, limit, async (st) => { + this.deps.signal.throwIfAborted(); + st.status = 'running'; + this.progress(`▸ ${st.role}: started`); + try { + const out = await this.deps.spawnSubagent({ + profileName: `swarm:${st.role}`, + systemPrompt: st.systemPrompt, + tools: st.toolAllowlist ?? DEFAULT_WORKER_TOOLS, + prompt: st.prompt, + description: st.role, + signal: this.deps.signal, + }); + st.result = out.result; + st.status = 'done'; + this.progress(`✓ ${st.role}: done`); + } catch (err) { + st.status = 'failed'; + st.error = err instanceof Error ? err.message : String(err); + this.progress(`✗ ${st.role}: failed (${st.error})`); + } + }); + } +} diff --git a/packages/agent-core/src/agent/swarm/prompts.ts b/packages/agent-core/src/agent/swarm/prompts.ts new file mode 100644 index 000000000..1d5e9bd2e --- /dev/null +++ b/packages/agent-core/src/agent/swarm/prompts.ts @@ -0,0 +1,57 @@ +import type { SwarmPlan } from './types'; + +/** Read-only default tool set for workers; planner may widen via toolAllowlist. */ +export const DEFAULT_WORKER_TOOLS: string[] = ['Read', 'Grep', 'Glob', 'WebSearch', 'FetchURL']; + +/** Tool names a worker is allowed to request (excludes Agent/Swarm dispatch tools). */ +export const ALLOWED_WORKER_TOOLS: string[] = [ + 'Read', + 'Write', + 'Edit', + 'Grep', + 'Glob', + 'Bash', + 'WebSearch', + 'FetchURL', + 'ReadMediaFile', +]; + +export const PLANNER_SYSTEM_PROMPT = [ + 'You are a swarm planner. Decompose the user task into independent subtasks that can run in parallel.', + 'For each subtask invent a short role name, a focused system prompt for that role, and a concrete prompt.', + 'Optionally specify toolAllowlist (a subset of the allowed tools) when a subtask needs more than read-only access.', + `Allowed tools: ${ALLOWED_WORKER_TOOLS.join(', ')}.`, + 'Output ONLY a JSON object, no prose, matching exactly:', + '{"subtasks":[{"id":"task-1","role":"...","systemPrompt":"...","prompt":"...","toolAllowlist":["Read"]}]}', + 'Keep it to at most 6 subtasks. Each subtask must be self-contained (workers cannot see each other).', +].join('\n'); + +export function renderPlannerPrompt(rootTask: string): string { + return `Task to decompose:\n${rootTask}\n\nReturn only the JSON plan.`; +} + +export function renderPlannerRetryPrompt(rootTask: string, previous: string): string { + return [ + `Task to decompose:\n${rootTask}`, + '', + 'Your previous response was not valid JSON in the required shape:', + previous.slice(0, 1000), + '', + 'Return ONLY the JSON object, with a non-empty "subtasks" array. No prose, no code fences.', + ].join('\n'); +} + +export const SYNTHESIZER_SYSTEM_PROMPT = [ + 'You are a swarm synthesizer. You are given the original task and the outputs of several worker subagents.', + 'Merge them into one coherent, complete answer for the user.', + 'If a subtask failed, note the gap explicitly instead of inventing its content.', +].join('\n'); + +export function renderSynthesizerPrompt(plan: SwarmPlan): string { + const blocks = plan.subtasks.map((st) => { + const body = + st.status === 'done' ? (st.result ?? '') : `[FAILED: ${st.error ?? 'unknown error'}]`; + return `### ${st.role} (${st.status})\n${body}`; + }); + return [`Original task:\n${plan.rootTask}`, '', 'Worker outputs:', '', ...blocks].join('\n'); +} diff --git a/packages/agent-core/test/swarm/coordinator.test.ts b/packages/agent-core/test/swarm/coordinator.test.ts new file mode 100644 index 000000000..55e796d25 --- /dev/null +++ b/packages/agent-core/test/swarm/coordinator.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { SwarmCoordinator } from '../../src/agent/swarm/coordinator'; +import type { SpawnSubagentFn } from '../../src/agent/swarm/types'; + +const PLAN_JSON = JSON.stringify({ + subtasks: [ + { role: 'Researcher', systemPrompt: 'sp-research', prompt: 'p-research' }, + { role: 'Analyst', systemPrompt: 'sp-analyst', prompt: 'p-analyst', toolAllowlist: ['Read'] }, + ], +}); + +function makeSpawner(byProfile: Record): SpawnSubagentFn { + return vi.fn(async (args) => { + if (args.profileName === 'swarm-planner') return { result: '```json\n' + PLAN_JSON + '\n```' }; + if (args.profileName === 'swarm-synthesizer') return { result: 'FINAL ANSWER' }; + const key = args.profileName; + return { result: byProfile[key] ?? `done:${args.description}` }; + }); +} + +describe('SwarmCoordinator.run', () => { + it('plans, runs workers concurrently, and synthesizes', async () => { + const spawn = makeSpawner({}); + const coordinator = new SwarmCoordinator({ + spawnSubagent: spawn, + signal: new AbortController().signal, + maxConcurrency: 4, + }); + + const result = await coordinator.run('do a thing'); + + expect(result).toBe('FINAL ANSWER'); + const calls = (spawn as ReturnType).mock.calls.map((c) => c[0]); + expect(calls).toHaveLength(4); + expect(calls[0].profileName).toBe('swarm-planner'); + expect(calls.some((c) => c.profileName === 'swarm:Researcher' && c.systemPrompt === 'sp-research')).toBe(true); + expect(calls.some((c) => c.profileName === 'swarm:Analyst' && c.tools.includes('Read'))).toBe(true); + expect(calls[calls.length - 1].profileName).toBe('swarm-synthesizer'); + }); + + it('retries planning once on invalid JSON, then succeeds', async () => { + let first = true; + const spawn: SpawnSubagentFn = vi.fn(async (args) => { + if (args.profileName === 'swarm-planner') { + if (first) { + first = false; + return { result: 'not json at all' }; + } + return { result: PLAN_JSON }; + } + if (args.profileName === 'swarm-synthesizer') return { result: 'OK' }; + return { result: 'worker-done' }; + }); + const coordinator = new SwarmCoordinator({ spawnSubagent: spawn, signal: new AbortController().signal }); + const result = await coordinator.run('x'); + expect(result).toBe('OK'); + }); + + it('throws when planning fails twice', async () => { + const spawn: SpawnSubagentFn = vi.fn(async () => ({ result: 'never json' })); + const coordinator = new SwarmCoordinator({ spawnSubagent: spawn, signal: new AbortController().signal }); + await expect(coordinator.run('x')).rejects.toThrow(/valid plan/i); + }); + + it('records a failed worker and still synthesizes', async () => { + const spawn: SpawnSubagentFn = vi.fn(async (args) => { + if (args.profileName === 'swarm-planner') return { result: PLAN_JSON }; + if (args.profileName === 'swarm-synthesizer') return { result: 'SYNTH' }; + if (args.profileName === 'swarm:Researcher') throw new Error('boom'); + return { result: 'analyst-done' }; + }); + const onProgress = vi.fn(); + const coordinator = new SwarmCoordinator({ + spawnSubagent: spawn, + signal: new AbortController().signal, + onProgress, + }); + const result = await coordinator.run('x'); + expect(result).toBe('SYNTH'); + expect(onProgress.mock.calls.some((c) => /failed/i.test(String(c[0])))).toBe(true); + }); +});