Merge pull request #260 from MervinPraison/develop

adding async callbacks

Dockerfile

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

agents.yaml

@@ -7,12 +7,8 @@ roles:
       information.
     goal: Gather information about Mars and cats
     role: Researcher
-    llm:
-      model: "gpt-4o"
-    reflect_llm:
-      model: "gpt-4o"
-    min_reflect: 2
-    max_reflect: 4
+    min_reflect: 1
+    max_reflect: 2
     tasks:
       research_task:
         description: Research about Mars, its environment, and the feasibility of
@@ -25,8 +21,6 @@ roles:
       stories.
     goal: Create a story concept for a movie about a cat in Mars
     role: Narrative Designer
-    llm:
-      model: "gpt-4o"
     tasks:
       story_concept_development:
         description: Use the research findings to craft a unique story concept for
@@ -39,8 +33,6 @@ roles:
       scripts.
     goal: Write a movie script about a cat in Mars
     role: Scriptwriter
-    llm:
-      model: "gpt-4o"
     tasks:
       scriptwriting_task:
         description: Turn the story concept into a production-ready movie script,
@@ -48,4 +40,4 @@ roles:
         expected_output: Final movie script with dialogue and scene details.
     tools:
     - ''
-dependencies: []
+dependencies: []

agents/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())

agents/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())

agents/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()

agents/praisonaiagents/__init__.py

