Async agent/task execution failure, gets stuck in infinite loop (no progress to next tasks)

[mzazakeith]

Description

When using PraisonAIAgents to run multiple async tasks (even with simple, local async tools and no LLM or network calls), no task is run in parallel and only the first task is executed. The agent appears to get stuck in an infinite loop, repeatedly running the first task, and never progresses to subsequent tasks or summary/analysis steps. This occurs regardless of the tool or agent configuration.

Environment

• Provider (select one):
  • Anthropic
  • OpenAI
  • Google Vertex AI
  • AWS Bedrock
  • Other:
• PraisonAI version: PraisonAI==2.2.53
• Operating System: macOS 15.4

Full Code

import asyncio
from pydantic import BaseModel
from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput


class MultiplyResult(BaseModel):
    a: int
    b: int
    product: int

async def multiply_tool(a: int, b: int) -> dict:
    """
    Multiplies two integers asynchronously.
    Args:
        a (int): First number
        b (int): Second number
    Returns:
        dict: {'a': a, 'b': b, 'product': a * b}
    """
    await asyncio.sleep(0.1)  # Simulate async work
    return {"a": a, "b": b, "product": a * b}


math_agent = Agent(
    name="MathHomeworkAgent",
    role="Math Homework Solver",
    goal="Solve multiplication problems for homework",
    backstory="Expert in basic arithmetic and helping students with homework.",
    tools=[multiply_tool],
    self_reflect=False,
    llm="gemini/gemini-2.5-flash-lite-preview-06-17",
    verbose=True,
    markdown=True
)

summary_agent = Agent(
    name="SummaryAgent",
    role="Math Results Analyst",
    goal="Analyze and present math results in bullet points",
    backstory="Expert in summarizing and presenting math results clearly.",
    self_reflect=False,
    llm="gemini/gemini-2.5-flash-lite-preview-06-17",
    verbose=True,
    markdown=True
)

problems = [(3, 4), (7, 6), (9, 5)]

math_tasks = [
    Task(
        name=f"multiply_{a}_by_{b}",
        description=f"Multiply {a} by {b} and return the result.",
        expected_output="MultiplyResult model with a, b, and product.",
        agent=math_agent,
        async_execution=True,
        output_pydantic=MultiplyResult,
    ) for a, b in problems
]

summary_task = Task(
    name="summary_task",
    description="Summarize the multiplication results in bullet points.",
    expected_output="A bullet-point summary of all multiplication results.",
    agent=summary_agent,
    async_execution=False,  # Run after math tasks
)

# 5. Run all tasks
async def main():
    agents = PraisonAIAgents(
        agents=[math_agent, summary_agent],
        tasks=math_tasks + [summary_task],
        verbose=1,
        process="sequential"
    )
    results = await agents.astart()
    print(f"Tasks Results: {results}")

if __name__ == "__main__":
    asyncio.run(main()) 

or

import asyncio
from typing import List, Dict
from praisonaiagents import Agent, Task, PraisonAIAgents, TaskOutput
from pydantic import BaseModel
from custom_web_search_tool import google_web_search_llm

# 1. Define SearchResult model matching google_web_search_llm output
class SearchResult(BaseModel):
    """
    Represents the top web search result from the custom search tool.

    Attributes:
        title (str): The title of the top search result.
        url (str): The URL of the top search result.
        description (str): The description/snippet of the top search result.
    """
    title: str
    url: str
    description: str

# 1b. Async wrapper for google_web_search_llm (return only top result)
async def async_search_tool(query: str) -> dict:
    """
    Asynchronously performs a Google Custom Search using the custom search tool and returns only the top result.

    Args:
        query (str): The search query string.

    Returns:
        dict: A dictionary with keys 'title', 'url', and 'description' for the top search result.
    """
    loop = asyncio.get_event_loop()
    # Run the synchronous search in a thread pool to avoid blocking
    result = await loop.run_in_executor(None, google_web_search_llm, query)
    # Extract the top result (first in 'results["top"]')
    top = (result.get("results", {}).get("top") or [{}])[0]
    return {
        "title": top.get("title", ""),
        "url": top.get("url", ""),
        "description": top.get("description", "")
    }

# 2. Define async callback (optional)
async def async_callback(output: TaskOutput):
    await asyncio.sleep(1)
    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,
    llm="gemini/gemini-2.5-flash-lite-preview-06-17",
    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.",
    self_reflect=False,
    verbose=True,
    markdown=True
)

# 4. Create async search tasks for different topics
search_topics = [
    "Latest AI Developments 2024",
    "Machine Learning Best Practices",
    "Neural Networks Architecture"
]

parallel_tasks = [
    Task(
        name=f"search_task_{i}",
        description=f"Search for '{topic}' and return results in JSON format.",
        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)
]

# 5. Create a summary task (runs after all searches)
summary_task = Task(
    name="summary_task",
    description="Summarize the findings from all search tasks.",
    expected_output="A comprehensive research synthesis.",
    agent=summary_agent,
    async_execution=False,  # Run after search tasks
    callback=async_callback
)

# 6. Run all tasks
async def main():
    agents = PraisonAIAgents(
        agents=[async_agent, summary_agent],
        tasks=parallel_tasks + [summary_task],
        verbose=1,
        process="sequential"
    )
    results = await agents.astart()
    print(f"Tasks Results: {results}")

if __name__ == "__main__":
    asyncio.run(main())

Steps to Reproduce

1. Install the library
2. Copy the examples above into different script files
3. Run the scripts and observe the behaviour

Expected Behavior

Multiple instances of the agents each running the different tasks async

Actual Behavior

Only the first async task is executed.
The process loops, repeatedly running the first task.
No other tasks (including subsequent math problems or summary/analysis tasks) are ever executed.
The process must be killed manually; it never completes.

[MervinPraison]

@claude review this issue and do a detailed analysis and fix this if the existing code doesn't have the solution implemented. Making sure it has backward compatibility, no existing features removed. After making those changes, again review the applied changes. Use @web to search if you dont know any information or to find the latest documentation or to find the latest version. Run the code if you think you need to run it to test it. Minimal code change to start with if required any changes.

[claude]

Claude encountered an error —— View job

---

I'll analyze this and get back to you.

[praison-ai-bot]

Hi! I've analyzed and fixed this issue.

Problem:
The agent gets stuck in an infinite loop when running multiple asynchronous tasks in a workflow. This was caused by a flawed logic in the arun_all_tasks method, which was not correctly batching parallel tasks.

Solution:
I have refactored the arun_all_tasks method in src/praisonai-agents/praisonaiagents/agents/agents.py to correctly handle asynchronous and synchronous tasks in a workflow. The new logic ensures that tasks are executed in the correct order and in parallel when possible, which resolves the infinite loop issue.

Changes Made:

• Modified src/praisonai-agents/praisonaiagents/agents/agents.py

Pull Request:
#863

The fix is ready for review. Please test and let me know if you encounter any issues!