`RunnableParallel` ignores `max_concurrency` on `ainvoke`/`astream` while `invoke`/`stream` honour it

Submission checklist

• This is a bug, not a usage question.
• I added a clear and descriptive title that summarizes this issue.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain rather than my code.
• The bug is not resolved by updating to the latest stable version of LangChain (or the specific integration package).
• This is not related to the langchain-community package.
• I posted a self-contained, minimal, reproducible example. A maintainer can copy it and run it AS IS.

Package (Required)

• langchain
• langchain-openai
• langchain-anthropic
• langchain-classic
• langchain-core
• langchain-model-profiles
• langchain-tests
• langchain-text-splitters
• langchain-chroma
• langchain-deepseek
• langchain-exa
• langchain-fireworks
• langchain-groq
• langchain-huggingface
• langchain-mistralai
• langchain-nomic
• langchain-ollama
• langchain-openrouter
• langchain-perplexity
• langchain-qdrant
• langchain-typesafe
• langchain-xai
• Other / not sure / general

Related Issues / PRs

No response

Reproduction Steps / Example Code (Python)

"""Repro: RunnableParallel ignores max_concurrency on ainvoke/astream.

 

Prints the peak number of branches running at the same time for invoke and ainvoke.

invoke caps it through its thread pool; ainvoke never reads the key.

Exit code 1 means the two paths disagree.

"""

import asyncio

import time

 

from langchain_core.runnables import RunnableLambda, RunnableParallel

 

BRANCHES, CAP = 4, 1

state = {"n": 0, "peak": 0}

 

 

def bump(delta):

    state["n"] += delta

    state["peak"] = max(state["peak"], state["n"])

 

 

def sync_branch(_):

    bump(1)

    time.sleep(0.1)

    bump(-1)

    return "x"

 

 

async def async_branch(_):

    bump(1)

    await asyncio.sleep(0.1)

    bump(-1)

    return "x"

 

 

def measure(run):

    state["n"] = state["peak"] = 0

    run()

    return state["peak"]

 

 

sync_par = RunnableParallel(**{f"b{i}": RunnableLambda(sync_branch) for i in range(BRANCHES)})

async_par = RunnableParallel(**{f"b{i}": RunnableLambda(async_branch) for i in range(BRANCHES)})

 

sync_peak = measure(lambda: sync_par.invoke({}, {"max_concurrency": CAP}))

async_peak = measure(lambda: asyncio.run(async_par.ainvoke({}, {"max_concurrency": CAP})))

 

print(f"max_concurrency={CAP}, {BRANCHES} branches, each sleeping 0.1s")

print(f"  invoke  peak concurrent branches: {sync_peak}")

print(f"  ainvoke peak concurrent branches: {async_peak}")

diverged = sync_peak != async_peak

print("\nFAIL: ainvoke ignored max_concurrency" if diverged else "\nOK: both paths honoured it")

raise SystemExit(1 if diverged else 0)

Error Message and Stack Trace (if applicable)

max_concurrency=1, 4 branches, each sleeping 0.1s

  invoke  peak concurrent branches: 1

  ainvoke peak concurrent branches: 4

FAIL: ainvoke ignored max_concurrency

Description

I was setting max_concurrency to keep a downstream API happy and it worked on invoke but not on ainvoke. With four branches and max_concurrency=1 the sync path peaks at 1 concurrent branch while the async path peaks at 4; at max_concurrency=2 it is 2 vs 4. So the async side never reads the key.

RunnableParallel.invoke (libs/core/langchain_core/runnables/base.py, lines 4139-4196) gets its executor from get_executor_for_config(config), where max_workers is config.get("max_concurrency") (runnables/config.py:673). On the async side RunnableParallel.ainvoke (lines 4199-4251) just does await asyncio.gather(...) at line 4232, and _atransform (lines 4319-4367) creates one task per branch at lines 4345 and 4364. Neither one reads the key, and neither uses gather_with_concurrency (runnables/utils.py:65), which abatch and abatch_as_completed already do.

This matters because async is the path most things deploy on. A config that reads like a limit is not one there, and if the cap was there to stay under a downstream rate limit, the burst is exactly what you were trying to avoid.

I would expect the async paths to go through gather_with_concurrency or an equivalent semaphore. If parallel branches are meant to be exempt from this key, then the sync path should probably not honour it either, and either way it would help to have that written down.

System Info

langchain-core: 1.6.5

langchain: 1.4.2

langgraph: 1.2.11

pydantic: 2.12.5

Python: 3.11.15

upstream commit tested: 1ef23d6b7f6d83508a8cb531feeb88860e4dec40

OS: Windows 10 + WSL

No API key or network access is needed for the reproduction.

Social handles (optional)

No response