Conditional Execution

Unified condition syntax for controlling task and workflow execution

Conditional execution gates tasks and workflow steps on runtime values using one when syntax that works in both AgentFlow pipelines and Task teams.

Runtime support: Task(when=..., then_task=..., else_task=...) and Task(routing=...) are wired into the markdown workflow engine (WorkflowManager, .praisonai/workflows/*.md). Run them with praisonai workflow run <file>.md or WorkflowManager.execute(...) — the runtime that honours these fields. See Markdown Workflow Branches for a runnable branch example.

from praisonaiagents import Agent, Task, when

agent = Agent(name="conditional", instructions="Run tasks only when conditions match.")
agent.start("Execute the follow-up step if the score is above 0.8.")

The user defines workflows; when expressions gate tasks and AgentFlow steps on runtime variables.

Since PraisonAI PR #4020, when/then_task/else_task routing is wired into PraisonAIAgents. On earlier releases the API existed on Task but the Process orchestrator never consulted it — when-only tasks silently stalled or fell through to unrelated tasks.

Overview

Conditional execution allows you to control workflow branching based on variables, scores, or other runtime values. PraisonAI supports:

• String expression conditions - Simple {{variable}} syntax for comparisons
• Dictionary routing - Map decision values to next tasks
• Callable conditions - Custom Python functions

Quick Start

1. Task or AgentFlow

   Task with when (Recommended)

   from praisonaiagents import Task
   
   # Simple condition with then/else routing
   task = Task(
       name="score_check",
       description="Check if score passes threshold",
       when="{{score}} > 80",
       then_task="approve",
       else_task="reject"
   )
   
   # Evaluate the condition
   result = task.evaluate_when({"score": 90})  # True
   next_task = task.get_next_task({"score": 90})  # "approve"
   

Condition Syntax

String Expression Conditions

Use {{variable}} placeholders with comparison operators:

Operator	Example	Description
>	{{score}} > 80	Greater than
>=	{{score}} >= 80	Greater than or equal
<	{{score}} < 50	Less than
<=	{{score}} <= 50	Less than or equal
==	{{status}} == approved	Equal to
!=	{{status}} != rejected	Not equal to
in	{{word}} in {{text}}	Contains (substring)
contains	{{list}} contains {{item}}	Contains (list)

String comparisons don’t require quotes: {{status}} == approved works correctly.

Examples

# Numeric comparisons
"{{score}} > 80"
"{{count}} >= 10"
"{{price}} < 100.50"

# String comparisons
"{{status}} == approved"
"{{category}} != spam"

# Contains checks
"{{text}} contains error"
"{{tags}} in important"

# Boolean checks
"{{is_valid}}"  # True if truthy

Task Condition Parameters

when Parameter

The when parameter accepts a string expression condition:

from praisonaiagents import Task

task = Task(
    name="quality_check",
    description="Check content quality",
    when="{{quality_score}} >= 7",
    then_task="publish",
    else_task="revise"
)

then_task and else_task

Route to different tasks based on condition result:

task = Task(
    name="review",
    description="Review submission",
    when="{{approved}} == true",
    then_task="finalize",    # Run if condition is True
    else_task="request_changes"  # Run if condition is False
)

routing Parameter (Advanced)

For LLM-driven decisions, use the routing parameter (formerly condition). Pass a bare dict, or the TaskRoutingConfig dataclass for clarity:

from praisonaiagents import Task
from praisonaiagents.workflows.workflow_configs import TaskRoutingConfig

# Dict form (still supported)
Task(
    name="decide",
    description="Decide next action based on content",
    routing={"approved": ["publish"], "rejected": ["edit"]},
)

# Dataclass form (recommended for clarity)
Task(
    name="decide",
    description="Decide next action based on content",
    routing=TaskRoutingConfig(
        branches={"approved": ["publish"], "rejected": ["edit"]},
        next_steps=["cleanup"],
    ),
)

TaskRoutingConfig(...) unpacks onto Task.branch_condition and Task.next_steps — the attributes the executor reads — so Task.condition stays a plain dict/str.

Precedence Ladder

The routing parameter resolves in this order (only the last two forms are supported today):

# Dict (map decision value → next steps)
Task(routing={"approved": ["publish"], "rejected": ["edit"]})

# Config class (full control over branches + next_steps)
Task(routing=TaskRoutingConfig(
    branches={"approved": ["publish"]},
    next_steps=["cleanup"],
))

Precedence: Bool > String > Dict > Config — the markdown engine honours the Dict and Config forms.

The condition parameter still works for backward compatibility, but routing is preferred for clarity.

End-to-End Markdown Example

A branch actually being taken in a markdown workflow:

classify_and_route.md

---
name: classify_and_route
---

## step1
Classify the request.

```branch
success: [handle_success]
failure: [handle_failure]
```

## handle_failure
Escalate.

## handle_success
Reply.

praisonai workflow run .praisonai/workflows/classify_and_route.md

When step1’s output contains success, execution jumps to handle_success, skipping handle_failure.

should_run Callable

For complex logic, use a callable:

def check_prerequisites(context):
    return context.get("data_ready", False) and context.get("approved", False)

task = Task(
    name="process",
    description="Process data",
    should_run=check_prerequisites
)

As of PraisonAI PR #4907, should_run is honoured uniformly by both PraisonAIAgents / AgentTeam and the standalone Workflow engine — a falsy return skips the task and records an empty result so downstream tasks can tell a skip from a real empty run. Earlier releases silently ignored should_run under PraisonAIAgents.

AgentFlow Conditions

when() Function

from praisonaiagents import when

flow = AgentFlow(
    steps=[
        "step1",
        when(
            condition="{{result}} == success",
            then_steps=["success_handler"],
            else_steps=["error_handler"]
        ),
        "final_step"
    ]
)

Nested Conditions

flow = AgentFlow(
    steps=[
        when(
            condition="{{score}} >= 80",
            then_steps=[
                when(
                    condition="{{premium}} == true",
                    then_steps=["premium_path"],
                    else_steps=["standard_path"]
                )
            ],
            else_steps=["low_score_path"]
        )
    ]
)

Multi-agent Workflows (PraisonAIAgents)

when/then_task/else_task also routes multi-agent workflows built with PraisonAIAgents.

from praisonaiagents import Agent, Task, PraisonAIAgents

reviewer  = Agent(name="reviewer",  instructions="Score the draft 0-100 as JSON: {\"score\": <int>}")
publisher = Agent(name="publisher", instructions="Publish the approved draft")
writer    = Agent(name="writer",    instructions="Revise the draft")

review  = Task(name="review",  description="Score the draft", agent=reviewer)
publish = Task(name="publish", description="Publish it",       agent=publisher)
revise  = Task(name="revise",  description="Revise the draft", agent=writer)

gate = Task(
    name="gate",
    description="Route based on score",
    agent=reviewer,
    when="{{score}} > 80",
    then_task="publish",
    else_task="revise",
)

PraisonAIAgents(
    agents=[reviewer, publisher, writer],
    tasks=[review, gate, publish, revise],
).start()

should_run under PraisonAIAgents

A should_run gate composes directly with PraisonAIAgents(...).start() — a falsy return skips the task.

from praisonaiagents import Agent, Task, PraisonAIAgents

reviewer  = Agent(name="reviewer",  instructions="Score the draft 0-100 as JSON: {\"score\": <int>}")
publisher = Agent(name="publisher", instructions="Publish the approved draft")

review = Task(name="review", description="Score the draft", agent=reviewer, output_variable="score")

publish = Task(
    name="publish",
    description="Publish it",
    agent=publisher,
    context=[review],
    should_run=lambda ctx: ctx.variables.get("score", 0) > 80,
)

PraisonAIAgents(
    agents=[reviewer, publisher],
    tasks=[review, publish],
).start()

An async def gate is awaited natively on the async path and resolved to completion on the sync path.

async def check(ctx):
    return ctx.variables.get("score", 0) > 80

publish = Task(name="publish", description="Publish it", agent=publisher, should_run=check)

How the routing context is built — the when expression is evaluated against the current task (the one that just completed), not the target:

• The context is result.to_dict() (from json_dict / pydantic) merged with previous_output = result.raw.
• Access structured fields as {{field_name}} (e.g. {{score}} when the agent returns {"score": 90}).
• Access raw text as {{previous_output}}.

Priority — when both when and next_tasks are set on the same task, when-routing wins.

Clean termination — if the taken branch resolves to None (e.g. only then_task is set and the condition is false, with no next_tasks fallback), the workflow ends cleanly on that path. It does not pick an unrelated not-started task.

Flow Diagram

How It Works

---

Best Practices

Use simple conditions

Keep conditions readable and simple. Complex logic should go in should_run callables.

# ✅ Good - Simple and clear
when="{{score}} > 80"

# ❌ Avoid - Too complex
when="{{score}} > 80 and {{status}} == approved and {{count}} < 10"

Provide both then_task and else_task

Specify both branches when you want an explicit fork:

# ✅ Both branches defined - explicit fork
Task(
    when="{{approved}}",
    then_task="proceed",
    else_task="wait"
)

# ✅ One-branch routing (early-exit) - intentional shape
# When the condition is false, the branch resolves to None and the
# workflow ends cleanly on this path instead of picking an unrelated task.
Task(
    when="{{approved}}",
    then_task="proceed"
)

Use routing for LLM decisions

When the LLM needs to make a decision, use routing with task_type="decision":

Task(
    name="classifier",
    description="Classify the input",
    task_type="decision",
    routing={
        "positive": ["positive_handler"],
        "negative": ["negative_handler"],
        "neutral": ["neutral_handler"]
    }
)

Keep condition keys lowercase and whitespace-free

The router normalises the model’s output with .lower().strip() before the lookup, so any key that carries uppercase letters, surrounding whitespace, or punctuation falls through to the silent exit branch.

# ✅ Do — lowercase, no whitespace, no punctuation
Task(
    name="judge",
    description="Reply with exactly one word: valid or invalid.",
    task_type="decision",
    routing={"valid": [], "invalid": ["rewrite"]},
)

# ❌ Don't — capitalised / padded / punctuated keys never match
Task(
    name="judge",
    description="Judge the draft.",
    task_type="decision",
    routing={"Valid": [], "Invalid.": ["rewrite"], " retry ": ["rewrite"]},
)

Silent-exit pitfall — an unmatched decision and a deliberate exit take the same branch, with no exception and no warning. If your workflow finishes on the first pass when you expected a retry loop, the judge’s verdict probably didn’t match any key in routing/condition. Check the Workflow exit condition met on decision: log line to see the exact normalised key the router tried, then add it (lowercase, whitespace-free) to your routing map. See Task Validation & Feedback for the full deep dive.

Migration Guide

From condition to routing

# Old syntax (still works)
Task(condition={"yes": ["next"], "no": ["stop"]})

# New syntax (recommended)
Task(routing={"yes": ["next"], "no": ["stop"]})

Adding when to existing Tasks

# Before - Using should_run callable
Task(
    should_run=lambda ctx: ctx.get("score", 0) > 80
)

# After - Using when expression (simpler)
Task(
    when="{{score}} > 80"
)

API Reference

Task Parameters

Parameter	Type	Description
when	str	String expression condition
then_task	str	Task name to run if condition is True
else_task	str	Task name to run if condition is False
routing	Dict[str, List[str]] | TaskRoutingConfig	Map decision values to task names; TaskRoutingConfig unpacks to branch_condition / next_steps
should_run	Callable	Custom condition function

Task Methods

Method	Returns	Description
evaluate_when(context)	bool	Evaluate the when condition
get_next_task(context)	str	Get next task based on condition

Related

Markdown Workflow Branches

Route markdown-workflow steps on output

Tasks

Task fields including when / then_task / else_task

AgentFlow

Learn about deterministic pipelines

AgentTeam

Multi-agent task orchestration