Skip to main content

​
Module praisonaiagents.agents.agents

​
Functions

​
encode_file_to_base64(file_path: str) → str

Base64-encode a file.

​
process_video(video_path: str, seconds_per_frame=2)

Split video into frames (base64-encoded).

​
Classes

​
LoopItems

A Pydantic model for handling loop items.

​
Class Variables

  • items: List[Any] - List of items to loop through
  • model_config - Model configuration

​
PraisonAIAgents

The main class for managing agents and tasks execution.

​
Parameters

  • agents: List[Agent] - List of agents
  • tasks: List[Task] - List of tasks
  • verbose: int = 0 - Verbosity level
  • completion_checker: Optional[Callable] = None - Custom completion checker
  • max_retries: int = 5 - Maximum retry attempts
  • process: str = "sequential" - Process type (sequential, workflow, hierarchical)
  • manager_llm: Optional[str] = None - Language model for manager agent

​
Methods

  • add_task(self, task) - Add a task to the crew
  • clean_json_output(self, output: str) → str - Clean JSON output
  • clear_state(self) → None - Clear all state values
  • default_completion_checker(self, task, agent_output) - Default completion checker
  • execute_task(self, task_id) - Execute a specific task
  • get_agent_details(self, agent_name) - Get agent details
  • get_all_tasks_status(self) - Get status of all tasks
  • get_state(self, key: str, default: Any = None) → Any - Get a state value
  • get_task_details(self, task_id) - Get task details
  • get_task_result(self, task_id) - Get task result
  • get_task_status(self, task_id) - Get task status
  • run_all_tasks(self) - Execute tasks based on execution mode
  • run_task(self, task_id) - Run a specific task
  • save_output_to_file(self, task, task_output) - Save task output to file
  • set_state(self, key: str, value: Any) → None - Set a state value
  • start(self) - Start the crew execution
  • update_state(self, updates: Dict) → None - Update multiple state values

​
Async Methods

  • astart(self) - Async version of start method
  • aexecute_task(self, task_id) - Async version of execute_task
  • arun_task(self, task_id) - Async version of run_task
  • arun_all_tasks(self) - Async version of run_all_tasks

​
Async Support

The PraisonAIAgents class provides comprehensive async support through the following methods: 
class PraisonAIAgents:
    async def astart(self):
        """Async version of start method"""
        await self.arun_all_tasks()
        return {
            "task_status": self.get_all_tasks_status(),
            "task_results": self.get_task_results()
        }
        
    async def aexecute_task(self, task_id):
        """Async version of execute_task method"""
        # Async task execution implementation
        
    async def arun_task(self, task_id):
        """Async version of run_task method"""
        # Async task running with retries
        
    async def arun_all_tasks(self):
        """Async version of run_all_tasks method"""
        # Async execution of all tasks
Example usage: 
async def main():
    # Initialize agents and tasks
    agents = PraisonAIAgents(
        agents=[async_agent1, async_agent2],
        tasks=[async_task1, async_task2],
        process="workflow"
    )
    
    # Start async execution
    result = await agents.astart()
    print(result)

# Run the async workflow
asyncio.run(main())

​
Async Features

  • Full async/await support
  • Parallel task execution
  • Non-blocking operations
  • Efficient resource utilization
  • Mixed sync/async task handling
  • Async process management