Workflow Orchestration Example

Build complex graph-based workflows with LLM integration

Overview

This comprehensive example demonstrates MoFA’s graph-based workflow orchestration system. It includes:

• Linear, conditional, and parallel workflow patterns
• ETL data pipelines
• LLM agent workflow integration (Dify-style)
• Event listening and monitoring
• ReAct agent workflow integration
• Multi-agent parallel analysis
• Intelligent data pipelines

What You’ll Learn

• Using WorkflowBuilder for fluent workflow construction
• Implementing conditional branching
• Parallel execution and result aggregation
• State management and data passing
• LLM integration in workflows
• Event monitoring and checkpoints

Prerequisites

• Rust 1.75 or higher
• OpenAI API key (for LLM examples)
• Understanding of async/await and graph concepts

Architecture

Source Code Overview

The example contains 1061 lines with 10 different workflow scenarios:

Linear Workflow

use mofa_sdk::workflow::{WorkflowBuilder, WorkflowValue};

let graph = WorkflowBuilder::new("linear", "Linear Workflow")
    .description("Simple linear data processing")
    .start()
    .task("fetch", "Fetch Data", |_ctx, input| async move {
        info!("Fetching data...");
        let data = format!("Data: {}", input.as_str().unwrap_or("default"));
        Ok(WorkflowValue::String(data))
    })
    .task("process", "Process Data", |_ctx, input| async move {
        info!("Processing: {:?}", input);
        let processed = format!("Processed - {}", input.as_str().unwrap_or(""));
        Ok(WorkflowValue::String(processed))
    })
    .task("save", "Save Result", |_ctx, input| async move {
        info!("Saving: {:?}", input);
        Ok(WorkflowValue::String("Save successful".to_string()))
    })
    .end()
    .build();

let executor = WorkflowExecutor::new(ExecutorConfig::default());
let result = executor.execute(&graph, WorkflowValue::String("API".into())).await?;

Running the Example

1. Run Basic Examples (1-5)
2. cd examples/workflow_orchestration
   cargo run
   

3. Run with LLM Examples (6-10)
4. export OPENAI_API_KEY="your-api-key"
   cargo run
   

Workflow Examples

1. Linear

Pattern: Start → Fetch → Process → Save → End

Simple sequential processing pipeline.

Use Case: ETL, data transformation

2. Conditional

Pattern: Start → Check → Branch A / Branch B → End

Conditional routing based on data values.

Use Case: Decision trees, rule engines

3. Parallel

Pattern: Start → Fork → [A, B, C] → Join → End

Concurrent task execution with aggregation.

Use Case: Batch processing, fan-out/fan-in

4. Data Pipeline

Pattern: Extract → Transform → Load

ETL pipeline with state management.

Use Case: Data warehousing, analytics

5. Event Monitoring

Pattern: Workflow with event listeners

Monitor execution with checkpoints.

Use Case: Observability, debugging

6. ReAct Agent

Pattern: Context → ReAct Agent → LLM Synthesis

Integrate ReAct reasoning into workflows.

Use Case: Complex problem solving

7. Multi-Agent

Pattern: Fork → [Tech, Business, User] → LLM Synthesis

Parallel expert analysis with LLM aggregation.

Use Case: Multi-perspective analysis

8. Conditional LLM

Pattern: LLM Classify → Simple / Complex → Decision

LLM-driven conditional routing.

Use Case: Dynamic task routing

9. Intelligent Pipeline

Pattern: Extract → Transform → LLM Analysis

ETL + intelligent insights.

Use Case: Business intelligence

10. Tool Chain

Pattern: [Calculator, DateTime] → LLM Summary

Parallel tools with LLM synthesis.

Use Case: Multi-tool orchestration

Key Concepts

WorkflowBuilder API

WorkflowBuilder::new(id, name)
    .description(desc)              // Workflow description
    .start()                        // Add start node
    .task(id, name, handler)        // Add task node
    .condition(id, name, predicate) // Add conditional node
    .parallel(id, name)             // Start parallel section
    .branch(id, name, handler)      // Add parallel branch
    .join(id, name)                 // Join parallel branches
    .llm_agent(id, name, agent)     // Add LLM node
    .end()                          // Add end node
    .build()                        // Build graph

Node Types

Start Node

Entry point of the workflow:

WorkflowNode::start("start")

Task Node

Executes custom logic:

WorkflowNode::task("task_id", "Task Name", |ctx, input| async move {
    // Your logic
    Ok(WorkflowValue::String("result".to_string()))
})

Condition Node

