Skip to main content
Sign In
Agents

Multi-Agent Systems

Split a large task across an orchestrator agent and specialist agents that work in parallel.

See the full example on GitHub. View on GitHub
Your appstarts a joborchestratorplan and answersresearcherone per subtaskreviewerchecks the reportsubmit1ask_specialist2answers3requestReview4getReview5One per job or subtaskAgentCallReply

An orchestrator agent splits a job into subtasks, hands each one to its own specialist agent, and combines what comes back. Here a research job fans out to researchers that work in parallel, and a reviewer checks the finished report.

What we’ll build:

  • An orchestrator agent per job that breaks the job into subtasks and keeps the plan and the collected answers in its conversation.
  • An ask_specialist tool that starts one researcher agent per subtask, runs them in parallel, and waits for every answer.
  • A reviewer agent per job that gets the finished report as an agent-to-agent message, without the orchestrator waiting for the review.
  • A client that starts each job by calling its orchestrator directly and reads the review from its reviewer.
  • Recovery after a crash, so a finished subtask is not asked again and an unfinished one carries on where it stopped.

Install

npm add @rivet-dev/pi @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chord rivetkit

Create the researcher

Each subtask gets its own researcher, with its own tools and a fresh conversation:

import { Type } from "@earendil-works/pi-ai";
import { createRegistry, defineExtension, defineTool, section } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";

const fetchPage = defineTool({
	name: "fetch_page",
	description: "Fetch a web page and return the start of its text.",
	parameters: Type.Object({ url: Type.String() }),
	// A fetch changes nothing, so running it again after a crash is harmless.
	replay: "safe",
	execute: async ({ url }, _api, context) => {
		const response = await fetch(url, { signal: context.abortSignal });
		if (!response.ok) throw new Error(`${url} returned ${response.status}.`);
		const text = (await response.text()).slice(0, 20_000);
		return { content: [{ type: "text", text }] };
	},
});

const extensions = createRegistry();
extensions.install(
	defineExtension({
		name: "research",
		tools: [fetchPage],
		sections: [
			section(
				"role",
				() => "You research one question. Answer it in a few short paragraphs and list the URLs you used.",
			),
		],
	}),
);

// One researcher per subtask, with a fresh conversation.
export const researcher = pi({ model: "openai/gpt-5.4-mini", registry: extensions });
  • The researcher only sees the question it’s given, not the orchestrator’s conversation, so its context stays small and on topic.
  • fetch_page reads a web page. Give a specialist only the tools its job needs. See Custom Tools.
  • It runs on a smaller, cheaper model than the orchestrator, because each subtask is narrow.

Fan out to researchers

The orchestrator’s ask_specialist tool takes every subtask at once and starts one researcher for each:

import { Type } from "@earendil-works/pi-ai";
import {
	createRegistry,
	defineExtension,
	defineTool,
	ROOT_CONVERSATION_ID,
	section,
} from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const askSpecialist = defineTool({
	name: "ask_specialist",
	description: "Give each subtask to its own researcher. They work in parallel, and you get every answer back.",
	parameters: Type.Object({
		subtasks: Type.Array(
			Type.String({ description: "One self-contained question. The researcher can't see this conversation." }),
			{ minItems: 1, maxItems: 8 },
		),
	}),
	// A rerun after a crash reaches the same researchers, through the memo below.
	replay: "safe",
	execute: async ({ subtasks }, api, context) => {
		// One id per subtask, saved with this tool call, so a rerun gets the same ids.
		const ids = await api.memo("researcher-ids", subtasks.map(() => crypto.randomUUID()), context);

		const answers = await Promise.all(
			ids.map(async (id, i) => {
				const researcher = client.researcher.getOrCreate([id]);
				context.abortSignal?.addEventListener("abort", () => void researcher.conversation.abort(ROOT_CONVERSATION_ID));
				// A researcher that already answered returns its saved answer for the same requestId.
				const result = await researcher.prompt(subtasks[i], { requestId: id });
				if (result.status === "unanswered") return `No answer: ${result.reason}`;
				return result.text || "The researcher finished without an answer.";
			}),
		);

		const text = subtasks.map((subtask, i) => `## ${subtask}\n\n${answers[i]}`).join("\n\n");
		return { content: [{ type: "text", text }] };
	},
});

const extensions = createRegistry();
extensions.install(
	defineExtension({
		name: "orchestrator",
		tools: [askSpecialist],
		sections: [
			section(
				"role",
				() =>
					"You run research jobs. Split the job into independent subtasks, send them all to ask_specialist in one call, and combine the answers into one report.",
			),
		],
	}),
);

