Parallel Agents

Execute multiple agents concurrently for improved performance and throughput in complex workflows.

Overview

Parallel agent execution enables processing multiple independent tasks simultaneously, dramatically reducing total execution time for complex workflows.

┌─────────────────────────────────────────────────────────────────────┐
│                    PARALLEL VS SEQUENTIAL                            │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  SEQUENTIAL (Total: 9 seconds)                                      │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                             │
│  │ Agent A │─►│ Agent B │─►│ Agent C │                             │
│  │  3 sec  │  │  3 sec  │  │  3 sec  │                             │
│  └─────────┘  └─────────┘  └─────────┘                             │
│  ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━                             │
│                                                                      │
│  PARALLEL (Total: 3 seconds)                                        │
│  ┌─────────┐                                                        │
│  │ Agent A │                                                        │
│  │  3 sec  │                                                        │
│  └─────────┘                                                        │
│  ┌─────────┐                                                        │
│  │ Agent B │                                                        │
│  │  3 sec  │  ←── All run simultaneously                           │
│  └─────────┘                                                        │
│  ┌─────────┐                                                        │
│  │ Agent C │                                                        │
│  │  3 sec  │                                                        │
│  └─────────┘                                                        │
│  ━━━━━━━━━━━                                                        │
│             3x faster!                                              │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Basic Parallel Execution

Using asyncio.gather

python

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def run_agent(name: str, task: str, system_prompt: str) -> dict:
    """Run a single agent asynchronously."""
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": task}
        ]
    )
    return {
        "agent": name,
        "task": task,
        "result": response.choices[0].message.content
    }

async def run_parallel_agents(tasks: list[dict]) -> list[dict]:
    """Run multiple agents in parallel."""

    coroutines = [
        run_agent(
            name=task["name"],
            task=task["task"],
            system_prompt=task["system_prompt"]
        )
        for task in tasks
    ]

    results = await asyncio.gather(*coroutines, return_exceptions=True)

    # Handle any exceptions
    processed_results = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            processed_results.append({
                "agent": tasks[i]["name"],
                "error": str(result)
            })
        else:
            processed_results.append(result)

    return processed_results

# Usage
async def main():
    tasks = [
        {
            "name": "Analyst",
            "task": "Analyze Q3 sales data trends",
            "system_prompt": "You are a data analyst."
        },
        {
            "name": "Writer",
            "task": "Write executive summary for Q3 report",
            "system_prompt": "You are a business writer."
        },
        {
            "name": "Designer",
            "task": "Suggest data visualizations for Q3 metrics",
            "system_prompt": "You are a data visualization expert."
        }
    ]

    results = await run_parallel_agents(tasks)

    for result in results:
        print(f"\n{result['agent']}:")
        print(result.get('result', result.get('error')))

asyncio.run(main())

Fan-Out/Fan-In Pattern

Distribute work across agents, then combine results:

python

from dataclasses import dataclass
from typing import Any

@dataclass
class ParallelWorkflow:
    name: str
    fan_out_tasks: list[dict]
    fan_in_prompt: str

async def fan_out_fan_in(workflow: ParallelWorkflow) -> dict:
    """Execute fan-out/fan-in workflow."""

    # Fan-out: Run all tasks in parallel
    print(f"Starting {len(workflow.fan_out_tasks)} parallel tasks...")

    parallel_results = await run_parallel_agents(workflow.fan_out_tasks)

    # Prepare results for fan-in
    combined_input = "\n\n".join([
        f"## {r['agent']} Analysis:\n{r.get('result', 'Error: ' + r.get('error', 'Unknown'))}"
        for r in parallel_results
    ])

    # Fan-in: Synthesize results
    print("Synthesizing results...")

    synthesis = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": workflow.fan_in_prompt},
            {"role": "user", "content": f"Synthesize these analyses:\n\n{combined_input}"}
        ]
    )

    return {
        "workflow": workflow.name,
        "individual_results": parallel_results,
        "synthesis": synthesis.choices[0].message.content
    }

