{"record":{"id":"f55a5d9f982220c7","repo":"langchain-ai/langchain","slug":"cannot-stream-from-a-generator-function-asynchrono","errorCode":null,"errorMessage":"Cannot stream from a generator function asynchronously.Use .stream() instead.","messagePattern":"Cannot stream from a generator function asynchronously\\.Use \\.stream\\(\\) instead\\.","errorType":"exception","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"libs/core/langchain_core/runnables/base.py","lineNumber":5468,"sourceCode":"            # So we'll iterate until we get to the last chunk!\n            if not got_first_val:\n                final = ichunk\n                got_first_val = True\n            else:\n                try:\n                    final = final + ichunk  # type: ignore[operator]\n                except TypeError:\n                    final = ichunk\n\n        if hasattr(self, \"afunc\"):\n            afunc = self.afunc\n        else:\n            if inspect.isgeneratorfunction(self.func):\n                msg = (\n                    \"Cannot stream from a generator function asynchronously.\"\n                    \"Use .stream() instead.\"\n                )\n                raise TypeError(msg)\n\n            def func(\n                input_: Input,\n                run_manager: AsyncCallbackManagerForChainRun,\n                config: RunnableConfig,\n                **kwargs: Any,\n            ) -> Output:\n                return call_func_with_variable_args(\n                    self.func, input_, config, run_manager.get_sync(), **kwargs\n                )\n\n            @wraps(func)\n            async def f(*args: Any, **kwargs: Any) -> Any:\n                return await run_in_executor(config, func, *args, **kwargs)\n\n            afunc = f\n\n        if is_async_generator(afunc):","sourceCodeStart":5450,"sourceCodeEnd":5486,"githubUrl":"https://github.com/langchain-ai/langchain/blob/e32fa9a52eab3b61ad7a45399bfde59b3e580fc4/libs/core/langchain_core/runnables/base.py#L5450-L5486","documentation":"In `RunnableLambda._atransform`, when no `afunc` was provided the class wraps the sync `func` for async use — but only if `func` is a normal function. A sync generator function cannot be adapted into the async streaming path (its lazy pull semantics do not map onto `async for`), so attempting `astream()` raises a `TypeError` pointing to `.stream()`.","triggerScenarios":"`RunnableLambda(sync_generator_fn).astream(x)` or `.ainvoke(x)`; `async for` over a chain containing a lambda defined with `def f(x): yield ...`; async servers wrapping sync generator lambdas.","commonSituations":"Writing a sync generator lambda for token streaming, then serving it from FastAPI with `.astream`; refactoring sync chains to async while keeping generator lambdas.","solutions":["Call the sync stream from a thread: `run_in_executor` around `list(r.stream(x))`.","Write an async generator for the async path: `async def f(x): ... yield` and pass it as `func` (single async generator) or `afunc` with a sync `func`.","Chunk-bridge: iterate the sync generator in a worker thread pushing into an `asyncio.Queue`.","Reserve generator lambdas for sync pipelines only."],"exampleFix":"# before\ndef gen(x):\n    for t in x.split():\n        yield t\nr = RunnableLambda(gen)\nasync for c in r.astream('a b'): ...  # TypeError\n\n# after\nasync def agen(x):\n    for t in x.split():\n        yield t\nr = RunnableLambda(agen)\nasync for c in r.astream('a b'): ...","handlingStrategy":"type-guard","validationCode":"import inspect\n\ndef supports_async_stream(r) -> bool:\n    return hasattr(r, 'afunc') or not inspect.isgeneratorfunction(getattr(r, 'func', None))","typeGuard":"import inspect\n\ndef is_safe_for_astream(r) -> bool:\n    if hasattr(r, 'afunc'):\n        return True\n    fn = getattr(r, 'func', None)\n    return fn is not None and not inspect.isgeneratorfunction(fn)","tryCatchPattern":"try:\n    async for chunk in r.astream(x):\n        ...\nexcept TypeError as e:\n    if 'Use .stream()' in str(e):\n        chunks = await asyncio.to_thread(lambda: list(r.stream(x)))\n    else:\n        raise","preventionTips":["Do not pass sync generator functions to RunnableLambda for async pipelines.","Write async generator variants (afunc) for async streaming.","Bridge sync generators with asyncio.to_thread."],"tags":["runnable","runnable-lambda","streaming","sync-async-mismatch","asyncio"],"backgroundTag":null,"analyzedSha":"e32fa9a52eab3b61ad7a45399bfde59b3e580fc4","analyzedAt":"2026-08-14T18:42:09.092Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}