Routes based on predicate:

WorkflowNode::condition("check", "Check", |ctx, input| async move {
    input.as_i64().unwrap_or(0) > 50
})

LLM Agent Node

Integrates LLM agent:

WorkflowNode::llm_agent("llm", "LLM Node", agent)

End Node

Workflow termination:

WorkflowNode::end("end")

Context and State

// Access workflow context
.task("process", "Process", |ctx, input| async move {
    // Set variable
    ctx.set_variable("count", WorkflowValue::Int(10)).await;
    
    // Get variable
    let count = ctx.get_variable("count").await
        .and_then(|v| v.as_i64())
        .unwrap_or(0);
    
    Ok(WorkflowValue::Int(count + 1))
})

Event Monitoring

use mofa_sdk::workflow::ExecutionEvent;

let (event_tx, mut event_rx) = mpsc::channel(100);

let executor = WorkflowExecutor::new(config)
    .with_event_sender(event_tx);

tokio::spawn(async move {
    while let Some(event) = event_rx.recv().await {
        match event {
            ExecutionEvent::WorkflowStarted { workflow_id, .. } => {
                info!("Started: {}", workflow_id);
            }
            ExecutionEvent::NodeCompleted { node_id, result } => {
                info!("Completed: {} - {:?}", node_id, result.status);
            }
            _ => {}
        }
    }
});

Expected Output

=== MoFA Workflow Orchestration Example ===

--- Example 1: Linear Workflow ---
  [fetch_data] Fetching data...
  [process] Processing data
  [save] Saving result
  Workflow status: Completed
  Number of executed nodes: 5

--- Example 2: Conditional Branch Workflow ---
  [check_value] Checking value: 75 (threshold: 50)
  [high_path] Executing high-value path
  Workflow status: Completed

--- Example 3: Parallel Execution Workflow ---
  [task_a] Starting Task A...
  [task_b] Starting Task B...
  [task_c] Starting Task C...
  [task_b] Task B complete
  [task_c] Task C complete
  [task_a] Task A complete
  [join] Aggregating all results
  Workflow status: Completed

=== Dify-style LLM/Agent Workflow Examples ===

--- Example 6: ReAct Agent Decision Workflow ---
  [gather_context] Gathering context information...
  [react_agent] Starting ReAct reasoning...
  [react_agent] Reasoning complete, iterations: 3
  [final_synthesis] LLM synthesizing results...
  Workflow status: Completed

=== All examples have finished executing ===

Advanced Features

Checkpoints and Recovery

let executor = WorkflowExecutor::new(ExecutorConfig {
    enable_checkpoints: true,
    checkpoint_interval: 3,  // Checkpoint every 3 nodes
    ..Default::default()
});

// Resume from checkpoint
let result = executor.resume_from_checkpoint(
    &checkpoint_data,
    &graph
).await?;

Timeout and Retries

let config = ExecutorConfig {
    max_execution_time: Duration::from_secs(300),
    retry_on_failure: true,
    max_retries: 3,
    ..Default::default()
};

Custom Aggregation

.join_with_transform("join", "Aggregate", |results| async move {
    // Custom aggregation logic
    let sum: i64 = results.values()
        .filter_map(|v| v.as_i64())
        .sum();
    WorkflowValue::Int(sum)
})

Common Use Cases

ETL Pipelines

Extract, transform, load data workflows

ML Pipelines

Training, validation, deployment flows

CI/CD

Build, test, deploy automation

Business Process

Multi-stage approval workflows

Troubleshooting

Node Execution Failure

Problem: Node returns error

Solution: Add error handling:

.task("task", "Task", |ctx, input| async move {
    match process(input).await {
        Ok(result) => Ok(result),
        Err(e) => {
            // Log and return default
            warn!("Task failed: {}", e);
            Ok(WorkflowValue::String("default".into()))
        }
    }
})

Workflow Deadlock

Problem: Workflow hangs

Solution: Check for circular dependencies and add timeout:

let config = ExecutorConfig {
    max_execution_time: Duration::from_secs(60),
    ..Default::default()
};

Memory Issues

Problem: Large workflows consume too much memory

Solution: Use streaming or chunking:

.task("process", "Process", |ctx, input| async move {
    // Process in chunks
    for chunk in input.chunks(1000) {
        process_chunk(chunk).await?;
    }
    Ok(WorkflowValue::Null)
})

Next Steps

ReAct Agent

Integrate reasoning agents

Multi-Agent

Multi-agent workflows

Workflow Guide

Advanced workflow patterns

Workflow API

Complete API reference