What causes "RunnableEach does not support astream_events yet." in langchain-ai/langchain?

Use plain streaming instead: `async for chunk in mapped.astream(x): ...`.

langchain-ai/langchain · error · NotImplementedError

RunnableEach does not support astream_events yet.

Error message

RunnableEach does not support astream_events yet.

What it means

`RunnableEachBase` (the base behind `RunnableEach`, produced by `Runnable.map()`) deliberately does not implement event streaming for the v3 protocol: `astream_events` raises `NotImplementedError` via `_astream_events_unsupported_v3`. Event capture requires per-run event instrumentation that the map wrapper does not provide, so any v3 request fails fast rather than silently dropping events.

Source

Thrown at libs/core/langchain_core/runnables/base.py:5714

        )

    @override
    def astream_events(  # type: ignore[override]
        self,
        input: Input,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2", "v3"] = "v2",
        **kwargs: Any | None,
    ) -> AsyncIterator[StreamEvent] | Awaitable[Any]:
        del input, config, kwargs
        if version == "v3":
            return self._astream_events_unsupported_v3()
        return self._astream_events_unsupported_v1_v2()

    async def _astream_events_unsupported_v3(self) -> Any:
        msg = "RunnableEach does not support astream_events yet."
        raise NotImplementedError(msg)

    async def _astream_events_unsupported_v1_v2(self) -> AsyncIterator[StreamEvent]:
        msg = "RunnableEach does not support astream_events yet."
        raise NotImplementedError(msg)
        yield  # type: ignore[unreachable] # makes this an async generator (never reached)


class RunnableEach(RunnableEachBase[Input, Output]):
    """RunnableEach class.

    `Runnable` that calls another `Runnable` for each element of the input sequence.

    It allows you to call multiple inputs with the bounded `Runnable`.

    `RunnableEach` makes it easy to run multiple inputs for the `Runnable`.
    In the below example, we associate and run three inputs
    with a `Runnable`:

View on GitHub (pinned to e32fa9a52e)

Solutions

Use plain streaming instead: `async for chunk in mapped.astream(x): ...`.
Emit events from the inner runnable: call `astream_events` on `my_runnable` per element rather than on the mapped wrapper.
Replace `.map()` with explicit `asyncio.gather` over `runnable.astream_events(...)` per item.
Request a supported version only if the object actually implements it — for `RunnableEach` no version is supported, so avoid `astream_events` on it entirely.

Example fix

# before
async for ev in my_runnable.map().astream_events(items, version='v3'):  # NotImplementedError
    print(ev)

# after
results = await asyncio.gather(*[
    consume(my_runnable.astream_events(item, version='v3')) for item in items
])

Defensive patterns

Strategy: try-catch

Validate before calling

from langchain_core.runnables.base import RunnableEachBase

def supports_astream_events(r) -> bool:
    return not isinstance(r, RunnableEachBase)

Type guard

from langchain_core.runnables.base import RunnableEachBase

def is_runnable_each(r) -> bool:
    return isinstance(r, RunnableEachBase)

Try / catch

try:
    async for ev in mapped.astream_events(x, version='v3'):
        handle(ev)
except NotImplementedError as e:
    if 'astream_events' in str(e):
        for item in x:
            async for ev in inner.astream_events(item, version='v3'):
                handle(ev)
    else:
        raise

Prevention

Never call astream_events on `.map()` results; target the inner runnable.
Check isinstance(r, RunnableEachBase) before event streaming.
Use astream() plus callbacks for visibility into mapped stages.

When it happens

Trigger: `some_runnable.map().astream_events(input, version='v3')`; calling `astream_events(version='v3')` on a chain that contains a `.map()` stage; tracing tools that default to v3 and walk into mapped runnables.

Common situations: Debugging batch pipelines with event tracing; observability integrations (LangSmith-style event capture) traversing chains containing `.map()`; migrating tracing code from v1/v2 to v3.

Related errors

AI-assisted analysis of langchain-ai/langchain@e32fa9a52e (2026-08-14). Data as JSON: /api/errors/e626a2b5d1c18b5d. Report an issue: GitHub.