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.
Process Types Overview
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
The simplest form of task execution where tasks are performed one after another.
Card### 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.
Card### 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()### Error Recovery
Hierarchical processes include automatic error recovery mechanisms:
- 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.
- Invalid task IDs are re-prompted — if the manager picks a
task_idnot in the workflow, the manager is asked again with the list of valid IDs (up to 3 attempts before aborting). - Loop subtask file errors are now visible — failures to read a loop input file mark the task
status="failed"and attach aTaskOutputwith the error message instead of failing silently.
Workflow Process (AgentFlow)
### 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
- Install PraisonAI
Install the core package:
Terminal pip install praisonaiagents - Configure Environment
Terminal export OPENAI_API_KEY=your_openai_keyGenerate your OpenAI API key from OpenAI Use other LLM providers like Ollama, Anthropic, Groq, Google, etc. Please refer to the Models for more information.
- 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()Multiple Agents Hierarchical Process 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
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
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
Progress Tracking
Real-time visibility into task completion status
Performance Metrics
Measure execution time and resource efficiency
Resource Usage
Monitor system resource utilization
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
Core Async Methods
Async version of sequential process for non-blocking linear execution
aworkflowAsync workflow process for complex parallel task execution
ahierarchicalAsync hierarchical process for distributed task management
Process-Specific Features
Sequential Process
Card- 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
Card- 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
Card- 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()