AI workflow orchestration

import { EventEmitter } from 'events';

interface NodeState {
  status: 'pending' | 'running' | 'success' | 'failed' | 'skipped';
  input: Record<string, unknown>;
  output?: Record<string, unknown>;
  retryCount: number;
  lastError?: string;
}

interface WorkflowNode {
  id: string;
  dependencies: string[];
  execute: (input: Record<string, unknown>) => Promise<Record<string, unknown>>;
  maxRetries?: number;
  backoffMs?: number;
}

interface WorkflowDAG {
  nodes: WorkflowNode[];
  initialState: Record<string, unknown>;
}

export class AIWorkflowOrchestrator extends EventEmitter {
  private states: Map<string, NodeState> = new Map();
  private results: Map<string, Record<string, unknown>> = new Map();

  constructor(private dag: WorkflowDAG) {
    super();
    this.initializeStates();
  }

  private initializeStates(): void {
    for (const node of this.dag.nodes) {
      this.states.set(node.id, {
        status: 'pending',
        input: this.dag.initialState,
        retryCount: 0,
      });
    }
  }

  async execute(): Promise<Record<string, unknown>> {
    const executionQueue = this.buildExecutionQueue();
    for (const nodeId of executionQueue) {
      await this.runNode(nodeId);
    }
    return this.aggregateResults();
  }

  private buildExecutionQueue(): string[] {
    const queue: string[] = [];
    const visited = new Set<string>();
    const visit = (id: string) => {
      if (visited.has(id)) return;
      visited.add(id);
      const node = this.dag.nodes.find(n => n.id === id)!;
      for (const dep of node.dependencies) visit(dep);
      queue.push(id);
    };

for (const node of this.dag.nodes) visit(node.id);
return queue;

}

private async runNode(nodeId: string): Promise<void> { const node = this.dag.nodes.find(n => n.id === nodeId)!; const state = this.states.get(nodeId)!;

// Skip if dependencies failed
const depsFailed = node.dependencies.some(depId => {
  const depState = this.states.get(depId)!;
  return depState.status === 'failed' || depState.status === 'skipped';
});
if (depsFailed) {
  state.status = 'skipped';
  this.emit('node:skipped', nodeId);
  return;
}

state.status = 'running';
this.emit('node:start', nodeId);

const maxRetries = node.maxRetries ?? 3;
const backoff = node.backoffMs ?? 1000;

for (let attempt = 0; attempt <= maxRetries; attempt++) {
  try {
    const mergedInput = this.mergeDependencyOutputs(node.dependencies);
    const output = await node.execute(mergedInput);
    state.status = 'success';
    state.output = output;
    this.results.set(nodeId, output);
    this.emit('node:success', nodeId, output);
    return;
  } catch (err) {
    state.lastError = (err as Error).message;
    state.retryCount = attempt + 1;
    if (attempt === maxRetries) {
      state.status = 'failed';
      this.emit('node:failed', nodeId, err);
      throw err;
    }
    const delay = backoff * Math.pow(2, attempt) + Math.random() * 500;
    this.emit('node:retry', nodeId, attempt + 1, delay);
    await new Promise(res => setTimeout(res, delay));
  }
}

}

private mergeDependencyOutputs(depIds: string[]): Record<string, unknown> { const merged: Record<string, unknown> = {}; for (const id of depIds) { const output = this.results.get(id); if (output) Object.assign(merged, output); } return merged; }

private aggregateResults(): Record<string, unknown> { const final: Record<string, unknown> = {}; for (const [id, output] of this.results) { final[id] = output; } return final; } }

### Usage Example

