2. Parallel Function Calling

Capsule overview

In the previous capsule (01) you saw the whole landscape of function calling patterns. This is the first one: parallel function calling. The concept is straightforward: when a user asks for "weather in NYC and SF", a naive agent runs the two calls one after the other (serial). An agent with parallel function calling runs them simultaneously. The result is the same, but latency drops by half.

Why does it matter? Because in production, users don't ask for one thing at a time. They ask "find flights to Madrid, the weather for the week, and hotels near downtown." That's 3 calls to external APIs. Serial: 3-5 seconds each = 9-15 seconds. Parallel: 3-5 seconds total. The difference between an agent that feels slow and one that feels fast.

Parallel function calling has two halves. The model does the first: when it detects independent sub-tasks, it returns multiple tool_calls in a single AIMessage. You do the second: your system takes those N calls and runs them simultaneously with ThreadPoolExecutor or asyncio. If you only do the first half but run them one by one in a for loop, you have half a parallel calling — the model did its job, but your code didn't.

The problem: serial execution

The scenario

Imagine this prompt: "What's the weather in Madrid, Barcelona and Buenos Aires?" The model detects 3 independent sub-tasks and emits 3 tool_calls in a single AIMessage. Your code receives them. What do you do?

Naive implementation: a for loop

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
import time

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    time.sleep(2)  # Simulates real API latency
    climates = {
        "Madrid": "22°C, sunny", "Barcelona": "20°C, cloudy",
        "Buenos Aires": "18°C, rainy", "NYC": "12°C, partly cloudy",
        "San Francisco": "16°C, foggy", "Tokyo": "19°C, clear",
    }
    return f"Weather in {city}: {climates.get(city, '18°C, variable')}"

model = init_chat_model("openai:gpt-4.1-mini")
tools = [get_weather]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)

def agent_loop_serial(user_input: str, max_iterations: int = 5) -> str:
    """Loop that runs tool calls one by one (serial)."""
    messages = [HumanMessage(content=user_input)]
    for i in range(max_iterations):
        response = model_with_tools.invoke(messages)
        messages.append(response)
        if not response.tool_calls:
            return response.content or "No response"
        for tc in response.tool_calls:
            try:
                result = tools_by_name[tc["name"]].invoke(tc["args"])
            except Exception as e:
                result = f"Error: {e}"
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return "Iteration limit reached"

The cost of going serial

start = time.time()
result = agent_loop_serial("Weather in Madrid, Barcelona and Buenos Aires?")
elapsed = time.time() - start
print(f"Result: {result}")
print(f"Time: {elapsed:.1f}s")
# Time: ~6.0s  ← 3 tools × 2s each = 6s serial

The diagram of the problem

Serial:
  get_weather("Madrid")    ─────▶ 2s
                           get_weather("Barcelona") ─────▶ 2s
                                                    get_weather("Buenos Aires") ─────▶ 2s
  |──────────────────────────────────────────────────────────────|
  0s                                                           6s

Parallel:
  get_weather("Madrid")         ─────▶ 2s
  get_weather("Barcelona")      ─────▶ 2s
  get_weather("Buenos Aires")   ─────▶ 2s
  |───────────────────|
  0s                 2s   ← Total: the slowest one

How parallel function calling works

Part 1: The model emits multiple tool_calls

When you ask the model for something with independent sub-tasks, it emits all the tool_calls in a single AIMessage:

response = model_with_tools.invoke([
    HumanMessage(content="Weather in Madrid, Barcelona and Buenos Aires?")
])

print(f"Number of tool_calls: {len(response.tool_calls)}")
for tc in response.tool_calls:
    print(f"  → {tc['name']}({tc['args']}) id={tc['id']}")
# Number of tool_calls: 3
#   → get_weather({'city': 'Madrid'}) id=call_abc123
#   → get_weather({'city': 'Barcelona'}) id=call_def456
#   → get_weather({'city': 'Buenos Aires'}) id=call_ghi789

Part 2: Your system runs them simultaneously

