Praison AI

Get Started

Agent Tasks

Understanding Tasks in PraisonAI

Next

Tasks are units of work that agents execute. Each task has a description, expected output, and optional configuration for routing, retries, and advanced features.

Diagram

📥 Input

📋 Task

🤖 Agent

📤 Output

Quick Start

Basic Task

python
from praisonaiagents import Agent, Task, AgentTeam

agent = Agent(name="Researcher", instructions="Research topics")

task = Task(
    action="Research AI trends in 2024",
    expected_output="A summary report",
    agent=agent
)

team = AgentTeam(agents=[agent], tasks=[task])
team.start()

With Handler

python
# Custom function instead of agent
def process_data(context):
    return {"result": "Processed"}

task = Task(
    action="Process the data",
    handler=process_data
)

Core Parameters

ParameterTypeDescription
actionstrWhat the task should accomplish
expected_outputstrWhat the output should look like
agentAgentAgent to execute the task
namestrOptional identifier
toolslistTools available to the task

Task Dependencies

Use context to chain tasks:

python
task1 = Task(
    name="research",
    action="Research AI trends",
    agent=researcher
)

task2 = Task(
    name="write",
    action="Write a report based on research",
    agent=writer,
    context=[task1]  # or depends_on=[task1]
)
Diagram

📋 Research

📝 Write

Conditional Execution

Simple Condition (should_run)

python
task = Task(
    action="Optional step",
    should_run=lambda ctx: ctx.get("enabled", False)
)

When/Then/Else Routing

python
task = Task(
    action="Evaluate score",
    when="{{score}} > 80",      # Condition expression
    then_task="celebrate",       # If true
    else_task="retry"           # If false
)

Loop Support

Iterate over a list:

python
task = Task(
    action="Process {{item}}",
    loop_over="items",    # Variable containing the list
    loop_var="item"       # Name for each item
)

Callbacks

Use on_task_complete for task completion notifications:

python
def on_complete(output):
    print(f"Task done: {output.raw}")

task = Task(
    action="Do something",
    agent=agent,
    on_task_complete=on_complete
)

Guardrails

Validate task output before accepting:

python
def validate_output(output):
    if len(output.raw) < 100:
        return (False, "Output too short")
    return (True, output)

task = Task(
    action="Write detailed report",
    agent=writer,
    guardrails=validate_output,
    max_retries=3
)

Robustness Features

ParameterTypeDescription
max_retriesintMaximum retry attempts (default: 3)
retry_delayfloatSeconds between retries
skip_on_failureboolContinue workflow if task fails
python
task = Task(
    action="Optional enrichment",
    skip_on_failure=True,   # Workflow continues if this fails
    retry_delay=1.0,        # Wait between retries
    max_retries=3
)

Output Configuration

ParameterTypeDescription
output_filestrSave output to file
output_jsonBaseModelParse output as JSON
output_pydanticBaseModelValidate with Pydantic
output_variablestrStore output in workflow variable
python
from pydantic import BaseModel

class Report(BaseModel):
    title: str
    summary: str

task = Task(
    action="Generate report",
    output_pydantic=Report,
    output_file="report.json"
)

Feature Configs

Enable advanced features per-task:

python
task = Task(
    action="Complex task",
    autonomy=True,         # Agent autonomy level
    knowledge=["docs/"],   # Knowledge base paths
    web=True,              # Web search enabled
    reflection=True,       # Self-reflection
    planning=True,         # Task planning
    caching=True           # Response caching
)

Task Types

TypeDescription
taskStandard task (default)
decisionBranching logic with conditions
loopIterate over items
python
# Decision task
task = Task(
    action="Decide next step",
    task_type="decision",
    condition={
        "approve": ["next_task"],
        "reject": ["error_task"]
    }
)

Async Execution

python
task = Task(
    action="Long running task",
    agent=agent,
    async_execution=True
)

# Run async
result = await team.astart()

Full Parameter Reference

All Task Parameters
ParameterTypeDefaultDescription
actionstr–What the task should do (required)
expected_outputstr“Complete successfully”Expected output format
agentAgentNoneAgent to execute
namestrNoneTask identifier
toolslist[]Available tools
contextlist[]Dependent tasks
depends_onlist[]Alias for context
handlerCallableNoneCustom function
should_runCallableNoneCondition to run
loop_overstrNoneVariable containing list
loop_varstr“item”Loop variable name
whenstrNoneCondition expression
then_taskstrNoneTask if condition true
else_taskstrNoneTask if condition false
on_task_completeCallableNoneCompletion callback
guardrailsCallable/strNoneOutput validation
max_retriesint3Max retry attempts
retry_delayfloat0.0Retry delay seconds
skip_on_failureboolFalseContinue on failure
async_executionboolFalseRun async
output_filestrNoneSave to file
output_jsonBaseModelNoneJSON output model
output_pydanticBaseModelNonePydantic validation
output_variablestrNoneWorkflow variable name
task_typestr“task”task/decision/loop
routingdict{}Next task routing
imageslist[]Image inputs
input_filestrNoneInput file path
memoryMemoryNoneMemory instance
variablesdict{}Task variables
~~description~~str–Deprecated – use action

Related

Agents

Create and configure agents

 AgentTeam

Orchestrate multiple agents

 Callbacks

Task completion callbacks

 Guardrails

Output validation