Agent Delegation and Parallel Execution

Agent Delegation and Parallel Execution

The real power of multi-agent systems comes from running subagents concurrently. Sequential delegation is simple; parallel delegation is fast.

Sequential vs Parallel Delegation

import asyncio
import anthropic
import time

client = anthropic.Anthropic()

async def run_subagent(task: str, context: str = "") -> str:
    """Async wrapper for a subagent."""
    # Run in thread pool to avoid blocking the event loop
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,
        lambda: client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=512,
            system="Complete the assigned task concisely.",
            messages=[{
                "role": "user",
                "content": f"{context}\n\nTask: {task}" if context else f"Task: {task}"
            }],
        ).content[0].text
    )
    return result


async def parallel_research(topics: list[str]) -> dict[str, str]:
    """Research multiple topics simultaneously."""
    start = time.time()

    tasks = [run_subagent(f"Summarize key facts about: {topic}") for topic in topics]
    results = await asyncio.gather(*tasks)

    elapsed = time.time() - start
    print(f"Researched {len(topics)} topics in {elapsed:.1f}s (parallel)")

    return dict(zip(topics, results))


async def main():
    topics = [
        "Claude AI model capabilities",
        "OpenAI GPT-4 features",
        "Google Gemini architecture",
        "Meta LLaMA performance",
    ]

    # Parallel — all 4 requests fire simultaneously
    parallel_results = await parallel_research(topics)

    for topic, summary in parallel_results.items():
        print(f"\n{topic}:\n{summary[:150]}")


asyncio.run(main())

Fan-Out / Fan-In Pattern

async def fan_out_fan_in(goal: str, subtasks: list[str]) -> str:
    """
    Fan-out: execute all subtasks in parallel.
    Fan-in: synthesize results into a final answer.
    """
    # Fan-out: run all subtasks concurrently
    loop = asyncio.get_event_loop()
    subtask_results = await asyncio.gather(*[
        run_subagent(task) for task in subtasks
    ])

    # Fan-in: synthesize
    synthesis_prompt = f"Goal: {goal}\n\nSubtask results:\n" + "\n\n".join(
        f"[{i+1}] {task}:\n{result}"
        for i, (task, result) in enumerate(zip(subtasks, subtask_results))
    )

    final = await run_subagent(
        "Synthesize the subtask results into a comprehensive final answer for the original goal.",
        context=synthesis_prompt,
    )

    return final


result = asyncio.run(fan_out_fan_in(
    goal="Write a competitive analysis of AI coding tools",
    subtasks=[
        "List features of GitHub Copilot",
        "List features of Cursor IDE",
        "List features of Claude Code",
        "Identify pricing differences between the three",
    ],
))
print(result)

JavaScript Parallel Agents

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

async function runSubagent(task: string): Promise<string> {
  const response = await client.messages.create({
    model: "claude-haiku-4-5",
    max_tokens: 512,
    messages: [{ role: "user", content: task }],
  });
  return response.content[0].text;
}

// Run all subtasks in parallel with Promise.all
const results = await Promise.all([
  runSubagent("Summarize benefits of TypeScript"),
  runSubagent("Summarize benefits of Python"),
  runSubagent("Summarize benefits of Go"),
]);

Parallel execution multiplies the throughput of your agentic system without increasing total model calls.