Error Handling

Control what happens when a workflow step fails: stop the run, continue past it, or retry

Every workflow step decides what happens when it fails. By default a failure stops the whole run — but you can also continue past it or retry.

The simplest failing pipeline: on_error="stop" (the Task default) aborts the run the moment a step raises.

from praisonaiagents import Task, AgentFlow, WorkflowContext, StepResult

def boom(ctx: WorkflowContext) -> StepResult:
    raise RuntimeError("boom")

def after(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="this never runs")

workflow = AgentFlow(steps=[
    Task(name="boom", handler=boom, max_retries=0),
    Task(name="after", handler=after),
])
result = workflow.start("go")

print(result["status"])  # "failed" — after() never ran

How it Works

A step raises, on_error is consulted, and the run either aborts or keeps going — but the final status reflects the failure either way.

Which mode when?

Pick the mode that matches how much a single failure should affect the run.

• Stop → data integrity matters; one failure means the whole run is wrong.
• Continue → batch processing; partial results are useful.
• Retry → transient failures like network blips or rate limits.

on_error Options

Retry Configuration

Set max_retries on flaky steps. Retries use exponential backoff of 2 ** (attempt - 1) seconds between attempts.

from praisonaiagents import Task, AgentFlow

class TransientError(Exception):
    is_retryable = True   # honoured by the retry machinery

flow = AgentFlow(steps=[
    Task(name="fetch", handler=fetch_data, max_retries=3, on_error="retry"),
])
flow.start("go")

• max_retries — how many extra attempts before the step is considered failed (default 3).
• is_retryable — set False on your exception class to short-circuit retries for known non-retryable failures.
• Backoff between attempts is 2 ** (attempt - 1) seconds.

Guardrail Failures

A guardrail that never validates after exhausting its retries is treated like an exception — it surfaces as step_error and follows the same on_error rules. See Guardrails for validator configuration.

Errors in Nested Patterns

As of PR #4194, the same rules apply inside every nested pattern. A stopping step propagates out of the pattern and aborts the enclosing workflow.

from praisonaiagents import Task, AgentFlow, WorkflowContext, StepResult, loop

def boom(ctx: WorkflowContext) -> StepResult:
    raise RuntimeError("boom")

def after(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="this never runs")

workflow = AgentFlow(
    steps=[
        loop(steps=[Task(name="boom", handler=boom, max_retries=0)], over="items"),
        Task(name="after", handler=after),
    ],
    variables={"items": ["only"]},
)
result = workflow.start("go")

print(result["status"])                                     # "failed"
print(any(s["step"] == "after" for s in result["steps"]))   # False

from praisonaiagents import Task, AgentFlow, WorkflowContext, StepResult, parallel, loop

def boom(ctx: WorkflowContext) -> StepResult:
    raise RuntimeError("boom")

workflow = AgentFlow(steps=[
    parallel(
        [loop(steps=[Task(name="boom", handler=boom, max_retries=0)], over="items")],
        on_failure="fail_fast",
    )
], variables={"items": ["only"]})

result = workflow.start("go")
print(result["status"])  # "failed"

on_error="continue" keeps running

from praisonaiagents import Task, AgentFlow, WorkflowContext, StepResult, loop

def boom(ctx: WorkflowContext) -> StepResult:
    raise RuntimeError("boom")

workflow = AgentFlow(
    steps=[loop(
        steps=[Task(name="boom", handler=boom, on_error="continue", max_retries=0)],
        over="items",
    )],
    variables={"items": ["a", "b", "c"]},
)
result = workflow.start("go")

print(result["status"])  # "failed" — the run keeps going but reports failure

The exception text is never fed to the next step. A downstream step receives the last successful output (or the initial input), never "Error: boom".

Reading the Result

Inspect result["status"] and each step’s error field to find what failed.

result = workflow.start("go")
if result["status"] == "failed":
    for step in result["steps"]:
        if step.get("error"):
            print(f"{step['step']} failed: {step['error']}")

You can also read workflow.step_statuses for a per-step status map after the run.

parallel(on_failure=...) Modes

For the full merge rule (declaration order, collision warning, delta-only merge), see Named outputs from branches.

In every mode the whole run is marked failed. With partial_ok, the failing branch is recorded as {"output": None, "status": "failed", "error": str(e)} — the exception text is not folded into the branch output.

As of PraisonAI #4203, Parallel propagates a nested on_error="stop" → halts the workflow (matches Route / Loop / Repeat / If), even under partial_ok. The None recorded for the failed branch stays index-aligned with its step, so parallel_outputs[i] is always the result of parallel_step.steps[i]. See Stop propagation from nested steps.

Behavior Change

Best Practices

Related