Merge pull request #292 from MervinPraison/develop

Update PraisonAI and praisonaiagents to version 2.0.45 and 0.0.35 res…

Dockerfile

@@ -1,6 +1,6 @@
 FROM python:3.11-slim
 WORKDIR /app
 COPY . .
-RUN pip install flask praisonai==2.0.41 gunicorn markdown
+RUN pip install flask praisonai==2.0.45 gunicorn markdown
 EXPOSE 8080
 CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]

agents/autonomous-agent.py

@@ -0,0 +1,59 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import time
+
+def get_environment_feedback():
+    """Simulates environment feedback based on time"""
+    current_time = int(time.time())
+    feedback = "positive" if current_time % 2 == 0 else "negative"
+    print(f"Environment feedback: {feedback}")
+    return feedback
+
+# Create autonomous agent
+autonomous_agent = Agent(
+    name="Autonomous Agent",
+    role="Environment interactor",
+    goal="Interact with environment and learn from feedback",
+    instructions="Take actions based on environment state and adapt from feedback",
+    tools=[get_environment_feedback]
+)
+
+# Create tasks for the autonomous loop
+action_task = Task(
+    name="take_action",
+    description="Analyze environment and take appropriate action",
+    expected_output="Action taken and its rationale",
+    agent=autonomous_agent,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["process_feedback"],
+    condition={
+        "continue": ["process_feedback"],  # Continue to get feedback
+        "stop": [""]  # Stop when goal is achieved
+    }
+)
+
+feedback_task = Task(
+    name="process_feedback",
+    description="Process environment feedback and adapt strategy",
+    expected_output="Adapted strategy based on feedback",
+    agent=autonomous_agent,
+    next_tasks=["take_action"],
+    context=[action_task]  # Access to previous action for learning
+)
+
+# Create workflow manager
+workflow = PraisonAIAgents(
+    agents=[autonomous_agent],
+    tasks=[action_task, feedback_task],
+    process="workflow",
+    verbose=True
+)
+
+# Run autonomous workflow
+results = workflow.start()
+
+# Print results
+print("\nAutonomous Agent Results:")
+for task_id, result in results["task_results"].items():
+    if result:
+        print(f"Task {task_id}: {result.raw}")

agents/evaluator-optimiser.py

@@ -0,0 +1,70 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Create generator and evaluator agents
+generator = Agent(
+    name="Generator",
+    role="Solution generator",
+    goal="Generate initial solutions and incorporate feedback",
+    instructions=(
+        "1. Look at the context from previous tasks.\n"
+        "2. If you see that you have already produced 2 points, then add another 2 new points "
+        "   so that the total becomes 10.\n"
+        "3. Otherwise, just produce the first 2 points.\n"
+        "4. Return only the final list of points, with no extra explanation."
+    )
+)
+
+evaluator = Agent(
+    name="Evaluator",
+    role="Solution evaluator",
+    goal="Evaluate solutions and provide improvement feedback",
+    instructions=(
+        "1. Count how many lines in the response start with a number and a period (like '1. ' or '2. ').\n"
+        "2. If there are 10 or more, respond with 'done'.\n"
+        "3. Otherwise, respond with 'more'.\n"
+        "4. Return only the single word: 'done' or 'more'."
+    )
+)
+
+
+# Create tasks for the feedback loop
+generate_task = Task(
+    name="generate",
+    description="Write 2 points about AI incuding if anything exiting from previous points",
+    expected_output="2 points",
+    agent=generator,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["evaluate"]
+)
+
+evaluate_task = Task(
+    name="evaluate",
+    description="Check if there are 10 points about AI",
+    expected_output="more or done",
+    agent=evaluator,
+    next_tasks=["generate"],
+    context=[generate_task],
+    task_type="decision",
+    condition={
+        "more": ["generate"],  # Continue to generate
+        "done": [""]  # Exit when optimization complete
+    }
+)
+
+# Create workflow manager
+workflow = PraisonAIAgents(
+    agents=[generator, evaluator],
+    tasks=[generate_task, evaluate_task],
+    process="workflow",
+    verbose=True
+)
+
+# Run optimization workflow
+results = workflow.start()
+
+# Print results
+print("\nEvaluator-Optimizer Results:")
+for task_id, result in results["task_results"].items():
+    if result:
+        print(f"Task {task_id}: {result.raw}")

agents/orchestrator-workers.py

@@ -0,0 +1,110 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import time
+
+def get_time_check():
+    current_time = int(time.time())
+    if current_time % 3 == 0:
+        result = 1
+    elif current_time % 3 == 1:
+        result = 2
+    else:
+        result = 3
+    print(f"Time: {current_time}, Result: {result}")
+    return result
+
+# Create orchestrator and worker agents
+router = Agent(
+    name="Router",
+    role="Task router",
+    goal="Distribute tasks to based on response from get_time_check",
+    tools=[get_time_check]
+)
+
+worker1 = Agent(
+    name="Worker 1",
+    role="Specialized worker",
+    goal="Handle specific subtask type 1",
+)
+
+worker2 = Agent(
+    name="Worker 2",
+    role="Specialized worker",
+    goal="Handle specific subtask type 2",
+)
+
+worker3 = Agent(
+    name="Worker 3",
+    role="Specialized worker",
+    goal="Handle specific subtask type 3",
+)
+
+synthesizer = Agent(
+    name="Synthesizer",
+    role="Result synthesizer",
+    goal="Combine and synthesize worker outputs",
+)
+
+# Create orchestrated workflow tasks
+router_task = Task(
+    name="route_task",
+    description="Analyze input from get_time_check and route to appropriate workers",
+    expected_output="Task routing decision, 1 , 2 or 3",
+    agent=router,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["worker1_task", "worker2_task", "worker3_task"],
+    condition={
+        "1": ["worker1_task"],
+        "2": ["worker2_task"],
+        "3": ["worker3_task"]
+    }
+)
+
+worker1_task = Task(
+    name="worker1_task",
+    description="Process type 1 operation",
+    expected_output="Worker 1 result",
+    agent=worker1,
+    next_tasks=["synthesize_task"]
+)
+
+worker2_task = Task(
+    name="worker2_task",
+    description="Process type 2 operation",
+    expected_output="Worker 2 result",
+    agent=worker2,
+    next_tasks=["synthesize_task"]
+)
+
+worker3_task = Task(
+    name="worker3_task",
+    description="Process type 3 operation",
+    expected_output="Worker 3 result",
+    agent=worker3,
+    next_tasks=["synthesize_task"]
+)
+
+synthesize_task = Task(
+    name="synthesize_task",
+    description="Synthesize worker results into final output",
+    expected_output="Final synthesized result",
+    agent=synthesizer,
+    context=[worker1_task, worker2_task, worker3_task]
+)
+
+# Create workflow manager
+workflow = PraisonAIAgents(
+    agents=[router, worker1, worker2, worker3, synthesizer],
+    tasks=[router_task, worker1_task, worker2_task, worker3_task, synthesize_task],
+    process="workflow",
+    verbose=True
+)
+
+# Run orchestrated workflow
+results = workflow.start()
+
+# Print results
+print("\nOrchestrator-Workers Results:")
+for task_id, result in results["task_results"].items():
+    if result:
+        print(f"Task {task_id}: {result.raw}")

agents/parallelisation.py

@@ -0,0 +1,95 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+from datetime import datetime
+import asyncio
+
+def process_time():
+    """Simulate processing"""
+    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    print(f"Processing at: {current_time}")
+    return f"Processed at {current_time}"
+
+# Create parallel processing agents
+agent1 = Agent(
+    name="Processor 1",
+    role="Time collector",
+    goal="Get the time and return it",
+    tools=[process_time]
+)
+
+agent2 = Agent(
+    name="Processor 2",
+    role="Time collector",
+    goal="Get the time and return it",
+    tools=[process_time]
+)
+
+agent3 = Agent(
+    name="Processor 3",
+    role="Time collector",
+    goal="Get the time and return it",
+    tools=[process_time]
+)
+
+aggregator = Agent(
+    name="Aggregator",
+    role="Result aggregator",
+    goal="Collect all the processed time from all tasks"
+)
+
+# Create parallel tasks with memory disabled
+task1 = Task(
+    name="process_1",
+    description="Use process_time tool to get the time",
+    expected_output="processed time",
+    agent=agent1,
+    is_start=True,
+    async_execution=True
+)
+
+task2 = Task(
+    name="process_2",
+    description="Use process_time tool to get the time",
+    expected_output="processed time",
+    agent=agent2,
+    is_start=True,
+    async_execution=True
+)
+
+task3 = Task(
+    name="process_3",
+    description="Use process_time tool to get the time",
+    expected_output="processed time",
+    agent=agent3,
+    is_start=True,
+    async_execution=True
+)
+
+aggregate_task = Task(
+    name="aggregate",
+    description="Collect all the processed time from all tasks",
+    expected_output="Output all the processed time from all tasks and just the time",
+    agent=aggregator,
+    context=[task1, task2, task3]
+)
+
+async def main():
+
+    # Create workflow manager
+    workflow = PraisonAIAgents(
+        agents=[agent1, agent2, agent3, aggregator],
+        tasks=[task1, task2, task3, aggregate_task],
+        process="workflow"
+    )
+
+    # Run parallel workflow
+    results = await workflow.astart()
+
+    # Print results
+    print("\nParallel Processing Results:")
+    for task_id, result in results["task_results"].items():
+        if result:
+            print(f"Task {task_id}: {result.raw}")
+
+# Run the async main function
+if __name__ == "__main__":
+    asyncio.run(main())

agents/praisonaiagents/agent/agent.py

@@ -16,7 +16,8 @@ from ..main import (
     display_self_reflection,
     ReflectionOutput,
     client,
-    error_logs
+    error_logs,
+    adisplay_instruction
 )
 import inspect
 

agents/praisonaiagents/agents/agents.py

@@ -211,6 +211,11 @@ class PraisonAIAgents:
 
         executor_agent = task.agent
 
