Non-Fatal Errors

See which callbacks or memory operations failed without stopping your workflow

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.

from praisonaiagents import Agent, Task, PraisonAIAgents

agent = Agent(name="Reporter", instructions="Summarise the input.")
task = Task(description="Summarise 'hello'", agent=agent)
PraisonAIAgents(agents=[agent], tasks=[task]).start()

The user runs a workflow; callback or memory failures surface on the task output without stopping the run.

Quick Start

1. Simple Usage

   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. With Configuration

   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,
       fail_on_callback_error=True,  # re-raise instead of capturing
   )
   
   workflow = PraisonAIAgents(agents=[agent], tasks=[task])
   try:
       workflow.start()
   except RuntimeError as e:
       print("Callback blocked workflow:", e)

How It Works

Error Classification

Non-Fatal Errors (Captured)

These errors are captured and stored in non_fatal_errors without stopping task execution:

• Callback failures: Exceptions in task callback functions
• Memory operation failures: Memory storage, retrieval, or quality check failures
• Non-critical integrations: Optional service failures

Fatal Errors (Raised)

These errors now raise exceptions and stop task execution:

• LLM errors: Chat completion failures (now raise LLMError)
• Tool execution failures: Tool failures raise ToolExecutionError on both sync (chat()) and async (achat()) paths — the error is not swallowed or relabelled as an LLMError (fixes in PR #4252 for sync and PR #6109 for async). See Error Handling → Tool Failure Behavior.
• Validation errors: Configuration or input validation failures

from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import LLMError

def failing_callback(output):
    raise Exception("Webhook failed")  # Non-fatal: captured

agent = Agent(name="Test Agent", instructions="Process input")
task = Task(
    description="Process data", 
    agent=agent, 
    callback=failing_callback
)

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

try:
    workflow.start()
    # Check non-fatal errors
    if task.non_fatal_errors:
        print("Non-fatal errors:", task.non_fatal_errors)
        # ["callback: Webhook failed"]
except LLMError as e:
    # Fatal LLM errors are raised, not captured
    print(f"Fatal LLM error: {e.message}")

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

if task.output.non_fatal_errors:
    print("Task completed with issues:", task.output.non_fatal_errors)
    # Decide whether to continue or retry

# 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)

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

Related