Concurrent Orchestration: Parallel Specialists

Lesson 6 — Multi-agent Pattern 2: fan-out/fan-in, multiple agents working simultaneously

The Pattern

Source: Microsoft Azure — AI Agent Orchestration Patterns

Two Sub-Patterns

Source: Anthropic — Building Effective Agents (Parallelization section)

Real-World Example: Stock Analysis

Microsoft's guide describes a financial services firm using concurrent agents for stock evaluation:

Build It: Parallel Stock Analysis

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import os

llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"),
                 base_url="https://api.deepseek.com", temperature=0.3)

# 4 specialist agents — each gets the SAME input
agents = {
    "fundamental": ChatPromptTemplate.from_messages([
        ("system", "You are a fundamental analyst. Analyze the stock's financials, revenue trends, "
                   "and competitive positioning. Be concise (3-4 sentences)."),
        ("user", "Analyze: {ticker}"),
    ]) | llm,
    
    "technical": ChatPromptTemplate.from_messages([
        ("system", "You are a technical analyst. Analyze price patterns, momentum, and indicators. "
                   "Be concise (3-4 sentences)."),
        ("user", "Analyze: {ticker}"),
    ]) | llm,
    
    "sentiment": ChatPromptTemplate.from_messages([
        ("system", "You are a sentiment analyst. Assess market sentiment, news, and analyst opinions. "
                   "Be concise (3-4 sentences)."),
        ("user", "Analyze: {ticker}"),
    ]) | llm,
    
    "risk": ChatPromptTemplate.from_messages([
        ("system", "You are a risk analyst. Assess downside risks, regulatory concerns, and worst-case scenarios. "
                   "Be concise (3-4 sentences)."),
        ("user", "Analyze: {ticker}"),
    ]),
}

async def run_analysis(ticker: str):
    # Fan-out: all agents run concurrently
    tasks = {name: (chain if isinstance(chain, type(llm)) else chain).ainvoke({"ticker": ticker}) 
             for name, chain in agents.items()}
    results = {}
    for name, task in tasks.items():
        results[name] = await task
    
    # Fan-in: aggregate results
    summary = "\n\n".join(f"### {name.title()} Analysis\n{r.content}" 
                          for name, r in results.items())
    
    # Synthesize with a final LLM call
    synthesis = llm.invoke(
        f"Based on these 4 analyses, give a HOLD/SELL recommendation:\n\n{summary}"
    )
    return synthesis.content

result = asyncio.run(run_analysis("SBIN"))

When to Use vs Avoid

Practice Drill

1. Create stock-research-agent/concurrent_analysis.py with the code above
2. Run it for SBIN. How long does it take? Compare with your sequential agent from Lesson 5.
3. Add a 5th agent: "competitive positioning" agent
4. Try the voting pattern: run 3 copies of the SAME agent and compare outputs. Are they consistent?