> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# 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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
subgraph AgentTeamProcess["AgentTeam Processes"]
direction LR
SEQ["➡️ sequential"]
HIER["👔 hierarchical"]
end
subgraph AgentFlowPatterns["AgentFlow Patterns"]
direction LR
ROUTE["🔀 route()"]
PAR["⚡ parallel()"]
LOOP["🔁 loop()"]
end
classDef process fill:#189AB4,stroke:#7C90A0,color:#fff
classDef pattern fill:#2E8B57,stroke:#7C90A0,color:#fff
class SEQ,HIER process
class ROUTE,PAR,LOOP pattern
```
Linear task execution in a predefined order via `AgentTeam`
Manager-coordinated task execution with dynamic assignment via `AgentTeam`
Complex flows with routing, parallel, and loops via `AgentFlow`
## Sequential Process
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
Input([▶]) --> A1[🤖 Agent 1] --> A2[🤖 Agent 2] --> A3[🤖 Agent 3] --> Output([✓])
classDef io fill:#8B0000,stroke:#7C90A0,color:#fff
classDef agent fill:#189AB4,stroke:#7C90A0,color:#fff
class Input,Output io
class A1,A2,A3 agent
```
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, AgentTeam
agents = AgentTeam(
agents=[agent1, agent2, agent3],
tasks=[task1, task2, task3],
process="sequential" # Default
)
result = agents.start()
```
## Hierarchical Process
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Input([▶]) --> Manager[👔 Manager Agent]
Manager -->|assign| W1[🤖 Worker 1]
Manager -->|assign| W2[🤖 Worker 2]
Manager -->|assign| W3[🤖 Worker 3]
W1 -->|report| Manager
W2 -->|report| Manager
W3 -->|report| Manager
Manager --> Output([✓])
classDef io fill:#8B0000,stroke:#7C90A0,color:#fff
classDef manager fill:#6366F1,stroke:#7C90A0,color:#fff
classDef worker fill:#189AB4,stroke:#7C90A0,color:#fff
class Input,Output io
class Manager manager
class W1,W2,W3 worker
```
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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()
```
### Error Recovery
Hierarchical processes include automatic error recovery mechanisms:
1. **Manager LLM parse failures retry up to 3 times** with exponential backoff (2s, 4s, 8s). After 3 failures, the workflow stops and logs the final exception.
2. **Invalid task IDs are re-prompted** — if the manager picks a `task_id` not in the workflow, the manager is asked again with the list of valid IDs (up to 3 attempts before aborting).
3. **Loop subtask file errors are now visible** — failures to read a loop input file mark the task `status="failed"` and attach a `TaskOutput` with the error message instead of failing silently.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TD
Start([Manager Request]) --> Parse{Parse Response}
Parse -->|Success| Execute[Execute Task]
Parse -->|Parse Error| Retry{Retry < 3?}
Retry -->|Yes| Wait[Exponential Backoff]
Wait --> Start
Retry -->|No| Stop[❌ Stop Workflow]
Execute --> ValidID{Valid Task ID?}
ValidID -->|Yes| Success[✅ Task Executed]
ValidID -->|No| Reprompt{Reprompt < 3?}
Reprompt -->|Yes| Manager[Re-ask Manager]
Manager --> Parse
Reprompt -->|No| Stop
classDef success fill:#10B981,stroke:#7C90A0,color:#fff
classDef error fill:#8B0000,stroke:#7C90A0,color:#fff
classDef retry fill:#F59E0B,stroke:#7C90A0,color:#fff
class Success success
class Stop error
class Retry,Reprompt,Wait retry
```
The retry constants (`MAX_MANAGER_RETRIES = 3`, `MAX_INVALID_SELECTIONS = 3`) are not user-configurable today; they are SDK defaults for reliability.
***
## Workflow Process (AgentFlow)
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
Input([▶]) --> Start[Step 1]
Start --> C{Condition}
C -->|route A| A[Handler A]
C -->|route B| B[Handler B]
A --> Join[Step 3]
B --> Join
Join --> Output([✓])
classDef io fill:#8B0000,stroke:#7C90A0,color:#fff
classDef step fill:#189AB4,stroke:#7C90A0,color:#fff
classDef condition fill:#6366F1,stroke:#7C90A0,color:#fff
class Input,Output io
class Start,A,B,Join step
class C condition
```
`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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
subgraph Sequential["Sequential"]
S1[Step] --> S2[Step] --> S3[Step]
end
subgraph Parallel["⚡ parallel()"]
P1[Step]
P2[Step]
P3[Step]
end
subgraph Route["🔀 route()"]
R{Condition}
R --> RA[Step A]
R --> RB[Step B]
end
subgraph Loop["🔁 loop()"]
L1[Step] --> L2{Check}
L2 -->|repeat| L1
L2 -->|done| L3[Done]
end
classDef step fill:#189AB4,stroke:#7C90A0,color:#fff
classDef condition fill:#6366F1,stroke:#7C90A0,color:#fff
classDef done fill:#10B981,stroke:#7C90A0,color:#fff
class S1,S2,S3,P1,P2,P3,RA,RB,L1 step
class R,L2 condition
class L3 done
```
## Getting Started
Install the core package:
```bash Terminal theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
pip install praisonaiagents
```
```bash Terminal theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
export OPENAI_API_KEY=your_openai_key
```
Generate your OpenAI API key from [OpenAI](https://platform.openai.com/api-keys)
Use other LLM providers like Ollama, Anthropic, Groq, Google, etc. Please refer to the [Models](/models) for more information.
Create `app.py`:
```python Single Agent Sequential Process theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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()
```
```python Multiple Agents Hierarchical Process theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, Task, AgentTeam
# Create multiple agents
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",
)
writer = Agent(
name="Writer",
role="Tech Content Strategist",
goal="Craft compelling content on tech advancements",
backstory="You are a content strategist",
llm="gpt-4o",
)
# Define multiple tasks
task1 = Task(
name="research_task",
description="Analyze 2024's AI advancements",
expected_output="A detailed report",
agent=researcher
)
task2 = Task(
name="writing_task",
description="Create a blog post about AI advancements",
expected_output="A blog post",
agent=writer
)
# Run with hierarchical process
agents = AgentTeam(
agents=[researcher, writer],
tasks=[task1, task2],
output="silent",
process="hierarchical",
manager_llm="gpt-4o"
)
result = agents.start()
```
## Advanced Features
Monitor and manage the progress of each task in real-time
Maintain important information across different stages of execution
Manage how data moves between tasks and agents efficiently
Automatically handle failures and continue execution
Switch to backup plans when primary execution fails
Detailed error logs and diagnostic information
Real-time visibility into task completion status
Measure execution time and resource efficiency
Monitor system resource utilization
Connect with other services and platforms
Maintain data consistency across systems
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
Async version of sequential process for non-blocking linear execution
Async workflow process for complex parallel task execution
Async hierarchical process for distributed task management
### Process-Specific Features
* Tasks execute in order but don't block
* Maintains sequence while allowing async operations
* Perfect for I/O-heavy tasks
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
async def main():
agents = AgentTeam(
process="sequential",
async_mode=True
)
await agents.astart()
```
* Parallel execution with `parallel()`
* Conditional routing with `route()`
* Loop iteration with `loop()`
* Repeat patterns with `repeat()`
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow
from praisonaiagents import parallel
workflow = AgentFlow(
steps=[
parallel([step1, step2, step3]),
aggregator_step
]
)
result = workflow.start("input")
```
* Async manager delegation
* Parallel subtask execution
* Dynamic worker assignment
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
async def main():
agents = AgentTeam(
process="hierarchical",
async_mode=True,
dynamic_assignment=True
)
await agents.astart()
```
### Key Benefits
* Efficient resource utilization
* Reduced waiting time
* Better throughput
* Mix sync and async tasks
* Adaptable execution patterns
* Easy scaling