langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

astream_events does not stream data with LangGraph nested in a Runnable #967

New Issue
Closed
opened 2026-02-20 17:42:33 -05:00 by yindo · 2 comments
yindo commented 2026-02-20 17:42:33 -05:00
Owner
Copy Link
Copy Source

Originally created by @sundaraa-deshaw on GitHub (Sep 8, 2025).

Checked other resources

  • This is a bug, not a usage question. For questions, please use the LangChain Forum (https://forum.langchain.com/).
  • I added a clear and detailed title that summarizes the issue.
  • I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
  • I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

from langgraph.graph import END, START, MessagesState, StateGraph
from langchain_core.runnables import Runnable
from langgraph.config import get_stream_writer
from langchain_core.callbacks.manager import (
    adispatch_custom_event,
    dispatch_custom_event,
)
from langchain_core.messages import AIMessage
from langchain_core.runnables import Runnable
from langchain_core.callbacks import CallbackManager, CallbackManagerForChainRun
from langchain_core.callbacks.manager import (
    adispatch_custom_event,
    dispatch_custom_event,
)
from langchain_core.runnables import Runnable, ensure_config
from langchain_core.runnables.config import RunnableConfig, patch_config
from langchain.chains.base import Chain
from langchain_core.load.dump import dumpd

from rich.pretty import pprint

def node_1(state: MessagesState):
    return {
        "messages": AIMessage(content="message from node_1"),
    }

def get_graph():
    return (
        StateGraph(MessagesState)
        .add_node(node_1)
        .add_edge(START, "node_1")
        .compile()
    )

class OuterRunnable(Runnable):
    def invoke(self, inputs):
        return {"foo": "from outer runnable"}

    async def astream(self, inputs, config=None, **kwargs):
        async for chunk in get_graph().astream(inputs, config=config, **kwargs):
            pass # yield chunk

        yield {"foo": "bar"}

async for chunk in OuterRunnable().astream({}):
    pprint(chunk)

print('----')

async for chunk in OuterRunnable().astream_events({'messages': []}, version='v2', stream_mode=['custom', 'values']):
    if chunk['event'] == 'on_chain_stream':
        pprint(chunk['data'])

Error Message and Stack Trace (if applicable)

{'foo': 'bar'}
----
{'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}}


Expected:

{'foo': 'bar'}
----
{'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}}
{'chunk': {'data: {'foo': 'bar'}}}

Description

I am trying to stream chunks from a LangGraph graph inside a Runnable with astream_events (to get stream data and events)

While the following works:
Streaming the graph results from the outer runnable with astream_events

class OuterRunnable(Runnable):
    def invoke(self, inputs):
        return {"foo": "from outer runnable"}

    async def astream(self, inputs, config=None, **kwargs):
        async for chunk in get_graph().astream(inputs, config=config, **kwargs):
            yield chunk

        yield {"foo": "bar"}

and produces:

{
│   'node_1': {
│   │   'messages': AIMessage(
│   │   │   content='message from node_1',
│   │   │   additional_kwargs={},
│   │   │   response_metadata={},
│   │   │   id='6e06f484-b26d-46f5-87f4-de1fc4eeda09'
│   │   )
│   }
}
{'foo': 'bar'}
----
{'chunk': ('values', {'messages': []})}
{'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}}
{
│   'chunk': (
│   │   'values',
│   │   {
│   │   │   'messages': [
│   │   │   │   AIMessage(
│   │   │   │   │   content='message from node_1',
│   │   │   │   │   additional_kwargs={},
│   │   │   │   │   response_metadata={},
│   │   │   │   │   id='fc0c4296-4f12-47ff-8249-554566e4fa75'
│   │   │   │   )
│   │   │   ]
│   │   }
│   )
}
{'chunk': {'foo': 'bar'}}

the reproducer which just streams the graph, and does not yield the graph results, does not produce the final yielded chunk from the outer runnable (the dict foo: bar)

System Info

System Information
------------------
> OS:  Linux
> OS Version:  #1 SMP Fri Jun 27 05:59:19 EDT 2025
> Python Version:  3.11.8 (main, Sep 17 2024, 14:28:20) [GCC 10.3.1 20210422 (Red Hat 10.3.1-1)]

Package Information
-------------------
> langchain_core: 0.3.55
> langchain: 0.3.14
> langchain_community: 0.3.14
> langsmith: 0.2.10
> langchain_anthropic: 0.2.4
> langchain_cli: 0.0.31
> langchain_experimental: 0.3.4
> langchain_google_vertexai: 2.0.21
> langchain_mcp_adapters: 0.1.8
> langchain_milvus: 0.2.1
> langchain_openai: 0.2.14
> langchain_text_splitters: 0.3.5
> langchainplus_sdk: 0.0.21
> langgraph_gen: 0.0.6
> langgraph_reflection: 0.0.1
> langgraph_sdk: 0.1.60
> langgraph_supervisor: 0.0.20
> langgraph_swarm: 0.0.11

Optional packages not installed
-------------------------------
> langserve

Other Dependencies
------------------
> aiohttp: 3.8.4
> anthropic: 0.30.0
> anthropic[vertexai]: Installed. No version info available.
> async-timeout: 4.0.3
> bottleneck: 1.3.7
> dataclasses-json: 0.6.3
> defusedxml: 0.7.1
> gitpython: 3.1.40
> google-cloud-aiplatform: 1.89.0
> google-cloud-storage: 2.13.0
> gritql: Installed. No version info available.
> httpx: 0.27.0
> httpx-sse: 0.4.0
> jinja2>=3.1.5: Installed. No version info available.
> jsonpatch<2.0,>=1.33: Installed. No version info available.
> langchain-core<0.4,>=0.3.36: Installed. No version info available.
> langchain-core<0.4.0,>=0.3.40: Installed. No version info available.
> langchain-mistralai: Installed. No version info available.
> langchain>=0.1.0: Installed. No version info available.
> langgraph: 0.4.6
> langgraph-prebuilt<0.2.0,>=0.1.7: Installed. No version info available.
> langgraph>=0.2.74: Installed. No version info available.
> langgraph>=0.3.5: Installed. No version info available.
> langserve[all]: Installed. No version info available.
> langsmith-pyo3: Installed. No version info available.
> langsmith<0.4,>=0.1.125: Installed. No version info available.
> mcp>=1.9.2: Installed. No version info available.
> mypy>=1.8.0: Installed. No version info available.
> numexpr: 2.8.7
> numpy: 1.26.4+deshaw5
> openai: 1.81.0
> orjson: 3.10.16+deshaw2
> packaging<25,>=23.2: Installed. No version info available.
> pyarrow: 18.0.0
> pydantic: 2.9.2
> pydantic-settings: 2.6.0
> pydantic<3.0.0,>=2.5.2;: Installed. No version info available.
> pydantic<3.0.0,>=2.7.4;: Installed. No version info available.
> pymilvus: 2.5.3+desco.3
> PyYAML: 6.0.2
> PyYAML>=5.3: Installed. No version info available.
> pyyaml>=6.0.2: Installed. No version info available.
> requests: 2.32.3
> requests-toolbelt: 1.0.0
> SQLAlchemy: 1.4.50
> tenacity: 8.2.3
> tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available.
> tiktoken: 0.7.0
> tomlkit: 0.12.3
> typer[all]: Installed. No version info available.
> typing-extensions>=4.14.0: Installed. No version info available.
> typing-extensions>=4.7: Installed. No version info available.
> uvicorn: 0.26.0
> validators: 0.22.0
> zstandard: 0.22.0
Originally created by @sundaraa-deshaw on GitHub (Sep 8, 2025). ### Checked other resources - [x] This is a bug, not a usage question. For questions, please use the LangChain Forum (https://forum.langchain.com/). - [x] I added a clear and detailed title that summarizes the issue. - [x] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example). - [x] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue. ### Example Code ```python from langgraph.graph import END, START, MessagesState, StateGraph from langchain_core.runnables import Runnable from langgraph.config import get_stream_writer from langchain_core.callbacks.manager import ( adispatch_custom_event, dispatch_custom_event, ) from langchain_core.messages import AIMessage from langchain_core.runnables import Runnable from langchain_core.callbacks import CallbackManager, CallbackManagerForChainRun from langchain_core.callbacks.manager import ( adispatch_custom_event, dispatch_custom_event, ) from langchain_core.runnables import Runnable, ensure_config from langchain_core.runnables.config import RunnableConfig, patch_config from langchain.chains.base import Chain from langchain_core.load.dump import dumpd from rich.pretty import pprint def node_1(state: MessagesState): return { "messages": AIMessage(content="message from node_1"), } def get_graph(): return ( StateGraph(MessagesState) .add_node(node_1) .add_edge(START, "node_1") .compile() ) class OuterRunnable(Runnable): def invoke(self, inputs): return {"foo": "from outer runnable"} async def astream(self, inputs, config=None, **kwargs): async for chunk in get_graph().astream(inputs, config=config, **kwargs): pass # yield chunk yield {"foo": "bar"} async for chunk in OuterRunnable().astream({}): pprint(chunk) print('----') async for chunk in OuterRunnable().astream_events({'messages': []}, version='v2', stream_mode=['custom', 'values']): if chunk['event'] == 'on_chain_stream': pprint(chunk['data']) ``` ### Error Message and Stack Trace (if applicable) ```shell {'foo': 'bar'} ---- {'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}} Expected: {'foo': 'bar'} ---- {'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}} {'chunk': {'data: {'foo': 'bar'}}} ``` ### Description I am trying to stream chunks from a LangGraph graph inside a Runnable with astream_events (to get stream data and events) While the following works: Streaming the graph results from the outer runnable with `astream_events` ``` class OuterRunnable(Runnable): def invoke(self, inputs): return {"foo": "from outer runnable"} async def astream(self, inputs, config=None, **kwargs): async for chunk in get_graph().astream(inputs, config=config, **kwargs): yield chunk yield {"foo": "bar"} ``` and produces: ``` { │ 'node_1': { │ │ 'messages': AIMessage( │ │ │ content='message from node_1', │ │ │ additional_kwargs={}, │ │ │ response_metadata={}, │ │ │ id='6e06f484-b26d-46f5-87f4-de1fc4eeda09' │ │ ) │ } } {'foo': 'bar'} ---- {'chunk': ('values', {'messages': []})} {'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}} { │ 'chunk': ( │ │ 'values', │ │ { │ │ │ 'messages': [ │ │ │ │ AIMessage( │ │ │ │ │ content='message from node_1', │ │ │ │ │ additional_kwargs={}, │ │ │ │ │ response_metadata={}, │ │ │ │ │ id='fc0c4296-4f12-47ff-8249-554566e4fa75' │ │ │ │ ) │ │ │ ] │ │ } │ ) } {'chunk': {'foo': 'bar'}} ``` the reproducer which just streams the graph, and does not yield the graph results, does not produce the final yielded chunk from the outer runnable (the dict foo: bar) ### System Info ``` System Information ------------------ > OS: Linux > OS Version: #1 SMP Fri Jun 27 05:59:19 EDT 2025 > Python Version: 3.11.8 (main, Sep 17 2024, 14:28:20) [GCC 10.3.1 20210422 (Red Hat 10.3.1-1)] Package Information ------------------- > langchain_core: 0.3.55 > langchain: 0.3.14 > langchain_community: 0.3.14 > langsmith: 0.2.10 > langchain_anthropic: 0.2.4 > langchain_cli: 0.0.31 > langchain_experimental: 0.3.4 > langchain_google_vertexai: 2.0.21 > langchain_mcp_adapters: 0.1.8 > langchain_milvus: 0.2.1 > langchain_openai: 0.2.14 > langchain_text_splitters: 0.3.5 > langchainplus_sdk: 0.0.21 > langgraph_gen: 0.0.6 > langgraph_reflection: 0.0.1 > langgraph_sdk: 0.1.60 > langgraph_supervisor: 0.0.20 > langgraph_swarm: 0.0.11 Optional packages not installed ------------------------------- > langserve Other Dependencies ------------------ > aiohttp: 3.8.4 > anthropic: 0.30.0 > anthropic[vertexai]: Installed. No version info available. > async-timeout: 4.0.3 > bottleneck: 1.3.7 > dataclasses-json: 0.6.3 > defusedxml: 0.7.1 > gitpython: 3.1.40 > google-cloud-aiplatform: 1.89.0 > google-cloud-storage: 2.13.0 > gritql: Installed. No version info available. > httpx: 0.27.0 > httpx-sse: 0.4.0 > jinja2>=3.1.5: Installed. No version info available. > jsonpatch<2.0,>=1.33: Installed. No version info available. > langchain-core<0.4,>=0.3.36: Installed. No version info available. > langchain-core<0.4.0,>=0.3.40: Installed. No version info available. > langchain-mistralai: Installed. No version info available. > langchain>=0.1.0: Installed. No version info available. > langgraph: 0.4.6 > langgraph-prebuilt<0.2.0,>=0.1.7: Installed. No version info available. > langgraph>=0.2.74: Installed. No version info available. > langgraph>=0.3.5: Installed. No version info available. > langserve[all]: Installed. No version info available. > langsmith-pyo3: Installed. No version info available. > langsmith<0.4,>=0.1.125: Installed. No version info available. > mcp>=1.9.2: Installed. No version info available. > mypy>=1.8.0: Installed. No version info available. > numexpr: 2.8.7 > numpy: 1.26.4+deshaw5 > openai: 1.81.0 > orjson: 3.10.16+deshaw2 > packaging<25,>=23.2: Installed. No version info available. > pyarrow: 18.0.0 > pydantic: 2.9.2 > pydantic-settings: 2.6.0 > pydantic<3.0.0,>=2.5.2;: Installed. No version info available. > pydantic<3.0.0,>=2.7.4;: Installed. No version info available. > pymilvus: 2.5.3+desco.3 > PyYAML: 6.0.2 > PyYAML>=5.3: Installed. No version info available. > pyyaml>=6.0.2: Installed. No version info available. > requests: 2.32.3 > requests-toolbelt: 1.0.0 > SQLAlchemy: 1.4.50 > tenacity: 8.2.3 > tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available. > tiktoken: 0.7.0 > tomlkit: 0.12.3 > typer[all]: Installed. No version info available. > typing-extensions>=4.14.0: Installed. No version info available. > typing-extensions>=4.7: Installed. No version info available. > uvicorn: 0.26.0 > validators: 0.22.0 > zstandard: 0.22.0 ```
yindo added the bugpending labels 2026-02-20 17:42:33 -05:00
yindo closed this issue 2026-02-20 17:42:34 -05:00
yindo commented 2026-02-20 17:42:34 -05:00
Author
Owner
Copy Link
Copy Source

@harshit-agarwal7 commented on GitHub (Sep 13, 2025):

What you are seeing is the expected behaviour.

astream_events only forwards events reported to the CallbackManager. In your custom astream implementation, the CallbackManager is not being invoked, so the final yield never shows up in the event stream.

You do see the graph’s streamed output because the astream implementation of the Pregel class is unmodified — it integrates with the CallbackManager correctly, and therefore its messages appear in the event stream.

If you also want your runnable’s final output to appear in astream_events, you need to wrap your async generator in a RunnableGenerator. This ensures that the yielded values are surfaced as on_chain_stream events when stream_mode=['values'] is enabled.

I have updated your code to use a RunnableGenerator below:

from langgraph.graph import END, START, MessagesState, StateGraph
from langchain_core.messages import AIMessage

from rich.pretty import pprint

def node_1(state: MessagesState):
    return {
        "messages": AIMessage(content="message from node_1"),
    }

def get_graph():
    return (
        StateGraph(MessagesState)
        .add_node(node_1)
        .add_edge(START, "node_1")
        .compile()
    )

from langchain_core.runnables import RunnableGenerator

async def outer_stream(inputs, config=None, **kwargs):
    async for input in inputs:
        async for _ in get_graph().astream(input, config=config, **kwargs):
            pass

    yield {"foo": "bar"}

OuterRunnable = RunnableGenerator(outer_stream)

import asyncio  

async def main():
    async for chunk in OuterRunnable.astream({}):
        pprint(chunk)

asyncio.run(main())

print('----')

async def main2():
    async for chunk in OuterRunnable.astream_events({'messages': []}, version='v2', stream_mode=['custom', 'values']):
        if chunk['event'] == 'on_chain_stream':
            pprint(chunk['data'])

asyncio.run(main2())

This will give you the desired result:

{'foo': 'bar'}
----
{'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}}
{'chunk': {'foo': 'bar'}}
@harshit-agarwal7 commented on GitHub (Sep 13, 2025): What you are seeing is the expected behaviour. `astream_events` only forwards events reported to the `CallbackManager`. In your custom `astream` implementation, the `CallbackManager` is not being invoked, so the final yield never shows up in the event stream. You do see the graph’s streamed output because the `astream` implementation of the `Pregel` class is unmodified — it integrates with the `CallbackManager` correctly, and therefore its messages appear in the event stream. If you also want your runnable’s final output to appear in `astream_events`, you need to wrap your async generator in a `RunnableGenerator`. This ensures that the yielded values are surfaced as `on_chain_stream` events when `stream_mode=['values']` is enabled. I have updated your code to use a RunnableGenerator below: ```python from langgraph.graph import END, START, MessagesState, StateGraph from langchain_core.messages import AIMessage from rich.pretty import pprint def node_1(state: MessagesState): return { "messages": AIMessage(content="message from node_1"), } def get_graph(): return ( StateGraph(MessagesState) .add_node(node_1) .add_edge(START, "node_1") .compile() ) from langchain_core.runnables import RunnableGenerator async def outer_stream(inputs, config=None, **kwargs): async for input in inputs: async for _ in get_graph().astream(input, config=config, **kwargs): pass yield {"foo": "bar"} OuterRunnable = RunnableGenerator(outer_stream) import asyncio async def main(): async for chunk in OuterRunnable.astream({}): pprint(chunk) asyncio.run(main()) print('----') async def main2(): async for chunk in OuterRunnable.astream_events({'messages': []}, version='v2', stream_mode=['custom', 'values']): if chunk['event'] == 'on_chain_stream': pprint(chunk['data']) asyncio.run(main2()) ``` This will give you the desired result: ``` {'foo': 'bar'} ---- {'chunk': {'messages': AIMessage(content='message from node_1', additional_kwargs={}, response_metadata={})}} {'chunk': {'foo': 'bar'}} ```
yindo commented 2026-02-20 17:42:34 -05:00
Author
Owner
Copy Link
Copy Source

@sundaraa-deshaw commented on GitHub (Sep 14, 2025):

Thank you for the explanation. This is useful

@sundaraa-deshaw commented on GitHub (Sep 14, 2025): Thank you for the explanation. This is useful
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label bug pending
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#967
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?