// One orchestrator per job, keyed by the job id.
export const orchestrator = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	// Hands every finished report to the job's reviewer without waiting for the review.
	onAnswer: async (c, answer) => {
		const [jobId] = c.key;
		if (!answer.text) return;
		// The answer's timestamp is saved with it, so a repeated call sends the same requestId.
		await client.reviewer.getOrCreate([jobId]).requestReview(answer.text, `${jobId}:${answer.message.timestamp}`);
	},
});

const client = createClient<typeof registry>();
  • The model writes the plan as the list of subtasks. Each researcher is keyed by its own id, so every subtask gets a new agent.
  • Promise.all prompts every researcher at once and waits for all of them. The combined answers go back to the model as one tool result, which stays in the orchestrator’s conversation.
  • If the orchestrator’s run is cancelled, the tool cancels every researcher’s run too.
  • maxItems caps how many researchers one call can start.

An action waits up to ten minutes. A researcher that takes longer keeps working, but its prompt rejects and the tool call fails. Keep subtasks small enough to finish in that time.

Create the orchestrator

The orchestrator has ask_specialist and a prompt that tells it how to split a job:

import { Type } from "@earendil-works/pi-ai";
import {
	createRegistry,
	defineExtension,
	defineTool,
	ROOT_CONVERSATION_ID,
	section,
} from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const askSpecialist = defineTool({
	name: "ask_specialist",
	description: "Give each subtask to its own researcher. They work in parallel, and you get every answer back.",
	parameters: Type.Object({
		subtasks: Type.Array(
			Type.String({ description: "One self-contained question. The researcher can't see this conversation." }),
			{ minItems: 1, maxItems: 8 },
		),
	}),
	// A rerun after a crash reaches the same researchers, through the memo below.
	replay: "safe",
	execute: async ({ subtasks }, api, context) => {
		// One id per subtask, saved with this tool call, so a rerun gets the same ids.
		const ids = await api.memo("researcher-ids", subtasks.map(() => crypto.randomUUID()), context);

		const answers = await Promise.all(
			ids.map(async (id, i) => {
				const researcher = client.researcher.getOrCreate([id]);
				context.abortSignal?.addEventListener("abort", () => void researcher.conversation.abort(ROOT_CONVERSATION_ID));
				// A researcher that already answered returns its saved answer for the same requestId.
				const result = await researcher.prompt(subtasks[i], { requestId: id });
				if (result.status === "unanswered") return `No answer: ${result.reason}`;
				return result.text || "The researcher finished without an answer.";
			}),
		);

		const text = subtasks.map((subtask, i) => `## ${subtask}\n\n${answers[i]}`).join("\n\n");
		return { content: [{ type: "text", text }] };
	},
});

const extensions = createRegistry();
extensions.install(
	defineExtension({
		name: "orchestrator",
		tools: [askSpecialist],
		sections: [
			section(
				"role",
				() =>
					"You run research jobs. Split the job into independent subtasks, send them all to ask_specialist in one call, and combine the answers into one report.",
			),
		],
	}),
);

// One orchestrator per job, keyed by the job id.
export const orchestrator = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	// Hands every finished report to the job's reviewer without waiting for the review.
	onAnswer: async (c, answer) => {
		const [jobId] = c.key;
		if (!answer.text) return;
		// The answer's timestamp is saved with it, so a repeated call sends the same requestId.
		await client.reviewer.getOrCreate([jobId]).requestReview(answer.text, `${jobId}:${answer.message.timestamp}`);
	},
});

const client = createClient<typeof registry>();
  • Each job gets its own orchestrator, keyed by the job id, so jobs never share a conversation.
  • onAnswer runs for every final answer, with the orchestrator’s own context, so c.key is the job id. It sends the report to that job’s reviewer.
  • It runs inside the orchestrator’s run, so a report that finishes just before a crash still reaches the reviewer. If it runs twice for the same answer, the reviewer gets the same requestId and reviews it once.

Create the reviewer

The reviewer takes the report as a message and returns before reviewing it:

import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { createRegistry, defineExtension, section } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";

const extensions = createRegistry();
extensions.install(
	defineExtension({
		name: "review",
		sections: [
			section(
				"role",
				() =>
					"You review research reports. List claims without a source, gaps, and contradictions, then end with APPROVE or REVISE.",
			),
		],
	}),
);

