Forem: Ivan Holovach

A DAG-Based Approach to LLM Workflow Orchestration

Ivan Holovach — Fri, 07 Nov 2025 20:03:59 +0000

When processing large batches of items through multiple LLM analyses, you quickly run into orchestration complexity. Consider a customer review pipeline that needs to:

Run sentiment analysis and topic extraction (independent tasks)
Generate summaries that depend on both sentiment and topics
Filter spam before expensive operations
Handle failures gracefully
Route to different models based on cost/capability
Track spending in real-time

The Problem

The typical progression:

Sequential approach - Simple but slow:

for (const review of reviews) {
  const sentiment = await runSentiment(review);
  const topics = await runTopics(review);
  const summary = await runSummary(review, sentiment, topics);
}

Parallel approach - Faster but fragile:

const [sentiments, topics] = await Promise.all([
  Promise.all(reviews.map(r => runSentiment(r))),
  Promise.all(reviews.map(r => runTopics(r)))
]);

const summaries = await Promise.all(
  reviews.map((r, i) => runSummary(r, sentiments[i], topics[i]))
);

This works until you need error handling, retries, rate limiting, concurrency controls, and cost tracking. At scale, you end up with substantial orchestration code.

Existing Solutions

Several tools address this:

LangChain/LCEL - Has parallel execution with RunnableParallel, good for chaining operations
Prefect/Temporal - Durable workflow execution with state persistence, excellent for long-running workflows but requires infrastructure
LangGraph - State-based approach that can model DAGs through state transitions

These are production-ready tools. The gap I found: I wanted pure dependency-based orchestration focused specifically on LLM batch processing, without the infrastructure overhead or state management abstractions.

The DAG Approach

The core observation: for batch LLM processing, you're primarily describing analysis dependencies:

sentiment ─┐
           ├─→ summary
topics ────┘

Dependencies define execution order. Parallelization emerges automatically.

I built dagengine around this principle.

Implementation

Here's the complete working example:

import { DagEngine, Plugin } from '@dagengine/core';

class ReviewAnalyzer extends Plugin {
  constructor() {
    super('analyzer', 'Review Analyzer', 'Analyze reviews');
    this.dimensions = ['sentiment', 'topics', 'summary'];
  }

  // Declare dependencies
  defineDependencies() {
    return {
      summary: ['sentiment', 'topics']
    };
  }

  // Define prompts for each dimension
  createPrompt(context) {
    const review = context.sections[0]?.content;

    if (context.dimension === 'sentiment') {
      return `Analyze sentiment: "${review}"

Return JSON: {
  "sentiment": "positive|negative|neutral",
  "confidence": 0.0-1.0
}`;
    }

    if (context.dimension === 'topics') {
      return `Extract topics: "${review}"

Return JSON: {
  "topics": ["topic1", "topic2", ...]
}`;
    }

    if (context.dimension === 'summary') {
      const sentiment = context.dependencies.sentiment.data;
      const topics = context.dependencies.topics.data;

      return `Summarize with context:
- Sentiment: ${sentiment.sentiment} (${sentiment.confidence})
- Topics: ${topics.topics.join(', ')}
- Review: "${review}"

Return JSON: {
  "summary": "...",
  "key_points": ["point1", "point2", ...]
}`;
    }
  }

  // Select model
  selectProvider() {
    return {
      provider: 'anthropic',
      options: { model: 'claude-3-5-haiku-20241022' }
    };
  }
}

// Run
const engine = new DagEngine({
  plugin: new ReviewAnalyzer(),
  providers: {
    anthropic: { apiKey: process.env.ANTHROPIC_API_KEY }
  }
});

const result = await engine.process(reviews);

What happens automatically:

✅ Parallel execution of independent dimensions
✅ Dependency waiting
✅ Concurrent section processing (configurable batch size)
✅ Automatic retries with exponential backoff
✅ Cost tracking
✅ JSON parsing and validation

What you control:

