PraisonAI/examples/python/general/async_example_full.py

# praisonai: skip=true
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 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],
    reflection=False,
)

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",
    reflection=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 Agents instance with both agents
    agents = AgentTeam(
        agents=[async_agent, summary_agent],
        tasks=parallel_tasks + [summary_task],
        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())