// One reviewer per job, keyed by the job id.
export const reviewer = pi({
	model: "openai/gpt-5.5",
	registry: extensions,
	state: { review: null as string | null },
	// Saves each review so your app can read it.
	onAnswer: (c, answer) => {
		c.state.review = answer.text ?? null;
	},
	actions: {
		// Saves the request and returns without waiting for the review.
		requestReview: async (c, report: string, requestId: string) => {
			const conversation = await c.pi.root(BACKGROUND_CONTEXT);
			await conversation.submit(
				{ type: "input", content: `Review this report:\n\n${report}`, requestId },
				BACKGROUND_CONTEXT,
			);
		},
		getReview: (c) => c.state.review,
	},
});
  • requestReview submits the report to the reviewer’s conversation and returns, so the orchestrator never waits for the review. The request is saved before the action returns. See Agent-to-Agent Messages.
  • The reviewer runs on a different model family than the orchestrator, so it doesn’t share the same blind spots.
  • Its onAnswer saves the review in the reviewer’s state, and getReview returns it to your app.

Register the agents

All three agents go in one registry:

import { setup } from "rivetkit";
import { orchestrator } from "./orchestrator";
import { researcher } from "./researcher";
import { reviewer } from "./reviewer";

export const registry = setup({ use: { orchestrator, researcher, reviewer } });

registry.start();

There are no routes to write. Clients call each agent’s actions by key through the RivetKit client.

Start a job

Your app calls submit on the orchestrator for the job:

import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const client = createClient<typeof registry>();

// Each job gets its own orchestrator, keyed by the job id.
const jobId = crypto.randomUUID();
const orchestrator = client.orchestrator.getOrCreate([jobId]);

// Starts the job and returns without waiting for it.
// The job id is also the requestId, so a retried call doesn't start the job twice.
await orchestrator.submit("Compare SQLite, Postgres, and DuckDB for an analytics dashboard with 50 million rows.", {
	requestId: jobId,
});
console.log(`Started job ${jobId}`);

// Check the job's reviewer every 30 seconds until the review is in.
const reviewer = client.reviewer.getOrCreate([jobId]);
let review = await reviewer.getReview();
while (review === null) {
	await new Promise((resolve) => setTimeout(resolve, 30_000));
	review = await reviewer.getReview();
}
console.log(review);
  • The app picks the job id and keys the orchestrator by it. The first call creates that job’s orchestrator.
  • submit starts the job and returns without waiting, so a job that takes many minutes never holds a call open.
  • The job id is also the requestId, so a retried call runs the job once.

Any client can start a job or read a review. In production, protect the Actors with authentication.

Recover from a crash

Nothing extra is needed. Each agent is its own Rivet Actor that saves every step to its own SQLite database, and the fan-out tool is written so a rerun is safe:

  • ask_specialist sets replay: "safe", so if the orchestrator crashes while researchers are working, the tool runs again when the orchestrator restarts.
  • api.memo saves the researcher ids with the tool call, so the rerun reaches the same researchers instead of starting new ones.
  • A researcher that already answered returns its saved answer for the same requestId without running again.
  • A researcher that was still working kept going in its own Actor, or resumed from its last saved step if it crashed too. The rerun waits for that run instead of asking again.

See Resuming a run for what happens to each kind of step.

Try it

Set the keys for both model providers and start the server:

export ANTHROPIC_API_KEY=sk-ant-...
export OPENAI_API_KEY=sk-...
npx tsx server.ts

In another terminal, start a job and wait for its review:

import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const client = createClient<typeof registry>();

// Each job gets its own orchestrator, keyed by the job id.
const jobId = crypto.randomUUID();
const orchestrator = client.orchestrator.getOrCreate([jobId]);

// Starts the job and returns without waiting for it.
// The job id is also the requestId, so a retried call doesn't start the job twice.
await orchestrator.submit("Compare SQLite, Postgres, and DuckDB for an analytics dashboard with 50 million rows.", {
	requestId: jobId,
});
console.log(`Started job ${jobId}`);

// Check the job's reviewer every 30 seconds until the review is in.
const reviewer = client.reviewer.getOrCreate([jobId]);
let review = await reviewer.getReview();
while (review === null) {
	await new Promise((resolve) => setTimeout(resolve, 30_000));
	review = await reviewer.getReview();
}
console.log(review);

npx tsx client.ts

To test recovery, kill the server with kill -9 while the researchers are working, then start it again. The job finishes, and the client prints the review.

Next steps

Edit this page