# Example: Multi-perspective document analysis
async def analyze_document(document: str):
    workflow = ParallelWorkflow(
        name="Document Analysis",
        fan_out_tasks=[
            {
                "name": "Sentiment Analyzer",
                "task": f"Analyze the sentiment of this document:\n\n{document}",
                "system_prompt": "You analyze text sentiment. Provide positive/negative/neutral assessment with confidence score."
            },
            {
                "name": "Key Point Extractor",
                "task": f"Extract key points from this document:\n\n{document}",
                "system_prompt": "You extract main points from documents. List the top 5 most important points."
            },
            {
                "name": "Entity Recognizer",
                "task": f"Identify entities in this document:\n\n{document}",
                "system_prompt": "You identify named entities (people, organizations, locations, dates, etc.)."
            },
            {
                "name": "Risk Assessor",
                "task": f"Identify potential risks mentioned in this document:\n\n{document}",
                "system_prompt": "You identify risks, concerns, and potential issues in documents."
            }
        ],
        fan_in_prompt="""You are a document analysis synthesizer.
        Combine the individual analyses into a comprehensive summary.
        Highlight areas of agreement and any conflicts between analyses.
        Provide actionable recommendations based on the combined insights."""
    )

    return await fan_out_fan_in(workflow)

Batch Processing with Concurrency Control

Process large batches with controlled parallelism:

python

import asyncio
from asyncio import Semaphore

class BatchProcessor:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = Semaphore(max_concurrent)
        self.client = AsyncOpenAI()

    async def process_single(self, item: dict) -> dict:
        """Process a single item with semaphore control."""
        async with self.semaphore:
            response = await self.client.chat.completions.create(
                model=item.get("model", "gpt-4o-mini"),
                messages=item["messages"]
            )
            return {
                "id": item.get("id"),
                "input": item,
                "output": response.choices[0].message.content
            }

    async def process_batch(
        self,
        items: list[dict],
        progress_callback=None
    ) -> list[dict]:
        """Process a batch of items with controlled concurrency."""

        completed = 0
        total = len(items)

        async def process_with_progress(item):
            nonlocal completed
            result = await self.process_single(item)
            completed += 1
            if progress_callback:
                progress_callback(completed, total)
            return result

        results = await asyncio.gather(
            *[process_with_progress(item) for item in items],
            return_exceptions=True
        )

        return results

# Usage
async def process_customer_feedback(feedbacks: list[str]):
    processor = BatchProcessor(max_concurrent=20)

    items = [
        {
            "id": i,
            "messages": [
                {"role": "system", "content": "Classify customer feedback as positive, negative, or neutral. Extract main topic."},
                {"role": "user", "content": feedback}
            ]
        }
        for i, feedback in enumerate(feedbacks)
    ]

    def progress(completed, total):
        print(f"Progress: {completed}/{total} ({100*completed/total:.1f}%)")

    results = await processor.process_batch(items, progress_callback=progress)
    return results

Parallel with Dependencies (DAG)

Handle workflows with dependencies between tasks:

python

from dataclasses import dataclass, field
from typing import Optional
import asyncio

@dataclass
class Task:
    id: str
    agent_config: dict
    dependencies: list[str] = field(default_factory=list)
    result: Optional[Any] = None
    status: str = "pending"

