Task Configuration

Complete reference for task configuration parameters and workflow control

Task Configuration

This page provides comprehensive documentation for task configuration parameters, including task types, conditional execution, task chaining, and workflow control.

Core Task Parameters

Task Type Configuration

`task_type`

Type: str
Default: "normal"
Options: "normal", "decision", "loop", "parallel", "sequential"
Description: Defines the execution behavior and flow control of the task

The task_type parameter determines how a task is executed within the workflow and how it interacts with other tasks.

python

from praisonaiagents import Task

# Normal task - standard execution
task = Task(
    description="Analyze data",
    task_type="normal",
    expected_output="Data analysis report"
)

# Decision task - conditional branching
decision_task = Task(
    description="Evaluate results",
    task_type="decision",
    condition={
        "field": "score",
        "operator": ">=",
        "value": 0.8
    },
    next_tasks=["success_task", "retry_task"]
)

Task Type Details

Normal Tasks

Standard tasks that execute sequentially with defined inputs and outputs.

python

normal_task = Task(
    name="data_processing",
    description="Process customer data",
    task_type="normal",
    agent=data_agent,
    expected_output="Processed data summary"
)

Decision Tasks

Tasks that evaluate conditions and determine workflow paths.

python

decision_task = Task(
    name="quality_check",
    description="Check data quality",
    task_type="decision",
    condition={
        "type": "compound",
        "operator": "AND",
        "conditions": [
            {"field": "completeness", "operator": ">=", "value": 0.95},
            {"field": "accuracy", "operator": ">=", "value": 0.98}
        ]
    },
    next_tasks={
        "true": "publish_task",
        "false": "cleanup_task"
    }
)

Loop Tasks

Tasks that repeat based on conditions or iterations.

python

loop_task = Task(
    name="iterative_improvement",
    description="Improve results iteratively",
    task_type="loop",
    loop_state={
        "max_iterations": 5,
        "condition": {
            "field": "quality_score",
            "operator": "<",
            "value": 0.9
        },
        "increment_field": "iteration_count"
    }
)

Workflow Control Parameters

`condition`

Type: Dict[str, Any]
Default: None
Description: Defines conditions for task execution or branching

Complex condition structures for advanced workflow control:

python

# Simple condition
simple_condition = {
    "field": "status",
    "operator": "==",
    "value": "completed"
}

# Compound condition with AND/OR logic
compound_condition = {
    "type": "compound",
    "operator": "OR",
    "conditions": [
        {
            "type": "compound",
            "operator": "AND",
            "conditions": [
                {"field": "priority", "operator": "==", "value": "high"},
                {"field": "deadline", "operator": "<", "value": "2024-12-31"}
            ]
        },
        {"field": "override", "operator": "==", "value": True}
    ]
}

# Function-based condition
function_condition = {
    "type": "function",
    "function": "evaluate_complex_criteria",
    "params": {
        "threshold": 0.85,
        "include_history": True
    }
}

`next_tasks`

Type: Union[List[str], Dict[str, str]]
Default: []
Description: Defines subsequent tasks in the workflow

Controls task sequencing and branching:

python

# Sequential flow
sequential_task = Task(
    name="step1",
    next_tasks=["step2", "step3"]  # Execute both after completion
)

# Conditional branching
branching_task = Task(
    name="router",
    task_type="decision",
    next_tasks={
        "success": "success_handler",
        "failure": "error_handler",
        "retry": "retry_task"
    }
)

# Dynamic task selection
dynamic_task = Task(
    name="analyzer",
    next_tasks={
        "type": "dynamic",
        "selector": "select_next_task",
        "params": {"strategy": "priority"}
    }
)

`is_start`

Type: bool
Default: False
Description: Marks the task as a workflow entry point

Identifies tasks that can initiate workflow execution:

python

# Entry point task
start_task = Task(
    name="initialization",
    description="Initialize workflow",
    is_start=True,
    next_tasks=["validation", "setup"]
)

# Multiple entry points
alternative_start = Task(
    name="quick_start",
    description="Fast initialization path",
    is_start=True,
    config={"skip_validation": True}
)

Advanced Task Configuration

Complex Workflow Patterns

Parallel Execution Pattern

python

# Parent task that spawns parallel tasks
parallel_parent = Task(
    name="parallel_processor",
    description="Process multiple data sources",
    task_type="parallel",
    config={
        "parallel_tasks": [
            {"name": "source_a", "timeout": 30},
            {"name": "source_b", "timeout": 45},
            {"name": "source_c", "timeout": 60}
        ],
        "wait_for_all": True,
        "failure_threshold": 1  # Allow 1 failure
    }
)

