{"record":{"id":"e626a2b5d1c18b5d","repo":"langchain-ai/langchain","slug":"runnableeach-does-not-support-astream-events-yet","errorCode":null,"errorMessage":"RunnableEach does not support astream_events yet.","messagePattern":"RunnableEach does not support astream_events yet\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"libs/core/langchain_core/runnables/base.py","lineNumber":5714,"sourceCode":"        )\n\n    @override\n    def astream_events(  # type: ignore[override]\n        self,\n        input: Input,\n        config: RunnableConfig | None = None,\n        *,\n        version: Literal[\"v1\", \"v2\", \"v3\"] = \"v2\",\n        **kwargs: Any | None,\n    ) -> AsyncIterator[StreamEvent] | Awaitable[Any]:\n        del input, config, kwargs\n        if version == \"v3\":\n            return self._astream_events_unsupported_v3()\n        return self._astream_events_unsupported_v1_v2()\n\n    async def _astream_events_unsupported_v3(self) -> Any:\n        msg = \"RunnableEach does not support astream_events yet.\"\n        raise NotImplementedError(msg)\n\n    async def _astream_events_unsupported_v1_v2(self) -> AsyncIterator[StreamEvent]:\n        msg = \"RunnableEach does not support astream_events yet.\"\n        raise NotImplementedError(msg)\n        yield  # type: ignore[unreachable] # makes this an async generator (never reached)\n\n\nclass RunnableEach(RunnableEachBase[Input, Output]):\n    \"\"\"RunnableEach class.\n\n    `Runnable` that calls another `Runnable` for each element of the input sequence.\n\n    It allows you to call multiple inputs with the bounded `Runnable`.\n\n    `RunnableEach` makes it easy to run multiple inputs for the `Runnable`.\n    In the below example, we associate and run three inputs\n    with a `Runnable`:\n","sourceCodeStart":5696,"sourceCodeEnd":5732,"githubUrl":"https://github.com/langchain-ai/langchain/blob/e32fa9a52eab3b61ad7a45399bfde59b3e580fc4/libs/core/langchain_core/runnables/base.py#L5696-L5732","documentation":"`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.","triggerScenarios":"`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.","commonSituations":"Debugging batch pipelines with event tracing; observability integrations (LangSmith-style event capture) traversing chains containing `.map()`; migrating tracing code from v1/v2 to v3.","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."],"exampleFix":"# before\nasync for ev in my_runnable.map().astream_events(items, version='v3'):  # NotImplementedError\n    print(ev)\n\n# after\nresults = await asyncio.gather(*[\n    consume(my_runnable.astream_events(item, version='v3')) for item in items\n])","handlingStrategy":"try-catch","validationCode":"from langchain_core.runnables.base import RunnableEachBase\n\ndef supports_astream_events(r) -> bool:\n    return not isinstance(r, RunnableEachBase)","typeGuard":"from langchain_core.runnables.base import RunnableEachBase\n\ndef is_runnable_each(r) -> bool:\n    return isinstance(r, RunnableEachBase)","tryCatchPattern":"try:\n    async for ev in mapped.astream_events(x, version='v3'):\n        handle(ev)\nexcept NotImplementedError as e:\n    if 'astream_events' in str(e):\n        for item in x:\n            async for ev in inner.astream_events(item, version='v3'):\n                handle(ev)\n    else:\n        raise","preventionTips":["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."],"tags":["runnable","runnable-each","map","astream-events","not-implemented","tracing"],"backgroundTag":null,"analyzedSha":"e32fa9a52eab3b61ad7a45399bfde59b3e580fc4","analyzedAt":"2026-08-14T18:42:09.092Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}