You have those 3 instructions. If you run them with a for loop, that's 6s. If you run them with ThreadPoolExecutor or asyncio.gather, that's 2s.

When the model emits parallel calls

The model makes the decision, not you. Your job is to be ready to receive 1 or N tool_calls.

The contract

1. An AIMessage can have 0, 1 or N tool_calls — your code handles all three cases
2. Each tool_call has a unique id — you need it for the response ToolMessage
3. You must send one ToolMessage per tool_call — if the model asked for 3, you return 3
4. The order of the ToolMessages doesn't matter — the model pairs them by tool_call_id

Implementation with ThreadPoolExecutor

The parallel execution function

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    time.sleep(2)
    climates = {
        "Madrid": "22°C, sunny", "Barcelona": "20°C, cloudy",
        "Buenos Aires": "18°C, rainy", "NYC": "12°C, partly cloudy",
        "San Francisco": "16°C, foggy", "Tokyo": "19°C, clear",
    }
    return f"Weather in {city}: {climates.get(city, '18°C, variable')}"

@tool
def get_stock_price(symbol: str) -> str:
    """Get the current price of a stock."""
    time.sleep(1.5)
    prices = {"AAPL": 178.50, "GOOGL": 141.20, "MSFT": 415.30, "TSLA": 245.80}
    price = prices.get(symbol.upper())
    if price:
        return f"{symbol.upper()}: ${price}"
    return f"Error: Symbol '{symbol}' not found."

@tool
def calculator(expression: str) -> str:
    """Evaluate a safe mathematical expression."""
    allowed = set("0123456789+-*/(). ")
    if not all(c in allowed for c in expression):
        return "Error: characters not allowed"
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
tools = [get_weather, get_stock_price, calculator]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)


def execute_tools_parallel(tool_calls, tools_by_name, timeout=30):
    """Run multiple tool calls in parallel with ThreadPoolExecutor."""
    results = []
    with ThreadPoolExecutor(max_workers=min(len(tool_calls), 10)) as executor:
        futures = {
            executor.submit(tools_by_name[tc["name"]].invoke, tc["args"]): tc
            for tc in tool_calls
            if tc["name"] in tools_by_name
        }
        for future in as_completed(futures):
            tc = futures[future]
            try:
                result = future.result(timeout=timeout)
            except TimeoutError:
                result = f"Error: {tc['name']} exceeded the {timeout}s timeout"
            except Exception as e:
                result = f"Error running {tc['name']}: {e}"
            results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

    for tc in tool_calls:
        if tc["name"] not in tools_by_name:
            results.append(ToolMessage(
                content=f"Error: tool '{tc['name']}' does not exist. Available: {list(tools_by_name.keys())}",
                tool_call_id=tc["id"]
            ))
    return results

The complete loop with parallel execution

def agent_loop_parallel(user_input: str, max_iterations: int = 5) -> str:
    """Tool execution loop with parallel tool execution."""
    messages = [HumanMessage(content=user_input)]
    for i in range(max_iterations):
        response = model_with_tools.invoke(messages)
        messages.append(response)
        if not response.tool_calls:
            return response.content or "No response"

        if len(response.tool_calls) == 1:
            tc = response.tool_calls[0]
            try:
                result = tools_by_name[tc["name"]].invoke(tc["args"])
            except Exception as e:
                result = f"Error: {e}"
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
        else:
            tool_messages = execute_tools_parallel(response.tool_calls, tools_by_name)
            messages.extend(tool_messages)

    return "Iteration limit reached"

Timing comparison

start = time.time()
result = agent_loop_parallel("Weather in Madrid, Barcelona and Buenos Aires?")
elapsed = time.time() - start
print(f"Time (parallel): {elapsed:.1f}s")
# Time (parallel): ~2.0s  ← vs 6.0s serial (3x faster)

start = time.time()
result = agent_loop_parallel("Weather in Tokyo, price of AAPL, and how much is 1500 * 0.16?")
elapsed = time.time() - start
print(f"Time (mixed tools): {elapsed:.1f}s")
# get_weather: 2.0s, get_stock_price: 1.5s, calculator: ~0s → Total: 2.0s