Dimensions to run
Dependencies between them
Prompt generation
Model selection

Design Choice: JSON Responses

The engine expects JSON responses. Include "Return JSON: {schema}" in prompts.

Rationale:

Works across all providers without provider-specific syntax
Reliable parsing without regex
Clean dependency data access
Enables skip logic based on structured fields
Consistent cost tracking

The engine automatically parses responses and makes them available in context.dependencies[dimension].data.

Practical Features

1. Skip Logic for Cost Savings

class ReviewAnalyzer extends Plugin {
  constructor() {
    super('analyzer', 'Review Analyzer', 'Analyze reviews');
    this.dimensions = [
      'filter_spam',
      'sentiment',
      'topics',
      'summary'
    ];
  }

  defineDependencies() {
    return {
      sentiment: ['filter_spam'],
      topics: ['filter_spam'],
      summary: ['sentiment', 'topics']
    };
  }

  shouldSkipSectionDimension(context) {
    if (context.dimension === 'sentiment' || 
        context.dimension === 'topics') {
      const spam = context.dependencies.filter_spam?.data;
      return spam?.is_spam === true;
    }
    return false;
  }
}

In testing with 20 reviews where 10 were spam, this skipped 20 unnecessary API calls.

2. Multi-Model Routing

selectProvider(context) {
  // Cheap model for filtering
  if (context.dimension === 'filter_spam') {
    return {
      provider: 'anthropic',
      options: { model: 'claude-3-5-haiku-20241022' }
    };
  }

  // More capable model for analysis
  return {
    provider: 'anthropic',
    options: { model: 'claude-3-7-sonnet-20250219' },
    fallbacks: [
      { 
        provider: 'openai',
        options: { model: 'gpt-4o' }
      }
    ]
  };
}

Using Haiku ($0.80/1M input) for filtering and Sonnet ($3/1M input) for analysis reduces costs compared to using Sonnet for everything.

3. Gateway Integration

const engine = new DagEngine({
  plugin: new ReviewAnalyzer(),
  providers: {
    anthropic: {
      apiKey: process.env.ANTHROPIC_API_KEY,
      gateway: 'portkey',
      gatewayApiKey: process.env.PORTKEY_API_KEY,
      gatewayConfig: {
        retry: { 
          attempts: 5,
          on_status_codes: [429, 500, 502, 503, 504]
        }
      }
    }
  }
});

Portkey handles rate limiting and retries transparently.

4. Partial Results on Failures

const engine = new DagEngine({
  plugin: new ReviewAnalyzer(),
  providers: { /* ... */ },
  execution: {
    maxRetries: 3,
    retryDelay: 1000,
    continueOnError: true
  }
});

// Provide fallback values
class ReviewAnalyzer extends Plugin {
  async handleDimensionFailure(context) {
    return {
      data: { 
        sentiment: 'neutral',
        confidence: 0.0
      },
      metadata: { 
        fallback: true,
        error: context.error.message
      }
    };
  }
}

If one item fails, you still get results for the rest.

5. Cost Tracking

const result = await engine.process(reviews);

console.log(result.costs);
// {
//   totalCost: 0.0282,
//   totalTokens: 12289,
//   byDimension: {
//     filter_spam: { 
//       cost: 0.0115,
//       tokens: { input: 6548, output: 1559 }
//     },
//     ...
//   }
// }

Cross-Item Analysis Pattern

For operations that need to aggregate across all items:

class CategoryAnalyzer extends Plugin {
  constructor() {
    super('category', 'Category Analyzer', 'Analyze by category');
    this.dimensions = [
      'classify',
      { name: 'group', scope: 'global' },
      'analyze'
    ];
  }

  defineDependencies() {
    return {
      group: ['classify'],
      analyze: ['group']
    };
  }

