Robustness

Build resilient AI workflows with graceful degradation and debugging

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

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

1. 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()

2. 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

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']}")

Conditional Branching

from praisonaiagents import AgentFlow, when

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

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, and LangfuseClient each hold a single lazily-constructed shared HTTP client; the passthrough capability (praisonai.capabilities.passthrough) holds one shared sync httpx.Client and one async httpx.AsyncClient per running event loop (so pooled connections never cross loops — this is what makes asyncio.run(apassthrough(...)) in a hot path safe). 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() (drops every per-loop async client and closes the sync one) / passthrough.aclose_clients() (await-closes the current loop’s async client from within its own loop), and JobExecutor.stop().
• Passthrough api_base SSRF guard. On the direct HTTP fallback, passthrough / apassthrough resolve api_base, screen every A/AAAA record (rejecting localhost, loopback, private, link-local, multicast, reserved and unspecified addresses), and pin the connection to the screened IP. This is the same check used for job webhooks in jobs/models.py. See Passthrough SSRF Guard.

Runtime Resilience

Three runtime guarantees keep long-running agents working after failures.

• Hooks: hooked async tools inherit the caller’s ContextVars (session identity, cancel, handoff chain).
• MCP stdio: crashed stdio MCP servers auto-respawn up to 3 times; the dying call is not replayed. See MCP stdio.
• Cancellation: /stop interrupts LLM retry/backoff sleeps within ~200 ms. See Cooperative Cancellation.

Best Practices

Related