Spaces:
Running
Running
File size: 1,312 Bytes
0e38162 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | """
Parallel Workflow: runs a wave of agents concurrently using asyncio.gather.
"""
import asyncio
from typing import Any, List
from workflows.base_workflow import BaseWorkflow
from agents.base_agent import AgentContext, AgentResult, AgentStatus
from utils.logger import get_logger
logger = get_logger("parallel_workflow")
class ParallelWorkflow(BaseWorkflow):
"""Run all provided agents concurrently; collect results."""
name = "parallel"
async def execute(self, agents: List[Any], ctx: AgentContext) -> List[AgentResult]:
if not agents:
return []
if len(agents) == 1:
return [await agents[0].run(ctx)]
logger.info(f"Running {len(agents)} agents in parallel: {[a.name for a in agents]}")
tasks = [agent.run(ctx) for agent in agents]
results: List[AgentResult] = await asyncio.gather(*tasks, return_exceptions=True)
# Unwrap any exceptions from gather
cleaned = []
for i, r in enumerate(results):
if isinstance(r, Exception):
cleaned.append(AgentResult(
agent_name=agents[i].name,
status=AgentStatus.FAILED,
error=str(r),
))
else:
cleaned.append(r)
return cleaned
|