  createPrompt(context) {
    if (context.dimension === 'classify') {
      return `Classify: "${context.sections[0].content}"
Return JSON: { "category": "product|service|delivery|other" }`;
    }

    if (context.dimension === 'group') {
      // Access all classification results
      const allClassifications = context.sections.map((section, i) => ({
        review: section.content,
        category: context.results.classify[i].data.category
      }));

      return `Group ${allClassifications.length} reviews by category.
${JSON.stringify(allClassifications, null, 2)}

Return JSON: { "categories": [...] }`;
    }

    if (context.dimension === 'analyze') {
      const category = context.sections[0].metadata.category;
      return `Analyze ${category} reviews:
${context.sections[0].content}

Return JSON: { "insights": [...], "themes": [...] }`;
    }
  }

  transformSections(context) {
    if (context.dimension !== 'group') {
      return context.currentSections;
    }

    // Transform: N reviews → M category groups
    const groups = context.result.data.categories;
    return groups.map(group => ({
      content: group.reviews.join('\n\n'),
      metadata: { 
        category: group.name,
        count: group.count
      }
    }));
  }
}

This handles: classify all items → group by category → analyze each group.

Performance Observations

These numbers are from my testing environment and will vary based on infrastructure, models, and batch sizes:

Small batch (20 reviews):

API calls: ~50
Time: ~4 seconds
Pattern: Skip logic reduced unnecessary calls

Medium batch (100 items):

API calls: ~290
Time: ~4 seconds
Gateway handled rate limits transparently

Cross-item analysis (10 reviews, 5 categories):

API calls: 16 (10 classify + 1 group + 5 analyze)
Time: ~10 seconds

Your mileage will vary.

When to Use This Approach

✅ Good fit:

Batch LLM processing with dependencies
Multi-model workflows
Need built-in retry/fallback
Want real-time cost tracking
Processing reviews, emails, documents
Requiring structured JSON outputs

❌ Not a good fit:

Complex state machines → Use Temporal or LangGraph
General data pipelines without LLMs → Use Prefect or Airflow
Conversational agents with multi-turn state → Use LangChain
Streaming responses → Not supported (coming)
Simple sequential workflows → Use async/await
Long-running jobs needing durable execution → Use Temporal

Tradeoffs

Advantages:

Declarative dependency model
Automatic parallelization
Built-in LLM-specific features

Limitations:

Requires thinking in DAG terms (learning curve)
All responses must be JSON (by design)

Current Status

This is beta software. I use it in production, but be aware:

✅ Supported: Anthropic, OpenAI, Gemini
🚧 Coming: Azure OpenAI, Cohere
🚧 In development: Streaming support, enhanced observability
🎯 Needed: Better debugging tools for complex DAGs

Getting Started

npm install @dagengine/core

Minimal example:

import { DagEngine, Plugin } from '@dagengine/core';

class SimpleAnalyzer extends Plugin {
  constructor() {
    super('simple', 'Simple Analyzer', 'Analyze items');
    this.dimensions = ['analysis'];
  }

  createPrompt(context) {
    return `Analyze: ${context.sections[0].content}
Return JSON: { "result": "...", "score": 0.0-1.0 }`;
  }

  selectProvider() {
    return {
      provider: 'anthropic',
      options: { model: 'claude-3-5-haiku-20241022' }
    };
  }
}

const engine = new DagEngine({
  plugin: new SimpleAnalyzer(),
  providers: {
    anthropic: { apiKey: process.env.ANTHROPIC_API_KEY }
  }
});

const result = await engine.process([
  { content: 'First item' },
  { content: 'Second item' }
]);

Start simple, add features as needed.

Resources

📦 GitHub: github.com/dagengine/dagengine
📖 Docs: dagengine.ai
💡 Examples: github.com/dagengine/dagengine/tree/main/examples

Conclusion

For batch LLM processing where you can express the workflow as dependencies between analyses, a DAG approach reduces orchestration code while handling parallelization, retries, and cost tracking automatically.

Whether it fits your use case depends on your specific requirements. The "When to Use" section should help you decide.

Feedback welcome via GitHub discussions or LinkedIn.

