Conditional Execution

PraisonAI provides a unified, protocol-driven approach to conditional execution across both AgentFlow (deterministic pipelines) and AgentTeam (multi-agent task graphs).

The ConditionProtocol enables DRY condition handling - write conditions once, use them everywhere.

Architecture Overview

Two Condition Paradigms

System	Condition Type	Use Case
AgentFlow	String expressions	`"{{score}} > 80"`
AgentTeam	Dict routing	`{"approved": ["publish"]}`

Both use the shared ConditionProtocol for consistent evaluation.

AgentFlow Conditions

AgentFlow uses string-based conditions with {{variable}} placeholders.

Basic Usage

from praisonaiagents import AgentFlow, Agent, when

researcher = Agent(name="researcher", instructions="Research topics")
approver = Agent(name="approver", instructions="Approve content")
writer = Agent(name="writer", instructions="Write content")

flow = AgentFlow(
    steps=[
        researcher,
        when(
            condition="{{score}} > 80",
            then_steps=[approver],
            else_steps=[writer]
        )
    ]
)

result = flow.run("Research AI trends")

Supported Condition Formats

Numeric Comparisons

# Greater than
when(condition="{{score}} > 80", then_steps=[...])

# Greater or equal
when(condition="{{score}} >= 80", then_steps=[...])

# Less than
when(condition="{{score}} < 50", then_steps=[...])

# Less or equal
when(condition="{{score}} <= 50", then_steps=[...])

# Equal
when(condition="{{score}} == 100", then_steps=[...])

# Not equal
when(condition="{{score}} != 0", then_steps=[...])

String Comparisons

# String equality
when(condition="{{status}} == approved", then_steps=[...])

# String inequality
when(condition="{{status}} != rejected", then_steps=[...])

Contains Checks

# Using 'in' keyword
when(condition="error in {{message}}", then_steps=[...])

# Using 'contains' keyword
when(condition="{{message}} contains success", then_steps=[...])

Boolean Evaluation

# Truthy check
when(condition="{{flag}}", then_steps=[...])

# Boolean literals
when(condition="{{is_valid}}", then_steps=[...])

Nested Properties

# Access nested dict values
when(condition="{{item.score}} >= 60", then_steps=[...])

# Deep nesting
when(condition="{{result.data.status}} == success", then_steps=[...])

AgentTeam Conditions

AgentTeam uses dict-based conditions for task routing.

Basic Usage

from praisonaiagents import AgentTeam, Agent, Task

reviewer = Agent(name="reviewer", instructions="Review content")
publisher = Agent(name="publisher", instructions="Publish content")
editor = Agent(name="editor", instructions="Edit content")

review_task = Task(
    description="Review the content quality",
    agent=reviewer,
    task_type="decision",
    condition={
        "approved": ["publish_task"],
        "rejected": ["edit_task"]
    }
)

publish_task = Task(
    name="publish_task",
    description="Publish the content",
    agent=publisher
)

edit_task = Task(
    name="edit_task", 
    description="Edit and improve content",
    agent=editor
)

team = AgentTeam(
    agents=[reviewer, publisher, editor],
    tasks=[review_task, publish_task, edit_task]
)

result = team.start()

Decision Flow

Using the Protocol Directly

For advanced use cases, you can use the condition classes directly.

ExpressionCondition

from praisonaiagents import ExpressionCondition

# Create a reusable condition
score_check = ExpressionCondition("{{score}} > 80")

# Evaluate against different contexts
result1 = score_check.evaluate({"score": 90})  # True
result2 = score_check.evaluate({"score": 70})  # False

DictCondition

from praisonaiagents import DictCondition

# Create a routing condition
router = DictCondition(
    routes={"approved": ["publish"], "rejected": ["revise"]},
    key="decision"
)

# Check if decision is valid
is_valid = router.evaluate({"decision": "approved"})  # True

# Get target tasks
targets = router.get_target({"decision": "approved"})  # ["publish"]

evaluate_condition Function

from praisonaiagents import evaluate_condition

# Direct evaluation
result = evaluate_condition(
    condition="{{score}} > 80",
    variables={"score": 90},
    previous_output="Previous step output"
)

Custom Conditions

Implement ConditionProtocol for custom condition logic.

from praisonaiagents import ConditionProtocol
from typing import Dict, Any

class TimeBasedCondition:
    """Execute only during business hours."""
    
    def __init__(self, start_hour: int = 9, end_hour: int = 17):
        self.start_hour = start_hour
        self.end_hour = end_hour
    
    def evaluate(self, context: Dict[str, Any]) -> bool:
        from datetime import datetime
        current_hour = datetime.now().hour
        return self.start_hour <= current_hour < self.end_hour

# Use with AgentFlow
business_hours = TimeBasedCondition()
if business_hours.evaluate({}):
    # Execute workflow
    pass

Best Practices

Use Descriptive Variables

Name variables clearly: {{approval_status}} not {{s}}

Handle Missing Variables

Missing variables return False - design for this

Keep Conditions Simple

Complex logic belongs in agents, not conditions

Test Conditions

Unit test conditions with various inputs

API Reference

ConditionProtocol

class ConditionProtocol(Protocol):
    def evaluate(self, context: Dict[str, Any]) -> bool:
        """Evaluate condition against context."""
        ...

ExpressionCondition

class ExpressionCondition:
    def __init__(self, expression: str): ...
    def evaluate(self, context: Dict[str, Any]) -> bool: ...

DictCondition

class DictCondition:
    def __init__(self, routes: Dict[str, List[str]], key: str = "decision"): ...
    def evaluate(self, context: Dict[str, Any]) -> bool: ...
    def get_target(self, context: Dict[str, Any]) -> List[str]: ...

evaluate_condition

def evaluate_condition(
    condition: str,
    variables: Dict[str, Any],
    previous_output: Optional[str] = None
) -> bool: ...

AgentFlow

Deterministic workflow pipelines

AgentTeam

Multi-agent task coordination

Task Reference

Task configuration options

Orchestration

Workflow orchestration patterns

​Conditional Execution

​Architecture Overview

​Two Condition Paradigms

​AgentFlow Conditions

​Basic Usage

​Supported Condition Formats

​AgentTeam Conditions

​Basic Usage

​Decision Flow

​Using the Protocol Directly

​ExpressionCondition

​DictCondition

​evaluate_condition Function

​Custom Conditions

​Best Practices

Use Descriptive Variables

Handle Missing Variables

Keep Conditions Simple

Test Conditions

​API Reference

​ConditionProtocol

​ExpressionCondition

​DictCondition

​evaluate_condition

​Related

AgentFlow

AgentTeam

Task Reference

Orchestration

Conditional Execution

Architecture Overview

Two Condition Paradigms

AgentFlow Conditions

Basic Usage

Supported Condition Formats

AgentTeam Conditions

Basic Usage

Decision Flow

Using the Protocol Directly

ExpressionCondition

DictCondition

evaluate_condition Function

Custom Conditions

Best Practices

API Reference

ConditionProtocol

ExpressionCondition

DictCondition

evaluate_condition

Related