Long Running Workflows

When building production applications with Vellum Workflows, you may encounter scenarios where Workflows take several minutes or longer to complete. This guide covers best practices for handling these long-running Workflows effectively.

Understanding the Challenge

Long-running Workflows can present challenges in production environments:

• Client timeouts: HTTP clients may timeout before Workflow completion
• Resource constraints: Keeping connections open for extended periods
• Error handling: Managing failures in long-running processes
• User experience: Providing feedback during extended operations

Recommended Approaches

Option 1: Webhooks (Recommended)

The most robust approach for long-running Workflows is to execute them asynchronously and use webhooks to receive completion notifications. See our Webhooks documentation for detailed setup instructions.

1
import vellum
2
3
client = vellum.VellumClient(api_key="your-api-key")
4
5
# Use async endpoint to get execution_id immediately
6
response = client.execute_workflow_async(
7
    workflow_deployment_name="your-workflow",
8
    inputs=[
9
        vellum.WorkflowRequestStringInput(
10
            name="user_query",
11
            value="Process this complex request"
12
        )
13
    ],
14
    external_id="task-12345"  # Your internal task ID for correlation
15
)
16
17
execution_id = response.execution_id
18
print(f"Workflow started with execution_id: {execution_id}")
19
# Store execution_id for later correlation with webhook

1
from flask import Flask, request, jsonify
2
import hmac
3
import hashlib
4
5
app = Flask(__name__)
6
7
@app.route('/webhook/vellum', methods=['POST'])
8
def handle_vellum_webhook():
9
    # Verify HMAC signature (recommended)
10
    signature = request.headers.get('X-Vellum-Signature')
11
    timestamp = request.headers.get('X-Vellum-Timestamp')
12
    
13
    if not verify_hmac_signature(request.data, signature, timestamp):
14
        return jsonify({'error': 'Invalid signature'}), 401
15
    
16
    event = request.json
17
    
18
    if event['type'] == 'workflow.execution.fulfilled':
19
        # Workflow completed successfully
20
        external_id = event['data']['parent']['external_id']
21
        outputs = event['data']['outputs']
22
        
23
        # Update your internal task status
24
        update_task_status(external_id, 'completed', outputs)
25
        
26
    elif event['type'] == 'workflow.execution.rejected':
27
        # Workflow failed
28
        external_id = event['data']['parent']['external_id']
29
        error = event['data']['error']
30
        
31
        # Update your internal task status
32
        update_task_status(external_id, 'failed', error)
33
    
34
    return jsonify({'status': 'received'}), 200
35
36
def verify_hmac_signature(payload, signature, timestamp):
37
    # Implement HMAC verification
38
    # See our HMAC Authentication documentation for implementation details:
39
    # https://docs.vellum.ai/product/security/hmac-authentication
40
    pass
41
42
def update_task_status(external_id, status, data):
43
    # Update your database/system with the Workflow result
44
    print(f"Task {external_id} status: {status}")

Option 2: Async Execution with Status Polling

Use the async execution endpoint to initiate a Workflow and immediately receive an execution_id, then poll for the execution status using the Check Workflow Execution Status endpoint. This is the recommended approach for polling-based async execution.

Async executions automatically queue when you exceed your concurrency limit, making this endpoint ideal for batch jobs where you don’t need everything to complete at once. You can initiate many executions quickly and they’ll process as capacity becomes available:

1
import vellum
2
import time
3
4
client = vellum.VellumClient(api_key="your-api-key")
5
6
# Step 1: Start Workflow asynchronously and get execution_id
7
response = client.execute_workflow_async(
8
    workflow_deployment_name="your-workflow",
9
    inputs=[
10
        vellum.WorkflowRequestStringInput(
11
            name="user_query",
12
            value="Process this complex request"
13
        )
14
    ],
15
    external_id="task-12345"  # Optional: your internal ID for tracking
16
)
17
18
execution_id = response.execution_id
19
print(f"Workflow started with execution_id: {execution_id}")
20
21
# Step 2: Poll for execution status
22
while True:
23
    try:
24
        # Check execution status
25
        status_response = client.check_workflow_execution_status(
26
            execution_id=execution_id
27
        )
28
        
29
        if status_response.status == "FULFILLED":
30
            print("Workflow completed successfully!")
31
            print(f"Results: {status_response.outputs}")
32
            if status_response.execution_detail_url:
33
                print(f"View details: {status_response.execution_detail_url}")
34
            break
35
        elif status_response.status == "REJECTED":
36
            print("Workflow failed!")
37
            break
38
        elif status_response.status == "PENDING":
39
            print("Workflow is still pending...")
40
            time.sleep(5)  # Poll every 5 seconds for pending
41
        else:
42
            print(f"Workflow still running... Status: {status_response.status}")
43
            time.sleep(30)  # Poll every 30 seconds for running workflows
44
            
45
    except Exception as e:
46
        print(f"Error checking execution status: {e}")
47
        time.sleep(30)

Option 3: API Node

This approach is beneficial when you want to specify a specific callback URL or payload format. You can use an API Node at the end of your Workflow to send results directly to your system:

1
{
2
  "task_id": "{{workflow_inputs.task_id}}",
3
  "status": "completed",
4
  "results": {
5
    "output": "{{final_output_node.output}}",
6
    "metadata": {
7
      "execution_time": "{{workflow_metadata.duration}}",
8
      "cost": "{{workflow_metadata.cost}}"
9
    }
10
  }
11
}