Sequential Pipeline Pattern

python

# Pipeline configuration
pipeline_task = Task(
    name="data_pipeline",
    description="Multi-stage data processing",
    task_type="sequential",
    config={
        "stages": [
            {"name": "extract", "retry": 3},
            {"name": "transform", "timeout": 120},
            {"name": "load", "checkpoint": True}
        ],
        "stop_on_failure": True
    }
)

Task Dependencies and Context

python

# Task with complex dependencies
dependent_task = Task(
    name="aggregator",
    description="Aggregate results from multiple sources",
    context=["analyzer_task", "validator_task"],  # Require outputs
    config={
        "wait_timeout": 300,  # 5 minutes
        "partial_results_allowed": True,
        "minimum_dependencies": 1
    }
)

Error Handling and Recovery

python

# Robust task configuration
robust_task = Task(
    name="critical_operation",
    description="Critical business logic",
    config={
        "error_handling": {
            "retry_count": 3,
            "retry_delay": 5,  # seconds
            "exponential_backoff": True,
            "fallback_task": "emergency_handler",
            "error_types": {
                "NetworkError": {"retry": True, "delay": 10},
                "ValidationError": {"retry": False, "escalate": True}
            }
        },
        "timeout": 180,
        "checkpoint": True,
        "rollback_on_failure": True
    }
)

Task State Management

Loop State Configuration

python

# Advanced loop configuration
iterative_task = Task(
    name="optimizer",
    task_type="loop",
    loop_state={
        "variables": {
            "iteration": 0,
            "best_score": 0.0,
            "history": []
        },
        "conditions": {
            "continue": {
                "type": "compound",
                "operator": "AND",
                "conditions": [
                    {"field": "iteration", "operator": "<", "value": 10},
                    {"field": "improvement", "operator": ">", "value": 0.01}
                ]
            }
        },
        "updates": {
            "iteration": {"operation": "increment", "value": 1},
            "history": {"operation": "append", "field": "current_score"},
            "best_score": {"operation": "max", "field": "current_score"}
        }
    }
)

Configuration Best Practices

Task Naming Conventions

python

# Good naming practices
task = Task(
    name="validate_user_input",  # Descriptive, action-oriented
    description="Validate and sanitize user-provided data",
    config={
        "naming_convention": "verb_noun_qualifier",
        "log_prefix": "[VALIDATION]"
    }
)

Performance Optimization

python

# Optimized task configuration
optimized_task = Task(
    name="batch_processor",
    config={
        "batch_size": 100,
        "parallel_workers": 4,
        "memory_limit": "2GB",
        "cpu_affinity": [0, 1, 2, 3],
        "priority": "high",
        "preload_resources": True
    }
)

Environment Variables

Task behavior can be configured via environment variables:

bash

# Task execution settings
export PRAISONAI_TASK_DEFAULT_TIMEOUT=300
export PRAISONAI_TASK_MAX_RETRIES=3
export PRAISONAI_TASK_RETRY_DELAY=5

# Workflow settings
export PRAISONAI_WORKFLOW_MAX_DEPTH=10
export PRAISONAI_WORKFLOW_PARALLEL_LIMIT=5

# Debug settings
export PRAISONAI_TASK_DEBUG=true
export PRAISONAI_TASK_TRACE_EXECUTION=true

Validation Rules

Parameter Constraints

Parameter	Validation	Constraints
`task_type`	Must be valid type	One of: normal, decision, loop, parallel, sequential
`condition`	Valid structure	Must have field, operator, value or be compound
`next_tasks`	Valid task names	Tasks must exist in workflow
`is_start`	Boolean	At least one task must be marked as start

Condition Operators

Operator	Description	Example
`==`	Equals	`{"field": "status", "operator": "==", "value": "done"}`
`!=`	Not equals	`{"field": "type", "operator": "!=", "value": "error"}`
`>`, `<`	Greater/Less than	`{"field": "score", "operator": ">", "value": 0.8}`
`>=`, `<=`	Greater/Less or equal	`{"field": "count", "operator": ">=", "value": 10}`
`in`	Contains	`{"field": "category", "operator": "in", "value": ["A", "B"]}`
`regex`	Pattern match	`{"field": "email", "operator": "regex", "value": ".*@company.com"}`

Task Configuration