​

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