langchain-ai/langchain · error · TypeError

Cannot stream from a generator function asynchronously.Use .

Error message

Cannot stream from a generator function asynchronously.Use .stream() instead.

What it means

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()`.

Source

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

            # So we'll iterate until we get to the last chunk!
            if not got_first_val:
                final = ichunk
                got_first_val = True
            else:
                try:
                    final = final + ichunk  # type: ignore[operator]
                except TypeError:
                    final = ichunk

        if hasattr(self, "afunc"):
            afunc = self.afunc
        else:
            if inspect.isgeneratorfunction(self.func):
                msg = (
                    "Cannot stream from a generator function asynchronously."
                    "Use .stream() instead."
                )
                raise TypeError(msg)

            def func(
                input_: Input,
                run_manager: AsyncCallbackManagerForChainRun,
                config: RunnableConfig,
                **kwargs: Any,
            ) -> Output:
                return call_func_with_variable_args(
                    self.func, input_, config, run_manager.get_sync(), **kwargs
                )

            @wraps(func)
            async def f(*args: Any, **kwargs: Any) -> Any:
                return await run_in_executor(config, func, *args, **kwargs)

            afunc = f

        if is_async_generator(afunc):

View on GitHub (pinned to e32fa9a52e)

Solutions

1. Call the sync stream from a thread: `run_in_executor` around `list(r.stream(x))`.
2. 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`.
3. Chunk-bridge: iterate the sync generator in a worker thread pushing into an `asyncio.Queue`.
4. Reserve generator lambdas for sync pipelines only.

Example fix

# before
def gen(x):
    for t in x.split():
        yield t
r = RunnableLambda(gen)
async for c in r.astream('a b'): ...  # TypeError

# after
async def agen(x):
    for t in x.split():
        yield t
r = RunnableLambda(agen)
async for c in r.astream('a b'): ...

import inspect

def supports_async_stream(r) -> bool:
    return hasattr(r, 'afunc') or not inspect.isgeneratorfunction(getattr(r, 'func', None))

import inspect

def is_safe_for_astream(r) -> bool:
    if hasattr(r, 'afunc'):
        return True
    fn = getattr(r, 'func', None)
    return fn is not None and not inspect.isgeneratorfunction(fn)

try:
    async for chunk in r.astream(x):
        ...
except TypeError as e:
    if 'Use .stream()' in str(e):
        chunks = await asyncio.to_thread(lambda: list(r.stream(x)))
    else:
        raise

When it happens

Trigger: `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.

Common situations: Writing a sync generator lambda for token streaming, then serving it from FastAPI with `.astream`; refactoring sync chains to async while keeping generator lambdas.

Related errors

• {self!r} only supports sync methods.
• Cannot invoke a coroutine function synchronously.Use `ainvok
• Cannot stream a coroutine function synchronously.Use `astrea
• {self!r} only supports async methods.
• Func was provided as a coroutine function, but afunc was als

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