```typescript
const workflow: WorkflowDAG = {
  initialState: { query: 'Extract entities from this text and summarize.', text: '...' },
  nodes: [
    {
      id: 'extract',
      dependencies: [],
      execute: async (input) => {
        // Call LLM with structured output schema
        return { entities: ['Alice', 'ProjectX'], confidence: 0.94 };
      },
      maxRetries: 2,
      backoffMs: 800
    },
    {
      id: 'summarize',
      dependencies: ['extract'],
      execute: async (input) => {
        // LLM call using extracted entities
        return { summary: 'Alice leads ProjectX with high confidence.' };
      }
    }
  ]
};

const orchestrator = new AIWorkflowOrchestrator(workflow);
orchestrator.on('node:retry', (id, attempt, delay) => 
  console.warn(`[Orchestrator] ${id} retry ${attempt} in ${delay}ms`)
);

try {
  const result = await orchestrator.execute();
  console.log('Workflow complete:', result);
} catch (err) {
  console.error('Workflow failed:', err);
}

Architecture Rationale

• Topological sort guarantees dependency resolution without cycles.
• State isolation ensures each node receives deterministic inputs regardless of execution order or retries.
• Exponential backoff with jitter prevents thundering herd during provider rate limits.
• Event emitter pattern enables seamless integration with OpenTelemetry, logging pipelines, or alerting systems.
• Schema-enforced LLM outputs (not shown for brevity but required in production) eliminate JSON parsing failures and enable type-safe downstream consumption.

Pitfall Guide

1. Treating LLMs as Deterministic Functions

LLMs return probabilistic outputs. Assuming consistent JSON structure or identical reasoning paths across runs causes silent data corruption. Always validate outputs against JSON Schema or Zod before passing to downstream nodes.

2. Ignoring State Serialization

Workflow state lost during worker scaling, crashes, or cold starts forces full re-execution. Serialize node states to Redis, PostgreSQL, or durable queues. Include input hashes, retry counts, and timestamps for auditability.

3. Synchronous Blocking Chains

Chaining await llm.generate() calls sequentially multiplies latency and token costs. Parallelize independent branches, use streaming for user-facing endpoints, and batch non-critical tool calls.

4. Missing Cost and Token Guards

Unbounded retries and verbose prompts inflate costs rapidly. Implement per-workflow token budgets, early termination on low-confidence outputs, and fallback to smaller models for routing/classification tasks.

5. Hardcoded Routing Logic

Static if/else branches break when model behavior shifts. Use LLM-as-router patterns with explicit confidence thresholds, or switch to rule-based dispatchers for deterministic steps. Cache routing decisions when inputs repeat.

6. Neglecting Observability

Without traces, you cannot distinguish between model degradation, prompt drift, and infrastructure failures. Emit OpenTelemetry spans per node, log token usage, capture raw LLM responses, and track success/failure ratios by node ID.

7. No Human-in-the-Loop Fallback

High-stakes workflows (compliance, finance, healthcare) cannot rely solely on probabilistic outputs. Insert manual review nodes that pause execution, expose intermediate state, and allow approval/rejection with audit trails.

Production Bundle

Action Checklist

• Define DAG topology: Map all AI steps, dependencies, and parallel branches before coding.
• Enforce structured outputs: Validate every LLM response against JSON Schema or Zod.
• Implement state persistence: Store node states, inputs, and outputs in durable storage.
• Add retry semantics: Configure max retries, exponential backoff, and jitter per node.
• Instrument observability: Emit spans, logs, and metrics for every node execution.
• Set cost boundaries: Define token budgets, model fallbacks, and early termination rules.
• Insert review gates: Add human-in-the-loop nodes for compliance or high-risk decisions.
• Test failure modes: Simulate provider outages, rate limits, and malformed outputs.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
High-volume batch processing	Stateful DAG with parallel nodes	Maximizes throughput, isolates failures	-30% vs linear chaining
Real-time user interaction	Streamed DAG + async background nodes	Reduces perceived latency, preserves state	+15% infrastructure, -40% token waste
Compliance/audit workflows	DAG + human review gates + state persistence	Ensures traceability and regulatory alignment	+20% operational cost, -90% risk exposure
Experimental/research pipelines	Linear chaining + lightweight retry	Fast iteration, minimal overhead	+50% cost if promoted to production

Configuration Template

workflow:
  id: ai-data-extraction-v1
  version: 1.2.0
  budget:
    max_tokens: 12000
    max_cost_usd: 0.50
  nodes:
    - id: classify
      type: llm
      model: gpt-4o-mini
      prompt_template: prompts/classify.j2
      max_retries: 2
      backoff_ms: 800
      output_schema: schemas/classify.json
      dependencies: []

    - id: extract
      type: llm
      model: gpt-4o
      prompt_template: prompts/extract.j2
      max_retries: 3
      backoff_ms: 1000
      output_schema: schemas/extract.json
      dependencies: [classify]

    - id: validate
      type: function
      handler: validators/structure.ts
      dependencies: [extract]

    - id: persist
      type: function
      handler: storage/write.ts
      dependencies: [validate]
      idempotency_key: "${extract.id}"

Quick Start Guide

1. Initialize project: npm init -y && npm install zod openai @opentelemetry/api
2. Define schema: Create Zod schemas for every LLM output to enforce structure.
3. Build DAG: Instantiate AIWorkflowOrchestrator with node definitions and dependencies.
4. Wire observability: Attach OpenTelemetry exporters to the node:start, node:success, and node:failed events.
5. Execute & monitor: Run orchestrator.execute(), track metrics via your observability stack, and iterate on retry thresholds and model routing.

Orchestration is not a framework dependency; it is an engineering discipline. Treat AI workflows as distributed systems, enforce state boundaries, measure everything, and design for failure. The models will improve; your architecture must be ready to absorb the variance.