+        # Ensure tools are available from both task and agent
+        tools = task.tools or []
+        if executor_agent and executor_agent.tools:
+            tools.extend(executor_agent.tools)
+
         task_prompt = f"""
 You need to do the following task: {task.description}.
 Expected Output: {task.expected_output}.
@@ -267,14 +272,14 @@ Here are the results of previous tasks that might be useful:\n
 
             agent_output = await executor_agent.achat(
                 _get_multimodal_message(task_prompt, task.images),
-                tools=task.tools,
+                tools=tools,
                 output_json=task.output_json,
                 output_pydantic=task.output_pydantic
             )
         else:
             agent_output = await executor_agent.achat(
                 task_prompt,
-                tools=task.tools,
+                tools=tools,
                 output_json=task.output_json,
                 output_pydantic=task.output_pydantic
             )
@@ -380,11 +385,25 @@ Here are the results of previous tasks that might be useful:\n
         )
         
         if self.process == "workflow":
+            # Collect all tasks that should run in parallel
+            parallel_tasks = []
             async for task_id in process.aworkflow():
-                if self.tasks[task_id].async_execution:
-                    await self.arun_task(task_id)
-                else:
-                    self.run_task(task_id)
+                if self.tasks[task_id].async_execution and self.tasks[task_id].is_start:
+                    parallel_tasks.append(task_id)
+                elif parallel_tasks:
+                    # Execute collected parallel tasks
+                    await asyncio.gather(*[self.arun_task(t) for t in parallel_tasks])
+                    parallel_tasks = []
+                    # Run the current non-parallel task
+                    if self.tasks[task_id].async_execution:
+                        await self.arun_task(task_id)
+                    else:
+                        self.run_task(task_id)
+            
+            # Execute any remaining parallel tasks
+            if parallel_tasks:
+                await asyncio.gather(*[self.arun_task(t) for t in parallel_tasks])
+                
         elif self.process == "sequential":
             async for task_id in process.asequential():
                 if self.tasks[task_id].async_execution:
@@ -596,8 +615,12 @@ Here are the results of previous tasks that might be useful:\n
                     task.status = "completed"
                     # Run execute_callback for memory operations
                     try:
-                        loop = asyncio.get_event_loop()
-                        loop.run_until_complete(task.execute_callback(task_output))
+                        if asyncio.get_event_loop().is_running():
+                            asyncio.create_task(task.execute_callback(task_output))
+                        else:
+                            loop = asyncio.new_event_loop()
+                            asyncio.set_event_loop(loop)
+                            loop.run_until_complete(task.execute_callback(task_output))
                     except Exception as e:
                         logger.error(f"Error executing memory callback for task {task_id}: {e}")
                         logger.exception(e)
@@ -606,7 +629,12 @@ Here are the results of previous tasks that might be useful:\n
                     if task.callback:
                         try:
                             if asyncio.iscoroutinefunction(task.callback):
-                                loop.run_until_complete(task.callback(task_output))
+                                if asyncio.get_event_loop().is_running():
+                                    asyncio.create_task(task.callback(task_output))
+                                else:
+                                    loop = asyncio.new_event_loop()
+                                    asyncio.set_event_loop(loop)
+                                    loop.run_until_complete(task.callback(task_output))
                             else:
                                 task.callback(task_output)
                         except Exception as e:

agents/praisonaiagents/process/process.py

@@ -10,14 +10,16 @@ class LoopItems(BaseModel):
     items: List[Any]
 
 class Process:
-    def __init__(self, tasks: Dict[str, Task], agents: List[Agent], manager_llm: Optional[str] = None, verbose: bool = False):
+    def __init__(self, tasks: Dict[str, Task], agents: List[Agent], manager_llm: Optional[str] = None, verbose: bool = False, max_iter: int = 10):
         self.tasks = tasks
         self.agents = agents
         self.manager_llm = manager_llm
         self.verbose = verbose
+        self.max_iter = max_iter
 
     async def aworkflow(self) -> AsyncGenerator[str, None]:
         """Async version of workflow method"""
+        current_iter = 0  # Track how many times we've looped
         # Build workflow relationships first
         for task in self.tasks.values():
             if task.next_tasks:
@@ -41,7 +43,12 @@ class Process:
         visited_tasks = set()
         loop_data = {}  # Store loop-specific data
         
-        while current_task and current_task.id not in visited_tasks:
+        while current_task:
+            current_iter += 1
+            if current_iter > self.max_iter:
+                logging.info(f"Max iteration limit {self.max_iter} reached, ending workflow.")
+                break
+
             task_id = current_task.id
             logging.info(f"Executing workflow task: {current_task.name if current_task.name else task_id}")
             
@@ -111,6 +118,11 @@ Return a JSON object with an 'items' array containing the items to process.
             yield task_id
             visited_tasks.add(task_id)
             
+            # Reset completed task to "not started" so it can run again
+            if self.tasks[task_id].status == "completed":
+                logging.debug(f"Task {task_id} was completed, resetting to 'not started' for next iteration.")
+                self.tasks[task_id].status = "not started"
+            
             # Handle loop progression
             if current_task.task_type == "loop":
                 loop_key = f"loop_{current_task.name}"
@@ -130,22 +142,28 @@ Return a JSON object with an 'items' array containing the items to process.
                     
             # Determine next task based on result
             next_task = None
-            if current_task.result:
+            if current_task and current_task.result:
                 if current_task.task_type in ["decision", "loop"]:
                     result = current_task.result.raw.lower()
                     # Check conditions
                     for condition, tasks in current_task.condition.items():
-                        if condition.lower() in result and tasks:
-                            next_task_name = tasks[0]
+                        if condition.lower() in result:
+                            # Handle both list and direct string values
+                            task_value = tasks[0] if isinstance(tasks, list) else tasks
+                            if not task_value or task_value == "exit":  # If empty or explicit exit
+                                logging.info("Workflow exit condition met, ending workflow")
+                                current_task = None
+                                break
+                            next_task_name = task_value
                             next_task = next((t for t in self.tasks.values() if t.name == next_task_name), None)
                             # For loops, allow revisiting the same task
                             if next_task and next_task.id == current_task.id:
                                 visited_tasks.discard(current_task.id)
                             break
             
-                if not next_task and current_task.next_tasks:
-                    next_task_name = current_task.next_tasks[0]
-                    next_task = next((t for t in self.tasks.values() if t.name == next_task_name), None)
+            if not next_task and current_task and current_task.next_tasks:
+                next_task_name = current_task.next_tasks[0]
+                next_task = next((t for t in self.tasks.values() if t.name == next_task_name), None)
             
             current_task = next_task
             if not current_task:
@@ -285,6 +303,7 @@ Provide a JSON with the structure:
 
     def workflow(self):
         """Synchronous version of workflow method"""
+        current_iter = 0  # Track how many times we've looped
         # Build workflow relationships first
         for task in self.tasks.values():
             if task.next_tasks:
@@ -308,7 +327,12 @@ Provide a JSON with the structure:
         visited_tasks = set()
         loop_data = {}  # Store loop-specific data
         
-        while current_task and current_task.id not in visited_tasks:
+        while current_task:
+            current_iter += 1
+            if current_iter > self.max_iter:
+                logging.info(f"Max iteration limit {self.max_iter} reached, ending workflow.")
+                break
+
             task_id = current_task.id
             logging.info(f"Executing workflow task: {current_task.name if current_task.name else task_id}")
             
@@ -372,6 +396,11 @@ Return a JSON object with an 'items' array containing the items to process.
             yield task_id
             visited_tasks.add(task_id)
             
+            # Reset completed task to "not started" so it can run again
+            if self.tasks[task_id].status == "completed":
+                logging.debug(f"Task {task_id} was completed, resetting to 'not started' for next iteration.")
+                self.tasks[task_id].status = "not started"
+            
             # Handle loop progression
             if current_task.task_type == "loop":
                 loop_key = f"loop_{current_task.name}"
@@ -391,22 +420,28 @@ Return a JSON object with an 'items' array containing the items to process.
                     
             # Determine next task based on result
             next_task = None
-            if current_task.result:
+            if current_task and current_task.result:
                 if current_task.task_type in ["decision", "loop"]:
                     result = current_task.result.raw.lower()
                     # Check conditions
                     for condition, tasks in current_task.condition.items():
-                        if condition.lower() in result and tasks:
-                            next_task_name = tasks[0]
+                        if condition.lower() in result:
+                            # Handle both list and direct string values
+                            task_value = tasks[0] if isinstance(tasks, list) else tasks
+                            if not task_value or task_value == "exit":  # If empty or explicit exit
+                                logging.info("Workflow exit condition met, ending workflow")
+                                current_task = None
+                                break
+                            next_task_name = task_value
                             next_task = next((t for t in self.tasks.values() if t.name == next_task_name), None)
                             # For loops, allow revisiting the same task
                             if next_task and next_task.id == current_task.id:
                                 visited_tasks.discard(current_task.id)
                             break
             
-                if not next_task and current_task.next_tasks:
-                    next_task_name = current_task.next_tasks[0]
-                    next_task = next((t for t in self.tasks.values() if t.name == next_task_name), None)
+            if not next_task and current_task and current_task.next_tasks:
+                next_task_name = current_task.next_tasks[0]
+                next_task = next((t for t in self.tasks.values() if t.name == next_task_name), None)
             
             current_task = next_task
             if not current_task:

agents/prompt_chaining copy.py

@@ -0,0 +1,91 @@
+from praisonaiagents.agent import Agent
+from praisonaiagents.task import Task
+from praisonaiagents.agents import PraisonAIAgents
+from typing import List, Dict
+import time
+
+def get_time_check():
+    current_time = int(time.time())
+    result = "even" if current_time % 2 == 0 else "odd"
+    print(f"Time check: {current_time} is {result}")
+    return result
+
+def create_prompt_chain():
+    # Create agents for each step in the chain
+    agent1 = Agent(
+        name="Time Checker",
+        role="Time checker",
+        goal="Check if the time is even or odd",
+        instructions="Check if the time is even or odd",
+        tools=[get_time_check]
+    )
+
+    agent2 = Agent(
+        name="Advanced Analyzer",
+        role="Advanced data analyzer",
+        goal="Perform in-depth analysis of processed data",
+        instructions="Analyze the processed data in detail"
+    )
+
+    agent3 = Agent(
+        name="Final Processor",
+        role="Final data processor",
+        goal="Generate final output based on analysis",
+        instructions="Create final output based on analyzed data"
+    )
+
+    # Create tasks for each step
+    initial_task = Task(
+        name="time_check",
+        description="Getting time check and checking if it is even or odd",
+        expected_output="Getting time check and checking if it is even or odd",
+        agent=agent1,
+        is_start=True,  # Mark as the starting task
+        task_type="decision",  # This task will make a decision
+        next_tasks=["advanced_analysis"],  # Next task if condition passes
+        condition={
+            "even": ["advanced_analysis"],  # If passes, go to advanced analysis
+            "odd": ["final_processing"]  # If fails, exit the chain
+        }
+    )
+
+    analysis_task = Task(
+        name="advanced_analysis",
+        description="Perform advanced analysis on the processed data",
+        expected_output="Analyzed data ready for final processing",
+        agent=agent2,
+        next_tasks=["final_processing"]
+    )
+
+    final_task = Task(
+        name="final_processing",
+        description="Generate final output",
+        expected_output="Final processed result",
+        agent=agent3
+    )
+
+    # Create the workflow manager
+    workflow = PraisonAIAgents(
+        agents=[agent1, agent2, agent3],
+        tasks=[initial_task, analysis_task, final_task],
+        process="workflow",  # Use workflow process type
+        verbose=True
+    )
+
+    return workflow
+
+def main():
+    # Create and run the prompt chain
+    workflow = create_prompt_chain()
+    
+    # Run the workflow
+    results = workflow.start()
+    
+    # Print results
+    print("\nWorkflow Results:")
+    for task_id, result in results["task_results"].items():
+        if result:
+            print(f"Task {task_id}: {result.raw}")
+
+if __name__ == "__main__":
+    main()

agents/prompt_chaining.py

@@ -0,0 +1,78 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import time
+
+def get_time_check():
+    current_time = int(time.time())
+    result = "even" if current_time % 2 == 0 else "odd"
+    print(f"Time check: {current_time} is {result}")
+    return result
+
+# Create agents for each step in the chain
+agent1 = Agent(
+    name="Time Checker",
+    role="Time checker",
+    goal="Check if the time is even or odd",
+    instructions="Check if the time is even or odd",
+    tools=[get_time_check]
+)
+
+agent2 = Agent(
+    name="Advanced Analyzer",
+    role="Advanced data analyzer",
+    goal="Perform in-depth analysis of processed data",
+    instructions="Analyze the processed data in detail"
+)
+
+agent3 = Agent(
+    name="Final Processor",
+    role="Final data processor",
+    goal="Generate final output based on analysis",
+    instructions="Create final output based on analyzed data"
+)
+
+# Create tasks for each step
+initial_task = Task(
+    name="time_check",
+    description="Getting time check and checking if it is even or odd",
+    expected_output="Getting time check and checking if it is even or odd",
+    agent=agent1,
+    is_start=True,  # Mark as the starting task
+    task_type="decision",  # This task will make a decision
+    next_tasks=["advanced_analysis"],  # Next task if condition passes
+    condition={
+        "even": ["advanced_analysis"],  # If passes, go to advanced analysis
+        "odd": "exit"  # If fails, exit the chain
+    }
+)
+
+analysis_task = Task(
+    name="advanced_analysis",
+    description="Perform advanced analysis on the processed data",
+    expected_output="Analyzed data ready for final processing",
+    agent=agent2,
+    next_tasks=["final_processing"]
+)
+
+final_task = Task(
+    name="final_processing",
+    description="Generate final output",
+    expected_output="Final processed result",
+    agent=agent3
+)
+
+# Create the workflow manager
+workflow = PraisonAIAgents(
+    agents=[agent1, agent2, agent3],
+    tasks=[initial_task, analysis_task, final_task],
+    process="workflow",  # Use workflow process type
+    verbose=True
+)
+
+# Run the workflow
+results = workflow.start()
+
+# Print results
+print("\nWorkflow Results:")
+for task_id, result in results["task_results"].items():
+    if result:
+        print(f"Task {task_id}: {result.raw}")

agents/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "praisonaiagents"
-version = "0.0.31"
+version = "0.0.35"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 authors = [
     { name="Mervin Praison" }

agents/routing copy.py

@@ -0,0 +1,82 @@
+from praisonaiagents.agent import Agent
+from praisonaiagents.task import Task
+from praisonaiagents.agents import PraisonAIAgents
+import time
+
+current_time = int(time.time())
+result = "even" if current_time % 2 == 0 else "odd"
+print(f"Time check: {current_time} is {result}")
+
+# Create specialized agents
+router = Agent(
+    name="Router",
+    role="Input Router",
+    goal="Evaluate input and determine routing path",
+    instructions="Analyze input and decide whether to proceed or exit",
+    tools=[get_time_check]
+)
+
+processor1 = Agent(
+    name="Processor 1",
+    role="Secondary Processor",
+    goal="Process valid inputs that passed initial check",
+    instructions="Process data that passed the routing check"
+)
+
+processor2 = Agent(
+    name="Processor 2",
+    role="Final Processor",
+    goal="Perform final processing on validated data",
+    instructions="Generate final output for processed data"
+)
+
+# Create tasks with routing logic
+routing_task = Task(
+    name="initial_routing",
+    description="check the time and return according to what is returned",
+    expected_output="pass or fail based on what is returned",
+    agent=router,
+    is_start=True,
+    task_type="decision",
+    condition={
+        "pass": ["process_valid"],
+        "fail": "exit"
+    }
+)
+
+processing_task = Task(
+    name="process_valid",
+    description="Process validated input",
+    expected_output="Processed data ready for final step",
+    agent=processor1,
+    next_tasks=["final_process"]
+)
+
+final_task = Task(
+    name="final_process",
+    description="Generate final output",
+    expected_output="Final processed result",
+    agent=processor2
+)
+
+# Create and run workflow
+workflow = PraisonAIAgents(
+    agents=[router, processor1, processor2],
+    tasks=[routing_task, processing_task, final_task],
+    process="workflow",
+    verbose=True
+)
+
+print("\nStarting Routing Workflow...")
+print("=" * 50)
+
+results = workflow.start()
+
+print("\nWorkflow Results:")
+print("=" * 50)
+for task_id, result in results["task_results"].items():
+    if result:
+        task_name = result.description
+        print(f"\nTask: {task_name}")
+        print(f"Result: {result.raw}")
+        print("-" * 50)

agents/routing.py

@@ -0,0 +1,83 @@
+from praisonaiagents.agent import Agent
+from praisonaiagents.task import Task
+from praisonaiagents.agents import PraisonAIAgents
+import time
+
+def get_time_check():
+    current_time = int(time.time())
+    result = "even" if current_time % 2 == 0 else "odd"
+    print(f"Time check: {current_time} is {result}")
+    return result
+
+# Create specialized agents
+router = Agent(
+    name="Router",
+    role="Input Router",
+    goal="Evaluate input and determine routing path",
+    instructions="Analyze input and decide whether to proceed or exit",
+    tools=[get_time_check]
+)
+
+processor1 = Agent(
+    name="Processor 1",
+    role="Secondary Processor",
+    goal="Process valid inputs that passed initial check",
+    instructions="Process data that passed the routing check"
+)
+
+processor2 = Agent(
+    name="Processor 2",
+    role="Final Processor",
+    goal="Perform final processing on validated data",
+    instructions="Generate final output for processed data"
+)
+
+# Create tasks with routing logic
+routing_task = Task(
+    name="initial_routing",
+    description="check the time and return according to what is returned",
+    expected_output="pass or fail based on what is returned",
+    agent=router,
+    is_start=True,
+    task_type="decision",
+    condition={
+        "pass": ["process_valid"],
+        "fail": ["process_invalid"]
+    }
+)
+
+processing_task = Task(
+    name="process_valid",
+    description="Process validated input",
+    expected_output="Processed data ready for final step",
+    agent=processor1,
+)
+
+final_task = Task(
+    name="process_invalid",
+    description="Generate final output",
+    expected_output="Final processed result",
+    agent=processor2
+)
+
+# Create and run workflow
+workflow = PraisonAIAgents(
+    agents=[router, processor1, processor2],
+    tasks=[routing_task, processing_task, final_task],
+    process="workflow",
+    verbose=True
+)
+
+print("\nStarting Routing Workflow...")
+print("=" * 50)
+
+results = workflow.start()
+
+print("\nWorkflow Results:")
+print("=" * 50)
+for task_id, result in results["task_results"].items():
+    if result:
+        task_name = result.description
+        print(f"\nTask: {task_name}")
+        print(f"Result: {result.raw}")
+        print("-" * 50)

agents/uv.lock

@@ -990,7 +990,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.31"
+version = "0.0.35"
 source = { editable = "." }
 dependencies = [
     { name = "openai" },

cookbooks/general/async_example.py

@@ -0,0 +1,233 @@
+import asyncio
+import time
+from typing import List, Dict
+from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
+from praisonaiagents.main import (
+    display_error,
+    display_interaction,
+    display_tool_call,
+    display_instruction,
+    error_logs,
+    Console
+)
+from duckduckgo_search import DDGS
+from pydantic import BaseModel
+
+console = Console()
+
+# 1. Define output model for structured results
+class SearchResult(BaseModel):
+    query: str
+    results: List[Dict[str, str]]
+    total_results: int
+
+# 2. Define both sync and async tools
+def sync_search_tool(query: str) -> List[Dict]:
+    """
+    Synchronous search using DuckDuckGo.
+    Args:
+        query (str): The search query.
+    Returns:
+        list: Search results
+    """
+    display_tool_call(f"Running sync search for: {query}", console)
+    time.sleep(1)  # Simulate network delay
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=5):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        return results
+    except Exception as e:
+        error_msg = f"Error during sync search: {e}"
+        display_error(error_msg, console)
+        error_logs.append(error_msg)
+        return []
+
+async def async_search_tool(query: str) -> List[Dict]:
+    """
+    Asynchronous search using DuckDuckGo.
+    Args:
+        query (str): The search query.
+    Returns:
+        list: Search results
+    """
+    display_tool_call(f"Running async search for: {query}", console)
+    await asyncio.sleep(1)  # Simulate network delay
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=5):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        return results
+    except Exception as e:
+        error_msg = f"Error during async search: {e}"
+        display_error(error_msg, console)
+        error_logs.append(error_msg)
+        return []
+
+# 3. Define both sync and async callbacks
+def sync_callback(output: TaskOutput):
+    display_interaction("Sync Callback", f"Processing output: {output.raw[:100]}...", markdown=True, console=console)
+    time.sleep(1)  # Simulate processing
+    if output.output_format == "JSON":
+        display_tool_call(f"Processed JSON result: {output.json_dict}", console)
+    elif output.output_format == "Pydantic":
+        display_tool_call(f"Processed Pydantic result: {output.pydantic}", console)
+
+async def async_callback(output: TaskOutput):
+    display_interaction("Async Callback", f"Processing output: {output.raw[:100]}...", markdown=True, console=console)
+    await asyncio.sleep(1)  # Simulate processing
+    if output.output_format == "JSON":
+        display_tool_call(f"Processed JSON result: {output.json_dict}", console)
+    elif output.output_format == "Pydantic":
+        display_tool_call(f"Processed Pydantic result: {output.pydantic}", console)
+
+# 4. Create agents with different tools
+sync_agent = Agent(
+    name="SyncAgent",
+    role="Synchronous Search Specialist",
+    goal="Perform synchronous searches and return structured results",
+    backstory="Expert in sync operations and data organization",
+    tools=[sync_search_tool],
+    self_reflect=False,
+    verbose=True,
+    markdown=True
+)
+
+async_agent = Agent(
+    name="AsyncAgent",
+    role="Asynchronous Search Specialist",
+    goal="Perform asynchronous searches and return structured results",
+    backstory="Expert in async operations and data organization",
+    tools=[async_search_tool],
+    self_reflect=False,
+    verbose=True,
+    markdown=True
+)
+
+# 5. Create tasks with different configurations
+sync_task = Task(
+    name="sync_search",
+    description="Search for 'Python programming' using sync tool and return structured results",
+    expected_output="SearchResult model with query details and results",
+    agent=sync_agent,
+    async_execution=False,
+    callback=sync_callback,
+    output_pydantic=SearchResult
+)
+
+async_task = Task(
+    name="async_search",
+    description="Search for 'Async programming' using async tool and return structured results",
+    expected_output="SearchResult model with query details and results",
+    agent=async_agent,
+    async_execution=True,
+    callback=async_callback,
+    output_pydantic=SearchResult
+)
+
+# 6. Create workflow tasks
+workflow_sync_task = Task(
+    name="workflow_sync",
+    description="Workflow sync search for 'AI trends' with structured output",
+    expected_output="SearchResult model with AI trends data",
+    agent=sync_agent,
+    async_execution=False,
+    is_start=True,
+    next_tasks=["workflow_async"],
+    output_pydantic=SearchResult
+)
+
+workflow_async_task = Task(
+    name="workflow_async",
+    description="Workflow async search for 'Future of AI' with structured output",
+    expected_output="SearchResult model with Future of AI data",
+    agent=async_agent,
+    async_execution=True,
+    output_pydantic=SearchResult
+)
+
+# 7. Example usage functions
+def run_sync_example():
+    """Run synchronous example"""
+    display_instruction("\nRunning Synchronous Example...", console)
+    agents = PraisonAIAgents(
+        agents=[sync_agent],
+        tasks=[sync_task],
+        verbose=1,
+        process="sequential"
+    )
+    result = agents.start()
+    display_interaction("Sync Example", f"Result: {result}", markdown=True, console=console)
+
+async def run_async_example():
+    """Run asynchronous example"""
+    display_instruction("\nRunning Asynchronous Example...", console)
+    agents = PraisonAIAgents(
+        agents=[async_agent],
+        tasks=[async_task],
+        verbose=1,
+        process="sequential"
+    )
+    result = await agents.astart()
+    display_interaction("Async Example", f"Result: {result}", markdown=True, console=console)
+
+async def run_mixed_example():
+    """Run mixed sync/async example"""
+    display_instruction("\nRunning Mixed Sync/Async Example...", console)
+    agents = PraisonAIAgents(
+        agents=[sync_agent, async_agent],
+        tasks=[sync_task, async_task],
+        verbose=1,
+        process="sequential"
+    )
+    result = await agents.astart()
+    display_interaction("Mixed Example", f"Result: {result}", markdown=True, console=console)
+
+async def run_workflow_example():
+    """Run workflow example with both sync and async tasks"""
+    display_instruction("\nRunning Workflow Example...", console)
+    agents = PraisonAIAgents(
+        agents=[sync_agent, async_agent],
+        tasks=[workflow_sync_task, workflow_async_task],
+        verbose=1,
+        process="workflow"
+    )
+    result = await agents.astart()
+    display_interaction("Workflow Example", f"Result: {result}", markdown=True, console=console)
+
+# 8. Main execution
+async def main():
+    """Main execution function"""
+    display_instruction("Starting PraisonAI Agents Examples...", console)
+    
+    try:
+        # Run sync example in a separate thread to not block the event loop
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(None, run_sync_example)
+        
+        # Run async examples
+        await run_async_example()
+        await run_mixed_example()
+        await run_workflow_example()
+
+        if error_logs:
+            display_error("\nErrors encountered during execution:", console)
+            for error in error_logs:
+                display_error(error, console)
+    except Exception as e:
+        display_error(f"Error in main execution: {e}", console)
+        error_logs.append(str(e))
+
+if __name__ == "__main__":
+    # Run the main function
+    asyncio.run(main()) 

cookbooks/general/async_example_full.py

@@ -0,0 +1,152 @@
+import asyncio
+import time
+from typing import List, Dict
+from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
+from duckduckgo_search import DDGS
+from pydantic import BaseModel
+
+# 1. Define output model for structured results
+class SearchResult(BaseModel):
+    query: str
+    results: List[Dict[str, str]]
+    total_results: int
+
+# 2. Define async tool
+async def async_search_tool(query: str) -> Dict:
+    """Perform asynchronous search and return structured results."""
+    await asyncio.sleep(1)  # Simulate network delay
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=5):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        
+        return {
+            "query": query,
+            "results": results,
+            "total_results": len(results)
+        }
+    except Exception as e:
+        print(f"Error during async search: {e}")
+        return {
+            "query": query,
+            "results": [],
+            "total_results": 0
+        }
+
+# 3. Define async callback
+async def async_callback(output: TaskOutput):
+    await asyncio.sleep(1)  # Simulate processing
+    if output.output_format == "JSON":
+        print(f"Processed JSON result: {output.json_dict}")
+    elif output.output_format == "Pydantic":
+        print(f"Processed Pydantic result: {output.pydantic}")
+
+# 4. Create specialized agents
+async_agent = Agent(
+    name="AsyncSearchAgent",
+    role="Search Specialist",
+    goal="Perform fast parallel searches with structured results",
+    backstory="Expert in efficient data retrieval and parallel search operations",
+    tools=[async_search_tool],
+    self_reflect=False,
+    verbose=True,
+    markdown=True
+)
+
+summary_agent = Agent(
+    name="SummaryAgent",
+    role="Research Synthesizer",
+    goal="Create concise summaries from multiple search results",
+    backstory="Expert in analyzing and synthesizing information from multiple sources",
+    self_reflect=True,
+    verbose=True,
+    markdown=True
+)
+
+# 5. Create async tasks
+async_task = Task(
+    name="async_search",
+    description="Search for 'Async programming' and return results in JSON format with query, results array, and total_results count.",
+    expected_output="SearchResult model with structured data",
+    agent=async_agent,
+    async_execution=True,
+    callback=async_callback,
+    output_json=SearchResult
+)
+
+async def run_parallel_tasks():
+    """Run multiple async tasks in parallel"""
+    print("\nRunning Parallel Async Tasks...")
+    
+    # Define different search topics
+    search_topics = [
+        "Latest AI Developments 2024",
+        "Machine Learning Best Practices",
+        "Neural Networks Architecture"
+    ]
+    
+    # Create tasks for different topics
+    parallel_tasks = [
+        Task(
+            name=f"search_task_{i}",
+            description=f"Search for '{topic}' and return structured results with query details and findings.",
+            expected_output="SearchResult model with search data",
+            agent=async_agent,
+            async_execution=True,
+            callback=async_callback,
+            output_json=SearchResult
+        ) for i, topic in enumerate(search_topics)
+    ]
+    
+    # Create summarization task
+    summary_task = Task(
+        name="summary_task",
+        description="Analyze all search results and create a concise summary highlighting key findings, patterns, and implications.",
+        expected_output="Structured summary with key findings and insights",
+        agent=summary_agent,
+        async_execution=False,
+        callback=async_callback,
+        context=parallel_tasks
+    )
+    
+    # Create a single PraisonAIAgents instance with both agents
+    agents = PraisonAIAgents(
+        agents=[async_agent, summary_agent],
+        tasks=parallel_tasks + [summary_task],
+        verbose=1,
+        process="sequential"
+    )
+    
+    # Run all tasks
+    results = await agents.astart()
+    print(f"Tasks Results: {results}")
+    
+    # Return results in a serializable format
+    return {
+        "search_results": {
+            "task_status": {k: v for k, v in results["task_status"].items() if k != summary_task.id},
+            "task_results": [str(results["task_results"][i]) if results["task_results"][i] else None 
+                           for i in range(len(parallel_tasks))]
+        },
+        "summary": str(results["task_results"][summary_task.id]) if results["task_results"].get(summary_task.id) else None,
+        "topics": search_topics
+    }
+
+# 6. Main execution
+async def main():
+    """Main execution function"""
+    print("Starting Async AI Agents Examples...")
+    
+    try:
+        await run_parallel_tasks()
+    except Exception as e:
+        print(f"Error in main execution: {e}")
+
+if __name__ == "__main__":
+    # Run the main function
+    asyncio.run(main())

cookbooks/general/async_example_full_multigroups.py

@@ -0,0 +1,248 @@
+import asyncio
+import time
+from typing import List, Dict
+from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
+from duckduckgo_search import DDGS
+from pydantic import BaseModel
+
+# 1. Define output model for structured results
+class SearchResult(BaseModel):
+    query: str
+    results: List[Dict[str, str]]
+    total_results: int
+
+# 2. Define async tool
+async def async_search_tool(query: str) -> Dict:
+    """
+    Asynchronous search using DuckDuckGo.
+    Args:
+        query (str): The search query.
+    Returns:
+        dict: Search results in SearchResult model format
+    """
+    await asyncio.sleep(1)  # Simulate network delay
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=5):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        
+        # Format response to match SearchResult model
+        return {
+            "query": query,
+            "results": results,
+            "total_results": len(results)
+        }
+    except Exception as e:
+        print(f"Error during async search: {e}")
+        return {
+            "query": query,
+            "results": [],
+            "total_results": 0
+        }
+
+# 3. Define async callback
+async def async_callback(output: TaskOutput):
+    await asyncio.sleep(1)  # Simulate processing
+    if output.output_format == "JSON":
+        print(f"Processed JSON result: {output.json_dict}")
+    elif output.output_format == "Pydantic":
+        print(f"Processed Pydantic result: {output.pydantic}")
+
+# 4. Create specialized agents
+async_agent = Agent(
+    name="AsyncSearchAgent",
+    role="Asynchronous Search Specialist",
+    goal="Perform fast and efficient asynchronous searches with structured results",
+    backstory="Expert in parallel search operations and data retrieval",
+    tools=[async_search_tool],
+    self_reflect=False,
+    verbose=True,
+    markdown=True
+)
+
+summary_agent = Agent(
+    name="SummaryAgent",
+    role="Research Synthesizer",
+    goal="Create comprehensive summaries and identify patterns across multiple search results",
+    backstory="""Expert in analyzing and synthesizing information from multiple sources.
+Skilled at identifying patterns, trends, and connections between different topics.
+Specializes in creating clear, structured summaries that highlight key insights.""",
+    self_reflect=True,  # Enable self-reflection for better summary quality
+    verbose=True,
+    markdown=True
+)
+
+# 5. Create async tasks
+async_task = Task(
+    name="async_search",
+    description="""Search for 'Async programming' and return results in the following JSON format:
+{
+    "query": "the search query",
+    "results": [
+        {
+            "title": "result title",
+            "url": "result url",
+            "snippet": "result snippet"
+        }
+    ],
+    "total_results": number of results
+}""",
+    expected_output="SearchResult model with query details and results",
+    agent=async_agent,
+    async_execution=True,
+    callback=async_callback,
+    output_json=SearchResult
+)
+
+# 6. Example usage functions
+async def run_single_task():
+    """Run single async task"""
+    print("\nRunning Single Async Task...")
+    agents = PraisonAIAgents(
+        agents=[async_agent],
+        tasks=[async_task],
+        verbose=1,
+        process="sequential"
+    )
+    result = await agents.astart()
+    print(f"Single Task Result: {result}")
+
+async def run_parallel_tasks():
+    """Run multiple async tasks in parallel"""
+    print("\nRunning Parallel Async Tasks...")
+    
+    # Define different search topics
+    search_topics = [
+        "Latest AI Developments 2024",
+        "Machine Learning Best Practices",
+        "Neural Networks Architecture"
+    ]
+    
+    # Create tasks for different topics
+    parallel_tasks = [
+        Task(
+            name=f"search_task_{i}",
+            description=f"""Search for '{topic}' and return results in the following JSON format:
+{{
+    "query": "{topic}",
+    "results": [
+        {{
+            "title": "result title",
+            "url": "result url",
+            "snippet": "result snippet"
+        }}
+    ],
+    "total_results": number of results
+}}""",
+            expected_output="SearchResult model with detailed information",
+            agent=async_agent,
+            async_execution=True,
+            callback=async_callback,
+            output_json=SearchResult
+        ) for i, topic in enumerate(search_topics)
+    ]
+    
+    # Create summarization task with the specialized summary agent
+    summary_task = Task(
+        name="summary_task",
+        description="""As a Research Synthesizer, analyze the search results and create a comprehensive summary. Your task:
+
+1. Analyze Results:
+   - Review all search results thoroughly
+   - Extract key findings from each topic
+   - Identify main themes and concepts
+
+2. Find Connections:
+   - Identify relationships between topics
+   - Spot common patterns or contradictions
+   - Note emerging trends across sources
+
+3. Create Structured Summary:
+   - Main findings per topic
+   - Cross-cutting themes
+   - Emerging trends
+   - Practical implications
+   - Future directions
+
+4. Quality Checks:
+   - Ensure all topics are covered
+   - Verify accuracy of connections
+   - Confirm clarity of insights
+   - Validate practical relevance
+
+Present the summary in a clear, structured format with sections for findings, patterns, trends, and implications.""",
+        expected_output="""A comprehensive research synthesis containing:
+- Detailed findings from each search topic
+- Cross-topic patterns and relationships
+- Emerging trends and their implications
+- Practical applications and future directions""",
+        agent=summary_agent,  # Use the specialized summary agent
+        async_execution=False,  # Run synchronously after search tasks
+        callback=async_callback
+    )
+    
+    # First run parallel search tasks
+    agents = PraisonAIAgents(
+        agents=[async_agent],
+        tasks=parallel_tasks,  # Only run search tasks first
+        verbose=1,
+        process="sequential"
+    )
+    search_results = await agents.astart()
+    print(f"Search Tasks Results: {search_results}")
+    
+    # Create task objects with results for context
+    completed_tasks = []
+    for i, topic in enumerate(search_topics):
+        task = Task(
+            name=f"search_task_{i}_result",
+            description=f"Search results for: {topic}",
+            expected_output="Search results from previous task",
+            agent=async_agent,
+            result=search_results["task_results"][i]
+        )
+        completed_tasks.append(task)
+    
+    # Update summary task with context from search results
+    summary_task.context = completed_tasks
+    
+    # Run summarization task with summary agent
+    summary_agents = PraisonAIAgents(
+        agents=[summary_agent],  # Use summary agent for synthesis
+        tasks=[summary_task],
+        verbose=1,
+        process="sequential"
+    )
+    summary_result = await summary_agents.astart()
+    print(f"Summary Task Result: {summary_result}")
+    
+    # Return results in a serializable format
+    return {
+        "search_results": {
+            "task_status": search_results["task_status"],
+            "task_results": [str(result) if result else None for result in search_results["task_results"]]
+        },
+        "summary": str(summary_result),
+        "topics": search_topics
+    }
+
+# 7. Main execution
+async def main():
+    """Main execution function"""
+    print("Starting Async AI Agents Examples...")
+    
+    try:
+        # Run different async patterns
+        await run_single_task()
+        await run_parallel_tasks()
+    except Exception as e:
+        print(f"Error in main execution: {e}")
+
+if __name__ == "__main__":
+    # Run the main function
+    asyncio.run(main())

cookbooks/general/auto_agents_example.py

@@ -0,0 +1,11 @@
+from praisonaiagents import AutoAgents
+from praisonaiagents.tools import duckduckgo
+
+agents = AutoAgents(
+    instructions="Search for information about AI Agents",
+    tools=[duckduckgo],
+    process="sequential",
+    verbose=True
+)
+
+agents.start()

cookbooks/general/code_agents_example.py

@@ -0,0 +1,50 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import json
+from e2b_code_interpreter import Sandbox
+
+def code_interpreter(code: str):
+    """
+    A function to demonstrate running Python code dynamically using e2b_code_interpreter.
+    """
+    print(f"\n{'='*50}\n> Running following AI-generated code:\n{code}\n{'='*50}")
+    exec_result = Sandbox().run_code(code)
+    if exec_result.error:
+        print("[Code Interpreter error]", exec_result.error)
+        return {"error": str(exec_result.error)}
+    else:
+        results = []
+        for result in exec_result.results:
+            if hasattr(result, '__iter__'):
+                results.extend(list(result))
+            else:
+                results.append(str(result))
+        logs = {"stdout": list(exec_result.logs.stdout), "stderr": list(exec_result.logs.stderr)}
+        return json.dumps({"results": results, "logs": logs})
+
+code_agent = Agent(
+    name="code_agent",
+    llm="gpt-4o-mini",
+    backstory="Expert in writing Python scripts",
+    self_reflect=False
+)
+execution_agent = Agent(
+    name="execution_agent",
+    llm="gpt-4o-mini",
+    backstory="Expert in executing Python scripts",
+    self_reflect=False,
+    tools=[code_interpreter]
+)
+
+code_agent_task = Task(
+    description="Write a simple Python script to print 'Hello, World!'",
+    expected_output="A Python script that prints 'Hello, World!'",
+    agent=code_agent
+)
+execution_agent_task = Task(
+    description="Execute the Python script",
+    expected_output="The output of the Python script",
+    agent=execution_agent
+)
+
+agents = PraisonAIAgents(agents=[code_agent, execution_agent], tasks=[code_agent_task, execution_agent_task])
+agents.start()

cookbooks/general/evaluator-optimiser.py

@@ -0,0 +1,70 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Create generator and evaluator agents
+generator = Agent(
+    name="Generator",
+    role="Solution generator",
+    goal="Generate initial solutions and incorporate feedback",
+    instructions=(
+        "1. Look at the context from previous tasks.\n"
+        "2. If you see that you have already produced 2 points, then add another 2 new points "
+        "   so that the total becomes 10.\n"
+        "3. Otherwise, just produce the first 2 points.\n"
+        "4. Return only the final list of points, with no extra explanation."
+    )
+)
+
+evaluator = Agent(
+    name="Evaluator",
+    role="Solution evaluator",
+    goal="Evaluate solutions and provide improvement feedback",
+    instructions=(
+        "1. Count how many lines in the response start with a number and a period (like '1. ' or '2. ').\n"
+        "2. If there are 10 or more, respond with 'done'.\n"
+        "3. Otherwise, respond with 'more'.\n"
+        "4. Return only the single word: 'done' or 'more'."
+    )
+)
+
+
+# Create tasks for the feedback loop
+generate_task = Task(
+    name="generate",
+    description="Write 2 points about AI incuding if anything exiting from previous points",
+    expected_output="2 points",
+    agent=generator,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["evaluate"]
+)
+
+evaluate_task = Task(
+    name="evaluate",
+    description="Check if there are 10 points about AI",
+    expected_output="more or done",
+    agent=evaluator,
+    next_tasks=["generate"],
+    context=[generate_task],
+    task_type="decision",
+    condition={
+        "more": ["generate"],  # Continue to generate
+        "done": [""]  # Exit when optimization complete
+    }
+)
+
+# Create workflow manager
+workflow = PraisonAIAgents(
+    agents=[generator, evaluator],
+    tasks=[generate_task, evaluate_task],
+    process="workflow",
+    verbose=True
+)
+
+# Run optimization workflow
+results = workflow.start()
+
+# Print results
+print("\nEvaluator-Optimizer Results:")
+for task_id, result in results["task_results"].items():
+    if result:
+        print(f"Task {task_id}: {result.raw}")

cookbooks/general/example_callback.py

@@ -0,0 +1,204 @@
+from praisonaiagents import (
+    Agent, 
+    Task, 
+    PraisonAIAgents, 
+    error_logs, 
+    register_display_callback,
+    sync_display_callbacks,
+    async_display_callbacks
+)
+from duckduckgo_search import DDGS
+from rich.console import Console
+import json
+from datetime import datetime
+import logging
+
+# Setup logging
+logging.basicConfig(
+    filename='ai_interactions.log',
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+# Callback functions for different display types
+def interaction_callback(message=None, response=None, markdown=None, generation_time=None):
+    """Callback for display_interaction"""
+    logging.info(f"""
+    === INTERACTION ===
+    Time: {datetime.now()}
+    Generation Time: {generation_time}s
+    Message: {message}
+    Response: {response}
+    Markdown: {markdown}
+    """)
+
+def error_callback(message=None):
+    """Callback for display_error"""
+    logging.error(f"""
+    === ERROR ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+def tool_call_callback(message=None):
+    """Callback for display_tool_call"""
+    logging.info(f"""
+    === TOOL CALL ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+def instruction_callback(message=None):
+    """Callback for display_instruction"""
+    logging.info(f"""
+    === INSTRUCTION ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+def self_reflection_callback(message=None):
+    """Callback for display_self_reflection"""
+    logging.info(f"""
+    === SELF REFLECTION ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+def generating_callback(content=None, elapsed_time=None):
+    """Callback for display_generating"""
+    logging.info(f"""
+    === GENERATING ===
+    Time: {datetime.now()}
+    Content: {content}
+    Elapsed Time: {elapsed_time}
+    """)
+
+# Register all callbacks
+register_display_callback('interaction', interaction_callback)
+register_display_callback('error', error_callback)
+register_display_callback('tool_call', tool_call_callback)
+register_display_callback('instruction', instruction_callback)
+register_display_callback('self_reflection', self_reflection_callback)
+# register_display_callback('generating', generating_callback)
+
+def task_callback(output):
+    """Callback for task completion"""
+    logging.info(f"""
+    === TASK COMPLETED ===
+    Time: {datetime.now()}
+    Description: {output.description}
+    Agent: {output.agent}
+    Output: {output.raw[:200]}...
+    """)
+
+def internet_search_tool(query) -> list:
+    """
+    Perform a search using DuckDuckGo.
+
+    Args:
+        query (str): The search query.
+
+    Returns:
+        list: A list of search result titles and URLs.
+    """
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=10):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        return results
+
+    except Exception as e:
+        print(f"Error during DuckDuckGo search: {e}")
+        return []
+
+def main():
+    # Create agents
+    researcher = Agent(
+        name="Researcher",
+        role="Senior Research Analyst",
+        goal="Uncover cutting-edge developments in AI and data science",
+        backstory="""You are an expert at a technology research group, 
+        skilled in identifying trends and analyzing complex data.""",
+        verbose=True,
+        allow_delegation=False,
+        tools=[internet_search_tool],
+        llm="gpt-4o",
+        markdown=True,
+        reflect_llm="gpt-4o",
+        min_reflect=2,
+        max_reflect=4
+    )
+    
+    writer = Agent(
+        name="Writer",
+        role="Tech Content Strategist",
+        goal="Craft compelling content on tech advancements",
+        backstory="""You are a content strategist known for 
+        making complex tech topics interesting and easy to understand.""",
+        verbose=True,
+        allow_delegation=True,
+        llm="gpt-4o",
+        tools=[],
+        markdown=True
+    )
+
+    # Create tasks with callbacks
+    task1 = Task(
+        name="research_task",
+        description="""Analyze 2024's AI advancements. 
+        Find major trends, new technologies, and their effects.""",
+        expected_output="""A detailed report on 2024 AI advancements""",
+        agent=researcher,
+        tools=[internet_search_tool],
+        callback=task_callback
+    )
+
+    task2 = Task(
+        name="writing_task",
+        description="""Create a blog post about major AI advancements using the insights you have.
+        Make it interesting, clear, and suited for tech enthusiasts. 
+        It should be at least 4 paragraphs long.""",
+        expected_output="A blog post of at least 4 paragraphs",
+        agent=writer,
+        context=[task1],
+        callback=task_callback,
+        tools=[]
+    )
+
+    task3 = Task(
+        name="json_task",
+        description="""Create a json object with a title of "My Task" and content of "My content".""",
+        expected_output="""JSON output with title and content""",
+        agent=researcher,
+        callback=task_callback
+    )
+
+    task4 = Task(
+        name="save_output_task",
+        description="""Save the AI blog post to a file""",
+        expected_output="""File saved successfully""",
+        agent=writer,
+        context=[task2],
+        output_file='test.txt',
+        create_directory=True,
+        callback=task_callback
+    )
+
+    # Create and run agents manager
+    agents = PraisonAIAgents(
+        agents=[researcher, writer],
+        tasks=[task1, task2, task3, task4],
+        verbose=True,
+        process="sequential",
+        manager_llm="gpt-4o"
+    )
+
+    agents.start()
+
+if __name__ == "__main__":
+    main()

cookbooks/general/example_custom_tools.py

@@ -0,0 +1,151 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+from typing import List, Dict, Union
+from duckduckgo_search import DDGS
+from langchain_community.tools import YouTubeSearchTool
+from langchain_community.utilities import WikipediaAPIWrapper
+
+# 1. Tool
+def internet_search_tool(query: str) -> List[Dict]:
+    """
+    Perform a search using DuckDuckGo.
+
+    Args:
+        query (str): The search query.
+
+    Returns:
+        list: A list of search result titles, URLs, and snippets.
+    """
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=10):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        return results
+
+    except Exception as e:
+        print(f"Error during DuckDuckGo search: {e}")
+        return []
+    
+def youtube_search_tool(query: str, inspect: bool = False, max_results: int = 2):
+    """
+    Provide a custom wrapper around the YouTubeSearchTool.
+    
+    Args:
+        query (str): The search query for YouTube.
+        inspect (bool): If True, returns tool inspection info instead of search results.
+        max_results (int): Maximum number of results to return (default: 2).
+    Returns:
+        Union[List[str], dict]: List of YouTube video URLs or tool inspection info.
+    """
+    yt = YouTubeSearchTool()
+    
+    if inspect:
+        inspection_info = {
+            "type": type(yt),
+            "attributes": [attr for attr in dir(yt) if not attr.startswith('_')],
+            "methods": {
+                "run": getattr(yt, 'run', None),
+                "arun": getattr(yt, 'arun', None)
+            },
+            "properties": {
+                "name": getattr(yt, 'name', 'youtube_search'),
+                "description": getattr(yt, 'description', 'Search YouTube videos'),
+                "return_direct": getattr(yt, 'return_direct', False)
+            }
+        }
+        return inspection_info
+    
+    # Format query with max_results
+    formatted_query = f"{query}, {max_results}"
+    return yt.run(formatted_query)
+
+def wikipedia_search_tool(query: str, inspect: bool = False, max_chars: int = 4000, top_k: int = 3):
+    """
+    Provide a custom wrapper around langchain_community's WikipediaAPIWrapper.
+
+    Args:
+        query (str): A search query for Wikipedia.
+        inspect (bool): If True, returns tool inspection info instead of search results.
+        max_chars (int): Maximum characters to return (default: 4000).
+        top_k (int): Number of top results to consider (default: 3).
+    Returns:
+        Union[str, dict]: Summary from Wikipedia or tool inspection info if inspect=True.
+    """
+    w = WikipediaAPIWrapper(
+        top_k_results=top_k,
+        doc_content_chars_max=max_chars,
+        lang='en'
+    )
+    
+    if inspect:
+        inspection_info = {
+            "type": type(w),
+            "attributes": [attr for attr in dir(w) if not attr.startswith('_')],
+            "methods": {
+                "run": getattr(w, 'run', None),
+                "arun": getattr(w, 'arun', None)
+            },
+            "properties": {
+                "name": "wikipedia",
+                "description": "Search and get summaries from Wikipedia",
+                "top_k": w.top_k_results,
+                "lang": w.lang,
+                "max_chars": w.doc_content_chars_max
+            }
+        }
+        return inspection_info
+    
+    try:
+        result = w.run(query)
+        return result
+    except Exception as e:
+        return f"Error searching Wikipedia: {str(e)}"
+
+# 2. Agent
+data_agent = Agent(
+    name="DataCollector",
+    role="Search Specialist",
+    goal="Perform internet searches to collect relevant information.",
+    backstory="Expert in finding and organizing internet data from multiple sources.",
+    tools=[internet_search_tool, youtube_search_tool, wikipedia_search_tool],
+    self_reflect=False
+)
+
+# 3. Tasks
+collect_task = Task(
+    description="Perform an internet search using the query: 'AI job trends in 2024'. Return results as a list of title, URL, and snippet.",
+    expected_output="List of search results with titles, URLs, and snippets.",
+    agent=data_agent,
+    name="collect_data",
+    is_start=True,
+    next_tasks=["validate_data"]
+)
+
+validate_task = Task(
+    description="""Validate the collected data. Check if:
+    1. At least 5 results are returned.
+    2. Each result contains a title and a URL.
+    Return validation_result as 'valid' or 'invalid' only no other text.""",
+    expected_output="Validation result indicating if data is valid or invalid.",
+    agent=data_agent,
+    name="validate_data",
+    task_type="decision",
+    condition={
+        "valid": [],  # End the workflow on valid data
+        "invalid": ["collect_data"]  # Retry data collection on invalid data
+    },
+)
+
+# 4. Workflow
+agents = PraisonAIAgents(
+    agents=[data_agent],
+    tasks=[collect_task, validate_task],
+    verbose=1,
+    process="workflow"
+)
+
+agents.start()

cookbooks/general/example_sequential.py

@@ -0,0 +1,141 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents, error_logs
+from duckduckgo_search import DDGS
+
+def my_callback(output):
+    print(f"Callback Task output: {output}")
+
+def internet_search_tool(query) -> list:
+    """
+    Perform a search using DuckDuckGo.
+
+    Args:
+        query (str): The search query.
+
+    Returns:
+        list: A list of search result titles and URLs.
+    """
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=10):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        return results
+
+    except Exception as e:
+        print(f"Error during DuckDuckGo search: {e}")
+        return []
+
+# Create agents
+researcher = Agent(
+    name="Researcher",
+    role="Senior Research Analyst",
+    goal="Uncover cutting-edge developments in AI and data science",
+    backstory="""You are an expert at a technology research group, 
+    skilled in identifying trends and analyzing complex data.""",
+    verbose=True,
+    allow_delegation=False,
+    tools=[internet_search_tool],
+    llm="gpt-4o",
+    markdown=True,
+    reflect_llm="gpt-4o",
+    min_reflect=2,
+    max_reflect=4
+)
+writer = Agent(
+    name="Writer",
+    role="Tech Content Strategist",
+    goal="Craft compelling content on tech advancements",
+    backstory="""You are a content strategist known for 
+    making complex tech topics interesting and easy to understand.""",
+    verbose=True,
+    allow_delegation=True,
+    llm="gpt-4o",
+    tools=[],
+    markdown=True
+)
+
+# Create tasks
+task1 = Task(
+    name="research_task",
+    description="""Analyze 2024's AI advancements. 
+    Find major trends, new technologies, and their effects.""",
+    expected_output="""A detailed report on 2024 AI advancements""",
+    agent=researcher,
+    tools=[internet_search_tool]
+)
+
+task2 = Task(
+    name="writing_task",
+    description="""Create a blog post about major AI advancements using the insights you have.
+    Make it interesting, clear, and suited for tech enthusiasts. 
+    It should be at least 4 paragraphs long. 
+    Also, call the get_weather tool to get the weather in Paris.""",
+    expected_output="A blog post of at least 4 paragraphs, and weather in Paris",
+    agent=writer,
+    context=[task1],
+    callback=my_callback,
+    tools=[]
+)
+
+task3 = Task(
+    name="json_task",
+    description="""Create a json object with a title of "My Task" and content of "My content".""",
+    expected_output="""JSON output with title and content""",
+    agent=researcher,
+)
+
+task4 = Task(
+    name="save_output_task",
+    description="""Save the AI blog post to a file""",
+    expected_output="""File saved successfully""",
+    agent=writer,
+    context=[task2],
+    output_file='outputs/ai_blog_post.txt',
+    create_directory=True
+)
+
+# Create and run agents manager
+agents = PraisonAIAgents(
+    agents=[researcher, writer],
+    tasks=[task1, task2, task3, task4],
+    verbose=False,
+    process="sequential",  # "sequential" or "hierarchical"
+    manager_llm="gpt-4o"
+)
+
+result = agents.start()
+
+# Print results and error summary
+print("\n=== Task Results ===")
+for task_id, task_status in result['task_status'].items():
+    print(f"Task {task_id}: {task_status}")
+    if task_result := result['task_results'].get(task_id):
+        print(f"Output: {task_result.raw[:200]}...")  # Show first 200 chars
+
+# Print task details
+print("\n=== Task Details ===")
+for i in range(4):
+    print(agents.get_task_details(i))
+
+# Print agent details
+print("\n=== Agent Details ===")
+print(agents.get_agent_details('Researcher'))
+print(agents.get_agent_details('Writer'))
+
+# Print any errors
+if error_logs:
+    print("\n=== Error Summary ===")
+    for err in error_logs:
+        print(f"- {err}")
+        if "parsing self-reflection json" in err:
+            print("  Reason: The self-reflection JSON response was not valid JSON.")
+        elif "Error: Task with ID" in err:
+            print("  Reason: Task ID referenced does not exist.")
+        elif "saving task output to file" in err:
+            print("  Reason: Possible file permissions or invalid path.")
+        else:
+            print("  Reason not identified")

cookbooks/general/langchain_example.py

@@ -0,0 +1,30 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+from langchain_community.tools import YouTubeSearchTool
+from langchain_community.utilities import WikipediaAPIWrapper
+
+# Create an agent with both tools
+agent = Agent(
+    name="SearchAgent",
+    role="Research Assistant",
+    goal="Search for information from multiple sources",
+    backstory="I am an AI assistant that can search YouTube and Wikipedia.",
+    tools=[YouTubeSearchTool, WikipediaAPIWrapper],
+    self_reflect=False
+)
+
+# Create tasks to demonstrate both tools
+task = Task(
+    name="search_task",
+    description="Search for information about 'AI advancements' on both YouTube and Wikipedia",
+    expected_output="Combined information from YouTube videos and Wikipedia articles",
+    agent=agent
+)
+
+# Create and start the workflow
+agents = PraisonAIAgents(
+    agents=[agent],
+    tasks=[task],
+    verbose=True
+)
+
+agents.start()

cookbooks/general/memory_example.py

@@ -0,0 +1,183 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import logging
+import os
+
+def main():
+    # Initialize memory config
+    memory_config = {
+        "provider": "rag",
+        "use_embedding": True,
+        "storage": {
+            "type": "sqlite",
+            "path": "./.praison/memory.db"
+        },
+        "rag_db_path": "./.praison/chroma_db"
+    }
+    
+    # Test facts
+    fact1 = "The capital city of Jujuha is Hahanu and its population is 102300"
+    fact2 = "Three main ingredients in a classic proloder are eggs, sugar, and flour"
+    fact3 = "The year the first Josinga was released is 2007"
+    
+    # # Check if database exists
+    # if os.path.exists("./memory.db"):
+    #     logger.info("Found existing memory database")
+    # else:
+    #     logger.info("Creating new memory database")
+    
+    # Create task config (without memory config since it's moved to PraisonAIAgents)
+    task_config = {}
+
+    # Create agents with different roles
+    researcher = Agent(
+        role="Research Analyst",
+        goal="Research and document key information about topics",
+        backstory="Expert at analyzing and storing information in memory",
+        llm="gpt-4o-mini"
+    )
+    
+    retriever = Agent(
+        role="Information Retriever",
+        goal="Retrieve and verify stored information from memory",
+        backstory="Specialist in searching and validating information from memory",
+        llm="gpt-4o-mini"
+    )
+
+    # Task 1: Process the facts
+    store_task = Task(
+        description=f"""
+        Process and analyze this information:
+        1. {fact1}
+        2. {fact2}
+        3. {fact3}
+
+        Provide a clear summary of each fact.
+        """,
+        expected_output="""
+        Clear statements summarizing each fact.
+        Example format:
+        1. [Summary of fact 1]
+        2. [Summary of fact 2]
+        3. [Summary of fact 3]
+        """,
+        agent=researcher
+    )
+
+    # Task 2: Write essay about AI
+    verify_task = Task(
+        description="""
+        write few points about AI
+        """,
+        expected_output="Points about AI",
+        agent=retriever
+    )
+
+    # Task 3: Query memory
+    query_task = Task(
+        description="""
+        Using ONLY information found in memory:
+        1. What is stored in memory about Hahanu?
+        2. What ingredients for proloder are recorded in memory?
+        3. What year is stored in memory for the Josinga release?
+
+        For each answer, cite the memory record you found.
+        """,
+        expected_output="Answers based solely on memory records with citations",
+        agent=retriever
+    )
+
+    # Task 4: Query both short-term and long-term memory
+    query_both_task = Task(
+        description="""
+        Using ONLY information found in memory:
+        1. What is stored in both short-term and long-term memory about Jujuha?
+        2. What ingredients for proloder are recorded in both short-term and long-term memory?
+        3. What year is stored in both short-term and long-term memory for the Josinga release?
+
+        For each answer, cite the memory record you found.
+        """,
+        expected_output="Answers based solely on memory records with citations",
+        agent=retriever
+    )
+
+    # Initialize PraisonAIAgents with memory configuration
+    agents = PraisonAIAgents(
+        agents=[researcher, retriever],
+        tasks=[store_task, verify_task, query_task, query_both_task],
+        verbose=True,  # Use same verbose level as memory
+        memory=True,
+        embedder={
+            "provider": "openai",
+            "config": {
+                "model": "text-embedding-3-small"
+            }
+        }
+    )
+
+    # agents = PraisonAIAgents(
+    #     agents=[researcher, retriever],
+    #     tasks=[store_task, verify_task, query_task, query_both_task],
+    #     verbose=True,  # Use same verbose level as memory
+    #     memory=True
+    # )
+    
+    # Execute tasks
+    print("\nExecuting Memory Test Tasks...")
+    print("-" * 50)
+    agents.start()
+    
+    # Use shared memory for final checks
+    memory = agents.shared_memory
+    
+    # Test memory retrieval with different quality thresholds
+    if memory:
+        print("\nFinal Memory Check:")
+        print("-" * 50)
+        
+        queries = ["Jujuha", "proloder", "Josinga"]
+        for query in queries:
+            print(f"\nSearching memory for: {query}")
+            
+            # Search in both short-term and long-term memory
+            print("\nShort-term memory results:")
+            stm_results = memory.search_short_term(query)
+            if stm_results:
+                for item in stm_results:
+                    print(f"Content: {item.get('content', '')[:200]}")
+                    if 'meta' in item:
+                        print(f"Metadata: {item['meta']}")
+                    print("-" * 20)
+            else:
+                print("No results found in short-term memory")
+
+            print("\nLong-term memory results:")
+            ltm_results = memory.search_long_term(query)
+            if ltm_results:
+                for item in ltm_results:
+                    print(f"Content: {item.get('text', '')[:200]}")
+                    if 'metadata' in item:
+                        print(f"Metadata: {item['metadata']}")
+                    print("-" * 20)
+            else:
+                print("No results found in long-term memory")
+                
+            # Also check ChromaDB if using RAG
+            if memory.use_rag and hasattr(memory, "chroma_col"):
+                print("\nChromaDB results:")
+                try:
+                    all_items = memory.chroma_col.get()
+                    print(f"Found {len(all_items['ids'])} items in ChromaDB")
+                    for i in range(len(all_items['ids'])):
+                        print(f"ID: {all_items['ids'][i]}")
+                        print(f"Content: {all_items['documents'][i][:200]}")
+                        print(f"Metadata: {all_items['metadatas'][i]}")
+                        print("-" * 20)
+                except Exception as e:
+                    print(f"Error querying ChromaDB: {e}")
+            
+            print("-" * 30)
+    else:
+        print("\nNo memory available for final checks")
+
+if __name__ == "__main__":
+    main()

cookbooks/general/memory_simple.py

@@ -0,0 +1,63 @@
+from praisonaiagents.agents.agents import Agent, Task, PraisonAIAgents
+from praisonaiagents.tools import duckduckgo
+
+# Test facts
+fact1 = "The capital city of Jujuha is Hahanu and its population is 102300"
+fact2 = "Three main ingredients in a classic proloder are eggs, sugar, and flour"
+fact3 = "The year the first Josinga was released is 2007"
+
+fact_agent = Agent(
+    name="Fact Agent",
+    instructions="You are a fact agent, you store and retrieve facts in memory",
+    llm="gpt-4o-mini"
+)
+
+research_agent = Agent(
+    name="Research Agent",
+    instructions="You are a research analyst, you research and document key points about topics",
+    llm="gpt-4o-mini"
+)
+
+blog_agent = Agent(
+    name="Blog Agent",
+    instructions="You are a blog writer, you write a blog post about the research",
+    llm="gpt-4o-mini"
+)
+
+fact_task = Task(
+    description="Store the following facts in memory: " + fact1 + ", " + fact2 + ", " + fact3,
+    agent=fact_agent
+)
+
+research_task = Task(
+    description="Research and document 2 key points about AI",
+    agent=research_agent
+)
+
+research_task2 = Task(
+    description="Research and document 2 key points about AI in healthcare",
+    agent=research_agent
+)
+
+research_task3 = Task(
+    description="Research and document 2 key points about AI in education",
+    agent=research_agent
+)
+
+research_task4 = Task(
+    description="Research and document 2 key points about AI in finance",
+    agent=research_agent
+)
+
+blog_task = Task(
+    description="Write a blog post about Jujuha",
+    agent=blog_agent
+)
+
+agents = PraisonAIAgents(
+    agents=[fact_agent, research_agent, blog_agent],
+    tasks=[fact_task, research_task, research_task2, research_task3, research_task4, blog_task],
+    memory=True
+)   
+
+agents.start()

cookbooks/general/mini_agents_example.py

@@ -0,0 +1,6 @@
+from praisonaiagents import Agent, Agents, Tools
+
+research_agent = Agent(instructions="You are a research agent to search internet about AI 2024", tools=[Tools.internet_search])
+summarise_agent = Agent(instructions="You are a summarize agent to summarise in points")
+agents = Agents(agents=[research_agent, summarise_agent])
+agents.start()

cookbooks/general/multimodal.py

@@ -0,0 +1,59 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Create Vision Analysis Agent
+vision_agent = Agent(
+    name="VisionAnalyst",
+    role="Computer Vision Specialist",
+    goal="Analyze images and videos to extract meaningful information",
+    backstory="""You are an expert in computer vision and image analysis.
+    You excel at describing images, detecting objects, and understanding visual content.""",
+    llm="gpt-4o-mini",
+    self_reflect=False
+)
+
+# 1. Task with Image URL
+task1 = Task(
+    name="analyze_landmark",
+    description="Describe this famous landmark and its architectural features.",
+    expected_output="Detailed description of the landmark's architecture and significance",
+    agent=vision_agent,
+    images=["https://upload.wikimedia.org/wikipedia/commons/b/bf/Krakow_-_Kosciol_Mariacki.jpg"]
+)
+
+# 2. Task with Local Image File
+task2 = Task(
+    name="analyze_local_image",
+    description="What objects can you see in this image? Describe their arrangement.",
+    expected_output="Detailed description of objects and their spatial relationships",
+    agent=vision_agent,
+    images=["image.jpg"] 
+)
+
+# 3. Task with Video File
+task3 = Task(
+    name="analyze_video",
+    description="""Watch this video and provide:
+    1. A summary of the main events
+    2. Key objects and people visible
+    3. Any text or important information shown
+    4. The overall context and setting""",
+    expected_output="Comprehensive analysis of the video content",
+    agent=vision_agent,
+    images=["video.mp4"]  
+)
+
+# Create PraisonAIAgents instance
+agents = PraisonAIAgents(
+    agents=[vision_agent],
+    tasks=[task1, task2, task3],
+    process="sequential",
+    verbose=1
+)
+
+# Run all tasks
+result = agents.start()
+
+# Print results
+for task_id, task_result in result["task_results"].items():
+    print(f"\nTask {task_id} Result:")
+    print(task_result.raw)

cookbooks/general/parallelisation.py

@@ -0,0 +1,95 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+from datetime import datetime
+import asyncio
+
+def process_time():
+    """Simulate processing"""
+    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    print(f"Processing at: {current_time}")
+    return f"Processed at {current_time}"
+
+# Create parallel processing agents
+agent1 = Agent(
+    name="Processor 1",
+    role="Time collector",
+    goal="Get the time and return it",
+    tools=[process_time]
+)
+
+agent2 = Agent(
+    name="Processor 2",
+    role="Time collector",
+    goal="Get the time and return it",
+    tools=[process_time]
+)
+
+agent3 = Agent(
+    name="Processor 3",
+    role="Time collector",
+    goal="Get the time and return it",
+    tools=[process_time]
+)
+
+aggregator = Agent(
+    name="Aggregator",
+    role="Result aggregator",
+    goal="Collect all the processed time from all tasks"
+)
+
+# Create parallel tasks with memory disabled
+task1 = Task(
+    name="process_1",
+    description="Use process_time tool to get the time",
+    expected_output="processed time",
+    agent=agent1,
+    is_start=True,
+    async_execution=True
+)
+
+task2 = Task(
+    name="process_2",
+    description="Use process_time tool to get the time",
+    expected_output="processed time",
+    agent=agent2,
+    is_start=True,
+    async_execution=True
+)
+
+task3 = Task(
+    name="process_3",
+    description="Use process_time tool to get the time",
+    expected_output="processed time",
+    agent=agent3,
+    is_start=True,
+    async_execution=True
+)
+
+aggregate_task = Task(
+    name="aggregate",
+    description="Collect all the processed time from all tasks",
+    expected_output="Output all the processed time from all tasks and just the time",
+    agent=aggregator,
+    context=[task1, task2, task3]
+)
+
+async def main():
+
+    # Create workflow manager
+    workflow = PraisonAIAgents(
+        agents=[agent1, agent2, agent3, aggregator],
+        tasks=[task1, task2, task3, aggregate_task],
+        process="workflow"
+    )
+
+    # Run parallel workflow
+    results = await workflow.astart()
+
+    # Print results
+    print("\nParallel Processing Results:")
+    for task_id, result in results["task_results"].items():
+        if result:
+            print(f"Task {task_id}: {result.raw}")
+
+# Run the async main function
+if __name__ == "__main__":
+    asyncio.run(main())

cookbooks/general/prompt_chaining.py

@@ -0,0 +1,81 @@
+from praisonaiagents.agent import Agent
+from praisonaiagents.task import Task
+from praisonaiagents.agents import PraisonAIAgents
+from typing import List, Dict
+import time
+
+def get_time_check():
+    current_time = int(time.time())
+    result = "even" if current_time % 2 == 0 else "odd"
+    print(f"Time check: {current_time} is {result}")
+    return result
+
+# Create agents for each step in the chain
+agent1 = Agent(
+    name="Time Checker",
+    role="Time checker",
+    goal="Check if the time is even or odd",
+    instructions="Check if the time is even or odd",
+    tools=[get_time_check]
+)
+
+agent2 = Agent(
+    name="Advanced Analyzer",
+    role="Advanced data analyzer",
+    goal="Perform in-depth analysis of processed data",
+    instructions="Analyze the processed data in detail"
+)
+
+agent3 = Agent(
+    name="Final Processor",
+    role="Final data processor",
+    goal="Generate final output based on analysis",
+    instructions="Create final output based on analyzed data"
+)
+
+# Create tasks for each step
+initial_task = Task(
+    name="time_check",
+    description="Getting time check and checking if it is even or odd",
+    expected_output="Getting time check and checking if it is even or odd",
+    agent=agent1,
+    is_start=True,  # Mark as the starting task
+    task_type="decision",  # This task will make a decision
+    next_tasks=["advanced_analysis"],  # Next task if condition passes
+    condition={
+        "even": ["advanced_analysis"],  # If passes, go to advanced analysis
+        "odd": ""  # If fails, exit the chain
+    }
+)
+
+analysis_task = Task(
+    name="advanced_analysis",
+    description="Perform advanced analysis on the processed data",
+    expected_output="Analyzed data ready for final processing",
+    agent=agent2,
+    next_tasks=["final_processing"]
+)
+
+final_task = Task(
+    name="final_processing",
+    description="Generate final output",
+    expected_output="Final processed result",
+    agent=agent3
+)
+
+# Create the workflow manager
+workflow = PraisonAIAgents(
+    agents=[agent1, agent2, agent3],
+    tasks=[initial_task, analysis_task, final_task],
+    process="workflow",  # Use workflow process type
+    verbose=True
+)
+
+# Run the workflow
+results = workflow.start()
+
+# Print results
+print("\nWorkflow Results:")
+for task_id, result in results["task_results"].items():
+    if result:
+        print(f"Task {task_id}: {result.raw}")

cookbooks/general/structured_agents_example.py

@@ -0,0 +1,56 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents, Tools
+from pydantic import BaseModel
+
+class AnalysisReport(BaseModel):
+    title: str
+    findings: str
+    summary: str
+
+# Create a researcher agent
+researcher = Agent(
+    name="AIResearcher",
+    role="Technology Research Analyst",
+    goal="Analyze and structure information about AI developments",
+    backstory="Expert analyst specializing in AI technology trends",
+    verbose=True,
+    llm="gpt-4o-mini",
+    tools=[Tools.internet_search],
+    self_reflect=False
+)
+
+# Create an analyst agent
+analyst = Agent(
+    name="DataAnalyst",
+    role="Data Insights Specialist",
+    goal="Structure and analyze research findings",
+    backstory="Senior data analyst with expertise in pattern recognition",
+    verbose=True,
+    llm="gpt-4o-mini",
+    self_reflect=False
+)
+
+# Define structured tasks
+research_task = Task(
+    name="gather_research",
+    description="Research recent AI developments in 2024",
+    agent=researcher,
+    expected_output="Research findings"
+)
+
+analysis_task = Task(
+    name="analyze_findings",
+    description="Analyze research findings and create structured report. No additional text or explanation.",
+    agent=analyst,
+    output_json=AnalysisReport,
+    expected_output="JSON object"
+)
+
+# Initialize and run agents
+agents = PraisonAIAgents(
+    agents=[researcher, analyst],
+    tasks=[research_task, analysis_task],
+    process="sequential",
+    verbose=True
+)
+result = agents.start()
+print(result)

cookbooks/general/structured_response_example.py

@@ -0,0 +1,95 @@
+from praisonaiagents import Agent
+from pydantic import BaseModel
+from typing import List, Optional
+
+# Define structured output models
+class SearchResult(BaseModel):
+    title: str
+    url: Optional[str]
+    snippet: str
+
+class AnalysisReport(BaseModel):
+    topic: str
+    key_findings: List[str]
+    search_results: List[SearchResult]
+    summary: str
+    confidence_score: float
+
+def get_structured_analysis(query: str, verbose: bool = True) -> AnalysisReport:
+    """
+    Performs a search and returns a structured analysis using an AI agent.
+    
+    Args:
+        query (str): The search query or topic to analyze
+        verbose (bool): Whether to show detailed output
+        
+    Returns:
+        AnalysisReport: A structured report containing the analysis
+    """
+    # Create an agent with search capabilities
+    agent = Agent(
+        name="StructuredAnalyst",
+        role="Research Analyst",
+        goal="Analyze topics and provide structured reports",
+        backstory="Expert at gathering information and providing structured analysis",
+        verbose=verbose,
+        self_reflect=True,  # Enable self-reflection for better quality
+        markdown=True,
+        llm="gpt-4o-mini"
+    )
+    
+    # Create the analysis prompt
+    prompt = f"""
+Analyze the following topic: "{query}"
+
+Provide a structured analysis in JSON format that matches this schema:
+{{
+    "topic": "string",
+    "key_findings": ["string"],
+    "search_results": [
+        {{
+            "title": "string",
+            "url": "string",
+            "snippet": "string"
+        }}
+    ],
+    "summary": "string",
+    "confidence_score": float (0-1)
+}}
+
+Requirements:
+1. Include at least 3 key findings
+2. Include at least 2 relevant search results
+3. Provide a comprehensive summary
+4. Set confidence score based on quality of sources (0-1)
+
+Return ONLY the JSON object, no other text.
+"""
+    
+    # Get structured response from agent
+    response = agent.chat(
+        prompt=prompt,
+        output_json=AnalysisReport  # This ensures response matches our model
+    )
+    
+    return AnalysisReport.model_validate_json(response)
+
+if __name__ == "__main__":
+    # Example usage
+    analysis = get_structured_analysis("Latest developments in AI agents and autonomous systems")
+    print(analysis)
+    # Print the structured results
+    print("\n=== Structured Analysis Report ===")
+    print(f"Topic: {analysis.topic}")
+    print("\nKey Findings:")
+    for i, finding in enumerate(analysis.key_findings, 1):
+        print(f"{i}. {finding}")
+    
+    print("\nSearch Results:")
+    for result in analysis.search_results:
+        print(f"\nTitle: {result.title}")
+        print(f"URL: {result.url}")
+        print(f"Snippet: {result.snippet}")
+    
+    print(f"\nSummary: {analysis.summary}")
+    print(f"Confidence Score: {analysis.confidence_score:.2f}") 

cookbooks/general/tools_example.py

@@ -0,0 +1,30 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+from praisonaiagents.tools import get_article, get_news_sources, get_articles_from_source, get_trending_topics
+
+# Create Wikipedia agent
+news_agent = Agent(
+    name="NewsAgent",
+    role="News Analyst",
+    goal="Collect and analyze news articles from various sources.",
+    backstory="Expert in news gathering and content analysis.",
+    tools=[get_article, get_news_sources, get_articles_from_source, get_trending_topics],
+    self_reflect=False
+)
+
+
+# Define research task
+news_task = Task(
+    description="Analyze news articles about 'AI developments' from major tech news sources.",
+    expected_output="Summary of key AI developments with source articles.",
+    agent=news_agent,
+    name="ai_news"
+)
+
+
+# Run agent
+agents = PraisonAIAgents(
+    agents=[news_agent],
+    tasks=[news_task],
+    process="sequential"
+)
+agents.start()

cookbooks/general/workflow_example_basic.py

@@ -0,0 +1,74 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+from typing import List, Dict
+from duckduckgo_search import DDGS
+
+# 1. Tool
+def internet_search_tool(query: str) -> List[Dict]:
+    """
+    Perform a search using DuckDuckGo.
+
+    Args:
+        query (str): The search query.
+
+    Returns:
+        list: A list of search result titles, URLs, and snippets.
+    """
+    try:
+        results = []
+        ddgs = DDGS()
+        for result in ddgs.text(keywords=query, max_results=10):
+            results.append({
+                "title": result.get("title", ""),
+                "url": result.get("href", ""),
+                "snippet": result.get("body", "")
+            })
+        return results
+
+    except Exception as e:
+        print(f"Error during DuckDuckGo search: {e}")
+        return []
+
+# 2. Agent
+data_agent = Agent(
+    name="DataCollector",
+    role="Search Specialist",
+    goal="Perform internet searches to collect relevant information.",
+    backstory="Expert in finding and organising internet data.",
+    tools=[internet_search_tool],
+    self_reflect=False
+)
+
+# 3. Tasks
+collect_task = Task(
+    description="Perform an internet search using the query: 'AI job trends in 2024'. Return results as a list of title, URL, and snippet.",
+    expected_output="List of search results with titles, URLs, and snippets.",
+    agent=data_agent,
+    name="collect_data",
+    is_start=True,
+    next_tasks=["validate_data"]
+)
+
+validate_task = Task(
+    description="""Validate the collected data. Check if:
+    1. At least 5 results are returned.
+    2. Each result contains a title and a URL.
+    Return validation_result as 'valid' or 'invalid' only no other text.""",
+    expected_output="Validation result indicating if data is valid or invalid.",
+    agent=data_agent,
+    name="validate_data",
+    task_type="decision",
+    condition={
+        "valid": [],  # End the workflow on valid data
+        "invalid": ["collect_data"]  # Retry data collection on invalid data
+    },
+)
+
+# 4. AI Agents Workflow
+agents = PraisonAIAgents(
+    agents=[data_agent],
+    tasks=[collect_task, validate_task],
+    verbose=1,
+    process="workflow"
+)
+
+agents.start()

cookbooks/general/workflow_example_detailed.py

@@ -0,0 +1,253 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import random
+from typing import List, Dict, Union
+import json
+from pydantic import BaseModel
+
+# Add Pydantic models for data validation
+class Person(BaseModel):
+    name: str
+    age: int
+    job: str
+    city: str
+    salary: int
+
+class ProcessedPerson(Person):
+    salary_category: str
+    age_group: str
+    processing_status: str
+
+class DataList(BaseModel):
+    items: List[Dict]
+
+class ValidationResult(BaseModel):
+    validation_result: str
+    details: str = ""
+
+def random_data_of_individuals() -> List[Dict]:
+    """Generate random individual data"""
+    names = ["John", "Jane", "Mike", "Sarah", "David", "Emma"]
+    jobs = ["Engineer", "Doctor", "Teacher", "Artist", "Developer"]
+    cities = ["New York", "London", "Tokyo", "Paris", "Berlin"]
+    
+    data = []
+    for _ in range(random.randint(3, 7)):
+        person = {
+            "name": random.choice(names),
+            "age": random.randint(25, 65),
+            "job": random.choice(jobs),
+            "city": random.choice(cities),
+            "salary": random.randint(30000, 120000)
+        }
+        data.append(person)
+    return data
+
+def process_data_of_individuals(data: Union[List[Dict], Dict, str]) -> Dict:
+    """Process individual data by adding categories and analysis"""
+    try:
+        print("\n[DEBUG] process_data_of_individuals input:", data)
+        print("[DEBUG] Current workflow state:")
+        print("- total_records:", workflow.get_state("total_records"))
+        print("- current_index:", workflow.get_state("current_index"))
+        print("- remaining:", workflow.get_state("remaining"))
+
+        # Get the items list from the collect_task result
+        collect_result = None
+        for task in workflow.tasks.values():
+            if task.name == "collect_data" and task.result:
+                try:
+                    collect_data = json.loads(task.result.raw)
+                    collect_result = collect_data.get("items", [])
+                    print("[DEBUG] Found collect_data items:", len(collect_result))
+                except:
+                    print("[DEBUG] Failed to parse collect_data result")
+
+        # Handle string input by trying to parse it as JSON
+        if isinstance(data, str):
+            if ":" in data and not data.strip().startswith("{"):
+                # Convert string format to dictionary
+                pairs = [pair.strip() for pair in data.split(",")]
+                data_dict = {}
+                for pair in pairs:
+                    key, value = pair.split(":")
+                    key = key.strip().lower()
+                    value = value.strip()
+                    if key == "age" or key == "salary":
+                        value = int(value)
+                    data_dict[key] = value
+                data = data_dict
+            else:
+                data = json.loads(data)
+            print("[DEBUG] Parsed data:", data)
+
+        # Handle single record
+        if isinstance(data, dict):
+            person = data
+            # Initialize total records if not set
+            total_records = workflow.get_state("total_records")
+            if total_records is None and collect_result:
+                total_records = len(collect_result)
+                workflow.set_state("total_records", total_records)
+                print(f"[DEBUG] Initialized total_records to {total_records}")
+
+            current_index = workflow.get_state("current_index", 0)
+            total_records = total_records or 1
+            remaining = total_records - (current_index + 1)
+            workflow.set_state("remaining", remaining)
+            print(f"[DEBUG] Processing record {current_index + 1}/{total_records}")
+
+        elif isinstance(data, list):
+            if len(data) == 0:
+                raise ValueError("Empty data list")
+            person = data[0]
+            workflow.set_state("total_records", len(data))
+            workflow.set_state("current_index", 0)
+            workflow.set_state("remaining", len(data) - 1)
+            print(f"[DEBUG] First record from list of {len(data)} items")
+        else:
+            raise ValueError("Input must be a dictionary or list with at least one record")
+
+        processed_person = person.copy()
+        
+        # Add salary category
+        salary = person.get("salary", 0)
+        if salary < 50000:
+            processed_person["salary_category"] = "entry"
+        elif salary < 90000:
+            processed_person["salary_category"] = "mid"
+        else:
+            processed_person["salary_category"] = "senior"
+        
+        # Add age group
+        age = person.get("age", 0)
+        if age < 35:
+            processed_person["age_group"] = "young"
+        elif age < 50:
+            processed_person["age_group"] = "mid"
+        else:
+            processed_person["age_group"] = "senior"
+        
+        # Add processing status using workflow state
+        remaining = workflow.get_state("remaining", 0)
+        current_index = workflow.get_state("current_index", 0)
+        total_records = workflow.get_state("total_records", 1)
+        
+        # Update current index for next iteration
+        workflow.set_state("current_index", current_index + 1)
+        
+        print(f"[DEBUG] Status check - remaining: {remaining}, current_index: {current_index}, total_records: {total_records}")
+        
+        if remaining <= 0 and current_index >= total_records - 1:
+            print("[DEBUG] Setting status to 'all records processed'")
+            processed_person["processing_status"] = "all records processed"
+        else:
+            print(f"[DEBUG] More records to process. Remaining: {remaining}")
+            processed_person["processing_status"] = f"more records to process ({remaining} remaining)"
+            
+        print("[DEBUG] Final processed person:", processed_person)
+        return processed_person
+        
+    except Exception as e:
+        print(f"[DEBUG] Error processing data: {str(e)}")
+        return {"error": str(e), "processing_status": "error occurred"}
+
+# Create agents
+data_agent = Agent(
+    name="DataCollector",
+    role="Data collection specialist",
+    goal="Collect and validate data about individuals",
+    backstory="Expert in gathering and validating demographic data",
+    tools=[random_data_of_individuals],
+    self_reflect=False
+)
+
+process_agent = Agent(
+    name="DataProcessor",
+    role="Data processor",
+    goal="Process and categorize individual data",
+    backstory="Expert in data analysis and categorization",
+    tools=[process_data_of_individuals],
+    self_reflect=False
+)
+
+# Modify tasks to use Pydantic models
+collect_task = Task(
+    description="Collect random individual data using the random_data_of_individuals tool. Return as a JSON object with 'items' array.",
+    expected_output="List of individual records with basic information",
+    agent=data_agent,
+    name="collect_data",
+    tools=[random_data_of_individuals],
+    is_start=True,
+    next_tasks=["validate_data"],
+    output_json=DataList
+)
+
+validate_task = Task(
+    description="""Validate the collected data. Check if:
+    1. All required fields are present (name, age, job, city, salary)
+    2. Age is between 25 and 65
+    3. Salary is between 30000 and 120000
+    Return validation_result as 'valid' or 'invalid' with optional details.""",
+    expected_output="Validation result indicating if data is valid or invalid",
+    agent=data_agent,
+    name="validate_data",
+    task_type="decision",
+    condition={
+        "valid": ["process_data"],
+        "invalid": ["collect_data"]
+    },
+    output_json=ValidationResult
+)
+
+process_task = Task(
+    description="""Process one record at a time from the input data.
+    Current progress will be shown in Loop Status.
+    
+    For the current record:
+    1. Use process_data_of_individuals tool to add categories
+    2. Return the processed record with remaining count
+    
+    Current remaining: {remaining}
+    Current item: {current_item}
+    
+    Process this record and indicate if more records remain.""",
+    expected_output="Processed record with categories and status",
+    agent=process_agent,
+    name="process_data",
+    tools=[process_data_of_individuals],
+    task_type="loop",
+    condition={
+        "more records to process": ["process_data"],
+        "all records processed": []
+    },
+    context=[collect_task],
+    output_json=ProcessedPerson
+)
+
+# Create PraisonAIAgents instance with workflow process
+workflow = PraisonAIAgents(
+    agents=[data_agent, process_agent],
+    tasks=[collect_task, validate_task, process_task],
+    verbose=1,
+    process="workflow"
+)
+
+# Run the workflow
+result = workflow.start()
+
+# Print results
+print("\nWorkflow Results:")
+print("----------------")
+for task_id, task in workflow.tasks.items():
+    print(f"\nTask: {task.name}")
+    print(f"Status: {task.status}")
+    if task.result:
+        print("Output:")
+        try:
+            # Try to format as pretty JSON
+            import json
+            output = json.loads(task.result.raw)
+            print(json.dumps(output, indent=2))
+        except:
+            # If not JSON, print raw output
+            print(task.result.raw[:500] + "..." if len(task.result.raw) > 500 else task.result.raw) 

docs/api/praisonai/deploy.html

@@ -110,7 +110,7 @@ Loads environment variables from .env file or system and sets them.</p>
             file.write(&#34;FROM python:3.11-slim\n&#34;)
             file.write(&#34;WORKDIR /app\n&#34;)
             file.write(&#34;COPY . .\n&#34;)
-            file.write(&#34;RUN pip install flask praisonai==2.0.41 gunicorn markdown\n&#34;)
+            file.write(&#34;RUN pip install flask praisonai==2.0.45 gunicorn markdown\n&#34;)
             file.write(&#34;EXPOSE 8080\n&#34;)
             file.write(&#39;CMD [&#34;gunicorn&#34;, &#34;-b&#34;, &#34;0.0.0.0:8080&#34;, &#34;api:app&#34;]\n&#39;)
             

docs/features/agenticrouting.mdx

@@ -0,0 +1,224 @@
+---
+title: "Agentic Routing"
+description: "Learn how to create AI agents with dynamic routing capabilities for workflow management."
+icon: "route"
+---
+
+## Quick Start
+
+<Steps>
+    <Step title="Install Package">
+        First, install the PraisonAI Agents package:
+        ```bash
+        pip install praisonaiagents
+        ```
+    </Step>
+
+    <Step title="Set API Key">
+        Set your OpenAI API key as an environment variable in your terminal:
+        ```bash
+        export OPENAI_API_KEY=your_api_key_here
+        ```
+    </Step>
+
+    <Step title="Create a file">
+        Create a new file `app.py` with the basic setup:
+        ```python
+        from praisonaiagents.agent import Agent
+        from praisonaiagents.task import Task
+        from praisonaiagents.agents import PraisonAIAgents
+        import time
+
+        def get_time_check():
+            current_time = int(time.time())
+            result = "even" if current_time % 2 == 0 else "odd"
+            print(f"Time check: {current_time} is {result}")
+            return result
+
+        # Create specialized agents
+        router = Agent(
+            name="Router",
+            role="Input Router",
+            goal="Evaluate input and determine routing path",
+            instructions="Analyze input and decide whether to proceed or exit",
+            tools=[get_time_check]
+        )
+
+        processor1 = Agent(
+            name="Processor 1",
+            role="Secondary Processor",
+            goal="Process valid inputs that passed initial check",
+            instructions="Process data that passed the routing check"
+        )
+
+        processor2 = Agent(
+            name="Processor 2",
+            role="Final Processor",
+            goal="Perform final processing on validated data",
+            instructions="Generate final output for processed data"
+        )
+
+        # Create tasks with routing logic
+        routing_task = Task(
+            name="initial_routing",
+            description="check the time and return according to what is returned",
+            expected_output="pass or fail based on what is returned",
+            agent=router,
+            is_start=True,
+            task_type="decision",
+            condition={
+                "pass": ["process_valid"],
+                "fail": ["process_invalid"]
+            }
+        )
+
+        processing_task = Task(
+            name="process_valid",
+            description="Process validated input",
+            expected_output="Processed data ready for final step",
+            agent=processor1,
+        )
+
+        final_task = Task(
+            name="process_invalid",
+            description="Generate final output",
+            expected_output="Final processed result",
+            agent=processor2
+        )
+
+        # Create and run workflow
+        workflow = PraisonAIAgents(
+            agents=[router, processor1, processor2],
+            tasks=[routing_task, processing_task, final_task],
+            process="workflow",
+            verbose=True
+        )
+
+        print("\nStarting Routing Workflow...")
+        print("=" * 50)
+
+        results = workflow.start()
+
+        print("\nWorkflow Results:")
+        print("=" * 50)
+        for task_id, result in results["task_results"].items():
+            if result:
+                task_name = result.description
+                print(f"\nTask: {task_name}")
+                print(f"Result: {result.raw}")
+                print("-" * 50)
+        ```
+    </Step>
+
+    <Step title="Start Agents">
+        Type this in your terminal to run your agents:
+        ```bash
+        python app.py
+        ```
+    </Step>
+</Steps>
+
+<Note>
+  **Requirements**
+  - Python 3.10 or higher
+  - OpenAI API key. Generate OpenAI API key [here](https://platform.openai.com/api-keys). Use Other models using [this guide](/models).   
+  - Basic understanding of Python
+</Note>
+
+<div className="relative w-full aspect-video">
+  <iframe
+    className="absolute top-0 left-0 w-full h-full"
+    src="https://www.youtube.com/embed/KNDVWGN3TpM"
+    title="YouTube video player"
+    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture"
+    allowFullScreen
+  ></iframe>
+</div>
+
+## Understanding Agentic Routing
+
+<Card title="What is Agentic Routing?" icon="question">
+  Agentic routing enables:
+  - Dynamic decision-making in workflows
+  - Conditional task execution paths
+  - Automated process branching
+  - Intelligent workflow management
+</Card>
+
+## Features
+
+<CardGroup cols={2}>
+  <Card title="Dynamic Routing" icon="route">
+    Route tasks based on real-time decisions and conditions.
+  </Card>
+  <Card title="Conditional Logic" icon="code-branch">
+    Implement complex branching logic in workflows.
+  </Card>
+  <Card title="Task Management" icon="tasks">
+    Handle task dependencies and execution order.
+  </Card>
+  <Card title="Process Control" icon="sliders">
+    Control workflow execution with detailed monitoring.
+  </Card>
+</CardGroup>
+
+## Configuration Options
+
+```python
+# Create a router agent
+router = Agent(
+    name="Router",
+    role="Input Router",
+    goal="Evaluate input and determine routing path",
+    instructions="Analyze input and decide whether to proceed or exit",
+    tools=[get_time_check],  # Custom tools for routing decisions
+    verbose=True  # Enable detailed logging
+)
+
+# Task with routing configuration
+routing_task = Task(
+    name="initial_routing",
+    description="Route based on conditions",
+    expected_output="Routing decision",
+    agent=router,
+    is_start=True,
+    task_type="decision",
+    condition={
+        "pass": ["next_task"],
+        "fail": ["alternate_task"]
+    }
+)
+```
+
+## Troubleshooting
+
+<CardGroup cols={2}>
+  <Card title="Routing Issues" icon="triangle-exclamation">
+    If routing doesn't work as expected:
+    - Verify condition mappings
+    - Check task dependencies
+    - Enable verbose mode for debugging
+  </Card>
+
+  <Card title="Workflow Flow" icon="diagram-project">
+    If workflow is unclear:
+    - Review task connections
+    - Verify agent configurations
+    - Check routing conditions
+  </Card>
+</CardGroup>
+
+## Next Steps
+
+<CardGroup cols={2}>
+  <Card title="AutoAgents" icon="robot" href="./autoagents">
+    Learn about automatically created and managed AI agents
+  </Card>
+  <Card title="Mini Agents" icon="microchip" href="./mini">
+    Explore lightweight, focused AI agents
+  </Card>
+</CardGroup>
+
+<Note>
+  For optimal results, ensure your routing conditions are well-defined and your task dependencies are properly configured.
+</Note>

docs/features/evaluator-optimiser.mdx

@@ -0,0 +1,216 @@
+---
+title: "Agentic Evaluator Optimiser Workflow"
+sidebarTitle: "Evaluator Optimiser"
+description: "Learn how to create AI agents that can generate and optimize solutions through iterative feedback."
+icon: "arrows-rotate"
+---
+
+## Quick Start
+
+<Steps>
+    <Step title="Install Package">
+        First, install the PraisonAI Agents package:
+        ```bash
+        pip install praisonaiagents
+        ```
+    </Step>
+
+    <Step title="Set API Key">
+        Set your OpenAI API key as an environment variable in your terminal:
+        ```bash
+        export OPENAI_API_KEY=your_api_key_here
+        ```
+    </Step>
+
+    <Step title="Create a file">
+        Create a new file `app.py` with the basic setup:
+        ```python
+        from praisonaiagents import Agent, Task, PraisonAIAgents
+
+        # Create generator and evaluator agents
+        generator = Agent(
+            name="Generator",
+            role="Solution generator",
+            goal="Generate initial solutions and incorporate feedback",
+            instructions=(
+                "1. Look at the context from previous tasks.\n"
+                "2. If you see that you have already produced 2 points, then add another 2 new points "
+                "   so that the total becomes 10.\n"
+                "3. Otherwise, just produce the first 2 points.\n"
+                "4. Return only the final list of points, with no extra explanation."
+            )
+        )
+
+        evaluator = Agent(
+            name="Evaluator",
+            role="Solution evaluator",
+            goal="Evaluate solutions and provide improvement feedback",
+            instructions=(
+                "1. Count how many lines in the response start with a number and a period (like '1. ' or '2. ').\n"
+                "2. If there are 10 or more, respond with 'done'.\n"
+                "3. Otherwise, respond with 'more'.\n"
+                "4. Return only the single word: 'done' or 'more'."
+            )
+        )
+
+        # Create tasks for the feedback loop
+        generate_task = Task(
+            name="generate",
+            description="Write 2 points about AI incuding if anything exiting from previous points",
+            expected_output="2 points",
+            agent=generator,
+            is_start=True,
+            task_type="decision",
+            next_tasks=["evaluate"]
+        )
+
+        evaluate_task = Task(
+            name="evaluate",
+            description="Check if there are 10 points about AI",
+            expected_output="more or done",
+            agent=evaluator,
+            next_tasks=["generate"],
+            context=[generate_task],
+            task_type="decision",
+            condition={
+                "more": ["generate"],  # Continue to generate
+                "done": [""]  # Exit when optimization complete
+            }
+        )
+
+        # Create workflow manager
+        workflow = PraisonAIAgents(
+            agents=[generator, evaluator],
+            tasks=[generate_task, evaluate_task],
+            process="workflow",
+            verbose=True
+        )
+
+        # Run optimization workflow
+        results = workflow.start()
+
+        # Print results
+        print("\nEvaluator-Optimizer Results:")
+        for task_id, result in results["task_results"].items():
+            if result:
+                print(f"Task {task_id}: {result.raw}")
+        ```
+    </Step>
+
+    <Step title="Start Agents">
+        Type this in your terminal to run your agents:
+        ```bash
+        python app.py
+        ```
+    </Step>
+</Steps>
+
+<Note>
+  **Requirements**
+  - Python 3.10 or higher
+  - OpenAI API key. Generate OpenAI API key [here](https://platform.openai.com/api-keys). Use Other models using [this guide](/models).   
+  - Basic understanding of Python
+</Note>
+
+## Understanding Evaluator-Optimizer
+
+<Card title="What is Evaluator-Optimizer?" icon="question">
+  Evaluator-Optimizer pattern enables:
+  - Iterative solution generation and refinement
+  - Automated quality evaluation
+  - Feedback-driven optimization
+  - Continuous improvement loops
+</Card>
+
+## Features
+
+<CardGroup cols={2}>
+  <Card title="Solution Generation" icon="wand-magic-sparkles">
+    Generate solutions based on requirements and feedback.
+  </Card>
+  <Card title="Quality Evaluation" icon="magnifying-glass-chart">
+    Automatically assess solution quality and completeness.
+  </Card>
+  <Card title="Feedback Loop" icon="rotate">
+    Implement iterative improvement through feedback cycles.
+  </Card>
+  <Card title="Process Control" icon="sliders">
+    Monitor and control the optimization process.
+  </Card>
+</CardGroup>
+
+## Configuration Options
+
+```python
+# Create a generator agent
+generator = Agent(
+    name="Generator",
+    role="Solution generator",
+    goal="Generate and improve solutions",
+    instructions="Step-by-step instructions for generation",
+    verbose=True  # Enable detailed logging
+)
+
+# Create an evaluator agent
+evaluator = Agent(
+    name="Evaluator",
+    role="Solution evaluator",
+    goal="Evaluate and provide feedback",
+    instructions="Evaluation criteria and feedback format"
+)
+
+# Create tasks with feedback loop
+generate_task = Task(
+    name="generate",
+    description="Generate solution",
+    agent=generator,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["evaluate"]
+)
+
+evaluate_task = Task(
+    name="evaluate",
+    description="Evaluate solution",
+    agent=evaluator,
+    context=[generate_task],
+    task_type="decision",
+    condition={
+        "more": ["generate"],  # Continue optimization
+        "done": [""]  # Exit when complete
+    }
+)
+```
+
+## Troubleshooting
+
+<CardGroup cols={2}>
+  <Card title="Generation Issues" icon="triangle-exclamation">
+    If generation is not improving:
+    - Review generator instructions
+    - Check feedback integration
+    - Enable verbose mode for debugging
+  </Card>
+
+  <Card title="Evaluation Flow" icon="diagram-project">
+    If evaluation cycle is incorrect:
+    - Verify evaluation criteria
+    - Check condition mappings
+    - Review feedback loop connections
+  </Card>
+</CardGroup>
+
+## Next Steps
+
+<CardGroup cols={2}>
+  <Card title="AutoAgents" icon="robot" href="./autoagents">
+    Learn about automatically created and managed AI agents
+  </Card>
+  <Card title="Mini Agents" icon="microchip" href="./mini">
+    Explore lightweight, focused AI agents
+  </Card>
+</CardGroup>
+
+<Note>
+  For optimal results, ensure your generator instructions and evaluation criteria are clear and well-defined to achieve the desired optimization outcomes.
+</Note>

docs/features/parallelisation.mdx

@@ -0,0 +1,231 @@
+---
+title: "Agentic Parallelisation"
+description: "Learn how to create AI agents that can execute tasks in parallel for improved performance."
+icon: "arrows-split-up-and-left"
+---
+
+## Quick Start
+
+<Steps>
+    <Step title="Install Package">
+        First, install the PraisonAI Agents package:
+        ```bash
+        pip install praisonaiagents
+        ```
+    </Step>
+
+    <Step title="Set API Key">
+        Set your OpenAI API key as an environment variable in your terminal:
+        ```bash
+        export OPENAI_API_KEY=your_api_key_here
+        ```
+    </Step>
+
+    <Step title="Create a file">
+        Create a new file `app.py` with the basic setup:
+        ```python
+        from praisonaiagents import Agent, Task, PraisonAIAgents
+        from datetime import datetime
+        import asyncio
+
+        def process_time():
+            """Simulate processing"""
+            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            print(f"Processing at: {current_time}")
+            return f"Processed at {current_time}"
+
+        # Create parallel processing agents
+        agent1 = Agent(
+            name="Processor 1",
+            role="Time collector",
+            goal="Get the time and return it",
+            tools=[process_time]
+        )
+
+        agent2 = Agent(
+            name="Processor 2",
+            role="Time collector",
+            goal="Get the time and return it",
+            tools=[process_time]
+        )
+
+        agent3 = Agent(
+            name="Processor 3",
+            role="Time collector",
+            goal="Get the time and return it",
+            tools=[process_time]
+        )
+
+        aggregator = Agent(
+            name="Aggregator",
+            role="Result aggregator",
+            goal="Collect all the processed time from all tasks"
+        )
+
+        # Create parallel tasks with memory disabled
+        task1 = Task(
+            name="process_1",
+            description="Use process_time tool to get the time",
+            expected_output="processed time",
+            agent=agent1,
+            is_start=True,
+            async_execution=True
+        )
+
+        task2 = Task(
+            name="process_2",
+            description="Use process_time tool to get the time",
+            expected_output="processed time",
+            agent=agent2,
+            is_start=True,
+            async_execution=True
+        )
+
+        task3 = Task(
+            name="process_3",
+            description="Use process_time tool to get the time",
+            expected_output="processed time",
+            agent=agent3,
+            is_start=True,
+            async_execution=True
+        )
+
+        aggregate_task = Task(
+            name="aggregate",
+            description="Collect all the processed time from all tasks",
+            expected_output="Output all the processed time from all tasks and just the time",
+            agent=aggregator,
+            context=[task1, task2, task3]
+        )
+
+        async def main():
+
+            # Create workflow manager
+            workflow = PraisonAIAgents(
+                agents=[agent1, agent2, agent3, aggregator],
+                tasks=[task1, task2, task3, aggregate_task],
+                process="workflow"
+            )
+
+            # Run parallel workflow
+            results = await workflow.astart()
+
+            # Print results
+            print("\nParallel Processing Results:")
+            for task_id, result in results["task_results"].items():
+                if result:
+                    print(f"Task {task_id}: {result.raw}")
+
+        # Run the async main function
+        if __name__ == "__main__":
+            asyncio.run(main())
+        ```
+    </Step>
+
+    <Step title="Start Agents">
+        Type this in your terminal to run your agents:
+        ```bash
+        python app.py
+        ```
+    </Step>
+</Steps>
+
+<Note>
+  **Requirements**
+  - Python 3.10 or higher
+  - OpenAI API key. Generate OpenAI API key [here](https://platform.openai.com/api-keys). Use Other models using [this guide](/models).   
+  - Basic understanding of Python and async programming
+</Note>
+
+## Understanding Parallelisation
+
+<Card title="What is Parallelisation?" icon="question">
+  Parallelisation enables:
+  - Concurrent execution of multiple tasks
+  - Improved performance through parallel processing
+  - Efficient handling of independent operations
+  - Aggregation of parallel task results
+</Card>
+
+## Features
+
+<CardGroup cols={2}>
+  <Card title="Parallel Execution" icon="arrows-split-up-and-left">
+    Run multiple tasks simultaneously for improved performance.
+  </Card>
+  <Card title="Async Support" icon="bolt">
+    Built-in support for asynchronous execution.
+  </Card>
+  <Card title="Result Aggregation" icon="layer-group">
+    Combine results from parallel tasks efficiently.
+  </Card>
+  <Card title="Process Control" icon="sliders">
+    Monitor and manage parallel task execution.
+  </Card>
+</CardGroup>
+
+## Configuration Options
+
+```python
+# Create a parallel task
+task = Task(
+    name="parallel_task",
+    description="Task to run in parallel",
+    expected_output="Task result",
+    agent=agent,
+    is_start=True,
+    async_execution=True  # Enable parallel execution
+)
+
+# Create an aggregator task
+aggregator_task = Task(
+    name="aggregate",
+    description="Aggregate results from parallel tasks",
+    expected_output="Combined results",
+    agent=aggregator,
+    context=[task1, task2, task3]  # Reference parallel tasks
+)
+
+# Async workflow execution
+async def run_workflow():
+    workflow = PraisonAIAgents(
+        agents=[agent1, agent2, aggregator],
+        tasks=[task1, task2, aggregator_task],
+        process="workflow",
+        verbose=True
+    )
+    results = await workflow.astart()
+```
+
+## Troubleshooting
+
+<CardGroup cols={2}>
+  <Card title="Execution Issues" icon="triangle-exclamation">
+    If parallel execution fails:
+    - Check async configuration
+    - Verify task independence
+    - Monitor resource usage
+  </Card>
+
+  <Card title="Result Aggregation" icon="diagram-project">
+    If aggregation is incorrect:
+    - Review task outputs
+    - Check context connections
+    - Verify aggregator logic
+  </Card>
+</CardGroup>
+
+## Next Steps
+
+<CardGroup cols={2}>
+  <Card title="AutoAgents" icon="robot" href="./autoagents">
+    Learn about automatically created and managed AI agents
+  </Card>
+  <Card title="Mini Agents" icon="microchip" href="./mini">
+    Explore lightweight, focused AI agents
+  </Card>
+</CardGroup>
+
+<Note>
+  For optimal results, ensure your parallel tasks are truly independent and your system has sufficient resources to handle concurrent execution.
+</Note>

docs/features/promptchaining.mdx

@@ -0,0 +1,248 @@
+---
+title: "Agentic Prompt Chaining"
+description: "Learn how to create AI agents with sequential prompt chaining for complex workflows."
+icon: "link"
+---
+
+## Quick Start
+
+<Steps>
+    <Step title="Install Package">
+        First, install the PraisonAI Agents package:
+        ```bash
+        pip install praisonaiagents
+        ```
+    </Step>
+
+    <Step title="Set API Key">
+        Set your OpenAI API key as an environment variable in your terminal:
+        ```bash
+        export OPENAI_API_KEY=your_api_key_here
+        ```
+    </Step>
+
+    <Step title="Create a file">
+        Create a new file `app.py` with the basic setup:
+        ```python
+        from praisonaiagents import Agent, Task, PraisonAIAgents
+        import time
+
+        def get_time_check():
+            current_time = int(time.time())
+            result = "even" if current_time % 2 == 0 else "odd"
+            print(f"Time check: {current_time} is {result}")
+            return result
+
+        # Create agents for each step in the chain
+        agent1 = Agent(
+            name="Time Checker",
+            role="Time checker",
+            goal="Check if the time is even or odd",
+            instructions="Check if the time is even or odd",
+            tools=[get_time_check]
+        )
+
+        agent2 = Agent(
+            name="Advanced Analyzer",
+            role="Advanced data analyzer",
+            goal="Perform in-depth analysis of processed data",
+            instructions="Analyze the processed data in detail"
+        )
+
+        agent3 = Agent(
+            name="Final Processor",
+            role="Final data processor",
+            goal="Generate final output based on analysis",
+            instructions="Create final output based on analyzed data"
+        )
+
+        # Create tasks for each step
+        initial_task = Task(
+            name="time_check",
+            description="Getting time check and checking if it is even or odd",
+            expected_output="Getting time check and checking if it is even or odd",
+            agent=agent1,
+            is_start=True,  # Mark as the starting task
+            task_type="decision",  # This task will make a decision
+            next_tasks=["advanced_analysis"],  # Next task if condition passes
+            condition={
+                "even": ["advanced_analysis"],  # If passes, go to advanced analysis
+                "odd": ""  # If fails, exit the chain
+            }
+        )
+
+        analysis_task = Task(
+            name="advanced_analysis",
+            description="Perform advanced analysis on the processed data",
+            expected_output="Analyzed data ready for final processing",
+            agent=agent2,
+            next_tasks=["final_processing"]
+        )
+
+        final_task = Task(
+            name="final_processing",
+            description="Generate final output",
+            expected_output="Final processed result",
+            agent=agent3
+        )
+
+        # Create the workflow manager
+        workflow = PraisonAIAgents(
+            agents=[agent1, agent2, agent3],
+            tasks=[initial_task, analysis_task, final_task],
+            process="workflow",  # Use workflow process type
+            verbose=True
+        )
+
+        # Run the workflow
+        results = workflow.start()
+
+        # Print results
+        print("\nWorkflow Results:")
+        for task_id, result in results["task_results"].items():
+            if result:
+                print(f"Task {task_id}: {result.raw}")
+        ```
+    </Step>
+
+    <Step title="Start Agents">
+        Type this in your terminal to run your agents:
+        ```bash
+        python app.py
+        ```
+    </Step>
+</Steps>
+
+<Note>
+  **Requirements**
+  - Python 3.10 or higher
+  - OpenAI API key. Generate OpenAI API key [here](https://platform.openai.com/api-keys). Use Other models using [this guide](/models).   
+  - Basic understanding of Python
+</Note>
+
+## Understanding Prompt Chaining
+
+<Card title="What is Prompt Chaining?" icon="question">
+  Prompt chaining enables:
+  - Sequential execution of prompts
+  - Data flow between agents
+  - Conditional branching in workflows
+  - Step-by-step processing of complex tasks
+</Card>
+
+## Features
+
+<CardGroup cols={2}>
+  <Card title="Sequential Processing" icon="arrow-right">
+    Execute tasks in a defined sequence with data passing between steps.
+  </Card>
+  <Card title="Decision Points" icon="code-branch">
+    Implement conditional logic to control workflow progression.
+  </Card>
+  <Card title="Data Flow" icon="arrows-up-down">
+    Pass data seamlessly between agents in the chain.
+  </Card>
+  <Card title="Process Control" icon="sliders">
+    Monitor and control the execution of each step in the chain.
+  </Card>
+</CardGroup>
+
+## Configuration Options to exit the chain
+
+```python
+# Task with chaining configuration
+task = Task(
+    name="time_check",
+    description="Check time and make decision",
+    expected_output="Time check result",
+    agent=agent,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["next_step"],
+    condition={
+        "even": ["next_step"],
+        "odd": ""
+    }
+)
+```
+
+```python
+task = Task(
+    name="time_check",
+    description="Check time and make decision",
+    expected_output="Time check result",
+    agent=agent,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["next_step"],
+    condition={
+        "even": ["next_step"],
+        "odd": "exit"
+    }
+)
+```
+
+```python
+task = Task(
+    name="time_check",
+    description="Check time and make decision",
+    expected_output="Time check result",
+    agent=agent,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["next_step"],
+    condition={
+        "even": ["next_step"],
+        "odd": ["exit"]
+    }
+)
+```
+
+```python
+task = Task(
+    name="time_check",
+    description="Check time and make decision",
+    expected_output="Time check result",
+    agent=agent,
+    is_start=True,
+    task_type="decision",
+    next_tasks=["next_step"],
+    condition={
+        "even": ["next_step"],
+        "odd": [""]
+    }
+)
+```
+
+## Troubleshooting
+
+<CardGroup cols={2}>
+  <Card title="Chain Issues" icon="triangle-exclamation">
+    If chain execution fails:
+    - Verify task connections
+    - Check condition logic
+    - Enable verbose mode for debugging
+  </Card>
+
+  <Card title="Data Flow" icon="diagram-project">
+    If data flow is incorrect:
+    - Review task outputs
+    - Check agent configurations
+    - Verify task dependencies
+  </Card>
+</CardGroup>
+
+## Next Steps
+
+<CardGroup cols={2}>
+  <Card title="AutoAgents" icon="robot" href="./autoagents">
+    Learn about automatically created and managed AI agents
+  </Card>
+  <Card title="Mini Agents" icon="microchip" href="./mini">
+    Explore lightweight, focused AI agents
+  </Card>
+</CardGroup>
+
+<Note>
+  For optimal results, ensure your chain is properly configured with clear task dependencies and conditions for branching logic.
+</Note>

docs/mint.json

@@ -90,7 +90,11 @@
         "features/codeagent",
         "features/mathagent",
         "features/structured",
-        "features/callbacks"
+        "features/callbacks",
+        "features/agenticrouting",
+        "features/promptchaining",
+        "features/parallelisation",
+        "features/evaluator-optimiser"
       ]
     },
     {

praisonai.rb

@@ -3,7 +3,7 @@ class Praisonai < Formula
   
     desc "AI tools for various AI applications"
     homepage "https://github.com/MervinPraison/PraisonAI"
-    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/2.0.41.tar.gz"
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/2.0.45.tar.gz"
     sha256 "1828fb9227d10f991522c3f24f061943a254b667196b40b1a3e4a54a8d30ce32"  # Replace with actual SHA256 checksum
     license "MIT"
   

praisonai/deploy.py

@@ -56,7 +56,7 @@ class CloudDeployer:
             file.write("FROM python:3.11-slim\n")
             file.write("WORKDIR /app\n")
             file.write("COPY . .\n")
-            file.write("RUN pip install flask praisonai==2.0.41 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.0.45 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "PraisonAI"
-version = "2.0.41"
+version = "2.0.45"
 description = "PraisonAI is an AI Agents Framework with Self Reflection. PraisonAI application combines PraisonAI Agents, AutoGen, and CrewAI into a low-code solution for building and managing multi-agent LLM systems, focusing on simplicity, customisation, and efficient human–agent collaboration."
 readme = "README.md"
 license = ""
@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=0.0.31",
+    "praisonaiagents>=0.0.35",
     "python-dotenv>=0.19.0",
     "instructor>=1.3.3",
     "PyYAML>=6.0",
@@ -84,7 +84,7 @@ autogen = ["pyautogen>=0.2.19", "praisonai-tools>=0.0.7", "crewai"]
 
 [tool.poetry]
 name = "PraisonAI"
-version = "2.0.41"
+version = "2.0.45"
 description = "PraisonAI is an AI Agents Framework with Self Reflection. PraisonAI application combines PraisonAI Agents, AutoGen, and CrewAI into a low-code solution for building and managing multi-agent LLM systems, focusing on simplicity, customisation, and efficient human–agent collaboration."
 authors = ["Mervin Praison"]
 license = ""
@@ -102,7 +102,7 @@ python = ">=3.10,<3.13"
 rich = ">=13.7"
 markdown = ">=3.5"
 pyparsing = ">=3.0.0"
-praisonaiagents = ">=0.0.31"
+praisonaiagents = ">=0.0.35"
 python-dotenv = ">=0.19.0"
 instructor = ">=1.3.3"
 PyYAML = ">=6.0"

uv.lock

@@ -3060,7 +3060,7 @@ wheels = [
 
 [[package]]
 name = "praisonai"
-version = "2.0.41"
+version = "2.0.45"
 source = { editable = "." }
 dependencies = [
     { name = "instructor" },
@@ -3197,7 +3197,7 @@ requires-dist = [
     { name = "plotly", marker = "extra == 'realtime'", specifier = ">=5.24.0" },
     { name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.7" },
     { name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.7" },
-    { name = "praisonaiagents", specifier = ">=0.0.31" },
+    { name = "praisonaiagents", specifier = ">=0.0.35" },
     { name = "pyautogen", marker = "extra == 'autogen'", specifier = ">=0.2.19" },
     { name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
     { name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -3250,16 +3250,16 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.31"
+version = "0.0.35"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "openai" },
     { name = "pydantic" },
     { name = "rich" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/e6/33/3dbf92fd64ee405b9382e79f688cef390ff9164d72ead030f4246fb9d2d1/praisonaiagents-0.0.31.tar.gz", hash = "sha256:9c372c51c741a0365c1186b087b3a72f4371e6ae2c203fa072b7a89a400e2560", size = 68508 }
+sdist = { url = "https://files.pythonhosted.org/packages/58/2e/f7899ee05c74b0f2ddd237490f361865b85be404f7888c6229e5292a9af4/praisonaiagents-0.0.35.tar.gz", hash = "sha256:23fdda1320f29cdc86162a8a0b8e77bad07ad793a2f08d6d8d7f7729a01f7ab7", size = 69306 }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/91/7e/6eea88722770bd5f2691d8a0669cc46746961db969ea58d1380ce9e31648/praisonaiagents-0.0.31-py3-none-any.whl", hash = "sha256:41445ca9ab38d26f01df0873837bf8436e7013020474808312eff06894f2ce8e", size = 85455 },
+    { url = "https://files.pythonhosted.org/packages/d2/11/678bfe9b57d8d0a3b0f5f01c2cdb52d12ed6dcd157cc6324b084f265ba1b/praisonaiagents-0.0.35-py3-none-any.whl", hash = "sha256:0c0f9650cb67449261738cd7225d5815434bc042cca319e7e7230196b47807cf", size = 86057 },
 ]
 
 [[package]]