Option 4: Streaming Updates

For Workflows where you want to provide real-time progress updates, use the streaming execution endpoint. This can still be problematic if updates are really long, but at least enables you to stream updated messaging from your LLMs as they invoke tools:

1
import vellum
2
3
client = vellum.VellumClient(api_key="your-api-key")
4
5
# Stream Workflow execution for real-time updates
6
stream = client.execute_workflow_stream(
7
    workflow_deployment_name="your-workflow",
8
    inputs=[
9
        vellum.WorkflowRequestStringInput(
10
            name="user_query",
11
            value="Process this complex request"
12
        )
13
    ],
14
    external_id="task-12345"
15
)
16
17
for event in stream:
18
    if event.type == "workflow.execution.initiated":
19
        print(f"Workflow started: {event.execution_id}")
20
    elif event.type == "workflow.execution.streaming":
21
        # Handle intermediate results
22
        print(f"Progress update: {event.data}")
23
    elif event.type == "workflow.execution.fulfilled":
24
        print(f"Workflow completed: {event.data.outputs}")
25
        break
26
    elif event.type == "workflow.execution.rejected":
27
        print(f"Workflow failed: {event.data.error}")
28
        break

Timeout Management

Client-Side Timeouts

When using synchronous execution, configure appropriate timeouts:

1
import vellum
2
from vellum.core import RequestOptions
3
4
client = vellum.VellumClient(api_key="your-api-key")
5
6
try:
7
    response = client.execute_workflow(
8
        workflow_deployment_name="your-workflow",
9
        inputs=[...],
10
        request_options=RequestOptions(
11
            timeout_in_seconds=600  # 10 minute timeout
12
        )
13
    )
14
except TimeoutError:
15
    print("Workflow execution timed out")
16
    # Handle timeout - workflow may still be running

Infrastructure Considerations

AWS Lambda

If using AWS Lambda, be aware of the 15-minute maximum execution time:

1
import json
2
import vellum
3
4
def lambda_handler(event, context):
5
    client = vellum.VellumClient(api_key=os.environ['VELLUM_API_KEY'])
6
    
7
    # For long Workflows, use async pattern
8
    response = client.execute_workflow(
9
        workflow_deployment_name="long-running-workflow",
10
        inputs=event['inputs'],
11
        external_id=event['task_id']  # Use for webhook correlation
12
    )
13
    
14
    # Return immediately with execution_id
15
    return {
16
        'statusCode': 200,
17
        'body': json.dumps({
18
            'execution_id': response.execution_id,
19
            'status': 'initiated',
20
            'message': 'Workflow started. Results will be sent via webhook.'
21
        })
22
    }

Container Environments

For containerized applications, ensure your containers can handle long-running connections if using streaming:

1
# Dockerfile
2
FROM python:3.11
3
# ... other setup ...
4
5
# Set appropriate timeouts
6
ENV REQUESTS_TIMEOUT=900
7
ENV WORKFLOW_TIMEOUT=900
8
9
# Health check for long-running processes
10
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
11
  CMD curl -f http://localhost:8000/health || exit 1

Error Handling and Retry Strategies

Workflow-Level Retry

For Workflows that may fail due to transient issues, implement retry logic:

1
import time
2
import vellum
3
from typing import Optional
4
5
def execute_workflow_with_retry(
6
    client: vellum.VellumClient,
7
    workflow_name: str,
8
    inputs: list,
9
    external_id: str,
10
    max_retries: int = 3,
11
    retry_delay: int = 60
12
) -> Optional[str]:
13
    """Execute Workflow with retry logic for transient failures."""
14
    
15
    for attempt in range(max_retries + 1):
16
        try:
17
            response = client.execute_workflow(
18
                workflow_deployment_name=workflow_name,
19
                inputs=inputs,
20
                external_id=f"{external_id}-attempt-{attempt}"
21
            )
22
            return response.execution_id
23
            
24
        except Exception as e:
25
            if attempt == max_retries:
26
                print(f"Workflow failed after {max_retries} retries: {e}")
27
                return None
28
            
29
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay}s...")
30
            time.sleep(retry_delay)
31
    
32
    return None

Node-Level Resilience

Prompts may experience nondeterministic errors from model providers. Longer running Workflows are particularly prone to these issues. To mitigate this, it’s a good idea to implement retry logic with Node Adornemnts. You can also disable streaming from the Model Settings while editing a Prompt to mitigate other nondeterministic connection issues.

Use Node Adornments to add resilience to individual Workflow nodes:

• Retry Node Adornments: Automatically retry failed nodes
• Try Node Adornments: Gracefully handle node failures with fallback paths

See our Node Adornments documentation for detailed configuration.

Monitoring and Observability

Execution Tracking

Monitor long-running Workflows through the Monitoring tab of your Workflow Deployment (see our monitoring documentation for details):

1. Executions Tab: View real-time execution status
2. Timeline View: Analyze execution flow and bottlenecks
3. Cost Tracking: Monitor resource usage for long Workflows

Best Practices Summary

Related Documentation

• Batch Jobs - Process large volumes of Workflow executions efficiently
• Webhooks Configuration
• HMAC Authentication
• Node Adornments
• Workflow Monitoring
• API Reference - Execute Workflow
• API Reference - Execute Workflow Async
• API Reference - Check Workflow Execution Status