I kept rewriting AI orchestration code, so I built dagengine

Ivan Holovach — Thu, 06 Nov 2025 16:28:45 +0000

Stop me if you've heard this one:

You're processing data with AI. You write some async code. It works.

Then you need to:

Add another analysis
Make some run in parallel
Add retry logic
Track costs

You rewrite 50 lines. Again. And again.

I did this five times before building dagengine.

The Problem

I was analyzing my book manuscript—checking emotional pacing across 30 chapters. Each chapter needed multiple analyses: emotion score, readability, consistency checks.

First attempt: Sequential processing. 45 seconds per chapter × 30 chapters = 22.5 minutes.

Then I tried Promise.all():

await Promise.all([
  analyzeEmotion(chapter).catch(retry),
  checkReadability(chapter).catch(retry),
  verifyConsistency(chapter).catch(retry),
  // ... more analyses
])

200 lines of orchestration code. Manual retry logic everywhere. Every new analysis meant rewriting coordination.

Week 5, refactoring again: This isn't a book problem.

The Pattern

Every AI workflow has the same structure:

Multiple analyses needed
Some can run in parallel
Some depend on others
All need retries
All need cost tracking

I was solving a general problem with one-off code.

What I Built

dagengine handles the plumbing. You define what you want analyzed and what depends on what. The engine figures out execution order and parallelization.

Here's the concept:

import { DagEngine, Plugin, type PromptContext, type ProviderSelection } from '@dagengine/core';

// Define result types (TypeScript helps!)
interface SentimentResult {
  sentiment: "positive" | "negative" | "neutral";
  score: number;
}

class ReviewAnalyzer extends Plugin {
  constructor() {
    super('review-analyzer', 'Review Analyzer', 'Analyze customer reviews');

    // Define your analysis tasks
    this.dimensions = [
      'filter_spam',      // Stage 1: Filter
      'sentiment',        // Stage 2: Parallel
      'categorize',       // Stage 2: Parallel
      'analyze_category'  // Stage 3: Deep analysis
    ];
  }

  // Declare what depends on what
  defineDependencies(): Record<string, string[]> {
    return {
      sentiment: ['filter_spam'],
      categorize: ['filter_spam'],
      analyze_category: ['categorize']
    };
  }

  // Build prompts (engine calls this for each dimension)
  createPrompt(context: PromptContext): string {
    const content = context.sections[0]?.content || '';

    if (context.dimension === 'filter_spam') {
      return `Is this spam? "${content}"
Return JSON: { "is_spam": boolean, "confidence": 0-1 }`;
    }

    if (context.dimension === 'sentiment') {
      return `Analyze sentiment: "${content}"
Return JSON: { "sentiment": "positive|negative|neutral", "score": 0-1 }`;
    }

    return '';
  }

  // Choose AI model per dimension
  selectProvider(dimension: string): ProviderSelection {
    // Cheap model for filtering (Haiku: $0.80/$4 per 1M tokens)
    if (dimension === 'filter_spam' || 
        dimension === 'sentiment' || 
        dimension === 'categorize') {
      return {
        provider: 'anthropic',
        options: { model: 'claude-3-5-haiku-20241022' }
      };
    }

    // Powerful model for deep analysis (Sonnet: $3/$15 per 1M tokens)
    return {
      provider: 'anthropic',
      options: { model: 'claude-3-7-sonnet-20250219' }
    };
  }

  // Skip expensive analysis on spam
  shouldSkipSectionDimension(context: any): boolean {
    if (context.dimension === 'sentiment' || context.dimension === 'categorize') {
      const spam = context.dependencies.filter_spam?.data?.is_spam;
      return spam;
    }
    return false;
  }
}

The engine analyzes the dependency graph and automatically:

Runs independent tasks in parallel (sentiment + categorize)
Waits for dependencies before execution
Retries failures with exponential backoff
Tracks costs per dimension

Full code example on GitHub →

How It Executes

The DAG determines execution order:

