Agent Process Types

Understanding Process Types in PraisonAI

Understanding Process Types

Process types in PraisonAI define how tasks are executed and how agents collaborate. Each process type offers different patterns for task execution and agent coordination.

Naming Convention: AgentTeam uses process parameter for sequential/hierarchical. AgentFlow is a separate class for workflow patterns with route(), parallel(), loop().

Process Types Overview

Sequential

Linear task execution in a predefined order via AgentTeam

Hierarchical

Manager-coordinated task execution with dynamic assignment via AgentTeam

Workflow

Complex flows with routing, parallel, and loops via AgentFlow

Sequential Process

The simplest form of task execution where tasks are performed one after another.

Characteristics

• Linear execution flow
• Each task receives output from previous task
• Predictable order
• Simple dependency management

Usage

from praisonaiagents import Agent, AgentTeam

agents = AgentTeam(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process="sequential"  # Default
)
result = agents.start()

Hierarchical Process

Uses a manager LLM to coordinate task execution and agent assignments dynamically.

Characteristics

• Manager-driven coordination using manager_llm
• Dynamic task assignment to best-suited agents
• Flexible execution order (manager decides)
• Intelligent resource allocation

Configuration

from praisonaiagents import Agent, Task, AgentTeam

agents = AgentTeam(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process="hierarchical",
    manager_llm="gpt-4o"  # Required for hierarchical
)
result = agents.start()

Workflow Process (AgentFlow)

AgentFlow is a separate class from AgentTeam. Use AgentFlow for deterministic step sequences with advanced patterns.

Features

• Sequential step execution (steps can be agents OR functions)
• Conditional routing with route()
• Parallel execution with parallel()
• Loop iteration with loop(from_csv="file.csv")
• Repeat until condition with repeat()
• Callbacks and guardrails

Implementation

from praisonaiagents import Agent, AgentFlow
from praisonaiagents import route, parallel, loop

researcher = Agent(instructions="Research topics")
writer = Agent(instructions="Write content")

# AgentFlow with agents as steps
flow = AgentFlow(
    steps=[
        researcher,
        route({
            "positive": [writer],
            "negative": [fallback_agent]
        })
    ]
)
result = flow.start("Research AI trends")

AgentFlow Patterns

Getting Started

1. Install PraisonAI

   Install the core package:

   Terminal

   pip install praisonaiagents
   
2. Configure Environment

   Terminal

   export OPENAI_API_KEY=your_openai_key
   

   Generate your OpenAI API key from OpenAI Use other LLM providers like Ollama, Anthropic, Groq, Google, etc. Please refer to the Models for more information.

3. Create Agent

   Create app.py:

   Single Agent Sequential Process

   from praisonaiagents import Agent, Task, AgentTeam
   
   # Create an agent
   researcher = Agent(
       name="Researcher",
       role="Senior Research Analyst",
       goal="Uncover cutting-edge developments in AI",
       backstory="You are an expert at a technology research group",
       
       llm="gpt-4o"
   )
   
   # Define a task
   task = Task(
       name="research_task",
       description="Analyze 2024's AI advancements",
       expected_output="A detailed report",
       agent=researcher
   )
   
   # Run the agents
   agents = AgentTeam(
       agents=[researcher],
       tasks=[task],
       output="silent",
       process="sequential"
   )
   
   result = agents.start()
   

Advanced Features

State Management

Task State Tracking

Monitor and manage the progress of each task in real-time

Context Preservation

Maintain important information across different stages of execution

Data Flow Control

Manage how data moves between tasks and agents efficiently

Error Handling

Graceful Recovery

Automatically handle failures and continue execution

Alternative Paths

Switch to backup plans when primary execution fails

Error Reporting

Detailed error logs and diagnostic information

Monitoring

Progress Tracking

Real-time visibility into task completion status

Performance Metrics

Measure execution time and resource efficiency

Resource Usage

Monitor system resource utilization

Integration

External Systems

Connect with other services and platforms

API Sync

Maintain data consistency across systems

Event Handling

React to system and external events

Async Processing

All process types support asynchronous execution through async generators, enabling efficient parallel processing and non-blocking operations.

Core Async Methods

asequential

Async version of sequential process for non-blocking linear execution

aworkflow

Async workflow process for complex parallel task execution

ahierarchical

Async hierarchical process for distributed task management

Process-Specific Features

Sequential Process

• Tasks execute in order but don’t block
• Maintains sequence while allowing async operations
• Perfect for I/O-heavy tasks

async def main():
    agents = AgentTeam(
        process="sequential",
        async_mode=True
    )
    await agents.astart()

Workflow Process

• Parallel execution with parallel()
• Conditional routing with route()
• Loop iteration with loop()
• Repeat patterns with repeat()

from praisonaiagents import AgentFlow
from praisonaiagents import parallel

workflow = AgentFlow(
    steps=[
        parallel([step1, step2, step3]),
        aggregator_step
    ]
)
result = workflow.start("input")

Hierarchical Process

• Async manager delegation
• Parallel subtask execution
• Dynamic worker assignment

async def main():
    agents = AgentTeam(
        process="hierarchical",
        async_mode=True,
        dynamic_assignment=True
    )
    await agents.astart()

Key Benefits

Performance

• Efficient resource utilization
• Reduced waiting time
• Better throughput

Flexibility

• Mix sync and async tasks
• Adaptable execution patterns
• Easy scaling