@@ -16,6 +16,9 @@ from .main import (
     display_generating,
     clean_triple_backticks,
     error_logs,
+    register_display_callback,
+    sync_display_callbacks,
+    async_display_callbacks,
 )
 
 __all__ = [
@@ -32,4 +35,7 @@ __all__ = [
     'display_generating',
     'clean_triple_backticks',
     'error_logs',
+    'register_display_callback',
+    'sync_display_callbacks',
+    'async_display_callbacks',
 ] 

agents/praisonaiagents/main.py

@@ -12,6 +12,7 @@ from rich.text import Text
 from rich.markdown import Markdown
 from rich.logging import RichHandler
 from rich.live import Live
+import asyncio
 
 LOGLEVEL = os.environ.get('LOGLEVEL', 'INFO').upper()
 
@@ -25,6 +26,50 @@ logging.basicConfig(
 # Global list to store error logs
 error_logs = []
 
+# Separate registries for sync and async callbacks
+sync_display_callbacks = {}
+async_display_callbacks = {}
+
+# At the top of the file, add display_callbacks to __all__
+__all__ = [
+    'error_logs',
+    'register_display_callback',
+    'sync_display_callbacks',
+    'async_display_callbacks',
+    # ... other exports
+]
+
+def register_display_callback(display_type: str, callback_fn, is_async: bool = False):
+    """Register a synchronous or asynchronous callback function for a specific display type.
+    
+    Args:
+        display_type (str): Type of display event ('interaction', 'self_reflection', etc.)
+        callback_fn: The callback function to register
+        is_async (bool): Whether the callback is asynchronous
+    """
+    if is_async:
+        async_display_callbacks[display_type] = callback_fn
+    else:
+        sync_display_callbacks[display_type] = callback_fn
+
+async def execute_callback(display_type: str, **kwargs):
+    """Execute both sync and async callbacks for a given display type.
+    
+    Args:
+        display_type (str): Type of display event
+        **kwargs: Arguments to pass to the callback functions
+    """
+    # Execute synchronous callback if registered
+    if display_type in sync_display_callbacks:
+        callback = sync_display_callbacks[display_type]
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(None, lambda: callback(**kwargs))
+    
+    # Execute asynchronous callback if registered
+    if display_type in async_display_callbacks:
+        callback = async_display_callbacks[display_type]
+        await callback(**kwargs)
+
 def _clean_display_content(content: str, max_length: int = 20000) -> str:
     """Helper function to clean and truncate content for display."""
     if not content or not str(content).strip():
@@ -46,21 +91,30 @@ def _clean_display_content(content: str, max_length: int = 20000) -> str:
     return content.strip()
 
 def display_interaction(message, response, markdown=True, generation_time=None, console=None):
-    """Display the interaction between user and assistant."""
+    """Synchronous version of display_interaction."""
     if console is None:
         console = Console()
-    if generation_time:
-        console.print(Text(f"Response generated in {generation_time:.1f}s", style="dim"))
-
-    # Handle multimodal content (list)
+    
     if isinstance(message, list):
-        # Extract just the text content from the multimodal message
         text_content = next((item["text"] for item in message if item["type"] == "text"), "")
         message = text_content
 
     message = _clean_display_content(str(message))
     response = _clean_display_content(str(response))
 
+    # Execute synchronous callback if registered
+    if 'interaction' in sync_display_callbacks:
+        sync_display_callbacks['interaction'](
+            message=message,
+            response=response,
+            markdown=markdown,
+            generation_time=generation_time
+        )
+
+    # Rest of the display logic...
+    if generation_time:
+        console.print(Text(f"Response generated in {generation_time:.1f}s", style="dim"))
+
     if markdown:
         console.print(Panel.fit(Markdown(message), title="Message", border_style="cyan"))
         console.print(Panel.fit(Markdown(response), title="Response", border_style="cyan"))
@@ -74,6 +128,11 @@ def display_self_reflection(message: str, console=None):
     if console is None:
         console = Console()
     message = _clean_display_content(str(message))
+    
+    # Execute callback if registered
+    if 'self_reflection' in sync_display_callbacks:
+        sync_display_callbacks['self_reflection'](message=message)
+    
     console.print(Panel.fit(Text(message, style="bold yellow"), title="Self Reflection", border_style="magenta"))
 
 def display_instruction(message: str, console=None):
@@ -82,6 +141,11 @@ def display_instruction(message: str, console=None):
     if console is None:
         console = Console()
     message = _clean_display_content(str(message))
+    
+    # Execute callback if registered
+    if 'instruction' in sync_display_callbacks:
+        sync_display_callbacks['instruction'](message=message)
+    
     console.print(Panel.fit(Text(message, style="bold blue"), title="Instruction", border_style="cyan"))
 
 def display_tool_call(message: str, console=None):
@@ -90,6 +154,11 @@ def display_tool_call(message: str, console=None):
     if console is None:
         console = Console()
     message = _clean_display_content(str(message))
+    
+    # Execute callback if registered
+    if 'tool_call' in sync_display_callbacks:
+        sync_display_callbacks['tool_call'](message=message)
+    
     console.print(Panel.fit(Text(message, style="bold cyan"), title="Tool Call", border_style="green"))
 
 def display_error(message: str, console=None):
@@ -98,19 +167,138 @@ def display_error(message: str, console=None):
     if console is None:
         console = Console()
     message = _clean_display_content(str(message))
+    
+    # Execute callback if registered
+    if 'error' in sync_display_callbacks:
+        sync_display_callbacks['error'](message=message)
+    
     console.print(Panel.fit(Text(message, style="bold red"), title="Error", border_style="red"))
-    # Store errors
     error_logs.append(message)
 
 def display_generating(content: str = "", start_time: Optional[float] = None):
     if not content or not str(content).strip():
-        return Panel("", title="", border_style="green")  # Return empty panel when no content
+        return Panel("", title="", border_style="green")
+    
     elapsed_str = ""
     if start_time is not None:
         elapsed = time.time() - start_time
         elapsed_str = f" {elapsed:.1f}s"
     
     content = _clean_display_content(str(content))
+    
+    # Execute callback if registered
+    if 'generating' in sync_display_callbacks:
+        sync_display_callbacks['generating'](
+            content=content,
+            elapsed_time=elapsed_str.strip() if elapsed_str else None
+        )
+    
+    return Panel(Markdown(content), title=f"Generating...{elapsed_str}", border_style="green")
+
+# Async versions with 'a' prefix
+async def adisplay_interaction(message, response, markdown=True, generation_time=None, console=None):
+    """Async version of display_interaction."""
+    if console is None:
+        console = Console()
+    
+    if isinstance(message, list):
+        text_content = next((item["text"] for item in message if item["type"] == "text"), "")
+        message = text_content
+
+    message = _clean_display_content(str(message))
+    response = _clean_display_content(str(response))
+
+    # Execute callbacks
+    await execute_callback(
+        'interaction',
+        message=message,
+        response=response,
+        markdown=markdown,
+        generation_time=generation_time
+    )
+
+    # Rest of the display logic...
+    if generation_time:
+        console.print(Text(f"Response generated in {generation_time:.1f}s", style="dim"))
+
+    if markdown:
+        console.print(Panel.fit(Markdown(message), title="Message", border_style="cyan"))
+        console.print(Panel.fit(Markdown(response), title="Response", border_style="cyan"))
+    else:
+        console.print(Panel.fit(Text(message, style="bold green"), title="Message", border_style="cyan"))
+        console.print(Panel.fit(Text(response, style="bold blue"), title="Response", border_style="cyan"))
+
+async def adisplay_self_reflection(message: str, console=None):
+    """Async version of display_self_reflection."""
+    if not message or not message.strip():
+        return
+    if console is None:
+        console = Console()
+    message = _clean_display_content(str(message))
+    
+    if 'self_reflection' in async_display_callbacks:
+        await async_display_callbacks['self_reflection'](message=message)
+    
+    console.print(Panel.fit(Text(message, style="bold yellow"), title="Self Reflection", border_style="magenta"))
+
+async def adisplay_instruction(message: str, console=None):
+    """Async version of display_instruction."""
+    if not message or not message.strip():
+        return
+    if console is None:
+        console = Console()
+    message = _clean_display_content(str(message))
+    
+    if 'instruction' in async_display_callbacks:
+        await async_display_callbacks['instruction'](message=message)
+    
+    console.print(Panel.fit(Text(message, style="bold blue"), title="Instruction", border_style="cyan"))
+
+async def adisplay_tool_call(message: str, console=None):
+    """Async version of display_tool_call."""
+    if not message or not message.strip():
+        return
+    if console is None:
+        console = Console()
+    message = _clean_display_content(str(message))
+    
+    if 'tool_call' in async_display_callbacks:
+        await async_display_callbacks['tool_call'](message=message)
+    
+    console.print(Panel.fit(Text(message, style="bold cyan"), title="Tool Call", border_style="green"))
+
+async def adisplay_error(message: str, console=None):
+    """Async version of display_error."""
+    if not message or not message.strip():
+        return
+    if console is None:
+        console = Console()
+    message = _clean_display_content(str(message))
+    
+    if 'error' in async_display_callbacks:
+        await async_display_callbacks['error'](message=message)
+    
+    console.print(Panel.fit(Text(message, style="bold red"), title="Error", border_style="red"))
+    error_logs.append(message)
+
+async def adisplay_generating(content: str = "", start_time: Optional[float] = None):
+    """Async version of display_generating."""
+    if not content or not str(content).strip():
+        return Panel("", title="", border_style="green")
+    
+    elapsed_str = ""
+    if start_time is not None:
+        elapsed = time.time() - start_time
+        elapsed_str = f" {elapsed:.1f}s"
+    
+    content = _clean_display_content(str(content))
+    
+    if 'generating' in async_display_callbacks:
+        await async_display_callbacks['generating'](
+            content=content,
+            elapsed_time=elapsed_str.strip() if elapsed_str else None
+        )
+    
     return Panel(Markdown(content), title=f"Generating...{elapsed_str}", border_style="green")
 
 def clean_triple_backticks(text: str) -> str:

agents/pyproject.toml

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

agents/uv.lock

@@ -186,7 +186,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.17"
+version = "0.0.20"
 source = { editable = "." }
 dependencies = [
     { name = "openai" },

chainlit.md

@@ -1 +0,0 @@
-# Welcome to Praison AI! 🚀🤖

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.23 gunicorn markdown\n&#34;)
+            file.write(&#34;RUN pip install flask praisonai==2.0.25 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/api/praisonaiagents/agent/agent.mdx

@@ -45,6 +45,35 @@ The main class representing an AI agent with specific role, goal, and capabiliti
 
 #### Methods
 - `chat(self, prompt, temperature=0.2, tools=None, output_json=None)` - Chat with the agent
+- `achat(self, prompt, temperature=0.2, tools=None, output_json=None)` - Async version of chat method
 - `clean_json_output(self, output: str) → str` - Clean and extract JSON from response text
 - `clear_history(self)` - Clear chat history
-- `execute_tool(self, function_name, arguments)` - Execute a tool dynamically based on the function name and arguments
+- `execute_tool(self, function_name, arguments)` - Execute a tool dynamically based on the function name and arguments
+- `_achat_completion(self, response, tools)` - Async version of _chat_completion method
+
+#### Async Support
+The Agent class provides async support through the following methods:
+
+- `achat`: Async version of the chat method for non-blocking communication
+- `_achat_completion`: Internal async method for handling chat completions
+
+Example usage:
+```python
+async def main():
+    agent = Agent(
+        name="AsyncAgent",
+        role="Async Specialist",
+        goal="Handle async operations",
+        backstory="Expert in async processing"
+    )
+    
+    # Use async chat
+    result = await agent.achat(
+        prompt="Your prompt here",
+        tools=your_tools,
+        output_json=your_schema
+    )
+    print(result)
+
+asyncio.run(main())
+```

docs/api/praisonaiagents/agents/agents.mdx

@@ -24,16 +24,16 @@ A Pydantic model for handling loop items.
 - `model_config` - Model configuration
 
 ### PraisonAIAgents
-The main class for managing AI agents and their tasks.
+The main class for managing agents and tasks execution.
 
 #### Parameters
-- `agents` - List of agents
-- `tasks` - List of tasks
-- `verbose=0` - Verbosity level
-- `completion_checker=None` - Custom completion checker
-- `max_retries=5` - Maximum retry attempts
-- `process='sequential'` - Process execution mode
-- `manager_llm=None` - Manager language model
+- `agents: List[Agent]` - List of agents
+- `tasks: List[Task]` - List of tasks
+- `verbose: int = 0` - Verbosity level
+- `completion_checker: Optional[Callable] = None` - Custom completion checker
+- `max_retries: int = 5` - Maximum retry attempts
+- `process: str = "sequential"` - Process type (sequential, workflow, hierarchical)
+- `manager_llm: Optional[str] = None` - Language model for manager agent
 
 #### Methods
 - `add_task(self, task)` - Add a task to the crew
@@ -52,4 +52,62 @@ The main class for managing AI agents and their tasks.
 - `save_output_to_file(self, task, task_output)` - Save task output to file
 - `set_state(self, key: str, value: Any) → None` - Set a state value
 - `start(self)` - Start the crew execution
-- `update_state(self, updates: Dict) → None` - Update multiple state values
+- `update_state(self, updates: Dict) → None` - Update multiple state values
+
+#### Async Methods
+- `astart(self)` - Async version of start method
+- `aexecute_task(self, task_id)` - Async version of execute_task
+- `arun_task(self, task_id)` - Async version of run_task
+- `arun_all_tasks(self)` - Async version of run_all_tasks
+
+#### Async Support
+The PraisonAIAgents class provides comprehensive async support through the following methods:
+
+```python
+class PraisonAIAgents:
+    async def astart(self):
+        """Async version of start method"""
+        await self.arun_all_tasks()
+        return {
+            "task_status": self.get_all_tasks_status(),
+            "task_results": self.get_task_results()
+        }
+        
+    async def aexecute_task(self, task_id):
+        """Async version of execute_task method"""
+        # Async task execution implementation
+        
+    async def arun_task(self, task_id):
+        """Async version of run_task method"""
+        # Async task running with retries
+        
+    async def arun_all_tasks(self):
+        """Async version of run_all_tasks method"""
+        # Async execution of all tasks
+```
+
+Example usage:
+```python
+async def main():
+    # Initialize agents and tasks
+    agents = PraisonAIAgents(
+        agents=[async_agent1, async_agent2],
+        tasks=[async_task1, async_task2],
+        process="workflow"
+    )
+    
+    # Start async execution
+    result = await agents.astart()
+    print(result)
+
+# Run the async workflow
+asyncio.run(main())
+```
+
+#### Async Features
+- Full async/await support
+- Parallel task execution
+- Non-blocking operations
+- Efficient resource utilization
+- Mixed sync/async task handling
+- Async process management

docs/api/praisonaiagents/process/process.mdx

@@ -22,6 +22,51 @@ The main class for handling task execution processes in different modes (sequent
 - `sequential(self)` - Execute tasks in sequential order
 - `hierarchical(self)` - Execute tasks in hierarchical order with a manager agent
 
+#### Async Methods
+- `aworkflow(self) -> AsyncGenerator[str, None]` - Async version of workflow process
+- `asequential(self) -> AsyncGenerator[str, None]` - Async version of sequential process
+- `ahierarchical(self) -> AsyncGenerator[str, None]` - Async version of hierarchical process
+
+#### Async Process Support
+Each process type has an async counterpart that supports non-blocking execution:
+
+```python
+class Process:
+    async def aworkflow(self):
+        """Async workflow process execution"""
+        # Async workflow implementation
+        
+    async def asequential(self):
+        """Async sequential process execution"""
+        # Async sequential implementation
+        
+    async def ahierarchical(self):
+        """Async hierarchical process execution"""
+        # Async hierarchical implementation
+```
+
+Example usage:
+```python
+# Create process instance
+process = Process(tasks=tasks, agents=agents)
+
+# Use async workflow
+async for task_id in process.aworkflow():
+    # Handle task execution
+    if tasks[task_id].async_execution:
+        await handle_async_task(task_id)
+    else:
+        handle_sync_task(task_id)
+```
+
+#### Async Process Features
+- Non-blocking task execution
+- Parallel task processing
+- Mixed sync/async task support
+- Efficient resource utilization
+- Async manager coordination (hierarchical)
+- Dynamic task scheduling (workflow)
+
 ### LoopItems
 A Pydantic model for handling loop data in workflow process.
 

docs/api/praisonaiagents/task/task.mdx

@@ -33,4 +33,46 @@ Represents a task to be executed by an agent.
 - `task_type: str = 'task'` - Task type
 - `condition: Dict[str, List[str]] | None = None` - Task conditions
 - `is_start: bool = False` - Is start task
-- `loop_state: Dict[str, str | int] | None = None` - Loop state
+- `loop_state: Dict[str, str | int] | None = None` - Loop state
+
+#### Async Support
+The Task class supports asynchronous execution through the following features:
+
+- `async_execution`: Boolean flag to enable async execution
+- `callback`: Supports both sync and async callback functions
+- `execute_callback`: Internal async method for handling callbacks
+
+Example usage:
+```python
+from typing import Coroutine
+from praisonaiagents import Task, TaskOutput
+
+# Define an async callback
+async def async_callback(output: TaskOutput):
+    await some_async_operation()
+    print(f"Processed: {output.raw}")
+
+# Create task with async execution
+task = Task(
+    description="Async task example",
+    expected_output="Async result",
+    async_execution=True,  # Enable async execution
+    callback=async_callback  # Async callback
+)
+```
+
+#### Callback Types
+The Task class supports both synchronous and asynchronous callbacks:
+```python
+# Type definition
+callback: Optional[Union[
+    Callable[[TaskOutput], Any],  # Sync callback
+    Callable[[TaskOutput], Coroutine[Any, Any, Any]]  # Async callback
+]]
+```
+
+#### Async Task States
+- `not started`: Initial state
+- `in progress`: Task is being executed
+- `completed`: Task finished successfully
+- `failed`: Task execution failed

docs/concepts/agents.mdx

@@ -158,6 +158,63 @@ graph LR
   </Card>
 </CardGroup>
 
+## Async Support
+Agents now support asynchronous operations through the following methods:
+
+- `achat`: Async version of the chat method
+- `astart`: Async version of start method
+- `aexecute_task`: Async version of execute_task method
+- `arun_task`: Async version of run_task method
+- `arun_all_tasks`: Async version of run_all_tasks method
+
+### Example Usage:
+
+```python
+import asyncio
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+async def main():
+    # Create an async agent
+    async_agent = Agent(
+        name="AsyncAgent",
+        role="Async Task Specialist",
+        goal="Perform async operations",
+        backstory="Expert in async operations",
+        tools=[async_search_tool],  # Your async tool
+        verbose=True
+    )
+
+    # Create an async task
+    async_task = Task(
+        description="Perform async operation",
+        expected_output="Async result",
+        agent=async_agent,
+        async_execution=True  # Enable async execution
+    )
+
+    # Create and start agents with async support
+    agents = PraisonAIAgents(
+        agents=[async_agent],
+        tasks=[async_task],
+        verbose=True
+    )
+    
+    # Start async execution
+    result = await agents.astart()
+    print(result)
+
+# Run the async main function
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+### Key Features:
+- Full async/await support
+- Parallel task execution
+- Async tool integration
+- Async callback support
+- Mixed sync/async operations
+
 ## Next Steps
 
 <CardGroup cols={2}>

docs/concepts/process.mdx

@@ -160,4 +160,61 @@ Advanced process type supporting complex task relationships and conditional exec
 ### Integration
 - External system coordination
 - API synchronization
-- Event handling 
+- Event handling 
+
+## Async Process Execution
+All process types (sequential, workflow, hierarchical) support async execution through async generators. This enables efficient handling of async tasks and parallel processing.
+
+### Async Process Methods
+- `asequential`: Async version of sequential process
+- `aworkflow`: Async version of workflow process
+- `ahierarchical`: Async version of hierarchical process
+
+### Example Usage:
+
+```python
+import asyncio
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+async def main():
+    # Create agents and tasks
+    agents = [async_agent1, async_agent2]
+    tasks = [async_task1, async_task2]
+
+    # Initialize PraisonAIAgents with async workflow
+    agents = PraisonAIAgents(
+        agents=agents,
+        tasks=tasks,
+        process="workflow"  # or "sequential" or "hierarchical"
+    )
+
+    # Start async execution
+    result = await agents.astart()
+    print(result)
+
+# Run the async workflow
+asyncio.run(main())
+```
+
+### Async Process Features:
+- Parallel task execution in workflow
+- Async task delegation in hierarchical process
+- Non-blocking task execution
+- Efficient resource utilization
+- Mixed sync/async task support
+
+### Process-Specific Async Behavior:
+1. **Sequential Process**:
+   - Tasks are executed in order
+   - Async tasks don't block other tasks
+   - Maintains sequential order while allowing async execution
+
+2. **Workflow Process**:
+   - Parallel execution of independent tasks
+   - Async branching and merging
+   - Dynamic task scheduling
+
+3. **Hierarchical Process**:
+   - Async manager delegation
+   - Parallel subtask execution
+   - Dynamic task assignment 

docs/concepts/tasks.mdx

@@ -252,4 +252,66 @@ task_b = Task(
   >
     Explore the complete Task API documentation
   </Card>
-</CardGroup> 
+</CardGroup>
+
+## Async Task Execution
+Tasks can be executed asynchronously by setting `async_execution=True`. This is particularly useful for tasks that involve I/O operations or when you want to run multiple tasks in parallel.
+
+### Async Task Configuration
+```python
+async_task = Task(
+    description="Perform async operation",
+    expected_output="Async result",
+    agent=async_agent,
+    async_execution=True,  # Enable async execution
+    callback=async_callback  # Optional async callback
+)
+```
+
+### Async Callbacks
+Tasks support both synchronous and asynchronous callbacks:
+
+```python
+async def async_callback(output: TaskOutput):
+    await asyncio.sleep(1)  # Simulate async operation
+    print(f"Processed output: {output.raw}")
+
+# Task with async callback
+task = Task(
+    description="Task with async callback",
+    expected_output="Result",
+    agent=agent,
+    async_execution=True,
+    callback=async_callback
+)
+```
+
+### Mixing Sync and Async Tasks
+You can mix synchronous and asynchronous tasks in the same workflow:
+
+```python
+# Sync task
+sync_task = Task(
+    description="Synchronous operation",
+    expected_output="Sync result",
+    agent=sync_agent,
+    async_execution=False
+)
+
+# Async task
+async_task = Task(
+    description="Asynchronous operation",
+    expected_output="Async result",
+    agent=async_agent,
+    async_execution=True
+)
+
+# Create agents with mixed tasks
+agents = PraisonAIAgents(
+    agents=[sync_agent, async_agent],
+    tasks=[sync_task, async_task]
+)
+
+# Run asynchronously
+result = await agents.astart()
+``` 

docs/features/async.mdx

@@ -0,0 +1,494 @@
+---
+title: "Async AI Agents"
+description: "Async AI Agents allow you to run AI tasks asynchronously, improving performance and efficiency in your applications."
+icon: "clock"
+---
+
+<Note>
+  **Requirements**
+  
+  - Python 3.10 or higher
+  - Basic understanding of async/await in Python
+</Note>
+
+## 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:
+    ```bash
+    export OPENAI_API_KEY=your_api_key_here
+    ```
+  </Step>
+
+  <Step title="Create Project">
+    Create a new file `app.py` with the basic setup:
+    ```python
+    import asyncio
+    from praisonaiagents import Agent, Task, PraisonAIAgents
+
+    # Create an agent
+    agent = Agent(
+        name="AsyncAgent",
+        role="Assistant",
+        goal="Help with tasks",
+        backstory="Expert in async operations"
+    )
+
+    # Create a task
+    task = Task(
+        name="hello_task",
+        description="Say hello and introduce yourself",
+        agent=agent,
+        async_execution=True,
+        expected_output="Answer to the question"
+    )
+
+    # Create agents manager
+    agents = PraisonAIAgents(
+        agents=[agent],
+        tasks=[task],
+        process="sequential"
+    )
+
+    # Main async function
+    async def main():
+        result = await agents.astart()
+        print(result)
+
+    # Run the program
+    if __name__ == "__main__":
+        asyncio.run(main())
+
+    ```
+  </Step>
+
+  <Step title="Run the Program">
+    Run your async AI agent:
+    ```bash
+    python app.py
+    ```
+  </Step>
+</Steps>
+
+<div className="relative w-full aspect-video">
+  <iframe
+    className="absolute top-0 left-0 w-full h-full"
+    src="https://www.youtube.com/embed/VhVQfgo00LE"
+    title="YouTube video player"
+    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture"
+    allowFullScreen
+  ></iframe>
+</div>
+
+## Understanding Async Execution
+
+<Card title="What is Asynchronous Execution?" icon="question">
+  Async execution lets your program continue running while waiting for operations to complete. This is ideal for:
+  - Making multiple AI API calls
+  - Processing large datasets
+  - Handling multiple tasks simultaneously
+  - Maintaining responsive applications
+</Card>
+
+<CodeGroup>
+  ```python Sync
+  # Synchronous execution (blocks until complete)
+  result = agent.chat("Hello")
+  ```
+
+  ```python Async
+  # Asynchronous execution (non-blocking)
+  result = await agent.achat("Hello")
+  ```
+</CodeGroup>
+
+## Features
+
+<CardGroup cols={2}>
+  <Card title="Async Tasks" icon="tasks">
+    Create tasks that run asynchronously with built-in error handling and retries.
+  </Card>
+  <Card title="Parallel Processing" icon="arrows-split-up-and-left">
+    Run multiple tasks in parallel using asyncio.gather().
+  </Card>
+  <Card title="Mixed Mode" icon="shuffle">
+    Mix sync and async tasks in the same workflow seamlessly.
+  </Card>
+  <Card title="Resource Management" icon="gear">
+    Built-in resource management and performance optimization.
+  </Card>
+</CardGroup>
+
+## Advanced Usage
+
+### Async Tasks with Callbacks and Tools Example Code
+
+```python
+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 async tool
+class SearchResult(BaseModel):
+    query: str
+    results: List[Dict[str, str]]
+    total_results: int
+
+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
+        }
+
+# 2. 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}")
+
+# 3. 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
+)
+
+# 4. 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_pydantic=SearchResult
+)
+
+# 5. 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_pydantic=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
+    )
+    
+    # Create a single PraisonAIAgents instance with both agents
+    agents = PraisonAIAgents(
+        agents=[async_agent, summary_agent],  # Include both agents
+        tasks=parallel_tasks + [summary_task],  # Include all tasks
+        verbose=1,
+        process="sequential"  # Tasks will run in sequence, with parallel tasks running first
+    )
+    
+    # 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:
+        # 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())
+
+```
+
+### 1. Async Tasks with Callbacks
+
+<CodeGroup>
+  ```python Task
+  # Create an async task with callback
+  async_task = Task(
+      name="weather_task",
+      description="Check weather conditions",
+      agent=agent,
+      async_execution=True,
+      callback=async_callback
+  )
+  ```
+
+  ```python Callback
+  async def async_callback(output):
+      await process_result(output)
+      await save_to_database(output)
+  ```
+</CodeGroup>
+
+### 2. Parallel Task Processing
+
+<Warning>
+  Be mindful of rate limits and resource usage when processing tasks in parallel.
+</Warning>
+
+```python
+async def process_multiple_tasks():
+    tasks = [
+        Task(
+            name=f"task_{i}",
+            description=f"Process item {i}",
+            async_execution=True
+        ) for i in range(5)
+    ]
+    
+    results = await asyncio.gather(
+        *[agent.achat(task.description) for task in tasks]
+    )
+    return results
+```
+
+## Best Practices
+
+<AccordionGroup>
+  <Accordion title="Error Handling">
+    ```python
+    async def safe_async_operation():
+        try:
+            result = await agent.achat("Process this")
+            return result
+        except Exception as e:
+            logging.error(f"Error: {e}")
+            return None
+    ```
+  </Accordion>
+
+  <Accordion title="Resource Management">
+    ```python
+    async def efficient_processing():
+        semaphore = asyncio.Semaphore(5)
+        async with semaphore:
+            result = await agent.achat("Process within limits")
+        return result
+    ```
+  </Accordion>
+
+  <Accordion title="Performance Tips">
+    - Use `asyncio.gather()` for parallel operations
+    - Implement proper error handling
+    - Monitor memory usage
+    - Use timeouts for long-running operations
+  </Accordion>
+</AccordionGroup>
+
+## Troubleshooting
+
+<CardGroup cols={2}>
+  <Card title="ChatCompletion Error" icon="triangle-exclamation">
+    If you see "ChatCompletion can't be used in await expression":
+    - Use AsyncOpenAI() client
+    - Ensure proper async context
+  </Card>
+
+  <Card title="Event Loop Error" icon="circle-exclamation">
+    If you get "Event loop is closed":
+    - Use asyncio.run() for main entry
+    - Check async context
+  </Card>
+</CardGroup>
+
+## API Reference
+
+### Async Methods
+
+<ResponseField name="achat()" type="async">
+  Async version of chat method
+</ResponseField>
+
+<ResponseField name="aexecute_tool()" type="async">
+  Async tool execution method
+</ResponseField>
+
+<ResponseField name="astart()" type="async">
+  Async task execution starter
+</ResponseField>
+
+### Task Properties
+
+<ResponseField name="async_execution" type="boolean">
+  Enable/disable async mode for tasks
+</ResponseField>
+
+<ResponseField name="callback" type="function">
+  Set sync or async callback function
+</ResponseField>
+
+## Next Steps
+
+<CardGroup cols={2}>
+  <Card title="Examples" icon="code" href="./examples">
+    Check out more examples in our examples directory
+  </Card>
+  <Card title="Advanced Topics" icon="book" href="./advanced">
+    Learn about advanced async patterns and optimizations
+  </Card>
+</CardGroup>
+
+<Note>
+  Remember to handle errors properly, manage resources efficiently, and test thoroughly in async contexts.
+</Note>

docs/features/callbacks.mdx

@@ -0,0 +1,782 @@
+---
+title: "Callbacks"
+description: "Learn how to implement callbacks to monitor and log AI agent interactions, errors, and task completions."
+icon: "bell"
+---
+
+<Note>
+  **Prerequisites**
+  - Python 3.8 or higher
+  - PraisonAI Agents package installed
+  - Basic understanding of Python functions
+</Note>
+
+## Quick Start
+
+<Steps>
+  <Step title="Install Package">
+    First, install the PraisonAI Agents package:
+    ```bash
+    pip install praisonaiagents
+    ```
+  </Step>
+
+  <Step title="Import Required Functions">
+    Import the necessary components:
+    ```python
+    from praisonaiagents import (
+        register_display_callback,
+        Agent, 
+        Task, 
+        PraisonAIAgents
+    )
+    ```
+  </Step>
+
+  <Step title="Create a Simple Callback">
+    Define your first callback function:
+    ```python
+    def simple_callback(message=None, response=None, **kwargs):
+        print(f"Received message: {message}")
+        print(f"Got response: {response}")
+    
+    # Register as synchronous callback
+    register_display_callback('interaction', simple_callback, is_async=False)
+
+    # For async callbacks
+    async def async_simple_callback(message=None, response=None, **kwargs):
+        await asyncio.sleep(0)  # Non-blocking pause
+        print(f"Received message: {message}")
+        print(f"Got response: {response}")
+    
+    # Register as async callback
+    register_display_callback('interaction', async_simple_callback, is_async=True)
+    ```
+  </Step>
+
+  <Step title="Use in Your Code">
+    Create and run an agent with callbacks enabled:
+    ```python
+    # Create an agent
+    agent = Agent(
+        name="MyAgent",
+        role="Assistant",
+        goal="Help with tasks",
+        backstory="I am a helpful assistant",
+        verbose=True  # Enable verbose mode to see callbacks in action
+    )
+
+    # Create a task
+    task = Task(
+        name="simple_task",
+        description="Say hello",
+        agent=agent,
+        expected_output="A greeting"
+    )
+
+    # Run the agent
+    agents = PraisonAIAgents(
+        agents=[agent],
+        tasks=[task]
+    )
+    agents.start()
+    ```
+  </Step>
+</Steps>
+
+<Note>
+  PraisonAI Agents supports both synchronous and asynchronous callbacks. When registering a callback, specify `is_async=True` for async callbacks. Both sync and async callbacks for the same display type can be executed simultaneously.
+</Note>
+
+## Understanding Callbacks
+
+<Card title="What are Callbacks?" icon="question">
+  Callbacks are functions that get called automatically when specific events occur in your AI agents:
+  - Interactions between user and agent
+  - Error messages
+  - Tool calls
+  - Self-reflection moments
+  - Task completion
+  - Generation progress
+</Card>
+
+## Available Callback Types
+
+<CardGroup cols={2}>
+  <Card title="Interaction Callback" icon="comments">
+    Triggered when the agent interacts with users
+  </Card>
+  <Card title="Error Callback" icon="triangle-exclamation">
+    Called when errors occur
+  </Card>
+  <Card title="Tool Call Callback" icon="wrench">
+    Activated when tools are used
+  </Card>
+  <Card title="Self Reflection Callback" icon="brain">
+    Triggered during agent self-reflection
+  </Card>
+  <Card title="Instruction Callback" icon="list-check">
+    Called when instructions are processed
+  </Card>
+  <Card title="Generating Callback" icon="spinner">
+    Activated during content generation
+  </Card>
+</CardGroup>
+
+## Basic Implementation
+
+### 1. Simple Logging Callback
+
+<CodeGroup>
+```python Basic
+import logging
+
+# Setup logging
+logging.basicConfig(level=logging.INFO)
+
+def log_callback(message=None, **kwargs):
+    logging.info(f"Agent message: {message}")
+
+# Register synchronous callback
+register_display_callback('interaction', log_callback, is_async=False)
+
+# Register asynchronous callback
+async def async_log_callback(message=None, **kwargs):
+    await asyncio.sleep(0)
+    logging.info(f"Agent message: {message}")
+
+# Register as async callback
+register_display_callback('interaction', async_log_callback, is_async=True)
+```
+
+```python Advanced
+import logging
+from datetime import datetime
+
+# Setup logging with file output
+logging.basicConfig(
+    filename='ai_interactions.log',
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s'
+)
+
+def detailed_callback(message=None, response=None, **kwargs):
+    logging.info(f"""
+    Time: {datetime.now()}
+    Message: {message}
+    Response: {response}
+    Additional Info: {kwargs}
+    """)
+
+register_display_callback('interaction', detailed_callback)
+```
+</CodeGroup>
+
+### 2. Multiple Callback Types
+
+```python
+# Error callback
+def error_callback(message=None):
+    logging.error(f"Error occurred: {message}")
+
+# Tool call callback
+def tool_callback(message=None):
+    logging.info(f"Tool called: {message}")
+
+# Register multiple callbacks
+register_display_callback('error', error_callback)
+register_display_callback('tool_call', tool_callback)
+```
+
+## Best Practices
+
+<AccordionGroup>
+  <Accordion title="Error Handling">
+    Always include error handling in your callbacks:
+    ```python
+    def safe_callback(message=None, **kwargs):
+        try:
+            # Process the callback
+            logging.info(message)
+        except Exception as e:
+            logging.error(f"Callback error: {e}")
+    ```
+  </Accordion>
+
+  <Accordion title="Performance">
+    Keep callbacks lightweight:
+    ```python
+    # Good - Quick logging
+    def good_callback(message=None):
+        logging.info(message)
+
+    # Bad - Heavy processing
+    def bad_callback(message=None):
+        time.sleep(1)  # Delays agent execution
+        process_heavy_task(message)
+    ```
+  </Accordion>
+</AccordionGroup>
+
+## Complete Example
+
+Here's a full implementation showing all callback types and proper logging:
+
+```python
+from praisonaiagents import Agent, Task, PraisonAIAgents, register_display_callback
+import logging
+from datetime import datetime
+
+# Setup logging
+logging.basicConfig(
+    filename='ai_interactions.log',
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+# Interaction callback
+def interaction_callback(message=None, response=None, markdown=None, generation_time=None):
+    logging.info(f"""
+    === INTERACTION ===
+    Time: {datetime.now()}
+    Generation Time: {generation_time}s
+    Message: {message}
+    Response: {response}
+    Markdown: {markdown}
+    """)
+
+# Error callback
+def error_callback(message=None):
+    logging.error(f"""
+    === ERROR ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+# Tool call callback
+def tool_call_callback(message=None):
+    logging.info(f"""
+    === TOOL CALL ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+# Register callbacks
+register_display_callback('interaction', interaction_callback)
+register_display_callback('error', error_callback)
+register_display_callback('tool_call', tool_call_callback)
+
+agent = Agent(
+    name="CallbackAgent",
+    role="Assistant",
+    goal="Demonstrate callbacks",
+    backstory="I am a helpful assistant",
+    verbose=True
+)
+
+task = Task(
+    name="callback_task",
+    description="Show how callbacks work",
+    agent=agent,
+    expected_output="Demonstration complete"
+)
+
+agents = PraisonAIAgents(
+    agents=[agent],
+    tasks=[task],
+    verbose=True
+)
+
+agents.start()
+```
+
+## Advanced Examples
+
+- All callback types
+- Comprehensive logging
+- Task callbacks
+- Tool integration
+- Multiple agents
+
+```python
+from praisonaiagents import Agent, Task, PraisonAIAgents, error_logs, register_display_callback
+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()
+
+## Async Callbacks
+
+Async callbacks allow you to handle events asynchronously, which is particularly useful for long-running operations or when dealing with multiple agents simultaneously.
+
+### Basic Async Callback Implementation
+
+```python
+import asyncio
+from praisonaiagents import register_display_callback
+
+async def async_interaction_callback(message=None, response=None, **kwargs):
+    """Async callback for handling interactions"""
+    await asyncio.sleep(0)  # Non-blocking pause
+    print(f"Async processing - Message: {message}")
+    print(f"Response: {response}")
+
+# Register the async callback
+register_display_callback('interaction', async_interaction_callback)
+```
+
+### Complete Async Example
+
+```python
+import asyncio
+from praisonaiagents import Agent, Task, PraisonAIAgents, register_display_callback
+import logging
+from datetime import datetime
+
+# Setup async logging
+async def setup_async_logging():
+    logging.basicConfig(
+        filename='async_ai_interactions.log',
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+
+# Async callbacks
+async def async_interaction_callback(message=None, response=None, markdown=None, generation_time=None):
+    await asyncio.sleep(0)
+    logging.info(f"""
+    === ASYNC INTERACTION ===
+    Time: {datetime.now()}
+    Generation Time: {generation_time}s
+    Message: {message}
+    Response: {response}
+    """)
+
+async def async_error_callback(message=None):
+    await asyncio.sleep(0)
+    logging.error(f"""
+    === ASYNC ERROR ===
+    Time: {datetime.now()}
+    Message: {message}
+    """)
+
+# Register async callbacks
+register_display_callback('interaction', async_interaction_callback, is_async=True)
+register_display_callback('error', async_error_callback, is_async=True)
+
+# Create async task callback
+async def async_task_callback(output):
+    await asyncio.sleep(0)
+    logging.info(f"""
+    === ASYNC TASK COMPLETED ===
+    Time: {datetime.now()}
+    Description: {output.description}
+    Agent: {output.agent}
+    Output: {output.raw[:200]}...
+    """)
+
+async def main():
+    await setup_async_logging()
+    
+    # Create agent with async capabilities
+    agent = Agent(
+        name="AsyncAgent",
+        role="Assistant",
+        goal="Demonstrate async callbacks",
+        backstory="I am an async-capable assistant",
+        verbose=True
+    )
+
+    # Create task with async callback
+    task = Task(
+        name="async_task",
+        description="Demonstrate async callbacks",
+        agent=agent,
+        expected_output="Async demonstration complete",
+        callback=async_task_callback
+    )
+
+    # Create and run agents with async support
+    agents = PraisonAIAgents(
+        agents=[agent],
+        tasks=[task],
+        verbose=True
+    )
+
+    await agents.start_async()
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+### Available Async Functions
+
+PraisonAI Agents provides several async versions of display functions, prefixed with 'a'. Here's the complete list:
+
+<CardGroup cols={2}>
+  <Card title="adisplay_interaction" icon="comments">
+    ```python
+    async def adisplay_interaction(
+        message, 
+        response, 
+        markdown=True, 
+        generation_time=None, 
+        console=None
+    )
+    ```
+    Async version of display_interaction for showing conversations.
+  </Card>
+  
+  <Card title="adisplay_self_reflection" icon="brain">
+    ```python
+    async def adisplay_self_reflection(
+        message: str, 
+        console=None
+    )
+    ```
+    Async version for displaying agent's self-reflection moments.
+  </Card>
+
+  <Card title="adisplay_instruction" icon="list-check">
+    ```python
+    async def adisplay_instruction(
+        message: str, 
+        console=None
+    )
+    ```
+    Async version for showing instructions.
+  </Card>
+
+  <Card title="adisplay_tool_call" icon="wrench">
+    ```python
+    async def adisplay_tool_call(
+        message: str, 
+        console=None
+    )
+    ```
+    Async version for displaying tool calls.
+  </Card>
+
+  <Card title="adisplay_error" icon="triangle-exclamation">
+    ```python
+    async def adisplay_error(
+        message: str, 
+        console=None
+    )
+    ```
+    Async version for error messages.
+  </Card>
+
+  <Card title="adisplay_generating" icon="spinner">
+    ```python
+    async def adisplay_generating(
+        content: str = "", 
+        start_time: Optional[float] = None
+    )
+    ```
+    Async version for showing generation progress.
+  </Card>
+</CardGroup>
+
+#### Example Usage
+
+```python
+import asyncio
+from praisonaiagents import (
+    adisplay_interaction,
+    adisplay_error,
+    adisplay_tool_call
+)
+
+async def main():
+    # Display an interaction
+    await adisplay_interaction(
+        message="What's the weather?",
+        response="Let me check that for you.",
+        generation_time=0.5
+    )
+
+    # Display a tool call
+    await adisplay_tool_call(
+        "Calling weather API for location data..."
+    )
+
+    # Handle an error
+    try:
+        raise Exception("API connection failed")
+    except Exception as e:
+        await adisplay_error(str(e))
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+### Best Practices for Async Callbacks
+
+<AccordionGroup>
+  <Accordion title="Error Handling">
+    Implement proper error handling in async callbacks:
+    ```python
+    async def safe_async_callback(message=None, **kwargs):
+        try:
+            await asyncio.sleep(0)
+            # Process the callback
+            await process_message(message)
+        except Exception as e:
+            logging.error(f"Async callback error: {e}")
+    ```
+  </Accordion>
+
+  <Accordion title="Performance">
+    Optimize async callbacks for better performance:
+    ```python
+    async def optimized_async_callback(message=None, **kwargs):
+        # Use asyncio.gather for concurrent operations
+        results = await asyncio.gather(
+            process_message(message),
+            update_status(),
+            log_interaction(message)
+        )
+        return results
+    ```
+  </Accordion>
+</AccordionGroup>
+
+### Key Considerations
+
+<CardGroup cols={2}>
+  <Card title="Concurrency" icon="arrows-split-up-and-left">
+    - Use asyncio.gather for parallel operations
+    - Avoid blocking calls in async callbacks
+    - Handle concurrent access to shared resources
+  </Card>
+  <Card title="Resource Management" icon="server">
+    - Implement proper cleanup in async contexts
+    - Use async context managers when appropriate
+    - Monitor memory usage in long-running callbacks
+  </Card>
+</CardGroup>
+
+<Note>
+  When using async callbacks, ensure all dependent operations are also async-compatible to prevent blocking the event loop.
+</Note>
+
+## Troubleshooting
+
+<CardGroup cols={2}>
+  <Card title="Callback Not Firing" icon="bug">
+    - Verify callback registration
+    - Check verbose mode is enabled
+    - Ensure correct callback name
+  </Card>
+  <Card title="Logging Issues" icon="file-lines">
+    - Check file permissions
+    - Verify logging configuration
+    - Monitor log file size
+  </Card>
+</CardGroup>
+
+## Next Steps
+
+<CardGroup cols={2}>
+  <Card title="Async Callbacks" icon="bolt" href="./async">
+    Learn about asynchronous callbacks
+  </Card>
+  <Card title="Advanced Patterns" icon="code" href="./advanced">
+    Explore advanced callback patterns
+  </Card>
+</CardGroup>
+
+<Note>
+  Remember to handle callbacks efficiently and implement proper error handling to ensure smooth agent operations.
+</Note>

docs/mint.json

@@ -101,11 +101,13 @@
       "pages": [
         "features/selfreflection",
         "features/reasoning",
+        "framework/multimodalagents",
+        "features/async",
+        "features/callbacks",
         "train",
         "framework/praisonaiagents",
         "framework/crewai",
-        "framework/autogen",
-        "framework/multimodalagents"
+        "framework/autogen"
       ]
     },
     {

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.23.tar.gz"
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/2.0.25.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.23 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.0.25 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

praisonai/ui/agents.py

@@ -5,7 +5,7 @@ import chainlit as cl
 import os
 from chainlit.types import ThreadDict
 from chainlit.input_widget import Select, TextInput
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, AsyncGenerator, List, Callable
 from dotenv import load_dotenv
 from datetime import datetime
 import json
@@ -18,6 +18,9 @@ from contextlib import redirect_stdout, asynccontextmanager
 from db import DatabaseManager
 import time
 import sqlite3
+from openai import AsyncOpenAI
+from functools import partial
+import yaml
 
 # Load environment variables
 load_dotenv()
@@ -354,37 +357,193 @@ async def on_chat_resume(thread: ThreadDict):
 # async def tool(data: Optional[str] = None, language: Optional[str] = None):
 #     return cl.Message(content=data, language=language)
 
+# Add callback handler class
+class ChainlitCallbackHandler:
+    """Callback handler for streaming agent execution to Chainlit"""
+    
+    def __init__(self, parent_step: Optional[cl.Step] = None):
+        self.parent_step = parent_step
+        self.current_step = None
+        self._steps = {}
+    
+    async def on_agent_start(self, agent_name: str):
+        """Called when an agent starts execution"""
+        self.current_step = cl.Step(
+            name=f"Agent: {agent_name}",
+            type="agent",
+            show_input=True,
+            parent_id=self.parent_step.id if self.parent_step else None
+        )
+        await self.current_step.start()
+        await self.current_step.stream_token(f"🤖 Agent {agent_name} started\n")
+        self._steps[agent_name] = self.current_step
+
+    async def on_agent_action(self, agent_name: str, action: str):
+        """Called when an agent performs an action"""
+        step = self._steps.get(agent_name, self.current_step)
+        if step:
+            await step.stream_token(f"⚡ {action}\n")
+
+    async def on_agent_finish(self, agent_name: str, output: Any):
+        """Called when an agent finishes execution"""
+        step = self._steps.get(agent_name)
+        if step:
+            await step.stream_token(f"\n✅ Agent {agent_name} finished\n")
+            step.output = str(output)
+            await step.end()
+            self._steps.pop(agent_name, None)
+
+    async def on_task_start(self, task_id: str, task_name: str):
+        """Called when a task starts execution"""
+        self.current_step = cl.Step(
+            name=f"Task: {task_name}",
+            type="task",
+            show_input=True,
+            parent_id=self.parent_step.id if self.parent_step else None
+        )
+        await self.current_step.start()
+        await self.current_step.stream_token(f"📋 Starting task: {task_name}\n")
+        self._steps[task_id] = self.current_step
+
+    async def on_task_finish(self, task_id: str, output: Any):
+        """Called when a task finishes execution"""
+        step = self._steps.get(task_id, self.current_step)
+        if step:
+            await step.stream_token(f"\n✅ Task completed\n")
+            step.output = str(output)
+            await step.end()
+            self._steps.pop(task_id, None)
+
+    async def on_error(self, error: str):
+        """Called when an error occurs"""
+        if self.current_step:
+            await self.current_step.stream_token(f"\n❌ Error: {error}\n")
+            await self.current_step.end()
+
 @cl.step(type="tool", show_input=False)
 async def run_agents(agent_file: str, framework: str):
     """Runs the agents and returns the result."""
     try:
         logger.debug(f"Running agents with file: {agent_file}, framework: {framework}")
-        agents_generator = AgentsGenerator(agent_file, framework, config_list)
-        current_step = cl.context.current_step
-        logger.debug(f"Current Step: {current_step}")
-
-        stdout_buffer = StringIO()
-        with redirect_stdout(stdout_buffer):
-            result = agents_generator.generate_crew_and_kickoff()
+        
+        # Create main execution step
+        async with cl.Step(name="Agents Execution", type="agents") as agents_step:
+            agents_step.input = f"Running agents from {agent_file}"
             
-        complete_output = stdout_buffer.getvalue()
-        logger.debug(f"Agent execution output: {complete_output}")
-
-        async with cl.Step(name="gpt4", type="llm", show_input=True) as step:
-            step.input = ""
+            # Initialize callback handler
+            callback_handler = ChainlitCallbackHandler(parent_step=agents_step)
             
-            for line in stdout_buffer.getvalue().splitlines():
-                logger.debug(f"Agent output line: {line}")
-                await step.stream_token(line)
+            try:
+                # Load YAML config first
+                with open(agent_file, 'r') as f:
+                    config = yaml.safe_load(f)
                 
-            tool_res = await output(complete_output)
+                # Get topic from message content
+                topic = cl.user_session.get("message_history", [{}])[-1].get("content", "")
+                
+                # Create agents generator with loaded config
+                agents_generator = AgentsGenerator(
+                    agent_file=agent_file,
+                    framework=framework,
+                    config_list=config_list,
+                    agent_yaml=yaml.dump(config)  # Pass the loaded config as YAML string
+                )
+                
+                # Execute based on framework
+                if framework == "crewai":
+                    result = agents_generator._run_crewai(config, topic, [])
+                elif framework == "autogen":
+                    result = agents_generator._run_autogen(config, topic, [])
+                elif framework == "praisonai":
+                    result = agents_generator._run_praisonai(config, topic, [])
+                else:
+                    raise ValueError(f"Unsupported framework: {framework}")
+                
+                # Process the result if it has tasks
+                if hasattr(result, 'tasks') and result.tasks:
+                    for task in result.tasks:
+                        task_id = getattr(task, 'id', str(id(task)))
+                        task_desc = getattr(task, 'description', 'Executing task...')
+                        
+                        # Signal task start
+                        await callback_handler.on_task_start(
+                            task_id,
+                            task_desc[:50] + "..." if len(task_desc) > 50 else task_desc
+                        )
+                        
+                        try:
+                            # Handle agent actions if present
+                            agent = getattr(task, 'agent', None)
+                            if agent:
+                                agent_name = getattr(agent, 'name', 'Unknown Agent')
+                                await callback_handler.on_agent_start(agent_name)
+                                await callback_handler.on_agent_action(
+                                    agent_name,
+                                    f"Working on task: {task_desc[:50]}..."
+                                )
+                            
+                            # Get task output
+                            task_output = getattr(task, 'output', str(task))
+                            
+                            # Signal agent completion if exists
+                            if agent:
+                                await callback_handler.on_agent_finish(agent_name, task_output)
+                            
+                            # Signal task completion
+                            await callback_handler.on_task_finish(task_id, task_output)
+                            
+                        except Exception as e:
+                            await callback_handler.on_error(f"Error in task {task_id}: {str(e)}")
+                            raise
+                
+                # Return the final result
+                agents_step.output = "Agents execution completed"
+                return result if isinstance(result, str) else str(result)
+                
+            except Exception as e:
+                await callback_handler.on_error(str(e))
+                raise
 
-        return result
     except Exception as e:
         error_msg = f"Error running agents: {str(e)}"
         logger.error(error_msg)
         raise Exception(error_msg)
 
+async def stream_agents_execution(agents) -> AsyncGenerator[tuple[str, str, str], None]:
+    """
+    Generator to stream agents execution status and messages.
+    Yields tuples of (task_id, status, message)
+    """
+    try:
+        for task_id in agents.tasks:
+            task = agents.tasks[task_id]
+            
+            # Signal task start
+            yield task_id, "start", ""
+            
+            if task.async_execution:
+                # Execute async task
+                result = await agents.aexecute_task(task_id)
+            else:
+                # Execute sync task in thread pool
+                result = await cl.make_async(agents.execute_task)(task_id)
+            
+            if result:
+                # Stream agent messages
+                if isinstance(result, str):
+                    yield task_id, "agent_message", result
+                else:
+                    yield task_id, "agent_message", result.raw
+                
+                # Signal completion
+                yield task_id, "complete", ""
+            else:
+                yield task_id, "error", "Task execution failed"
+                
+    except Exception as e:
+        logger.error(f"Error in stream_agents_execution: {e}")
+        yield task_id, "error", str(e)
+
 @cl.step(type="tool", show_input=False, language="yaml")
 async def output(output):
     return output
@@ -405,17 +564,38 @@ def task(output):
         {output}
     """)
 
+# Add retry decorator for database operations
+def with_retries(max_retries=3, delay=1):
+    def decorator(func):
+        async def wrapper(*args, **kwargs):
+            for attempt in range(max_retries):
+                try:
+                    return await func(*args, **kwargs)
+                except sqlite3.OperationalError as e:
+                    if "database is locked" in str(e) and attempt < max_retries - 1:
+                        await asyncio.sleep(delay)
+                        continue
+                    raise
+            return await func(*args, **kwargs)
+        return wrapper
+    return decorator
+
+@with_retries(max_retries=3, delay=1)
+async def update_thread_metadata(thread_id: str, metadata: dict):
+    """Update thread metadata with retry logic"""
+    await cl_data.update_thread(thread_id, metadata=metadata)
+
 @cl.on_message
 async def main(message: cl.Message):
-    """Run PraisonAI with the provided message as the topic."""
     try:
-        # Get or initialize message history
-        message_history = cl.user_session.get("message_history")
-        if message_history is None:
-            message_history = []
-            cl.user_session.set("message_history", message_history)
+        # Get settings and chat profile
+        settings = cl.user_session.get("settings")
+        chat_profile = cl.user_session.get("chat_profile")
         
-        # Add current message to history
+        # Get message history or initialize if not exists
+        message_history = cl.user_session.get("message_history", [])
+        
+        # Format user message with context
         now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         user_message = f"""
         Answer the question and use tools if needed:
@@ -424,54 +604,27 @@ async def main(message: cl.Message):
 
         User Question: {message.content}
         """
+        
+        # Add to message history
         message_history.append({"role": "user", "content": user_message})
         
-        # Get chat profile and process accordingly
+        # Get configuration
+        framework = settings["Framework"]
         topic = message.content
-        chat_profile = cl.user_session.get("chat_profile")
-        logger.debug(f"Processing message with chat profile: {chat_profile}")
 
         if chat_profile == "Auto":
             agent_file = "agents.yaml"
             logger.info(f"Generating agents for topic: {topic}")
             generator = AutoGenerator(topic=topic, agent_file=agent_file, framework=framework, config_list=config_list)
+            
             await cl.sleep(2)
             agent_file = generator.generate()
             
-            logger.debug("Starting agents execution")
-            agents_generator = AgentsGenerator(
-                agent_file, 
-                framework, 
-                config_list
-            )
+            # Run agents with streaming while preserving context
+            result = await run_agents(agent_file, framework)
+            await cl.Message(content=result).send()
             
-            # Capture stdout
-            stdout_buffer = StringIO()
-            with redirect_stdout(stdout_buffer):
-                result = agents_generator.generate_crew_and_kickoff()
-                
-            complete_output = stdout_buffer.getvalue()
-            logger.debug(f"Agents execution output: {complete_output}")
-            
-            tool_res = await output(complete_output)
-            msg = cl.Message(content=result)
-            await msg.send()
-            
-            # Save to message history
-            message_history.append({"role": "assistant", "content": result})
-            cl.user_session.set("message_history", message_history)
-            
-            # Update thread metadata if exists
-            thread_id = cl.user_session.get("thread_id")
-            if thread_id:
-                metadata = {
-                    "last_response": result,
-                    "timestamp": now,
-                    "mode": "auto"
-                }
-                await cl_data.update_thread(thread_id, metadata=metadata)
-                
-        else:  # chat_profile == "Manual"
+        else:  # Manual mode
             agent_file = "agents.yaml"
             full_agent_file_path = os.path.abspath(agent_file)
             full_tools_file_path = os.path.abspath("tools.py")
@@ -487,31 +640,25 @@ async def main(message: cl.Message):
                         tools_content = f.read()
                     msg_tools = cl.Message(content=tools_content, language="python")
                     await msg_tools.send()
-            else:
-                logger.info("Generating agents for manual mode")
-                generator = AutoGenerator(topic=topic, agent_file=agent_file, framework=framework, config_list=config_list)
-                agent_file = generator.generate()
+            
+            # Run agents with streaming while preserving context
+            result = await run_agents(agent_file, framework)
+            await cl.Message(content=result, actions=actions).send()
+
+        # Update message history
+        message_history.append({"role": "assistant", "content": result})
+        cl.user_session.set("message_history", message_history)
+        
+        # Update thread metadata with retry logic
+        thread_id = cl.user_session.get("thread_id")
+        if thread_id:
+            metadata = {
+                "last_response": result,
+                "timestamp": now,
+                "mode": chat_profile.lower()
+            }
+            await update_thread_metadata(thread_id, metadata)
 
-            logger.debug("Starting agents execution for manual mode")
-            agents_generator = AgentsGenerator(agent_file, framework, config_list)
-            result = agents_generator.generate_crew_and_kickoff()
-            msg = cl.Message(content=result, actions=actions)
-            await msg.send()
-            
-            # Save to message history
-            message_history.append({"role": "assistant", "content": result})
-            cl.user_session.set("message_history", message_history)
-            
-            # Update thread metadata if exists
-            thread_id = cl.user_session.get("thread_id")
-            if thread_id:
-                metadata = {
-                    "last_response": result,
-                    "timestamp": now,
-                    "mode": "manual"
-                }
-                await cl_data.update_thread(thread_id, metadata=metadata)
-                
     except Exception as e:
         error_msg = f"Error processing message: {str(e)}"
         logger.error(error_msg)

praisonai/ui/callbacks.py

@@ -0,0 +1,57 @@
+import logging
+from typing import Dict, Any, Callable, Optional, Union
+import asyncio
+
+logger = logging.getLogger(__name__)
+
+class CallbackManager:
+    """Manages callbacks for the PraisonAI UI"""
+    
+    def __init__(self):
+        self._callbacks: Dict[str, Dict[str, Union[Callable, bool]]] = {}
+    
+    def register(self, name: str, callback: Callable, is_async: bool = False) -> None:
+        """Register a callback function"""
+        self._callbacks[name] = {
+            'func': callback,
+            'is_async': is_async
+        }
+    
+    async def call(self, name: str, **kwargs) -> None:
+        """Call a registered callback"""
+        if name not in self._callbacks:
+            logger.warning(f"No callback registered for {name}")
+            return
+            
+        callback_info = self._callbacks[name]
+        func = callback_info['func']
+        is_async = callback_info['is_async']
+        
+        try:
+            if is_async:
+                await func(**kwargs)
+            else:
+                if asyncio.iscoroutinefunction(func):
+                    await func(**kwargs)
+                else:
+                    await asyncio.get_event_loop().run_in_executor(None, lambda: func(**kwargs))
+        except Exception as e:
+            logger.error(f"Error in callback {name}: {str(e)}")
+
+# Global callback manager instance
+callback_manager = CallbackManager()
+
+def register_callback(name: str, callback: Callable, is_async: bool = False) -> None:
+    """Register a callback with the global callback manager"""
+    callback_manager.register(name, callback, is_async)
+
+async def trigger_callback(name: str, **kwargs) -> None:
+    """Trigger a callback from the global callback manager"""
+    await callback_manager.call(name, **kwargs)
+
+# Decorator for registering callbacks
+def callback(name: str, is_async: bool = False):
+    def decorator(func):
+        register_callback(name, func, is_async)
+        return func
+    return decorator 

praisonai/ui/colab.py

@@ -0,0 +1,474 @@
+from praisonaiagents import Agent, Task, PraisonAIAgents
+import os
+import importlib
+import inspect
+import yaml
+import logging
+from .callbacks import trigger_callback
+import asyncio
+import chainlit as cl
+from queue import Queue
+
+logger = logging.getLogger(__name__)
+agent_file = "agents.yaml"
+
+with open(agent_file, 'r') as f:
+    config = yaml.safe_load(f)
+
+topic = "get from the message content from the chainlit user message"
+
+# Create a message queue
+message_queue = Queue()
+
+async def process_message_queue():
+    """Process messages in the queue and send them to Chainlit"""
+    while True:
+        try:
+            if not message_queue.empty():
+                msg_data = message_queue.get()
+                await cl.Message(**msg_data).send()
+            await asyncio.sleep(0.1)  # Small delay to prevent busy waiting
+        except Exception as e:
+            logger.error(f"Error processing message queue: {e}")
+
+def load_tools_from_tools_py():
+    """
+    Imports and returns all contents from tools.py file.
+    Also adds the tools to the global namespace.
+
+    Returns:
+        list: A list of callable functions with proper formatting
+    """
+    tools_list = []
+    try:
+        # Try to import tools.py from current directory
+        spec = importlib.util.spec_from_file_location("tools", "tools.py")
+        logger.info(f"Spec: {spec}")
+        if spec is None:
+            logger.info("tools.py not found in current directory")
+            return tools_list
+
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+
+        # Get all module attributes except private ones and classes
+        for name, obj in inspect.getmembers(module):
+            if (not name.startswith('_') and 
+                callable(obj) and 
+                not inspect.isclass(obj)):
+                # Add the function to global namespace
+                globals()[name] = obj
+                # Format the tool as an OpenAI function
+                tool = {
+                    "type": "function",
+                    "function": {
+                        "name": name,
+                        "description": obj.__doc__ or f"Function to {name.replace('_', ' ')}",
+                        "parameters": {
+                            "type": "object",
+                            "properties": {
+                                "query": {
+                                    "type": "string",
+                                    "description": "The search query to look up information about"
+                                }
+                            },
+                            "required": ["query"]
+                        }
+                    }
+                }
+                # Add formatted tool to tools list
+                tools_list.append(tool)
+                logger.info(f"Loaded and globalized tool function: {name}")
+
+        logger.info(f"Loaded {len(tools_list)} tool functions from tools.py")
+        logger.info(f"Tools list: {tools_list}")
+        
+    except Exception as e:
+        logger.warning(f"Error loading tools from tools.py: {e}")
+        
+    return tools_list
+
+async def step_callback(step_details):
+    """Callback for agent steps"""
+    logger.info(f"[CALLBACK DEBUG] Step callback triggered with details: {step_details}")
+    try:
+        # Queue message for agent response
+        if step_details.get("response"):
+            message_queue.put({
+                "content": f"Agent Response: {step_details.get('response')}",
+                "author": step_details.get("agent_name", "Agent")
+            })
+            logger.info("[CALLBACK DEBUG] Queued agent response message")
+        
+        # Queue message for tool usage
+        if step_details.get("tool_name"):
+            message_queue.put({
+                "content": f"🛠️ Using tool: {step_details.get('tool_name')}",
+                "author": "System"
+            })
+            logger.info("[CALLBACK DEBUG] Queued tool usage message")
+            
+    except Exception as e:
+        logger.error(f"[CALLBACK DEBUG] Error in step callback: {str(e)}", exc_info=True)
+
+async def task_callback(task_output):
+    """Callback for task completion"""
+    logger.info(f"[CALLBACK DEBUG] Task callback triggered with output: {task_output}")
+    try:
+        # Create message content
+        if hasattr(task_output, 'raw'):
+            content = task_output.raw
+        elif hasattr(task_output, 'content'):
+            content = task_output.content
+        else:
+            content = str(task_output)
+            
+        # Queue the message
+        message_queue.put({
+            "content": f"Task Output: {content}",
+            "author": "Task"
+        })
+        logger.info("[CALLBACK DEBUG] Queued task completion message")
+        
+    except Exception as e:
+        logger.error(f"[CALLBACK DEBUG] Error in task callback: {str(e)}", exc_info=True)
+
+async def step_callback_wrapper(step_details):
+    logger.info(f"[CALLBACK DEBUG] Step callback wrapper triggered with details: {step_details}")
+    try:
+        # Check if we have a Chainlit context
+        if not cl.context.context_var.get():
+            logger.warning("[CALLBACK DEBUG] No Chainlit context available in wrapper")
+            return
+        logger.info("[CALLBACK DEBUG] Chainlit context found in wrapper")
+
+        # Create a message for the agent's response
+        if step_details.get("response"):
+            logger.info(f"[CALLBACK DEBUG] Sending agent response from wrapper: {step_details.get('response')}")
+            try:
+                await cl.Message(
+                    content=f"{role_name}: {step_details.get('response')}",
+                    author=role_name,
+                ).send()
+                logger.info("[CALLBACK DEBUG] Successfully sent agent response message from wrapper")
+            except Exception as e:
+                logger.error(f"[CALLBACK DEBUG] Error sending agent response message from wrapper: {str(e)}")
+
+        # Create a message for any tool usage
+        if step_details.get("tool_name"):
+            logger.info(f"[CALLBACK DEBUG] Sending tool usage from wrapper: {step_details.get('tool_name')}")
+            try:
+                await cl.Message(
+                    content=f"🛠️ {role_name} is using tool: {step_details.get('tool_name')}",
+                    author="System",
+                ).send()
+                logger.info("[CALLBACK DEBUG] Successfully sent tool usage message from wrapper")
+            except Exception as e:
+                logger.error(f"[CALLBACK DEBUG] Error sending tool usage message from wrapper: {str(e)}")
+
+        # Create a message for any thoughts or reasoning
+        if step_details.get("thought"):
+            logger.info(f"[CALLBACK DEBUG] Sending thought from wrapper: {step_details.get('thought')}")
+            try:
+                await cl.Message(
+                    content=f"💭 {role_name}'s thought: {step_details.get('thought')}",
+                    author=role_name,
+                ).send()
+                logger.info("[CALLBACK DEBUG] Successfully sent thought message from wrapper")
+            except Exception as e:
+                logger.error(f"[CALLBACK DEBUG] Error sending thought message from wrapper: {str(e)}")
+
+    except Exception as e:
+        logger.error(f"[CALLBACK DEBUG] Error in step callback wrapper: {str(e)}", exc_info=True)
+        try:
+            await cl.Message(
+                content=f"Error in step callback: {str(e)}",
+                author="System",
+            ).send()
+        except Exception as send_error:
+            logger.error(f"[CALLBACK DEBUG] Error sending error message: {str(send_error)}")
+
+async def task_callback_wrapper(task_output):
+    logger.info(f"[CALLBACK DEBUG] Task callback wrapper triggered with output type: {type(task_output)}")
+    try:
+        # Check if we have a Chainlit context
+        if not cl.context.context_var.get():
+            logger.warning("[CALLBACK DEBUG] No Chainlit context available in task wrapper")
+            return
+        logger.info("[CALLBACK DEBUG] Chainlit context found in task wrapper")
+
+        # Create a message for task completion
+        if hasattr(task_output, 'raw'):
+            content = task_output.raw
+            logger.info("[CALLBACK DEBUG] Using raw output")
+        elif hasattr(task_output, 'content'):
+            content = task_output.content
+            logger.info("[CALLBACK DEBUG] Using content output")
+        else:
+            content = str(task_output)
+            logger.info("[CALLBACK DEBUG] Using string representation of output")
+            
+        logger.info(f"[CALLBACK DEBUG] Sending task completion message from wrapper: {content[:100]}...")
+        try:
+            await cl.Message(
+                content=f"✅ {role_name} completed task:\n{content}",
+                author=role_name,
+            ).send()
+            logger.info("[CALLBACK DEBUG] Successfully sent task completion message from wrapper")
+        except Exception as e:
+            logger.error(f"[CALLBACK DEBUG] Error sending task completion message from wrapper: {str(e)}")
+
+        # If there are any additional task details
+        if hasattr(task_output, 'details'):
+            logger.info("[CALLBACK DEBUG] Task has additional details")
+            try:
+                await cl.Message(
+                    content=f"📝 Additional details:\n{task_output.details}",
+                    author=role_name,
+                ).send()
+                logger.info("[CALLBACK DEBUG] Successfully sent additional details message")
+            except Exception as e:
+                logger.error(f"[CALLBACK DEBUG] Error sending additional details message: {str(e)}")
+
+    except Exception as e:
+        logger.error(f"[CALLBACK DEBUG] Error in task callback wrapper: {str(e)}", exc_info=True)
+        try:
+            await cl.Message(
+                content=f"Error in task callback: {str(e)}",
+                author="System",
+            ).send()
+        except Exception as send_error:
+            logger.error(f"[CALLBACK DEBUG] Error sending error message: {str(send_error)}")
+
+def sync_task_callback_wrapper(task_output):
+    logger.info("[CALLBACK DEBUG] Sync task callback wrapper triggered")
+    try:
+        # Create a new event loop for this thread if there isn't one
+        try:
+            loop = asyncio.get_event_loop()
+            logger.info("[CALLBACK DEBUG] Got existing event loop")
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            logger.info("[CALLBACK DEBUG] Created new event loop")
+        
+        if loop.is_running():
+            # If loop is running, schedule the callback
+            logger.info("[CALLBACK DEBUG] Loop is running, scheduling callback")
+            asyncio.run_coroutine_threadsafe(
+                task_callback_wrapper(task_output),
+                loop
+            )
+        else:
+            # If loop is not running, run it directly
+            logger.info("[CALLBACK DEBUG] Loop is not running, running callback directly")
+            loop.run_until_complete(task_callback_wrapper(task_output))
+            
+    except Exception as e:
+        logger.error(f"[CALLBACK DEBUG] Error in sync task callback: {str(e)}", exc_info=True)
+
+def sync_step_callback_wrapper(step_details):
+    logger.info("[CALLBACK DEBUG] Sync step callback wrapper triggered")
+    try:
+        # Create a new event loop for this thread if there isn't one
+        try:
+            loop = asyncio.get_event_loop()
+            logger.info("[CALLBACK DEBUG] Got existing event loop")
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            logger.info("[CALLBACK DEBUG] Created new event loop")
+        
+        if loop.is_running():
+            # If loop is running, schedule the callback
+            logger.info("[CALLBACK DEBUG] Loop is running, scheduling callback")
+            asyncio.run_coroutine_threadsafe(
+                step_callback_wrapper(step_details),
+                loop
+            )
+        else:
+            # If loop is not running, run it directly
+            logger.info("[CALLBACK DEBUG] Loop is not running, running callback directly")
+            loop.run_until_complete(step_callback_wrapper(step_details))
+            
+    except Exception as e:
+        logger.error(f"[CALLBACK DEBUG] Error in sync step callback: {str(e)}", exc_info=True)
+
+async def ui_run_praisonai(config, topic, tools_dict):
+    """Run PraisonAI with the given configuration and topic."""
+    logger = logging.getLogger(__name__)
+    logger.setLevel(logging.DEBUG)
+    agents = {}
+    tasks = []
+    tasks_dict = {}
+
+    try:
+        # Start message queue processor
+        queue_processor = asyncio.create_task(process_message_queue())
+        
+        # Create agents for each role
+        for role, details in config['roles'].items():
+            # Format the role name and other details
+            role_name = details.get('name', role).format(topic=topic)
+            role_filled = details.get('role', role).format(topic=topic)
+            goal_filled = details['goal'].format(topic=topic)
+            backstory_filled = details['backstory'].format(topic=topic)
+
+            # Test message to verify Chainlit is working
+            await cl.Message(
+                content=f"[DEBUG] Creating agent: {role_name}",
+                author="System"
+            ).send()
+
+            # Create a sync wrapper for the step callback
+            def step_callback_sync(step_details):
+                try:
+                    # Create a new event loop for this thread
+                    loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(loop)
+                    
+                    # Add agent name to step details
+                    step_details["agent_name"] = role_name
+                    
+                    # Run the callback
+                    loop.run_until_complete(step_callback(step_details))
+                    loop.close()
+                except Exception as e:
+                    logger.error(f"[CALLBACK DEBUG] Error in step callback: {str(e)}", exc_info=True)
+
+            agent = Agent(
+                name=role_name,
+                role=role_filled,
+                goal=goal_filled,
+                backstory=backstory_filled,
+                llm=details.get('llm', 'gpt-4o'),
+                verbose=True,
+                allow_delegation=details.get('allow_delegation', False),
+                max_iter=details.get('max_iter', 15),
+                max_rpm=details.get('max_rpm'),
+                max_execution_time=details.get('max_execution_time'),
+                cache=details.get('cache', True),
+                step_callback=step_callback_sync
+            )
+            agents[role] = agent
+
+        # Create tasks for each role
+        for role, details in config['roles'].items():
+            agent = agents[role]
+            tools_list = []
+            
+            # Get tools for this role
+            for tool_name in details.get('tools', []):
+                if tool_name in tools_dict:
+                    tool_func = tools_dict[tool_name]
+                    tools_list.append(tool_func)
+
+            # Create tasks for the agent
+            for task_name, task_details in details.get('tasks', {}).items():
+                description_filled = task_details['description'].format(topic=topic)
+                expected_output_filled = task_details['expected_output'].format(topic=topic)
+
+                # Test message to verify task creation
+                await cl.Message(
+                    content=f"[DEBUG] Created task: {task_name} for agent {role_name}",
+                    author="System"
+                ).send()
+
+                # Create a sync wrapper for the task callback
+                def task_callback_sync(task_output):
+                    try:
+                        # Create a new event loop for this thread
+                        loop = asyncio.new_event_loop()
+                        asyncio.set_event_loop(loop)
+                        
+                        # Run the callback
+                        loop.run_until_complete(task_callback(task_output))
+                        loop.close()
+                    except Exception as e:
+                        logger.error(f"[CALLBACK DEBUG] Error in task callback: {str(e)}", exc_info=True)
+
+                task = Task(
+                    description=description_filled,
+                    expected_output=expected_output_filled,
+                    agent=agent,
+                    tools=tools_list,
+                    async_execution=True,
+                    context=[],
+                    config=task_details.get('config', {}),
+                    output_json=task_details.get('output_json'),
+                    output_pydantic=task_details.get('output_pydantic'),
+                    output_file=task_details.get('output_file', ""),
+                    callback=task_callback_sync,
+                    create_directory=task_details.get('create_directory', False)
+                )
+
+                tasks.append(task)
+                tasks_dict[task_name] = task
+
+        # Set up task contexts
+        for role, details in config['roles'].items():
+            for task_name, task_details in details.get('tasks', {}).items():
+                task = tasks_dict[task_name]
+                context_tasks = [tasks_dict[ctx] for ctx in task_details.get('context', []) 
+                            if ctx in tasks_dict]
+                task.context = context_tasks
+
+        # Send the start message
+        await cl.Message(
+            content="Starting PraisonAI agents execution...",
+            author="System"
+        ).send()
+
+        # Create and run the PraisonAI agents
+        if config.get('process') == 'hierarchical':
+            crew = PraisonAIAgents(
+                agents=list(agents.values()),
+                tasks=tasks,
+                verbose=True,
+                process="hierarchical",
+                manager_llm=config.get('manager_llm', 'gpt-4o')
+            )
+        else:
+            crew = PraisonAIAgents(
+                agents=list(agents.values()),
+                tasks=tasks,
+                verbose=2
+            )
+
+        # Store the crew in the user session
+        cl.user_session.set("crew", crew)
+
+        # Run the agents in a separate thread
+        loop = asyncio.get_event_loop()
+        response = await loop.run_in_executor(None, crew.start)
+        
+        logger.debug(f"[CALLBACK DEBUG] Result: {response}")
+        
+        # Convert response to string if it's not already
+        if hasattr(response, 'raw'):
+            result = response.raw
+        elif hasattr(response, 'content'):
+            result = response.content
+        else:
+            result = str(response)
+        
+        # Send the completion message
+        await cl.Message(
+            content="PraisonAI agents execution completed.",
+            author="System"
+        ).send()
+        
+        # After getting the response, wait a bit for remaining messages
+        await asyncio.sleep(1)  # Give time for final messages to be processed
+        queue_processor.cancel()  # Stop the queue processor
+        
+        return result
+        
+    except Exception as e:
+        error_msg = f"Error in ui_run_praisonai: {str(e)}"
+        logger.error(error_msg, exc_info=True)
+        await cl.Message(
+            content=error_msg,
+            author="System"
+        ).send()
+        raise

praisonai/ui/colab_chainlit.py

@@ -0,0 +1,81 @@
+import os
+import logging
+from dotenv import load_dotenv
+import chainlit as cl
+from chainlit.types import ThreadDict
+import yaml
+import sys
+import os
+from datetime import datetime
+
+# Add the parent directory to sys.path to allow imports
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+from praisonai.ui.colab import ui_run_praisonai, load_tools_from_tools_py
+from praisonai.ui.callbacks import callback, trigger_callback
+
+# Load environment variables
+load_dotenv()
+
+# Set up logging
+logger = logging.getLogger(__name__)
+log_level = os.getenv("LOGLEVEL", "INFO").upper()
+logger.setLevel(log_level)
+
+# Load agent configuration
+agent_file = "agents.yaml"
+with open(agent_file, 'r') as f:
+    config = yaml.safe_load(f)
+
+# Load tools
+tools_dict = load_tools_from_tools_py()
+
+@cl.on_message
+async def main(message: cl.Message):
+    """Main message handler for Chainlit"""
+    try:
+        logger.info(f"Processing message: {message.content}")
+        await cl.Message(
+            content=f"🔄 Processing your request about: {message.content}...",
+            author="System"
+        ).send()
+
+        await cl.Message(
+            content="Using Running PraisonAI Agents...",
+            author="System"
+        ).send()
+
+        # Run PraisonAI with the message content as the topic
+        result = await ui_run_praisonai(config, message.content, tools_dict)
+        
+        # Send the final result
+        await cl.Message(
+            content=result,
+            author="System"
+        ).send()
+
+    except Exception as e:
+        error_msg = f"Error running PraisonAI agents: {str(e)}"
+        logger.error(error_msg, exc_info=True)
+        await cl.Message(
+            content=error_msg,
+            author="System"
+        ).send()
+
+@cl.on_chat_start
+async def start():
+    """Handler for chat start"""
+    await cl.Message(
+        content="👋 Welcome! I'm your AI assistant. What would you like to work on?",
+        author="System"
+    ).send()
+
+# Authentication setup (optional)
+if os.getenv("CHAINLIT_AUTH_SECRET"):
+    @cl.password_auth_callback
+    def auth_callback(username: str, password: str) -> cl.User:
+        # Replace with your authentication logic
+        if username == os.getenv("CHAINLIT_USERNAME", "admin") and \
+           password == os.getenv("CHAINLIT_PASSWORD", "admin"):
+            return cl.User(identifier=username, metadata={"role": "user"})
+        return None 

praisonai/ui/config/chainlit.md

@@ -1 +1 @@
-# Welcome to Praison AI! 🚀🤖
+# Welcome to PraisonAI! 🚀

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "PraisonAI"
-version = "2.0.23"
+version = "2.0.25"
 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.17",
+    "praisonaiagents>=0.0.20",
     "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.23"
+version = "2.0.25"
 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.17"
+praisonaiagents = ">=0.0.20"
 python-dotenv = ">=0.19.0"
 instructor = ">=1.3.3"
 PyYAML = ">=6.0"

uv.lock

@@ -3061,7 +3061,7 @@ wheels = [
 
 [[package]]
 name = "praisonai"
-version = "2.0.23"
+version = "2.0.25"
 source = { editable = "." }
 dependencies = [
     { name = "instructor" },
@@ -3198,7 +3198,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.17" },
+    { name = "praisonaiagents", specifier = ">=0.0.20" },
     { 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" },
@@ -3251,16 +3251,16 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.17"
+version = "0.0.20"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "openai" },
     { name = "pydantic" },
     { name = "rich" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/f5/23/5ec86c5d22dd1b97ada3e93ccc0371f834f7eb503359f69b6db5d3952191/praisonaiagents-0.0.17.tar.gz", hash = "sha256:e9a19e60c04c87ab8fad626576c0bebf4fc4773d876a918ec8317fa4e9993c83", size = 23090 }
+sdist = { url = "https://files.pythonhosted.org/packages/8e/89/0e9d17b7113772ab469f3d113af51fb3ff5c2f16eb0a1b70667a12478358/praisonaiagents-0.0.20.tar.gz", hash = "sha256:15e08f2c8c06643e091870ba0f15c1614f6052dea45986c877dad52e8fe7f06e", size = 23885 }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/37/fc/7aa577670aa254c1bc447019e0887570651b6d5d70e31025198408ecce8c/praisonaiagents-0.0.17-py3-none-any.whl", hash = "sha256:ef5788215c8560bdbbc8a6ad1c7c0ce5affb375a2172b8b587faa40b2aebb952", size = 29131 },
+    { url = "https://files.pythonhosted.org/packages/f8/94/5f62ded3aad4ee5d879e8b9712be4a3268abc2d89f71b10999109cef147c/praisonaiagents-0.0.20-py3-none-any.whl", hash = "sha256:3e9c62f3c4d8978bea8e0c95359a49e20db833511e17e122ccc06eb77996a618", size = 29883 },
 ]
 
 [[package]]