filter_spam (20 reviews)
    ↓
    ├─→ sentiment (parallel)
    └─→ categorize (parallel)
    ↓
group_by_category (10 reviews → 5 groups)
    ↓
analyze_category (5 groups)

No manual coordination. No Promise.all() hell.

Skip Logic Saves Money

Here's the key optimization - skip expensive analysis on spam:

shouldSkipSectionDimension(context: any): boolean {
  if (context.dimension === 'sentiment' || context.dimension === 'categorize') {
    const spam = context.dependencies.filter_spam?.data?.is_spam;
    return spam;  // Skip downstream analysis
  }
  return false;
}

Real impact: This saved 20 API calls (10 spam reviews × 2 dimensions) = 30% efficiency gain

Multi-Model Routing

Use cheap models for filtering, expensive models for analysis:

selectProvider(dimension: string): ProviderSelection {
  // Cheap: Haiku at $0.80/$4 per 1M tokens
  if (dimension === 'filter_spam' || 
      dimension === 'sentiment' || 
      dimension === 'categorize') {
    return {
      provider: 'anthropic',
      options: { model: 'claude-3-5-haiku-20241022' }
    };
  }

  // Powerful: Sonnet at $3/$15 per 1M tokens
  return {
    provider: 'anthropic',
    options: { model: 'claude-3-7-sonnet-20250219' }
  };
}

Real Numbers

Processing 20 customer reviews through this pipeline:

✅ Spam filter detected 10 spam reviews

✅ Skip logic prevented 20 unnecessary API calls

✅ Multi-model routing: Haiku for filtering, Sonnet for deep analysis

✅ Total cost: $0.0282 in 24 seconds

Compare to alternatives:

Using only Sonnet: $0.094 (3.3× more expensive)
No skip logic: 20 extra API calls wasted on spam

Key insight: 70% cost savings from smart model routing + skip logic

Using the Results

// Create and run engine
const engine = new DagEngine({
  plugin: new ReviewAnalyzer(),
  providers: {
    anthropic: { apiKey: process.env.ANTHROPIC_API_KEY }
  },
  execution: {
    pricing: {
      models: {
        'claude-3-5-haiku-20241022': { inputPer1M: 0.80, outputPer1M: 4.00 },
        'claude-3-7-sonnet-20250219': { inputPer1M: 3.00, outputPer1M: 15.00 }
      }
    }
  }
});

const result = await engine.process(reviews);

// Access results with type safety
result.sections.forEach((section) => {
  const sentiment = section.results.sentiment?.data as SentimentResult;
  console.log(`Review: ${sentiment.sentiment}`);
});

// Check costs
console.log(`Total cost: $${result.stats.totalCost}`);
console.log(`Total tokens: ${result.stats.totalTokens}`);

No spreadsheets. No manual token counting.

Key Features

✅ Automatic parallelization - DAG determines execution order

✅ Skip logic - Avoid wasting API calls on low-quality data

✅ Multi-model routing - Cheap models for filtering, expensive for analysis

✅ Built-in retries - 3 automatic retries with exponential backoff

✅ Cost tracking - Real-time breakdown by dimension and model

✅ Full TypeScript - Type-safe with excellent IDE support

✅ Data transformations - Reshape data between stages (10 reviews → 5 groups)

✅ Multi-provider - Anthropic, OpenAI, Google Gemini

Try It

npm install @dagengine/core

Need help?

💬 Ask a question

📖 Read the docs

🐛 Report bugs

💡 Request features

⭐ Star on GitHub

It's beta. I use it daily. Looking for feedback on what breaks for your use case.

Not asking for trust. Asking you to break it and tell me what's wrong.

What AI workflows are you building that could benefit from parallelization? Drop a comment below! 👇

What's Next?

Want to see more advanced patterns?

📖 Section vs Global dimensions - Handle per-item and cross-item analysis

⚡ High-volume processing - 100 parallel requests with Portkey gateway

Check out the full examples on GitHub.