Workflows

Multi-Step AI Workflows

Chain LLM calls and tool calls into one job that picks up at the failed step after a crash or a rate-limit error.

See the full example on GitHub. View on GitHub

What we’ll build:

An aiJob workflow per job that turns a meeting transcript into a short brief for your sales team.
Two model steps, extract and summarize, that retry with backoff when the provider is rate limited or down.
Enrich steps that look up each company in your CRM between the two model calls.
A publish step that posts the brief to your CMS exactly once.
A /jobs route that starts a job and returns its id right away, and progress events that tell your app which step a job is on.

Install

npm add @rivet-dev/workflows rivetkit hono @hono/node-server

Configure

Set your model key and the base URLs of your CRM and CMS APIs:

OPENAI_API_KEY=sk-...
CRM_API_URL=https://crm.internal.example.com
CMS_API_URL=https://cms.internal.example.com

OPENAI_BASE_URL and OPENAI_MODEL are optional. Point OPENAI_BASE_URL at any server that speaks the OpenAI chat completions format.

Call the model

Every model call goes through one function:

import { CriticalError } from "@rivet-dev/workflows";

// The only function that talks to a model. Swap the body to use another provider or SDK.
export async function callModel(opts: {
	system: string;
	prompt: string;
	json?: boolean;
}): Promise<string> {
	const baseUrl = process.env.OPENAI_BASE_URL ?? "https://api.openai.com/v1";
	const res = await fetch(`${baseUrl}/chat/completions`, {
		method: "POST",
		headers: {
			authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
			"content-type": "application/json",
		},
		body: JSON.stringify({
			model: process.env.OPENAI_MODEL ?? "gpt-4.1-mini",
			messages: [
				{ role: "system", content: opts.system },
				{ role: "user", content: opts.prompt },
			],
			response_format: opts.json ? { type: "json_object" } : undefined,
		}),
	});

	// Rate limits and server errors throw a plain Error, so the step retries with backoff.
	if (res.status === 429 || res.status >= 500) {
		throw new Error(`model returned ${res.status}`);
	}
	// Other errors, such as a bad API key, will not fix themselves, so the step fails without retrying.
	if (!res.ok) {
		throw new CriticalError(
			`model returned ${res.status}: ${await res.text()}`,
		);
	}

	const body = (await res.json()) as {
		choices: Array<{ message: { content: string } }>;
	};
	return body.choices[0].message.content;
}

Keeping the call in one file means the workflow never depends on a provider. To use Anthropic, Gemini, or an SDK, change only this function.
A 429 or 5xx throws a plain Error. The workflow step that called it retries with backoff.
Any other error throws CriticalError, which stops retries right away. A bad API key or a malformed request fails the same way every time, so retrying only wastes time.

Create the workflow

Each job is its own aiJob workflow, keyed by the job id. It waits for a start message, then runs the extract step:

