> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Async Agents

> Async AI Agents allow you to run AI tasks asynchronously, improving performance and efficiency in your applications.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
    In[In] --> Agent1[AI Agent]
    Agent1 --> Agent2[AI Agent]
    Agent1 --> Agent3[AI Agent]
    Agent1 --> Agent4[AI Agent]
    Agent2 --> Out[Out]
    Agent3 --> Out
    Agent4 --> Out
    
    style In fill:#8B0000,color:#fff
    style Agent1 fill:#2E8B57,color:#fff
    style Agent2 fill:#2E8B57,color:#fff
    style Agent3 fill:#2E8B57,color:#fff
    style Agent4 fill:#2E8B57,color:#fff
    style Out fill:#8B0000,color:#fff
```

Async AI Agents allow you to run AI tasks asynchronously, improving performance and efficiency in your applications.

## Quick Start

<Tabs>
  <Tab title="Code">
    <Steps>
      <Step title="Install Package">
        First, install the PraisonAI Agents package:

        ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        pip install praisonaiagents
        ```
      </Step>

      <Step title="Set API Key">
        Set your OpenAI API key as an environment variable:

        ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        export OPENAI_API_KEY=your_api_key_here
        ```
      </Step>

      <Step title="Create Project">
        Create a new file `app.py` with the basic setup:

        ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        import asyncio
        from praisonaiagents import Agent, Task, AgentTeam

        # 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 = AgentTeam(
            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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        python app.py
        ```
      </Step>
    </Steps>
  </Tab>

  <Tab title="No Code">
    <Steps>
      <Step title="Install Package">
        Install the PraisonAI package:

        ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        pip install praisonai
        ```
      </Step>

      <Step title="Set API Key">
        Set your OpenAI API key as an environment variable:

        ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        export OPENAI_API_KEY=your_api_key_here
        ```
      </Step>

      <Step title="Create Project">
        Create a new file `agents.yaml` with the basic setup:

        ```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        framework: praisonai
        process: sequential
        topic: async operations
        agents:  # Canonical: use 'agents' instead of 'roles'
          assistant:
            name: AsyncAgent
            role: Assistant
            goal: Help with tasks
            instructions:  # Canonical: use 'instructions' instead of 'backstory' Expert in async operations
            tasks:
              hello_task:
                description: Say hello and introduce yourself
                expected_output: Answer to the question
                async_execution: true
        ```
      </Step>

      <Step title="Run the Program">
        Run your async AI agent:

        ```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
        praisonai agents.yaml
        ```
      </Step>
    </Steps>
  </Tab>
</Tabs>

<Note>
  **Requirements**

  * Python 3.10 or higher
  * OpenAI API key. Generate OpenAI API key [here](https://platform.openai.com/api-keys). Use Other models using [this guide](/models).
  * Basic understanding of async/await in Python
</Note>

<div className="relative w-full aspect-video">
  <iframe className="absolute top-0 left-0 w-full h-full" src="https://www.youtube.com/embed/VhVQfgo00LE" title="YouTube video player" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowFullScreen />
</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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  # Synchronous execution (blocks until complete)
  result = agent.chat("Hello")
  ```

  ```python Async theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  # 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

<Tabs>
  <Tab title="Code">
    ### Async Tasks with Callbacks and Tools Example Code

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    import asyncio
    import time
    from typing import List, Dict
    from praisonaiagents import Agent, Task, AgentTeam, 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],
        reflection=False,
        
        
    )

    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.""",
        reflection=True,  # Enable self-reflection for better summary quality
        
        
    )

    # 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 = AgentTeam(
            agents=[async_agent],
            tasks=[async_task],
            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 Agents instance with both agents
        agents = AgentTeam(
            agents=[async_agent, summary_agent],  # Include both agents
            tasks=parallel_tasks + [summary_task],  # Include all tasks
            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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
      # 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
      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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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
    ```
  </Tab>

  <Tab title="No Code">
    ```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    framework: praisonai
    process: sequential
    topic: async operations
    agents:  # Canonical
      async_agent:
        name: AsyncSearchAgent
        role: Asynchronous Search Specialist
        goal: Perform fast and efficient asynchronous searches
        instructions:  # Canonical: use 'instructions' instead of 'backstory' Expert in parallel search operations
        tools:
          - async_search_tool
        tasks:
          search_task:
            description: Perform async search operations
            expected_output: Structured search results
            async_execution: true
    ```
  </Tab>
</Tabs>

## Best Practices

<AccordionGroup>
  <Accordion title="Error Handling">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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>

## Sync ↔ Async Bridge

The async bridge utilities help you safely move between sync and async contexts without deadlocking the event loop:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.utils.async_bridge import run_coroutine_from_any_context

def sync_tool():
    # Call async code from sync context
    result = run_coroutine_from_any_context(async_api_call())
    return result
```

See the [Async Bridge](/features/async-bridge) guide for complete documentation and usage patterns.

## 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>

  <Card title="Event Loop Running Error" icon="warning">
    `RuntimeError: This event loop is already running` from SDK internals:

    * Upgrade to the latest release with async bridge support
    * Internal call sites now use run\_coroutine\_from\_any\_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 (async callbacks are dispatched safely even when the caller is inside a running event loop)
</ResponseField>

## Next Steps

<CardGroup cols={2}>
  <Card title="AutoAgents" icon="robot" href="./autoagents">
    Learn about automatically created and managed AI agents
  </Card>

  <Card title="Mini Agents" icon="microchip" href="./mini">
    Explore lightweight, focused AI agents
  </Card>
</CardGroup>

<Note>
  Remember to handle errors properly, manage resources efficiently, and test thoroughly in async contexts.
</Note>