Skip to main content
This page covers task/workflow-level reliability (retry jitter, workflow_timeout, fail_on_callback_error). If you are looking for the gateway-level reliability preset (reliability="production" on BotOS / gateway YAML / CLI), see Gateway Reliability Preset — it is a separate feature that composes drain and admission control.
Make agents survive flaky LLMs, hung workflows, and broken callbacks with built-in retry jitter and configurable failure policies.
This page covers task/workflow reliability (retry, timeouts, failure policies). For gateway reliability (graceful drain + inbound admission control presets), see Gateway Reliability.
from praisonaiagents import Agent, Task, PraisonAIAgents

task = Task(
    description="Summarise the article",
    fail_on_callback_error=True,
    fail_on_memory_error=False,
)

workflow = PraisonAIAgents(
    agents=[Agent(name="Writer", instructions="Summarise clearly")],
    tasks=[task],
    workflow_timeout=120,
)
workflow.start()
The user starts a workflow; retries, timeouts, and failure policies keep flaky LLM or callback errors from aborting the run.

Quick Start

1

Simple Usage

from praisonaiagents import Agent, Task, PraisonAIAgents

task = Task(
    description="Summarise the article",
    fail_on_callback_error=True,
    fail_on_memory_error=False,
)

workflow = PraisonAIAgents(
    agents=[Agent(name="Writer", instructions="Summarise clearly")],
    tasks=[task],
    workflow_timeout=120,
)
workflow.start()
2

With Configuration

Strict mode for CI; lenient mode for production with non-fatal error inspection:
from praisonaiagents import Agent, Task, PraisonAIAgents

strict_task = Task(
    description="Validate the output",
    fail_on_callback_error=True,
    fail_on_memory_error=True,
)

lenient_task = Task(
    description="Generate content",
    fail_on_callback_error=False,
    fail_on_memory_error=False,
)

workflow = PraisonAIAgents(
    agents=[Agent(name="Validator", instructions="Check quality")],
    tasks=[strict_task, lenient_task],
    workflow_timeout=300,
)

result = workflow.start()
if result.non_fatal_errors:
    logger.warning(f"Non-fatal errors: {result.non_fatal_errors}")

How It Works

ComponentPurpose
Retry jitterDesynchronises multi-agent retries on rate limits
Workflow timeoutHard kill after specified seconds (sync and async)
Failure policiesSurface or swallow callback and memory exceptions

Configuration Options

ParamTypeDefaultEffect when True
fail_on_callback_errorboolFalseRe-raises exceptions inside task.callback
fail_on_memory_errorboolFalseRe-raises memory-store failures
workflow_timeoutintNoneSeconds before workflow cancellation

Retry categories

Error categoryBehaviourCap
RATE_LIMITexp backoff ×3 + full jitter60s
TRANSIENTexp backoff ×2 + full jitter30s
CONTEXT_LIMITdeterministic 0.5s0.5s
AUTH / INVALID_REQUEST / PERMANENTno retry
Jitter is automatic — there is no flag to turn it off.

Common Patterns

Strict CI mode

task = Task(
    description="Validate output",
    fail_on_callback_error=True,
    fail_on_memory_error=True,
)
workflow = PraisonAIAgents(tasks=[task], workflow_timeout=60)

Lenient production mode

result = workflow.start()
for error in result.non_fatal_errors:
    metrics.increment("task.non_fatal_error", tags={"error_type": type(error).__name__})

Multi-agent fan-out

agents = [Agent(name=f"Worker-{i}") for i in range(10)]
# Jitter prevents thundering herd — no config needed

Best Practices

Network calls can hang indefinitely. Use 60s for quick tasks, 300s for multi-step workflows.
Tests should surface bugs immediately; production should log and continue unless the callback is critical.
The SDK already applies exponential backoff with jitter. Catching RateLimitError and sleeping duplicates that work.
Non-fatal errors indicate latent issues — increment metrics and log them even when the workflow completes.

Task Retry Policy

Per-task retry with exponential backoff

Workflow Error Recovery

Recover from workflow failures gracefully