Parallel & Concurrent Execution

Why Parallel Execution Matters

Most agent tasks involve significant I/O: web searches, API calls, database queries, LLM calls. These are all async-friendly and can run in parallel. The serial vs parallel difference is stark:

flowchart LR
    subgraph Serial["Serial, 1800ms total"]
        direction TB
        SA["search A · 800ms"] --> SB["search B · 600ms"] --> SC["fetch C · 400ms"]
    end
    subgraph Parallel["Parallel, ~800ms total"]
        direction TB
        PA["search A · 800ms"] & PB["search B · 600ms"] & PC["fetch C · 400ms"] --> DONE([Done])
    end
    style Serial fill:#fff0f0,stroke:#f87171
    style Parallel fill:#f0fdf9,stroke:#0D9488

For an agent making 10 independent tool calls, parallelism is a 5-10x latency improvement.

Concurrent Tool Calls with asyncio

Basic Pattern

asyncio.gather is the standard way to run independent coroutines simultaneously. Each tool call runs in its own coroutine, and all start at the same time, total latency is the slowest single call, not the sum.

import asyncio
from typing import Any

async def parallel_tool_calls(tool_calls: list[dict]) -> list[dict]:
    """Execute all tool calls concurrently and return results in order."""
    
    async def execute_one(tc: dict) -> dict:
        func_name = tc["function"]["name"]
        func_args = json.loads(tc["function"]["arguments"])
        
        try:
            result = await TOOL_FUNCTIONS[func_name](**func_args)
            return {
                "tool_call_id": tc["id"],
                "role": "tool",
                "content": json.dumps(result),
            }
        except Exception as e:
            # return errors as tool results rather than raising: lets the LLM handle failures gracefully
            return {
                "tool_call_id": tc["id"],
                "role": "tool",
                "content": json.dumps({"error": str(e)}),
            }
    
    # asyncio.gather() runs all tool calls concurrently: if each takes 1s, total is ~1s not N×1s
    results = await asyncio.gather(*[execute_one(tc) for tc in tool_calls])
    return list(results)  # gather preserves input order even though tasks complete in arbitrary order

Bounded Concurrency with Semaphore

Unbounded parallelism can trigger rate limits. This class wraps a semaphore to cap how many coroutines run at once, letting you maximize throughput without overwhelming the API.

import asyncio

class BoundedParallelExecutor:
    """Execute tasks in parallel with a concurrency limit."""
    
    def __init__(self, max_concurrent: int = 10):
        # Semaphore(10) caps concurrency: prevents overwhelming the API with too many simultaneous requests
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_with_limit(self, coro):
        async with self.semaphore:  # blocks here if max_concurrent tasks are already running
            return await coro
    
    async def gather(self, tasks: list) -> list:
        bounded_tasks = [self.execute_with_limit(task) for task in tasks]
        # return_exceptions=True means a single failure doesn't cancel all other tasks
        return await asyncio.gather(*bounded_tasks, return_exceptions=True)

# Usage: limit to 5 concurrent API calls
executor = BoundedParallelExecutor(max_concurrent=5)

async def search_multiple_queries(queries: list[str]) -> list[dict]:
    tasks = [search_web(q) for q in queries]
    results = await executor.gather(tasks)
    
    # Handle exceptions from individual tasks without dropping successful results
    return [
        r if not isinstance(r, Exception) else {"error": str(r)}
        for r in results
    ]

Parallel Branches in LangGraph

Send() is LangGraph’s mechanism for fan-out parallelism. It creates one copy of a node per item and runs all copies concurrently, the results are then merged by a reducer function before the next node runs.

# Parallel fact-checking: verify multiple claims simultaneously
class FactCheckState(TypedDict):
    claims: list[str]               # input: list of claims to verify
    # operator.add as state reducer means each worker's output appends to the shared list
    verified: Annotated[list[dict], operator.add]
    final_verdict: str | None

def fan_out_claims(state: FactCheckState) -> list[Send]:
    """Start one verification branch per claim."""
    # Send() creates one copy of the worker node per claim: LangGraph fans out in parallel automatically
    return [
        Send("verify_one_claim", {"claim": claim, "result": None})
        for claim in state["claims"]
    ]

class SingleClaimState(TypedDict):
    claim: str
    result: dict | None

async def verify_one_claim(state: SingleClaimState) -> dict:
    """Verify a single claim by searching for evidence."""
    # This runs in parallel for each claim: all claims are searched simultaneously
    search_results = await search_web(f"fact check: {state['claim']}")
    
    verification_prompt = f"""
Claim: {state['claim']}
Evidence: {json.dumps(search_results[:3], indent=2)}

Verdict (true/false/uncertain) and brief explanation.
Return JSON: {{"verdict": "true|false|uncertain", "explanation": "...", "sources": []}}"""
    
    result = llm.invoke([HumanMessage(content=verification_prompt)])
    verdict = json.loads(result.content)
    
    # return a single-item list so operator.add can accumulate results across all parallel branches
    return {"result": {"claim": state["claim"], **verdict}}

