Skip to main content
Non-fatal errors expose callback and memory failures that used to be silently swallowed, so your agent can see what went wrong without crashing the workflow.

Quick Start

1

Reading non-fatal errors from a task output

from praisonaiagents import Agent, Task, PraisonAIAgents

def my_callback(output):
    raise RuntimeError("downstream webhook timed out")

agent = Agent(name="Reporter", instructions="Summarise the input.")
task = Task(description="Summarise 'hello'", agent=agent, callback=my_callback)

workflow = PraisonAIAgents(agents=[agent], tasks=[task])
workflow.start()

if task.output.callback_error:
    print("Callback failed:", task.output.callback_error)
if task.output.non_fatal_errors:
    for err in task.output.non_fatal_errors:
        print("Non-fatal:", err)
2

Inspecting accumulated errors on the Task itself

# Same workflow as above
for err in task.non_fatal_errors:
    print(err)   # e.g. "callback: downstream webhook timed out"

How It Works

SurfaceTypeWhen populated
TaskOutput.callback_errorOptional[str]Only if a callback function raises
TaskOutput.non_fatal_errorsOptional[list[str]]All non-fatal errors (memory ops + callback) from this run
Task.non_fatal_errorslist[str]Same list, available on the Task instance directly

Common Patterns

Pattern A: Skip to next task only when callback succeeded

from praisonaiagents import Agent, Task, PraisonAIAgents

def critical_callback(output):
    # This must succeed for the workflow to be valid
    pass

agent = Agent(name="Worker", instructions="Process the data.")
task = Task(description="Process input", agent=agent, callback=critical_callback)

workflow = PraisonAIAgents(agents=[agent], tasks=[task])
workflow.start()

# Only proceed if callback succeeded
if task.output.callback_error is None:
    print("Safe to continue to next task")
else:
    print("Workflow failed - callback error:", task.output.callback_error)

Pattern B: Route failures to alerting system

from praisonaiagents import Agent, Task, PraisonAIAgents

def alert_on_errors(task):
    if task.non_fatal_errors:
        # Send to monitoring system
        for error in task.non_fatal_errors:
            send_alert(f"Task {task.description}: {error}")

agent = Agent(name="Processor", instructions="Process data.")
task = Task(description="Important task", agent=agent)

workflow = PraisonAIAgents(agents=[agent], tasks=[task])
workflow.start()

alert_on_errors(task)

User Interaction Flow


Best Practices

Always inspect the output before chaining tasks. A task that completed with non-fatal errors may not have produced the expected result.
if task.output.non_fatal_errors:
    print("Task completed with issues:", task.output.non_fatal_errors)
    # Decide whether to continue or retry
If quality_check=True was requested but the memory adapter isn’t ready, the task still completes. Check non_fatal_errors for memory-related issues.
# Look for memory-specific errors
memory_errors = [err for err in task.non_fatal_errors 
                if "memory" in err.lower()]
if memory_errors:
    print("Memory operations failed:", memory_errors)
If a callback must block the workflow on failure, re-raise inside it. The framework currently captures the message but does NOT re-raise automatically.
def critical_callback(output):
    try:
        # Critical operation
        result = send_to_external_api(output.raw)
    except Exception as e:
        # Log the error and re-raise to stop the workflow
        log.error(f"Critical callback failed: {e}")
        raise  # This will stop the entire workflow

Task Output

TaskOutput reference and all available fields

Task Lifecycle

Task execution phases and configuration