class DAGExecutor:
    def __init__(self):
        self.tasks: dict[str, Task] = {}
        self.results: dict[str, Any] = {}

    def add_task(self, task: Task):
        self.tasks[task.id] = task

    def get_ready_tasks(self) -> list[Task]:
        """Get tasks whose dependencies are all satisfied."""
        ready = []
        for task in self.tasks.values():
            if task.status == "pending":
                deps_satisfied = all(
                    self.tasks[dep_id].status == "completed"
                    for dep_id in task.dependencies
                )
                if deps_satisfied:
                    ready.append(task)
        return ready

    async def execute_task(self, task: Task) -> Any:
        """Execute a single task."""
        task.status = "running"

        # Inject dependency results into the task
        dep_results = {
            dep_id: self.results[dep_id]
            for dep_id in task.dependencies
        }

        # Build messages with dependency context
        context = ""
        if dep_results:
            context = "\n\nPrevious results:\n" + "\n".join([
                f"- {k}: {v}"
                for k, v in dep_results.items()
            ])

        messages = task.agent_config["messages"].copy()
        if context:
            messages[-1]["content"] += context

        response = await client.chat.completions.create(
            model=task.agent_config.get("model", "gpt-4o"),
            messages=messages
        )

        result = response.choices[0].message.content
        task.result = result
        task.status = "completed"
        self.results[task.id] = result

        return result

    async def execute_all(self) -> dict[str, Any]:
        """Execute all tasks respecting dependencies."""

        while any(t.status == "pending" for t in self.tasks.values()):
            ready_tasks = self.get_ready_tasks()

            if not ready_tasks:
                # Check for circular dependencies
                pending = [t.id for t in self.tasks.values() if t.status == "pending"]
                raise Exception(f"Circular dependency detected. Stuck tasks: {pending}")

            # Execute ready tasks in parallel
            await asyncio.gather(*[
                self.execute_task(task)
                for task in ready_tasks
            ])

        return self.results

# Example: Research paper workflow
async def research_workflow(topic: str):
    executor = DAGExecutor()

    # Stage 1: Parallel research (no dependencies)
    executor.add_task(Task(
        id="literature_review",
        agent_config={
            "messages": [
                {"role": "system", "content": "You conduct literature reviews."},
                {"role": "user", "content": f"Review existing literature on: {topic}"}
            ]
        }
    ))

    executor.add_task(Task(
        id="methodology_survey",
        agent_config={
            "messages": [
                {"role": "system", "content": "You survey research methodologies."},
                {"role": "user", "content": f"Survey methodologies for studying: {topic}"}
            ]
        }
    ))

    executor.add_task(Task(
        id="data_sources",
        agent_config={
            "messages": [
                {"role": "system", "content": "You identify data sources."},
                {"role": "user", "content": f"Identify data sources for: {topic}"}
            ]
        }
    ))

    # Stage 2: Synthesis (depends on all Stage 1 tasks)
    executor.add_task(Task(
        id="research_plan",
        agent_config={
            "messages": [
                {"role": "system", "content": "You create research plans."},
                {"role": "user", "content": f"Create a research plan for: {topic}"}
            ]
        },
        dependencies=["literature_review", "methodology_survey", "data_sources"]
    ))

    # Stage 3: Final output (depends on Stage 2)
    executor.add_task(Task(
        id="executive_summary",
        agent_config={
            "messages": [
                {"role": "system", "content": "You write executive summaries."},
                {"role": "user", "content": f"Write executive summary for research on: {topic}"}
            ]
        },
        dependencies=["research_plan"]
    ))

    results = await executor.execute_all()
    return results

Error Handling and Retries

python

import asyncio
from typing import Optional

class RetryConfig:
    max_retries: int = 3
    base_delay: float = 1.0
    max_delay: float = 30.0
    exponential_base: float = 2.0

async def run_with_retry(
    coroutine_factory,
    config: RetryConfig = RetryConfig()
) -> Any:
    """Run async function with exponential backoff retry."""

    last_exception = None

    for attempt in range(config.max_retries):
        try:
            return await coroutine_factory()
        except Exception as e:
            last_exception = e
            if attempt < config.max_retries - 1:
                delay = min(
                    config.base_delay * (config.exponential_base ** attempt),
                    config.max_delay
                )
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                await asyncio.sleep(delay)

    raise last_exception

async def parallel_with_retry(tasks: list[dict]) -> list[dict]:
    """Run parallel tasks with individual retry logic."""

    async def run_single_with_retry(task):
        async def factory():
            return await run_agent(
                name=task["name"],
                task=task["task"],
                system_prompt=task["system_prompt"]
            )
        return await run_with_retry(factory)

    results = await asyncio.gather(
        *[run_single_with_retry(task) for task in tasks],
        return_exceptions=True
    )

    return results