def aggregate_verdicts(state: FactCheckState) -> dict:
    """Combine all verification results."""
    # state["verified"] contains results from all parallel branches, merged by operator.add
    true_count = sum(1 for r in state["verified"] if r.get("verdict") == "true")
    false_count = sum(1 for r in state["verified"] if r.get("verdict") == "false")
    uncertain_count = len(state["verified"]) - true_count - false_count
    
    summary = f"{true_count} claims verified, {false_count} false, {uncertain_count} uncertain"
    return {"final_verdict": summary}

# Build the parallel fact-checking graph
fc_builder = StateGraph(FactCheckState)
fc_builder.add_node("fan_out", lambda s: s)      # passthrough node, triggers the fan-out edge
fc_builder.add_node("verify_one_claim", verify_one_claim)
fc_builder.add_node("aggregate", aggregate_verdicts)

fc_builder.set_entry_point("fan_out")
fc_builder.add_conditional_edges("fan_out", fan_out_claims)  # fan_out_claims returns list[Send]
fc_builder.add_edge("verify_one_claim", "aggregate")         # all branches converge here
fc_builder.add_edge("aggregate", END)

fact_checker = fc_builder.compile()

Batched Parallel Processing

For large-scale processing (hundreds of items), batch them to avoid overwhelming APIs:

async def process_documents_in_batches(
    documents: list[str],
    batch_size: int = 20,          # process 20 documents at once within each batch
    max_concurrent_batches: int = 3,  # run up to 3 batches simultaneously
) -> list[dict]:
    """Process documents in parallel batches."""
    
    async def process_batch(batch: list[str]) -> list[dict]:
        # all documents in the batch run concurrently
        tasks = [process_single_document(doc) for doc in batch]
        return await asyncio.gather(*tasks, return_exceptions=True)
    
    # Split into batches of batch_size
    batches = [documents[i:i+batch_size] for i in range(0, len(documents), batch_size)]
    
    all_results = []
    # Semaphore limits how many batches run simultaneously: outer rate control
    batch_semaphore = asyncio.Semaphore(max_concurrent_batches)
    
    async def run_batch_with_limit(batch):
        async with batch_semaphore:
            return await process_batch(batch)
    
    # Run all batches with concurrency capped at max_concurrent_batches
    batch_results = await asyncio.gather(*[run_batch_with_limit(b) for b in batches])
    
    for batch_result in batch_results:
        for item in batch_result:
            if isinstance(item, Exception):
                all_results.append({"error": str(item)})  # preserve errors without stopping the rest
            else:
                all_results.append(item)
    
    return all_results

Async LLM Calls in Parallel

When you need multiple model opinions on the same prompt, run all the LLM calls simultaneously rather than waiting for each to finish before starting the next.

async def evaluate_with_multiple_models(prompt: str) -> dict:
    """Get opinions from multiple LLMs simultaneously."""
    
    async def call_openai():
        client = AsyncOpenAI()
        response = await client.chat.completions.create(
            model="gpt-5.6-sol",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=500,
        )
        return {"model": "gpt-5.6-sol", "response": response.choices[0].message.content}
    
    async def call_anthropic():
        client = AsyncAnthropic()
        response = await client.messages.create(
            model="claude-sonnet-5",
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}],
        )
        return {"model": "claude-sonnet-5", "response": response.content[0].text}
    
    # asyncio.gather() runs all three model calls concurrently: if each takes 1s, total is ~1s not 3s
    results = await asyncio.gather(
        call_openai(),
        call_anthropic(),
        return_exceptions=True,  # one model failing doesn't cancel the others
    )
    
    successful = [r for r in results if not isinstance(r, Exception)]
    return {"responses": successful, "failures": len(results) - len(successful)}

Measuring Parallel vs Sequential Performance

Benchmarking before and after parallelization confirms the speedup is real, always measure with realistic workloads before committing to architectural changes.

import time

async def benchmark_execution():
    queries = ["AI agent frameworks", "vector databases", "LLM pricing 2025",
               "Python async patterns", "Docker best practices"]
    
    # Sequential: each search waits for the previous one to finish
    start = time.perf_counter()
    sequential_results = []
    for q in queries:
        result = await search_web(q)
        sequential_results.append(result)
    sequential_time = time.perf_counter() - start
    
    # Parallel: all searches start simultaneously with asyncio.gather
    start = time.perf_counter()
    parallel_results = await asyncio.gather(*[search_web(q) for q in queries])
    parallel_time = time.perf_counter() - start
    
    speedup = sequential_time / parallel_time
    print(f"Sequential: {sequential_time:.2f}s")
    print(f"Parallel:   {parallel_time:.2f}s")
    print(f"Speedup:    {speedup:.1f}x")  # expect ~N× for N independent I/O-bound calls

The key insight: for any agent task where tool calls are independent of each other (not using each other’s results), parallelism is almost always worth the slightly more complex code.