Why max_workers=min(len(tool_calls), 10)

One worker per tool call guarantees real parallelism. The cap of 10 keeps you from spawning 50 threads if a model emits a lot of calls. For IO-bound operations (APIs, DB reads), creating threads is safe and cheap.

Implementation with asyncio

When asyncio instead of threads

If your tools are already async (they use aiohttp, httpx.AsyncClient, or LangChain's ainvoke), asyncio is the natural choice. It avoids thread overhead and scales better for many concurrent IO-bound calls.

Parallel execution with asyncio.gather

import asyncio
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
async def get_weather_async(city: str) -> str:
    """Get the current weather for a city."""
    await asyncio.sleep(2)
    climates = {
        "Madrid": "22°C, sunny", "Barcelona": "20°C, cloudy",
        "Buenos Aires": "18°C, rainy", "NYC": "12°C, partly cloudy",
    }
    return f"Weather in {city}: {climates.get(city, '18°C, variable')}"

@tool
async def get_stock_price_async(symbol: str) -> str:
    """Get the price of a stock."""
    await asyncio.sleep(1.5)
    prices = {"AAPL": 178.50, "GOOGL": 141.20, "MSFT": 415.30}
    price = prices.get(symbol.upper())
    return f"{symbol.upper()}: ${price}" if price else f"Error: '{symbol}' not found."


async def execute_tools_async(tool_calls, tools_by_name, timeout=30):
    """Run multiple tool calls in parallel with asyncio.gather."""
    async def run_one(tc):
        name = tc["name"]
        if name not in tools_by_name:
            return ToolMessage(content=f"Error: tool '{name}' does not exist.", tool_call_id=tc["id"])
        try:
            result = await asyncio.wait_for(
                tools_by_name[name].ainvoke(tc["args"]), timeout=timeout
            )
            return ToolMessage(content=str(result), tool_call_id=tc["id"])
        except asyncio.TimeoutError:
            return ToolMessage(content=f"Error: {name} exceeded the {timeout}s timeout", tool_call_id=tc["id"])
        except Exception as e:
            return ToolMessage(content=f"Error running {name}: {e}", tool_call_id=tc["id"])

    results = await asyncio.gather(*[run_one(tc) for tc in tool_calls])
    return list(results)


async def agent_loop_async(user_input: str, max_iterations: int = 5) -> str:
    """Async tool execution loop with parallel execution."""
    model = init_chat_model("openai:gpt-4.1-mini")
    async_tools = [get_weather_async, get_stock_price_async]
    tbn = {t.name: t for t in async_tools}
    mwt = model.bind_tools(async_tools)
    messages = [HumanMessage(content=user_input)]

    for i in range(max_iterations):
        response = await mwt.ainvoke(messages)
        messages.append(response)
        if not response.tool_calls:
            return response.content or "No response"
        tool_messages = await execute_tools_async(response.tool_calls, tbn)
        messages.extend(tool_messages)
    return "Iteration limit reached"

# Usage:
# result = await agent_loop_async("Weather in Madrid, Barcelona and Buenos Aires?")

Comparison: ThreadPoolExecutor vs asyncio

Recommendation: If your tools use requests, use ThreadPoolExecutor. If you're in an async codebase or need high concurrency, use asyncio.

Error handling in parallel execution

The problem: 1 out of 3 tools fails

In serial execution, you catch the error and move on. In parallel, all 3 are running at the same time. What do you do when one fails?

Strategy 1: Collect All (recommended)

Run all of them, catch errors individually, return successes + errors as ToolMessages. The model receives everything and decides.

def execute_tools_collect_all(tool_calls, tools_by_name, timeout=30):
    """Never loses results because of one individual error."""
    results = []
    with ThreadPoolExecutor(max_workers=min(len(tool_calls), 10)) as executor:
        futures = {}
        for tc in tool_calls:
            if tc["name"] not in tools_by_name:
                results.append(ToolMessage(
                    content=f"Error: tool '{tc['name']}' does not exist.",
                    tool_call_id=tc["id"]
                ))
                continue
            futures[executor.submit(tools_by_name[tc["name"]].invoke, tc["args"])] = tc

        for future in as_completed(futures):
            tc = futures[future]
            try:
                result = future.result(timeout=timeout)
                content = str(result)
            except TimeoutError:
                content = f"Error: {tc['name']} exceeded the {timeout}s timeout. Try different parameters."
            except Exception as e:
                content = f"Error running {tc['name']}: {type(e).__name__}: {e}. You can retry or answer without this data."
            results.append(ToolMessage(content=content, tool_call_id=tc["id"]))
    return results

The model receives ["22°C Madrid", "Error: timeout", "20°C Barcelona"] and answers: "Madrid is at 22°C, Barcelona at 20°C. I couldn't get the weather for the third city."

Strategy 2: Fail Fast

If every result is required (for example, a calculation that needs all 3 data points), cancel the rest when one fails:

def execute_tools_fail_fast(tool_calls, tools_by_name, timeout=30):
    """Cancels everything if one fails — use it when ALL results are mandatory."""
    with ThreadPoolExecutor(max_workers=len(tool_calls)) as executor:
        future_to_tc = {
            executor.submit(tools_by_name[tc["name"]].invoke, tc["args"]): tc
            for tc in tool_calls if tc["name"] in tools_by_name
        }
        results = []
        for future in as_completed(future_to_tc):
            tc = future_to_tc[future]
            try:
                results.append(ToolMessage(content=str(future.result(timeout=timeout)), tool_call_id=tc["id"]))
            except Exception as e:
                for f in future_to_tc:
                    f.cancel()
                return [ToolMessage(
                    content=f"Error: operation cancelled because {tc['name']} failed: {e}.",
                    tool_call_id=tc_orig["id"]
                ) for tc_orig in tool_calls]
    return results

Strategy 3: Partial retry

Run all of them in parallel, retry only the ones that failed:

def execute_tools_with_retry(tool_calls, tools_by_name, timeout=30, max_retries=2):
    """Retries the failing ones individually."""
    results = {}
    pending = list(tool_calls)
    for attempt in range(max_retries + 1):
        with ThreadPoolExecutor(max_workers=len(pending)) as executor:
            futures = {
                executor.submit(tools_by_name[tc["name"]].invoke, tc["args"]): tc
                for tc in pending if tc["name"] in tools_by_name
            }
            failed = []
            for future in as_completed(futures):
                tc = futures[future]
                try:
                    results[tc["id"]] = ToolMessage(content=str(future.result(timeout=timeout)), tool_call_id=tc["id"])
                except Exception as e:
                    if attempt == max_retries:
                        results[tc["id"]] = ToolMessage(
                            content=f"Error after {max_retries + 1} attempts: {e}", tool_call_id=tc["id"]
                        )
                    else:
                        failed.append(tc)
        pending = failed
        if not pending:
            break
    return [results[tc["id"]] for tc in tool_calls if tc["id"] in results]

When to use each strategy

When to use it and when NOT to

Decision table

Explicit trade-offs

The central trade-off: parallel calls = more throughput but more error handling complexity. If your agent calls 1-2 tools per turn, it isn't worth it. If it regularly calls 3+, the latency gain is worth the complexity.

A semaphore for rate limits

If you need parallel but with a concurrency limit:

import threading

api_semaphore = threading.Semaphore(3)

def rate_limited_invoke(tool, args):
    with api_semaphore:
        return tool.invoke(args)

Connection with the project

In this module's project (capsule 08), you'll build an extraction + routing system. Parallel function calling is key because:

• The system extracts multiple entities simultaneously (people, companies, dates, amounts) — 4 extraction calls in parallel
• Each entity is routed to a specialized processor — the processors are independent and run in parallel
• The combination of parallel extraction + parallel routing cuts latency significantly

In the evolving project (M4-10):

• M4: The pattern becomes the logic of a StateGraph node
• M5: Planning runs multiple searches simultaneously
• M8: Multi-agent systems run multiple agents in parallel
• M9: Performance tests verify that parallel reduces latency vs serial

Troubleshooting

Problem 1: "The results arrive in a different order"

Cause: as_completed returns futures in the order they finish, not the order they were submitted.

Solution: It doesn't matter — the model pairs them by tool_call_id. If you need order for logging:

results_by_id = {}
for future in as_completed(futures):
    tc = futures[future]
    results_by_id[tc["id"]] = ToolMessage(content=str(future.result()), tool_call_id=tc["id"])
ordered = [results_by_id[tc["id"]] for tc in tool_calls]

Problem 2: "ThreadPoolExecutor hangs"

Cause: A tool takes longer than the timeout or is stuck in an infinite loop.

Solution: Always use a timeout in future.result() and in the tool's internal requests:

result = future.result(timeout=30)           # In the executor
response = requests.get(url, timeout=10)     # Inside the tool

Problem 3: "Rate limit when running many tools in parallel"

Cause: N simultaneous calls to the same API trigger rate limiting (429).

Solution: A semaphore that caps concurrency. In async:

sem = asyncio.Semaphore(3)
async def rate_limited(tool, args):
    async with sem:
        return await tool.ainvoke(args)

Problem 4: "I don't know whether to use threads or asyncio"

Solution:

For agent tools, 95% are IO-bound → ThreadPoolExecutor is almost always the right answer.

Exercises

Exercise 1: Measure the serial vs parallel difference (Easy)

Create 4 tools that simulate latency with time.sleep() (1s, 2s, 1.5s, 3s). Implement serial and parallel execution. Measure and compare the times.

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain_core.tools import tool

@tool
def api_fast(query: str) -> str:
    """Fast API — 1s."""
    time.sleep(1)
    return f"Fast: {query}"

@tool
def api_medium(query: str) -> str:
    """Medium API — 2s."""
    time.sleep(2)
    return f"Medium: {query}"

@tool
def api_slow(query: str) -> str:
    """Slow API — 1.5s."""
    time.sleep(1.5)
    return f"Slow: {query}"

@tool
def api_very_slow(query: str) -> str:
    """Very slow API — 3s."""
    time.sleep(3)
    return f"Very slow: {query}"

fake_calls = [
    {"name": "api_fast", "args": {"query": "t"}, "id": "1"},
    {"name": "api_medium", "args": {"query": "t"}, "id": "2"},
    {"name": "api_slow", "args": {"query": "t"}, "id": "3"},
    {"name": "api_very_slow", "args": {"query": "t"}, "id": "4"},
]
tools_map = {t.name: t for t in [api_fast, api_medium, api_slow, api_very_slow]}

start = time.time()
for tc in fake_calls:
    tools_map[tc["name"]].invoke(tc["args"])
print(f"Serial:   {time.time() - start:.1f}s")  # ~7.5s

start = time.time()
with ThreadPoolExecutor(max_workers=4) as ex:
    futs = {ex.submit(tools_map[tc["name"]].invoke, tc["args"]): tc for tc in fake_calls}
    for f in as_completed(futs):
        f.result()
print(f"Parallel: {time.time() - start:.1f}s")  # ~3.0s (speedup ~2.5x)

Exercise 2: Parallel execution with logging (Medium)

Modify execute_tools_parallel to print when each tool starts and finishes, with relative timestamps. Expected output:

[0.00s] Starting get_weather(Madrid)
[0.00s] Starting get_weather(Barcelona)
[2.01s] Completed get_weather(Madrid) → "22°C, sunny"
[2.02s] Completed get_weather(Barcelona) → "20°C, cloudy"
[2.02s] Total: 2 tools in 2.02s

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain_core.messages import ToolMessage

def execute_tools_parallel_logged(tool_calls, tools_by_name, timeout=30):
    results = []
    t0 = time.time()

    def run_tool(tc):
        print(f"[{time.time()-t0:.2f}s] Starting {tc['name']}({tc['args']})")
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        print(f"[{time.time()-t0:.2f}s] Completed {tc['name']} → \"{result}\"")
        return result

    with ThreadPoolExecutor(max_workers=len(tool_calls)) as executor:
        futures = {executor.submit(run_tool, tc): tc for tc in tool_calls}
        for future in as_completed(futures):
            tc = futures[future]
            try:
                result = future.result(timeout=timeout)
            except Exception as e:
                result = f"Error: {e}"
            results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

    print(f"[{time.time()-t0:.2f}s] Total: {len(tool_calls)} tools in {time.time()-t0:.2f}s")
    return results

Exercise 3: Concurrency semaphore (Medium)

You have 5 tool calls to the same API with a rate limit of at most 2 concurrent. Implement a semaphore. Measure the time and verify it's ~3s (batches of 2 at 1s each) instead of ~1s (no limit) or ~5s (serial).

import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain_core.tools import tool

api_semaphore = threading.Semaphore(2)

@tool
def search_api(query: str) -> str:
    """Search for information. Rate limit: 2 concurrent."""
    with api_semaphore:
        time.sleep(1)
        return f"Result: {query}"

calls = [{"name": "search_api", "args": {"query": f"q{i}"}, "id": f"c{i}"} for i in range(5)]
tmap = {"search_api": search_api}

start = time.time()
with ThreadPoolExecutor(max_workers=5) as ex:
    futs = {ex.submit(tmap[c["name"]].invoke, c["args"]): c for c in calls}
    for f in as_completed(futs):
        f.result()
print(f"With semaphore(2): {time.time()-start:.1f}s")
# ~3s: batch 1 (2 calls, 1s), batch 2 (2 calls, 1s), batch 3 (1 call, 1s)
# vs ~1s without a semaphore, vs ~5s serial

Exercise 4: Async parallel with individual timeouts (Medium)

Implement parallel async execution where each tool has its own timeout. If slow_api has a 3s timeout but takes 5s, only that one fails — the rest complete normally.

import asyncio
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage

@tool
async def fast_api(query: str) -> str:
    """Fast API."""
    await asyncio.sleep(0.5)
    return f"Fast: {query}"

@tool
async def slow_api(query: str) -> str:
    """Slow API."""
    await asyncio.sleep(5)
    return f"Slow: {query}"

TOOL_TIMEOUTS = {"fast_api": 2, "slow_api": 3}

async def execute_with_individual_timeouts(tool_calls, tools_by_name):
    async def run(tc):
        timeout = TOOL_TIMEOUTS.get(tc["name"], 10)
        try:
            result = await asyncio.wait_for(
                tools_by_name[tc["name"]].ainvoke(tc["args"]), timeout=timeout
            )
            return ToolMessage(content=str(result), tool_call_id=tc["id"])
        except asyncio.TimeoutError:
            return ToolMessage(
                content=f"Error: {tc['name']} exceeded the {timeout}s timeout",
                tool_call_id=tc["id"]
            )

    return list(await asyncio.gather(*[run(tc) for tc in tool_calls]))

# Test
async def test():
    tmap = {t.name: t for t in [fast_api, slow_api]}
    calls = [
        {"name": "fast_api", "args": {"query": "a"}, "id": "1"},
        {"name": "slow_api", "args": {"query": "b"}, "id": "2"},
    ]
    for r in await execute_with_individual_timeouts(calls, tmap):
        print(f"  {r.tool_call_id}: {r.content}")
    # 1: Fast: a                         ← 0.5s < timeout 2s ✅
    # 2: Error: slow_api exceeded...     ← 5s > timeout 3s ❌

await test()

Exercise 5: Complete agent loop with parallel + metrics (Hard)

Build a loop that combines: (1) parallel execution with ThreadPoolExecutor, (2) retry for tools that fail, (3) time metrics per tool, total, and a count of errors/retries. Return a dict with the response + metrics.

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather."""
    time.sleep(2)
    climates = {"Madrid": "22°C, sunny", "Barcelona": "20°C, cloudy"}
    return f"{city}: {climates.get(city, '18°C, variable')}"

@tool
def calculator(expression: str) -> str:
    """Evaluate mathematical expressions."""
    allowed = set("0123456789+-*/(). ")
    if not all(c in allowed for c in expression):
        return "Error: characters not allowed"
    return str(eval(expression))

def run_with_retry(tool_fn, args, max_retries=2):
    start = time.time()
    for attempt in range(max_retries + 1):
        try:
            result = str(tool_fn.invoke(args))
            return result, attempt + 1, time.time() - start
        except Exception as e:
            if attempt == max_retries:
                return f"Error after {max_retries+1} attempts: {e}", attempt + 1, time.time() - start
            time.sleep(0.5 * (2 ** attempt))

def agent_loop_full(user_input, max_iterations=5, max_workers=10, max_retries=2):
    model = init_chat_model("openai:gpt-4.1-mini")
    all_tools = [get_weather, calculator]
    tbn = {t.name: t for t in all_tools}
    mwt = model.bind_tools(all_tools)
    messages = [HumanMessage(content=user_input)]
    metrics = {"tool_times": {}, "total_calls": 0, "retries": 0, "errors": []}
    t0 = time.time()

    for i in range(max_iterations):
        response = mwt.invoke(messages)
        messages.append(response)
        if not response.tool_calls:
            metrics["total_time"] = time.time() - t0
            metrics["iterations"] = i + 1
            return {"response": response.content or "No response", "metrics": metrics}

        metrics["total_calls"] += len(response.tool_calls)
        workers = min(len(response.tool_calls), max_workers)

        with ThreadPoolExecutor(max_workers=workers) as executor:
            futures = {
                executor.submit(run_with_retry, tbn[tc["name"]], tc["args"], max_retries): tc
                for tc in response.tool_calls if tc["name"] in tbn
            }
            for future in as_completed(futures):
                tc = futures[future]
                result, attempts, elapsed = future.result(timeout=60)
                metrics["tool_times"][tc["name"]] = elapsed
                metrics["retries"] += attempts - 1
                if result.startswith("Error"):
                    metrics["errors"].append({"tool": tc["name"], "error": result})
                messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

    metrics["total_time"] = time.time() - t0
    return {"response": "Iteration limit", "metrics": metrics}

output = agent_loop_full("Weather in Madrid and Barcelona, and how much is 99*77?")
print(f"Response: {output['response']}")
print(f"Total time: {output['metrics']['total_time']:.1f}s")
print(f"Tool calls: {output['metrics']['total_calls']}")
print(f"Retries: {output['metrics']['retries']}")
# Response: Madrid 22°C, Barcelona 20°C, 99×77 = 7623
# Total time: ~2.5s (parallel)
# Tool calls: 3, Retries: 0

Summary

In this capsule you learned:

• Parallel function calling has two halves: the model emits multiple tool_calls in one AIMessage, and your code runs them simultaneously
• Serial execution adds up all the latencies; parallel execution takes the latency of the slowest tool
• ThreadPoolExecutor is the direct solution for sync tools — as_completed to collect results as they finish
• asyncio.gather is the solution for async tools — lighter, better cancellation, scales to thousands of coroutines
• Error handling in parallel requires a strategy: Collect All (gather everything), Fail Fast (cancel if one fails), or Partial retry
• The central trade-off: more throughput but more complexity in error handling and debugging
• Semaphores control maximum concurrency so you respect rate limits without being fully serial

Next capsule: Forced Tool Calls and Routing — how to force the model to use a specific tool with tool_choice, and how to make different tools available depending on the context.

Additional resources

1. OpenAI — Parallel Function Calling — Official documentation on parallel tool calls
2. Python concurrent.futures — Reference for ThreadPoolExecutor and ProcessPoolExecutor
3. Python asyncio — Gathering Tasks — Reference for asyncio.gather
4. LangChain Tool Calling — How To — Official tool calling guide
5. Anthropic Tool Use — How Claude handles parallel tool calls
6. Semaphore Pattern — Using semaphores to control concurrency