import {
	event,
	queue,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";
import { callModel } from "./model";
import { type Company, lookupCompany, publishPost } from "./tools";

export type JobInput = { title: string; transcript: string };
type Extracted = { companies: string[]; actionItems: string[] };
type Stage =
	| "queued"
	| "extracting"
	| "enriching"
	| "summarizing"
	| "publishing"
	| "published"
	| "failed";
export type Progress = { stage: Stage; detail: string | null };

// Model calls get a long timeout and slow backoff, because rate limits clear in seconds to minutes.
const MODEL_STEP = {
	timeout: 120_000,
	retryOnTimeout: true,
	maxRetries: 6,
	retryBackoffBase: 2_000,
	retryBackoffMax: 60_000,
};

// One workflow per job, keyed by the job id.
export const aiJob = workflow(
	{
		state: {
			stage: "queued" as Stage,
			extracted: null as Extracted | null,
			companies: [] as Company[],
			summary: null as string | null,
			url: null as string | null,
		},
		queues: {
			start: queue<JobInput>(),
		},
		events: {
			progress: event<Progress>(),
		},
		run: async (ctx) => {
			const jobId = ctx.key[0];
			const { body: input } = await ctx.queue.next("wait-start", {
				names: ["start"],
			});

			const extracted = await ctx.step({
				name: "extract",
				...MODEL_STEP,
				run: async (step) => {
					setStage(step, "extracting");
					const text = await callModel({
						system: "You read meeting transcripts and reply with JSON only.",
						prompt: `Extract the companies mentioned and the action items. Reply as {"companies": string[], "actionItems": string[]}.\n\n${input.transcript}`,
						json: true,
					});
					// Bad JSON throws, so the step asks the model again.
					const result = JSON.parse(text) as Extracted;
					step.state.extracted = result;
					return result;
				},
			});

			// One step per company, so a failed lookup retries alone and finished lookups are not repeated.
			const companies: Company[] = [];
			for (const [index, name] of extracted.companies.entries()) {
				const company = await ctx.step({
					name: `enrich-${index}`,
					maxRetries: 5,
					run: async (step) => {
						setStage(step, "enriching", name);
						const found = await lookupCompany(name);
						step.state.companies.push(found);
						return found;
					},
				});
				companies.push(company);
			}

			const summary = await ctx.step({
				name: "summarize",
				...MODEL_STEP,
				run: async (step) => {
					setStage(step, "summarizing");
					const text = await callModel({
						system: "You write short meeting briefs for a sales team.",
						prompt: `Write a brief for "${input.title}".\n\nCRM data: ${JSON.stringify(companies)}\nAction items: ${extracted.actionItems.join("; ")}\n\nTranscript:\n${input.transcript}`,
					});
					step.state.summary = text;
					return text;
				},
			});

			await ctx.step({
				name: "publish",
				maxRetries: 5,
				run: async (step) => {
					setStage(step, "publishing");
					const url = await publishPost({
						jobId,
						title: input.title,
						body: summary,
					});
					step.state.url = url;
					setStage(step, "published", url);
				},
			});
		},
		actions: {
			getStatus: (c) => c.state,
		},
	},
	{
		// Tell the app about retries and failures. This hook only observes. It does not change the run.
		onError: (c, event) => {
			if (!("step" in event)) return;
			const { stepName, willRetry, retryDelay, error } = event.step;
			if (!willRetry) c.state.stage = "failed";
			const detail = willRetry
				? `${stepName} failed (${error.message}), retrying in ${retryDelay}ms`
				: `${stepName} failed: ${error.message}`;
			c.broadcast("progress", { stage: c.state.stage, detail });
		},
	},
);

// Saves the stage in state, so a client that connects later can read it, and pushes it to connected clients.
function setStage(
	step: WorkflowStepContextOf<typeof aiJob>,
	stage: Stage,
	detail: string | null = null,
) {
	step.state.stage = stage;
	step.broadcast("progress", { stage, detail });
}

ctx.step records the step’s return value. After a crash or a deploy, the workflow replays and gets the recorded value back instead of calling the model again, so you never pay twice for the same answer.
MODEL_STEP gives model calls up to 7 attempts. The first retry waits 4 seconds and the wait doubles up to 60 seconds, which is enough for most rate limits to clear. The workflow sleeps durably between attempts, so a long wait does not hold a process.
Model calls can be slow, so the step timeout is 2 minutes instead of the default 30 seconds. retryOnTimeout makes a timeout retry like any other error.
If the model replies with invalid JSON, JSON.parse throws and the step asks again.

Call your APIs

Between the two model calls, the workflow looks up each company in your CRM:

// One step per company, so a failed lookup retries alone and finished lookups are not repeated.
const companies: Company[] = [];
for (const [index, name] of extracted.companies.entries()) {
	const company = await ctx.step({
		name: `enrich-${index}`,
		maxRetries: 5,
		run: async (step) => {
			setStage(step, "enriching", name);
			const found = await lookupCompany(name);
			step.state.companies.push(found);
			return found;
		},
	});
	companies.push(company);
}

// Calls to your own APIs. Each one runs inside a workflow step, so a failure retries only that call.

export type Company = { name: string; plan: string; seats: number };

// Looks up a company in your CRM.
export async function lookupCompany(name: string): Promise<Company> {
	const crmUrl = process.env.CRM_API_URL ?? "https://crm.example.com";
	const res = await fetch(
		`${crmUrl}/companies?name=${encodeURIComponent(name)}`,
	);
	if (!res.ok) throw new Error(`CRM lookup failed with status ${res.status}`);
	return (await res.json()) as Company;
}

// Publishes a post to your CMS. The job id is the idempotency key, so a retry never publishes twice.
export async function publishPost(post: {
	jobId: string;
	title: string;
	body: string;
}): Promise<string> {
	const cmsUrl = process.env.CMS_API_URL ?? "https://cms.example.com";
	const res = await fetch(`${cmsUrl}/posts`, {
		method: "POST",
		headers: {
			"content-type": "application/json",
			"idempotency-key": post.jobId,
		},
		body: JSON.stringify({ title: post.title, body: post.body }),
	});
	if (!res.ok) throw new Error(`CMS publish failed with status ${res.status}`);
	return ((await res.json()) as { url: string }).url;
}

Each lookup is its own step, named by its position. If the third lookup fails, only that one retries, and the first two are not repeated.
The loop is safe to replay because the list of companies comes from the recorded extract step, so every replay sees the same list in the same order.

Summarize with the model

The summarize step sends the transcript, the action items, and the CRM data back to the model:

const summary = await ctx.step({
	name: "summarize",
	...MODEL_STEP,
	run: async (step) => {
		setStage(step, "summarizing");
		const text = await callModel({
			system: "You write short meeting briefs for a sales team.",
			prompt: `Write a brief for "${input.title}".\n\nCRM data: ${JSON.stringify(companies)}\nAction items: ${extracted.actionItems.join("; ")}\n\nTranscript:\n${input.transcript}`,
		});
		step.state.summary = text;
		return text;
	},
});

It uses the same MODEL_STEP retry settings as the extract step.
companies and extracted are recorded results from earlier steps, so this step can run in a later process than the ones that produced them.

Publish the result

The last step posts the brief to your CMS:

await ctx.step({
	name: "publish",
	maxRetries: 5,
	run: async (step) => {
		setStage(step, "publishing");
		const url = await publishPost({
			jobId,
			title: input.title,
			body: summary,
		});
		step.state.url = url;
		setStage(step, "published", url);
	},
});

A step can run more than once. If the process dies after your CMS saved the post but before the step was recorded, the step runs again.
publishPost sends the job id as an idempotency-key header, so your CMS can return the existing post instead of creating a second one. See tools.ts in the previous step.

Report progress

Every step calls setStage, and the onError hook reports retries:

// Saves the stage in state, so a client that connects later can read it, and pushes it to connected clients.
function setStage(
	step: WorkflowStepContextOf<typeof aiJob>,
	stage: Stage,
	detail: string | null = null,
) {
	step.state.stage = stage;
	step.broadcast("progress", { stage, detail });
}

// Tell the app about retries and failures. This hook only observes. It does not change the run.
onError: (c, event) => {
	if (!("step" in event)) return;
	const { stepName, willRetry, retryDelay, error } = event.step;
	if (!willRetry) c.state.stage = "failed";
	const detail = willRetry
		? `${stepName} failed (${error.message}), retrying in ${retryDelay}ms`
		: `${stepName} failed: ${error.message}`;
	c.broadcast("progress", { stage: c.state.stage, detail });
},

The stage lives in state, so a client that connects late can call getStatus and see where the job is.
broadcast pushes each change to every client connected to the job. No separate pub/sub service is needed.
onError sees each failed attempt with its retry delay. It only observes, so it is a good place for progress updates and error reporting, but not for work the job depends on.

Start jobs over HTTP

The HTTP server has one route. It creates a workflow for the job and hands it the input:

import { randomUUID } from "node:crypto";
import { serve } from "@hono/node-server";
import { setup } from "@rivet-dev/workflows";
import { Hono } from "hono";
import { createClient } from "rivetkit/client";
import { aiJob, type JobInput } from "./ai-job";

export const registry = setup({ use: { aiJob } });
const client = createClient<typeof registry>();

registry.start();

const app = new Hono();

// Starts a job and returns its id without waiting for it to finish.
app.post("/jobs", async (c) => {
	const input = await c.req.json<JobInput>();
	if (!input.title || !input.transcript)
		return c.text("title and transcript are required", 400);

	const jobId = randomUUID();
	await client.aiJob.getOrCreate([jobId]).send("start", input);
	return c.json({ jobId }, 202);
});

serve({ fetch: app.fetch, port: Number(process.env.PORT ?? 3000) });

send puts the input on the job’s start queue and returns as soon as it is stored, so the route answers in milliseconds even though the job takes much longer.
A new random id per request means each POST starts its own job. To make retries of the same request safe, derive the id from something stable, such as a document id, and getOrCreate returns the existing job.

Try it

Start the server:

npx tsx server.ts

Then submit a job and follow its progress:

import { createClient } from "rivetkit/client";
import type { Progress } from "./ai-job";
import type { registry } from "./server";

const serverUrl = process.env.SERVER_URL ?? "http://localhost:3000";
const client = createClient<typeof registry>();

// 1. Submit a job to the HTTP server.
const res = await fetch(`${serverUrl}/jobs`, {
	method: "POST",
	headers: { "content-type": "application/json" },
	body: JSON.stringify({
		title: "Pricing call with Acme and Globex",
		transcript:
			"Dana (Acme Corp): We need pricing for 1,200 seats. Lee (Globex): We are evaluating too. Sam: I will send pricing to Acme by Friday.",
	}),
});
const { jobId } = (await res.json()) as { jobId: string };
console.log("job", jobId);

// 2. Follow the job's progress until it finishes.
const job = client.aiJob.get([jobId]);
const conn = job.connect();
await new Promise<void>((resolve) => {
	conn.on("progress", (event) => {
		const progress = event as Progress;
		console.log(progress.stage, progress.detail ?? "");
		if (progress.stage === "published" || progress.stage === "failed")
			resolve();
	});
});
console.log(await job.getStatus());
await conn.dispose();

npx tsx client.ts

The client prints each stage as it happens, including any retries, and then the job’s final state.

Stop the server while a job is running and start it again. The job continues from the step it was on, and steps that already finished do not call the model or your APIs again.

Next steps

Human-in-the-Loop Approvals: have a person approve the brief before the publish step runs.
Timers & Concurrency: look up every company at once with join instead of one after another.
Failure & Recovery: undo earlier steps when a later one fails.

Steps: how steps are recorded and replayed.
Timers & Concurrency: step timeouts and retry settings.
Failure & Recovery: the onError hook.
Patterns: storing progress in state and broadcasting it.