Performance Monitoring

python

import time
from dataclasses import dataclass, field
from typing import Dict, List

@dataclass
class ParallelMetrics:
    total_tasks: int = 0
    completed_tasks: int = 0
    failed_tasks: int = 0
    total_time: float = 0.0
    task_times: List[float] = field(default_factory=list)
    errors: List[str] = field(default_factory=list)

    @property
    def success_rate(self) -> float:
        if self.total_tasks == 0:
            return 0.0
        return self.completed_tasks / self.total_tasks

    @property
    def avg_task_time(self) -> float:
        if not self.task_times:
            return 0.0
        return sum(self.task_times) / len(self.task_times)

    @property
    def parallelism_efficiency(self) -> float:
        """Ratio of sequential time to actual parallel time."""
        if self.total_time == 0:
            return 0.0
        sequential_time = sum(self.task_times)
        return sequential_time / self.total_time

class MonitoredParallelExecutor:
    def __init__(self):
        self.metrics = ParallelMetrics()

    async def execute(self, tasks: list[dict]) -> tuple[list[dict], ParallelMetrics]:
        """Execute tasks and collect metrics."""

        self.metrics.total_tasks = len(tasks)
        start_time = time.time()

        async def timed_task(task):
            task_start = time.time()
            try:
                result = await run_agent(**task)
                task_time = time.time() - task_start
                self.metrics.task_times.append(task_time)
                self.metrics.completed_tasks += 1
                return result
            except Exception as e:
                self.metrics.failed_tasks += 1
                self.metrics.errors.append(str(e))
                return {"error": str(e)}

        results = await asyncio.gather(*[timed_task(t) for t in tasks])

        self.metrics.total_time = time.time() - start_time

        return results, self.metrics

# Usage
async def monitored_analysis():
    executor = MonitoredParallelExecutor()

    tasks = [
        {"name": f"Agent{i}", "task": f"Task {i}", "system_prompt": "You are helpful."}
        for i in range(10)
    ]

    results, metrics = await executor.execute(tasks)

    print(f"Total time: {metrics.total_time:.2f}s")
    print(f"Average task time: {metrics.avg_task_time:.2f}s")
    print(f"Parallelism efficiency: {metrics.parallelism_efficiency:.1f}x")
    print(f"Success rate: {metrics.success_rate:.1%}")

Best Practices

When to Use Parallel Execution

Scenario	Use Parallel?	Reason
Independent analyses	✅ Yes	No dependencies
Sequential workflow	❌ No	Each step needs previous result
Large batch processing	✅ Yes	With concurrency limits
Real-time response needed	✅ Yes	Reduces total latency
Cost optimization	⚠️ Consider	Parallel may hit rate limits

Concurrency Guidelines

python

# Rule of thumb for max_concurrent setting
CONCURRENCY_GUIDELINES = {
    "tier_1": 5,    # Free tier / low rate limits
    "tier_2": 20,   # Standard API access
    "tier_3": 50,   # High-volume access
    "tier_4": 100,  # Enterprise / custom limits
}

Parallel Agents ​

Overview ​

Basic Parallel Execution ​

Using asyncio.gather ​

Fan-Out/Fan-In Pattern ​

Batch Processing with Concurrency Control ​

Parallel with Dependencies (DAG) ​

Error Handling and Retries ​

Performance Monitoring ​

Best Practices ​

When to Use Parallel Execution ​

Concurrency Guidelines ​

See Also ​

Parallel Agents

Overview

Basic Parallel Execution

Using asyncio.gather

Fan-Out/Fan-In Pattern

Batch Processing with Concurrency Control

Parallel with Dependencies (DAG)

Error Handling and Retries

Performance Monitoring

Best Practices

When to Use Parallel Execution

Concurrency Guidelines

See Also