Workflow-001: AI Agent Workflow

Fresh

Pattern: Orchestrator + parallel worker tasks Use Case: Fan-out AI tasks, content generation, fact-checking, translation pipelines

Overview

This workflow uses a parent "orchestrator" task that fans out to multiple worker tasks in parallel, collects results, and produces a final output. Common in AI pipelines where you want to process items independently and combine results.

Architecture

Implementation

Worker Task

// trigger/ai-worker.ts
import { task } from "@trigger.dev/sdk";
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

export const aiWorker = task({
  id: "ai-worker",
  retry: { maxAttempts: 3, minTimeoutInMs: 1000 },
  run: async (payload: { content: string; instruction: string }) => {
    const message = await client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 1024,
      messages: [{
        role: "user",
        content: `${payload.instruction}\n\nContent: ${payload.content}`
      }]
    });

    return {
      result: message.content[0].type === 'text' ? message.content[0].text : '',
      inputTokens: message.usage.input_tokens,
      outputTokens: message.usage.output_tokens
    };
  },
});

Orchestrator Task

// trigger/orchestrator.ts
import { task } from "@trigger.dev/sdk";
import { aiWorker } from "./ai-worker";

export const orchestratorTask = task({
  id: "orchestrator",
  run: async (payload: { articles: Array<{ id: string; content: string }> }) => {
    // Fan out — trigger all workers in parallel and wait for all results
    const results = await aiWorker.batchTriggerAndWait(
      payload.articles.map(article => ({
        payload: {
          content: article.content,
          instruction: "Summarize this article in 2 sentences"
        }
      }))
    );

    // Process results
    const summaries = [];
    for (const run of results.runs) {
      if (run.ok) {
        summaries.push(run.output.result);
      } else {
        console.error("Worker failed:", run.error);
        // Choose: throw to retry, or continue with partial results
      }
    }

    return {
      totalArticles: payload.articles.length,
      successCount: summaries.length,
      summaries
    };
  },
});

Trigger from Backend

import { tasks } from "@trigger.dev/sdk";
import type { orchestratorTask } from "~/trigger/orchestrator";

const handle = await tasks.trigger<typeof orchestratorTask>("orchestrator", {
  articles: [
    { id: "1", content: "Article one content..." },
    { id: "2", content: "Article two content..." },
    { id: "3", content: "Article three content..." },
  ]
});

console.log("Orchestration running:", handle.id);

Key Patterns

Pattern 1: Fan-out + Wait (above example)

Use batchTriggerAndWait when:

• All items can be processed independently
• You need ALL results before proceeding
• Max items: up to 1,000 per batch (SDK 4.3.1+)

Pattern 2: Sequential Chain

export const chainedTask = task({
  id: "chained-task",
  run: async (payload: { topic: string }) => {
    // Step 1: Generate content
    const generated = await generatorTask.triggerAndWait(payload.topic).unwrap();

    // Step 2: Translate content
    const translated = await translatorTask.triggerAndWait({
      content: generated.text,
      targetLanguage: "es"
    }).unwrap();

    // Step 3: Quality check
    const checked = await qualityTask.triggerAndWait({
      original: generated.text,
      translated: translated.text
    }).unwrap();

    return { final: checked.approved ? translated.text : generated.text };
  },
});

Pattern 3: Route to Different Models

export const routerTask = task({
  id: "route-question",
  run: async (payload: { question: string; complexity: "simple" | "complex" }) => {
    if (payload.complexity === "simple") {
      return await simpleModelTask.triggerAndWait(payload).unwrap();
    } else {
      return await complexModelTask.triggerAndWait(payload).unwrap();
    }
  },
});

Pattern 4: Streaming Batch (Large Datasets)

export const largeBatchTask = task({
  id: "large-batch",
  run: async (payload: { userIds: string[] }) => {
    // Stream items — no need to load all into memory
    async function* generateItems() {
      for (const userId of payload.userIds) {
        yield { payload: { userId } };
      }
    }

    const batchHandle = await workerTask.batchTrigger(generateItems());
    return { batchId: batchHandle.batchId };
  },
});

Handling Failures

const results = await aiWorker.batchTriggerAndWait(items);

const succeeded = [];
const failed = [];

for (const run of results.runs) {
  if (run.ok) {
    succeeded.push(run.output);
  } else {
    failed.push({ error: run.error, runId: run.id });
  }
}

// Option A: Fail entire batch if any failed
if (failed.length > 0) {
  throw new Error(`${failed.length} workers failed`);
}

// Option B: Continue with partial results
console.log(`${succeeded.length}/${results.runs.length} succeeded`);

Real-World Examples from Trigger.dev

Use Case	Pattern
Content moderation	Fan-out workers check in parallel while main task responds
Generate + translate	Sequential chain: generate → translate → evaluate
News verification	Fan-out to multiple fact-checking workers
Route to AI models	Router task selects model based on complexity

See Also

• Workflow-002: Batch Processing
• SOP-002: Writing Tasks
• Quick Reference: Commands