Workflow Error Handling

Workflow error handling provides structured exception handling for step failures, enabling robust parallel execution and graceful error recovery.

from praisonaiagents import Agent, AgentFlow, WorkflowStepError

agent = Agent(name="Research Agent", instructions="Research topics that might fail")
workflow = AgentFlow(steps=[agent])

try:
    workflow.start("Research invalid topic")
except WorkflowStepError as e:
    print(f"Step failed: {e}")

The user runs a workflow; WorkflowStepError surfaces the failing step and root cause for recovery.

Running the same workflow instance concurrently

An AgentFlow instance is not safe to run() (or astart()) concurrently on the same object — its per-run mutable state (status, step_statuses, _handoff_chain) would be corrupted. As of PraisonAI #3086, a second concurrent call raises a clear error instead:

from praisonaiagents import Agent, AgentFlow
import threading

flow = AgentFlow(steps=[Agent(name="Worker", instructions="Do the thing")])

def _worker():
    flow.run("input A")   # first thread wins the lock

t1 = threading.Thread(target=_worker)
t1.start()
try:
    flow.run("input B")   # ← RuntimeError while the first run is in flight
except RuntimeError as exc:
    print(exc)
    # This Workflow instance is already running; a Workflow is not safe to run()
    # concurrently on the same object. Create a separate Workflow instance per
    # concurrent run.
t1.join()

Fix: create one AgentFlow instance per concurrent run — either instantiate a fresh AgentFlow(steps=[...]) on each call, or use a lightweight factory:

from praisonaiagents import Agent, AgentFlow
import threading

def new_flow():
    return AgentFlow(steps=[Agent(name="Worker", instructions="Do the thing")])

# Safe: each thread has its own instance
threads = [threading.Thread(target=lambda x=x: new_flow().run(x)) for x in ("A", "B")]
for t in threads:
    t.start()
for t in threads:
    t.join()

Sequential re-use of the same instance is fine — the lock is released in a finally when a run completes (successfully or not), so the next run() call proceeds normally.

Error	When it fires	Fix
`RuntimeError("This Workflow instance is already running; ...")`	Second `run()` / `astart()` starts on the same `AgentFlow` while the first is still in flight	Create a separate `AgentFlow` instance per concurrent run

Quick Start

Basic Error Handling

from praisonaiagents import Agent, AgentFlow, WorkflowStepError

agent = Agent(
    name="Research Agent", 
    instructions="Research topics that might fail"
)

workflow = AgentFlow(steps=[agent])

try:
    result = workflow.start("Research invalid topic")
except WorkflowStepError as e:
    print(f"Workflow failed: {e}")
    print(f"Root cause: {e.cause}")

Handling Multiple Errors

from praisonaiagents import AgentFlow, parallel, WorkflowStepError

workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c], on_failure="fail_all"),
])

try:
    workflow.start("Process all branches")
except WorkflowStepError as e:
    print(f"Workflow failed: {e}")
    print(f"Root cause: {e.cause}")
    for err in e.errors:
        print(f"  Branch {err['step']}: {err['error']}")

How It Works

Component	Role
WorkflowStepError	Main exception class for workflow failures
cause	Original exception that triggered the failure
errors	List of multiple errors (for parallel failures)

Configuration Options

Attribute	Type	Default	Description
`cause`	`Exception \| None`	`None`	The underlying exception that triggered the failure (first error in `fail_all` mode)
`errors`	`List[dict]`	`[]`	List of `{"step": int, "error": Exception}` for `fail_all` mode. Empty for `fail_fast`

Failed Task Propagation

When tasks fail after exhausting retries, dependent tasks are automatically skipped instead of running with None context:

How It Works

Example

from praisonaiagents import Agent, Task, PraisonAIAgents

agent = Agent(name="Worker", instructions="Process data")

# Primary task that might fail
fetch_data = Task(
    description="Fetch data from unreliable API",
    agent=agent,
    max_retries=3
)

# Dependent task - will be skipped if fetch_data fails
process_data = Task(
    description="Process the fetched data",
    agent=agent,
    context=[fetch_data]  # Depends on fetch_data
)

workflow = PraisonAIAgents(
    agents=[agent], 
    tasks=[fetch_data, process_data]
)

result = workflow.start()

# Check task statuses
print(f"Fetch data status: {fetch_data.status}")
print(f"Process data status: {process_data.status}")

if fetch_data.status == "failed":
    # process_data.status will also be "failed" (skipped)
    print("Primary task failed, dependent task was skipped")

Failure Propagation Rules

Failed Task: When a task fails after max_retries, its status is set to "failed"
Dependent Detection: Tasks with context=[failed_task] are identified as dependents
Skip Execution: Dependent tasks are marked as "failed" without execution
No None Propagation: Dependent tasks don’t receive None values from failed dependencies

