Robustness

Build production-ready workflows that continue when optional steps fail and leave an execution trace for debugging.

As of PraisonAI #4334 (2026-08-25), skip_on_failure=True and on_error="continue" are honored across every failure-cascade path (AgentTeam sync + async, AgentFlow, Process.sequential / asequential). Previously these flags were inert on cascade paths and dependents were force-failed the moment any upstream dependency failed. If you were working around this by removing context=[...] dependencies, you can put them back.

from praisonaiagents import Agent, Task, AgentFlow

researcher = Agent(name="researcher", instructions="Research topics thoroughly")
enricher = Agent(name="enricher", instructions="Add real-world examples")

workflow = AgentFlow(
    steps=[
        Task(description="Research AI trends", agent=researcher, max_retries=3, retry_delay=1.0),
        Task(description="Add examples", agent=enricher, skip_on_failure=True, retry_delay=0.5),
    ],
    history=True,
)
result = workflow.start("AI agents 2025")

The user starts a multi-step workflow; optional steps can fail without aborting the run, with execution history for debugging.

How It Works

Quick Start

Simple Usage

from praisonaiagents import Agent, Task, AgentFlow

researcher = Agent(name="researcher", instructions="Research topics thoroughly")
enricher = Agent(name="enricher", instructions="Add real-world examples")

workflow = AgentFlow(
    steps=[
        Task(description="Research AI trends", agent=researcher, max_retries=3, retry_delay=1.0),
        Task(description="Add examples", agent=enricher, skip_on_failure=True),
    ],
    history=True,
)
result = workflow.start("AI agents 2025")
history = workflow.get_history()

With Configuration

Full YAML with retries, optional steps, and execution history:

name: Robust Workflow
history: true

steps:
  - agent: researcher
    action: Research AI trends
    max_retries: 3
    retry_delay: 1.0

  - agent: enricher
    action: Add examples
    skip_on_failure: true
    retry_delay: 0.5

How Graceful Degradation Now Works

A dependent task that opts in still runs even when an upstream dependency in its context fails.

from praisonaiagents import Agent, Task, AgentTeam

researcher = Agent(name="researcher", instructions="Research topics")
enricher   = Agent(name="enricher",   instructions="Add examples")

research = Task(name="research", action="Research AI trends", agent=researcher)
enrich   = Task(
    name="enrich",
    action="Add real-world examples",
    agent=enricher,
    context=[research],
    skip_on_failure=True,     # or on_error="continue"
)

team = AgentTeam(agents=[researcher, enricher], tasks=[research, enrich])
team.start()   # even if `research` fails, `enrich` still runs (in degraded mode)

Equivalent Flags

Two flags opt into graceful degradation — pick either one.

skip_on_failure=True — legacy boolean, still works, means “keep the workflow moving”.
on_error="continue" — richer three-state field ("stop" / "continue" / "retry"). Prefer this in new code.
Either flag alone is enough; setting both is redundant but harmless.

Configuration Options

Param	Type	Default	Description
`skip_on_failure`	`bool`	`False`	Continue workflow if this task fails (legacy; equivalent to `on_error="continue"`).
`on_error`	`str`	`"stop"`	Failure behaviour: `"stop"` fails the workflow, `"continue"` keeps it running (equivalent to `skip_on_failure=True`), `"retry"` retries per `max_retries`.
`retry_delay`	`float`	`0.0`	Seconds between retries.
`max_retries`	`int`	`3`	Maximum retry attempts.
`history`	`bool`	`False`	Enable execution trace on `AgentFlow`.

Task Parameters

skip_on_failure

task = Task(
    description="Optional data enrichment",
    skip_on_failure=True,
    agent=enricher,
)

on_error

task = Task(
    description="Optional data enrichment",
    on_error="continue",   # "stop" | "continue" | "retry"
    agent=enricher,
)

retry_delay

task = Task(
    description="API call with rate limiting",
    retry_delay=2.0,
    max_retries=5,
    agent=api_agent,
)

Workflow History

workflow = AgentFlow(steps=[task1, task2], history=True)
result = workflow.start("input")

for entry in workflow.get_history():
    print(f"{entry['step']}: {'ok' if entry['success'] else 'failed'}")
    if entry.get("error"):
        print(f"  Error: {entry['error']}")

Field	Type	Description
`step`	`str`	Step or agent name
`timestamp`	`str`	ISO timestamp
`success`	`bool`	Whether the step succeeded
`output`	`str`	Truncated output (500 chars)
`error`	`str`	Error message if failed

Conditional Branching

from praisonaiagents import AgentFlow, when

workflow = AgentFlow(steps=[
    scorer_agent,
    when(
        condition="{{score}} > 80",
        then_steps=[approve_agent],
        else_steps=[reject_agent],
    ),
])

if_() is deprecated. Use when() instead.

Concurrency & Durability

The recipe wrapper serialises shared state and reuses connections so concurrent runs stay correct and cheap.

Concurrent-safe run history. RunHistory.store() / delete() / cleanup() are serialised in-process with a reentrant lock and across processes with a POSIX fcntl advisory lock; _save_index is atomic (tmp file + os.replace). Concurrent completions from recipe.arun_background() cannot silently drop index entries, and a crash mid-write cannot truncate index.json. See Run History → Thread- and process-safety.
Connection pooling across the wrapper. JobHandle, JobExecutor webhooks, the passthrough capability (praisonai.capabilities.passthrough), and LangfuseClient each hold a single lazily-constructed shared HTTP client. Poll loops, per-agent passthrough calls, and per-lookup Langfuse fetches no longer redo a TLS handshake per call. Deterministic release is provided via JobHandle.close() / with ... as job:, LangfuseClient.close() / with ... as client:, passthrough.close_clients() / passthrough.aclose_clients(), and JobExecutor.stop().

Best Practices

Mark optional steps with skip_on_failure

Enhancement and enrichment tasks should not block the main workflow path.

Use higher retry_delay for rate-limited APIs

Set retry_delay=2.0 or more when calling external APIs with throttling.

Enable history during development

Always set history=True while building workflows — inspect get_history() after failures.

Separate critical from optional in YAML

Keep required steps strict (skip_on_failure: false) and flag enrichers as optional explicitly.

Workflows

Workflow patterns and orchestration

Workflow Error Recovery

Recover from workflow failures gracefully

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)

How It Works

Quick Start

How Graceful Degradation Now Works

Equivalent Flags

Configuration Options

Task Parameters

skip_on_failure

on_error

retry_delay

Workflow History

Conditional Branching

Concurrency & Durability

Best Practices

Workflows

Workflow Error Recovery

​How It Works

​Quick Start

​How Graceful Degradation Now Works

​Equivalent Flags

​Configuration Options

​Task Parameters

​skip_on_failure

​on_error

​retry_delay

​Workflow History

​Conditional Branching

​Concurrency & Durability

​Best Practices

​Related

Workflows

Workflow Error Recovery

How It Works

Quick Start

How Graceful Degradation Now Works

Equivalent Flags

Configuration Options

Task Parameters

skip_on_failure

on_error

retry_delay

Workflow History

Conditional Branching

Concurrency & Durability

Best Practices

Related