Conditional Execution

Unified condition evaluation for AgentFlow and AgentTeam workflows

Conditional Execution

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

Architecture Overview

Diagram

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

python

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

python

# 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

python

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

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

Contains Checks

python

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

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

Boolean Evaluation

python

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

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

Nested Properties

python

# 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

python

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

Diagram

Using the Protocol Directly

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

ExpressionCondition

python

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

python

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

python

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.

python

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

python

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

ExpressionCondition

python

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

DictCondition

python

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

python

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