Process Integration

This behavior works consistently across all process types:

# Sequential process - stops at first failure
workflow = PraisonAIAgents(
    agents=[agent],
    tasks=[task_a, task_b, task_c],
    process="sequential"  # Stops if task_a fails
)

# Workflow process - skips dependents of failed tasks  
workflow = PraisonAIAgents(
    agents=[agent],
    tasks=[fetch, process, save],
    process="workflow"  # Skips process+save if fetch fails
)

Common Patterns

Pattern 1: Single Step Recovery

from praisonaiagents import Agent, AgentFlow, WorkflowStepError

def with_retry():
    for attempt in range(3):
        try:
            workflow = AgentFlow(steps=[unreliable_agent])
            return workflow.start("Task")
        except WorkflowStepError as e:
            if attempt == 2:  # Last attempt
                raise
            print(f"Attempt {attempt + 1} failed: {e}")

Pattern 2: Parallel Error Analysis

def analyze_parallel_failures(workflow_errors):
    """Analyze which parallel branches failed and why."""
    failed_branches = []
    for error_info in workflow_errors.errors:
        step_idx = error_info['step']
        error = error_info['error']
        failed_branches.append({
            'branch': step_idx,
            'error_type': type(error).__name__,
            'message': str(error)
        })
    return failed_branches

Pattern 3: Graceful Degradation

def robust_workflow(input_data):
    """Run workflow with fallback strategies."""
    try:
        # Try optimal path
        return run_full_workflow(input_data)
    except WorkflowStepError as e:
        if "timeout" in str(e).lower():
            # Fallback to simpler workflow
            return run_simple_workflow(input_data)
        else:
            # Log and re-raise for other errors
            logger.error(f"Workflow failed: {e}")
            raise

Best Practices

Always Catch Specific Errors

Catch WorkflowStepError specifically rather than generic Exception to handle workflow failures appropriately while allowing other errors to bubble up.

try:
    result = workflow.start("Task")
except WorkflowStepError as e:
    # Handle workflow-specific failures
    handle_workflow_error(e)
except Exception as e:
    # Handle unexpected errors
    logger.exception("Unexpected error")
    raise

Inspect Error Details

Use the cause and errors attributes to understand what specifically went wrong and implement targeted recovery strategies.

except WorkflowStepError as e:
    if isinstance(e.cause, TimeoutError):
        # Retry with longer timeout
        retry_with_timeout()
    elif isinstance(e.cause, ConnectionError):
        # Switch to backup service
        use_backup_service()

Log Error Context

Include workflow context in error logs to help with debugging and monitoring.

except WorkflowStepError as e:
    logger.error(
        "Workflow failed",
        extra={
            "workflow_id": workflow.id,
            "step_count": len(workflow.steps),
            "error_count": len(e.errors),
            "root_cause": str(e.cause)
        }
    )

Design for Partial Success

When using parallel execution, design your aggregation logic to handle partial results gracefully.

def smart_aggregator(ctx):
    """Aggregate results even with some failures."""
    outputs = ctx.variables.get("parallel_outputs", [])
    valid_results = [o for o in outputs if not o.startswith("Error:")]
    
    if len(valid_results) >= 2:  # Minimum threshold
        return aggregate_partial_results(valid_results)
    else:
        raise WorkflowStepError("Insufficient successful results")

Workflow Parallel

Parallel execution with failure strategies

Workflow Patterns

Common workflow implementation patterns

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Workflow Error Handling

Running the same workflow instance concurrently

Quick Start

How It Works

Configuration Options

Failed Task Propagation

How It Works

Example

Failure Propagation Rules

Process Integration

Common Patterns

Pattern 1: Single Step Recovery

Pattern 2: Parallel Error Analysis

Pattern 3: Graceful Degradation

Best Practices

Workflow Parallel

Workflow Patterns

​Running the same workflow instance concurrently

​Quick Start

​How It Works

​Configuration Options

​Failed Task Propagation

​How It Works

​Example

​Failure Propagation Rules

​Process Integration

​Common Patterns

​Pattern 1: Single Step Recovery

​Pattern 2: Parallel Error Analysis

​Pattern 3: Graceful Degradation

​Best Practices

​Related

Workflow Parallel

Workflow Patterns

Running the same workflow instance concurrently

Quick Start

How It Works

Configuration Options

Failed Task Propagation

How It Works

Example

Failure Propagation Rules

Process Integration

Common Patterns

Pattern 1: Single Step Recovery

Pattern 2: Parallel Error Analysis

Pattern 3: Graceful Degradation

Best Practices

Related