Parallel Agent Execution: Running AI Agents Concurrently in a CrewAI Flow

What it is

A linear multi-agent pipeline — agent 1, then agent 2, then agent 3 — is simple but wastes wall-clock time whenever two steps don't actually depend on each other's output. Parallel agent execution turns that line into a partial-order graph: independent branches run concurrently, and a join step waits for all of them before continuing. This is standard workflow-engine practice (Airflow, Temporal, Step Functions all support it), applied here to LLM agent calls specifically.

How ProspectAI implements it

ProspectAI's ProspectAIFlow (a CrewAI Flow) runs Technical Analysis and Fundamental Analysis as two branches that both depend only on Market Analysis, not on each other. Both are declared with @listen(market_analysis), and because each method is async def and awaited via akickoff(), CrewAI's Flow engine fires both as soon as market_analysis resolves — they execute concurrently on the event loop rather than sequentially:

class ProspectAIFlow(Flow[ProspectAIFlowState]):
    """Runs Technical and Fundamental analysis in parallel after Market
    Analysis completes, then gates Draft Strategy on both via and_()."""

    @start()
    async def market_analysis(self):
        ...
        result = cast(CrewOutput, await self._make_crew(task, 0).akickoff())
        self.state.market_output = self._extract_pydantic(
            result, MarketAnalysisOutput, "market_analysis"
        )
        return result.raw or ""

    @listen(market_analysis)
    async def technical_analysis(self):
        ...
        result = cast(CrewOutput, await self._make_crew(task, 1).akickoff())
        self.state.technical_output = self._extract_pydantic(
            result, TechnicalAnalysisOutput, "technical_analysis"
        )
        return result.raw or ""

    @listen(market_analysis)
    async def fundamental_analysis(self):
        ...
        result = cast(CrewOutput, await self._make_crew(task, 2).akickoff())
        self.state.fundamental_output = self._extract_pydantic(
            result, FundamentalAnalysisOutput, "fundamental_analysis"
        )
        return result.raw or ""

    @listen(and_(technical_analysis, fundamental_analysis))
    async def draft_strategy(self):
        ...

The join is CrewAI's and_() combinator: draft_strategy is decorated with @listen(and_(technical_analysis, fundamental_analysis)), so it only fires once both branches have completed — guaranteeing correctness (draft strategy always has both technical and fundamental data available) without giving up the latency win from running them concurrently.

Why it matters

Each of the six phases in ProspectAI's pipeline is an LLM call with real network latency. Overlapping two independent, non-interfering calls (technical indicators vs. fundamental financials) instead of running them back-to-back removes one full phase's worth of latency from every analysis run, for free — no additional infrastructure, just structuring the dependency graph correctly. The trade-off is that reasoning about pipeline state gets slightly harder: both branches read the same upstream market_output but must not read each other's state before the join, which is why each branch validates its own output independently (see output validation) rather than assuming a particular execution order.